diff --git a/backend/app/utils/phash.py b/backend/app/utils/phash.py new file mode 100644 index 0000000..8d00cb3 --- /dev/null +++ b/backend/app/utils/phash.py @@ -0,0 +1,44 @@ +"""Perceptual-hash dedup helpers (ported from ImageRepo). + +hash_size=8 -> 64-bit hash -> 16-hex-char string, which fits the existing +ImageRecord.phash String(32) column (no image_record migration). IR uses +hash_size=16; the deliberate FC deviation keeps the schema unchanged. The +Hamming threshold is the operator-exposed dial (ImportSettings). +""" + +import imagehash + +HASH_SIZE = 8 + + +def compute_phash(pil_image) -> str | None: + """Perceptual hash of an opened PIL image, as a hex string. None on any + failure (videos/unreadable/non-image).""" + try: + return str(imagehash.phash(pil_image, hash_size=HASH_SIZE)) + except Exception: + return None + + +def find_similar( + phash_hex: str, + width: int, + height: int, + candidates: list[tuple[str, int, int, int]], + threshold: int, +) -> tuple[str, int | None]: + """candidates: (phash_hex, width, height, image_id). Returns one of + ("none", None) / ("larger_exists", id) / ("smaller_exists", id). + First qualifying candidate wins (IR loop order).""" + new_h = imagehash.hex_to_hash(phash_hex) + for cand_hex, cw, ch, cid in candidates: + try: + dist = new_h - imagehash.hex_to_hash(cand_hex) + except Exception: + continue + if dist <= threshold: + if cw >= width and ch >= height: + return ("larger_exists", cid) + if width > cw or height > ch: + return ("smaller_exists", cid) + return ("none", None) diff --git a/tests/test_phash_util.py b/tests/test_phash_util.py new file mode 100644 index 0000000..11e81d1 --- /dev/null +++ b/tests/test_phash_util.py @@ -0,0 +1,51 @@ +import io + +import imagehash +from PIL import Image + +from backend.app.utils.phash import compute_phash, find_similar + + +def _img(color, size=(64, 64)): + return Image.new("RGB", size, color) + + +def test_compute_phash_stable_and_hex(): + h1 = compute_phash(_img((10, 120, 200))) + h2 = compute_phash(_img((10, 120, 200))) + assert isinstance(h1, str) and len(h1) == 16 # hash_size=8 -> 64-bit -> 16 hex + assert h1 == h2 + + +def test_compute_phash_none_for_non_image(): + assert compute_phash(io.BytesIO(b"not an image")) is None + + +def test_find_similar_none_when_far(): + new = compute_phash(_img((0, 0, 0))) + cand = (compute_phash(_img((255, 255, 255))), 100, 100, 7) + rel, mid = find_similar(new, 50, 50, [cand], threshold=2) + assert rel == "none" and mid is None + + +def test_find_similar_larger_exists_skips_new(): + h = compute_phash(_img((123, 50, 7))) + rel, mid = find_similar(h, 100, 100, [(h, 400, 400, 9)], threshold=2) + assert rel == "larger_exists" and mid == 9 + + +def test_find_similar_smaller_exists_supersede(): + h = compute_phash(_img((123, 50, 7))) + rel, mid = find_similar(h, 800, 800, [(h, 100, 100, 4)], threshold=2) + assert rel == "smaller_exists" and mid == 4 + + +def test_find_similar_threshold_boundary_inclusive_and_first_match(): + h = compute_phash(_img((10, 10, 10))) + far = compute_phash(_img((240, 240, 240))) + d = imagehash.hex_to_hash(h) - imagehash.hex_to_hash(far) + # at exactly the distance, it counts as similar (<=) + rel, _ = find_similar(h, 10, 10, [(far, 999, 999, 1)], threshold=d) + assert rel == "larger_exists" + rel2, _ = find_similar(h, 10, 10, [(far, 999, 999, 1)], threshold=d - 1) + assert rel2 == "none"