"""Content hashing for identity and reconciliation. - ``sha256_file`` is exact byte identity: it drives move/copy/replace matching. - ``pixel_sha256`` hashes the normalized decoded pixels (EXIF orientation applied, converted to RGB), so an EXIF-only edit keeps the same value while a genuine content change does not. That distinction separates "metadata changed" from "replaced". Refactored from photo_analyzer._sha1_file/_phash_image: SHA-256 replaces SHA-1 as byte identity and a normalized pixel hash is added (see donor_ledger.yaml: pa-hashing). Perceptual hashing stays with the duplicate engine (US01-04). """ from __future__ import annotations import hashlib from pathlib import Path from PIL import Image, ImageOps PIXEL_HASH_VERSION = 1 PHASH_VERSION = 1 _CHUNK = 1 << 20 def sha256_file(path: Path | str) -> str: digest = hashlib.sha256() with open(path, "rb") as handle: for chunk in iter(lambda: handle.read(_CHUNK), b""): digest.update(chunk) return digest.hexdigest() def pixel_sha256(path: Path | str) -> str: with Image.open(path) as image: oriented = ImageOps.exif_transpose(image) rgb = oriented.convert("RGB") header = f"{PIXEL_HASH_VERSION}:{rgb.width}x{rgb.height}:".encode() payload = header + rgb.tobytes() return hashlib.sha256(payload).hexdigest() def safe_pixel_sha256(path: Path | str) -> str | None: """``pixel_sha256`` but returns None for undecodable images instead of raising.""" try: return pixel_sha256(path) except Exception: return None def phash(path: Path | str) -> str: """DCT perceptual hash → 16 hex chars (64 bits). Standard imagehash.phash recipe (extracted verbatim from photo_analyzer._phash_image): grayscale → 32x32 → 2D DCT-II → keep the top-left 8x8 low-frequency block → bit = coefficient > median. Resilient to resize and recompression; used only as evidence for review, never for automatic exclusion. """ import numpy as np from scipy.fftpack import dct with Image.open(path) as image: small = image.convert("L").resize((32, 32), Image.LANCZOS) matrix = np.asarray(small, dtype=np.float64) transformed = dct(dct(matrix, axis=0), axis=1) low = transformed[:8, :8] bits = (low > np.median(low)).flatten() value = 0 for bit in bits: value = (value << 1) | int(bit) return f"{value:016x}" def safe_phash(path: Path | str) -> str | None: """``phash`` but returns None for undecodable images instead of raising.""" try: return phash(path) except Exception: return None def phash_distance(a_hex: str, b_hex: str) -> int: """Hamming distance between two 64-bit hex perceptual hashes.""" return bin(int(a_hex, 16) ^ int(b_hex, 16)).count("1")