Compute SHA256 hash for a file path, bytes, or file-like object. Args: file_path_or_obj: File path, bytes, or file-like object. buffer_size_mb: Read buffer size in MB. Default 16MB. tqdm_desc: Progress bar description. disable_tqdm: Whether to disable progress ba
(
file_path_or_obj: Union[str, Path, bytes, BinaryIO],
buffer_size_mb: Optional[int] = 16,
tqdm_desc: Optional[str] = '[Calculating]',
disable_tqdm: Optional[bool] = True,
)
| 207 | |
| 208 | |
| 209 | def get_file_hash( |
| 210 | file_path_or_obj: Union[str, Path, bytes, BinaryIO], |
| 211 | buffer_size_mb: Optional[int] = 16, |
| 212 | tqdm_desc: Optional[str] = '[Calculating]', |
| 213 | disable_tqdm: Optional[bool] = True, |
| 214 | ) -> dict: |
| 215 | """Compute SHA256 hash for a file path, bytes, or file-like object. |
| 216 | |
| 217 | Args: |
| 218 | file_path_or_obj: File path, bytes, or file-like object. |
| 219 | buffer_size_mb: Read buffer size in MB. Default 16MB. |
| 220 | tqdm_desc: Progress bar description. |
| 221 | disable_tqdm: Whether to disable progress bar. |
| 222 | |
| 223 | Returns: |
| 224 | dict with keys: file_path_or_obj, file_hash, file_size. |
| 225 | """ |
| 226 | from tqdm.auto import tqdm |
| 227 | |
| 228 | declared_size = get_file_size(file_path_or_obj) |
| 229 | if declared_size > 1024 * 1024 * 1024: # 1GB |
| 230 | disable_tqdm = False |
| 231 | name = 'Large File' |
| 232 | if isinstance(file_path_or_obj, (str, Path)): |
| 233 | path = file_path_or_obj if isinstance( |
| 234 | file_path_or_obj, Path) else Path(file_path_or_obj) |
| 235 | name = path.name |
| 236 | tqdm_desc = f'[Validating Hash for {name}]' |
| 237 | |
| 238 | buffer_size = buffer_size_mb * 1024 * 1024 |
| 239 | file_hash = hashlib.sha256() |
| 240 | |
| 241 | progress = tqdm( |
| 242 | total=declared_size, |
| 243 | initial=0, |
| 244 | unit_scale=True, |
| 245 | dynamic_ncols=True, |
| 246 | unit='B', |
| 247 | desc=tqdm_desc, |
| 248 | disable=disable_tqdm, |
| 249 | ) |
| 250 | |
| 251 | if isinstance(file_path_or_obj, (str, Path)): |
| 252 | bytes_hashed = 0 |
| 253 | with open(file_path_or_obj, 'rb') as f: |
| 254 | while byte_chunk := f.read(buffer_size): |
| 255 | file_hash.update(byte_chunk) |
| 256 | bytes_hashed += len(byte_chunk) |
| 257 | progress.update(len(byte_chunk)) |
| 258 | file_hash = file_hash.hexdigest() |
| 259 | if bytes_hashed != declared_size: |
| 260 | logger.warning( |
| 261 | f'File size changed during hash computation: ' |
| 262 | f'declared {declared_size} bytes, actually hashed {bytes_hashed} bytes. ' |
| 263 | f'File may have been modified: {file_path_or_obj}') |
| 264 | file_size = bytes_hashed |
| 265 | |
| 266 | elif isinstance(file_path_or_obj, bytes): |
no test coverage detected
searching dependent graphs…