50 lines
1.8 KiB
Python
50 lines
1.8 KiB
Python
import math
|
|
import hashlib
|
|
from pathlib import Path
|
|
from typing import Tuple
|
|
|
|
def compute_file_sha256(filepath: Path, buffer_size: int = 1024 * 1024) -> str:
|
|
"""Calculates full SHA-256 checksum of a file using streaming buffers to avoid memory overhead."""
|
|
hasher = hashlib.sha256()
|
|
with open(filepath, "rb") as f:
|
|
while True:
|
|
chunk = f.read(buffer_size)
|
|
if not chunk:
|
|
break
|
|
hasher.update(chunk)
|
|
return hasher.hexdigest()
|
|
|
|
class FileChunker:
|
|
"""Handles splitting large files into discrete chunks and calculating block hashes."""
|
|
|
|
def __init__(self, filepath: Path, chunk_size: int = 4 * 1024 * 1024):
|
|
self.filepath = Path(filepath).resolve()
|
|
if not self.filepath.exists():
|
|
raise FileNotFoundError(f"File {filepath} not found")
|
|
|
|
self.file_size = self.filepath.stat().st_size
|
|
self.chunk_size = chunk_size
|
|
self.total_chunks = max(1, math.ceil(self.file_size / self.chunk_size))
|
|
self._cached_sha256 = None
|
|
|
|
@property
|
|
def full_sha256(self) -> str:
|
|
if self._cached_sha256 is None:
|
|
self._cached_sha256 = compute_file_sha256(self.filepath)
|
|
return self._cached_sha256
|
|
|
|
def get_chunk(self, chunk_index: int) -> Tuple[bytes, str]:
|
|
"""
|
|
Reads the bytes for chunk `chunk_index` and returns (chunk_data, chunk_sha256).
|
|
"""
|
|
if chunk_index < 0 or chunk_index >= self.total_chunks:
|
|
raise IndexError(f"Chunk index {chunk_index} out of bounds (total: {self.total_chunks})")
|
|
|
|
offset = chunk_index * self.chunk_size
|
|
with open(self.filepath, "rb") as f:
|
|
f.seek(offset)
|
|
data = f.read(self.chunk_size)
|
|
|
|
chunk_sha256 = hashlib.sha256(data).hexdigest()
|
|
return data, chunk_sha256
|