"""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). For animated images (multi-frame WebP/GIF/APNG), explicitly seek to frame 0 first. Without this, some PIL operations downstream of imagehash.phash (convert("L"), resize) can iterate all frames and blow past Celery's hard time limit on large animations (operator-flagged 2026-05-26 against animated WebPs). The pHash of frame 0 is the conventional choice for animated content. """ try: if getattr(pil_image, "is_animated", False): try: pil_image.seek(0) except Exception: pass 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)