feat(phash): compute_phash + find_similar util (hash_size=8, fits existing column)
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -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)
|
||||
@@ -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"
|
||||
Reference in New Issue
Block a user