FabledCurator/backend/app/utils/phash.py

"""Perceptual-hash dedup helpers (ported from ImageRepo).

hash_size=16 -> 256-bit hash -> 64-hex-char string, which is why
`ImageRecord.phash` is String(64) (widened in migration 0098).

## Why 16, after running at 8 from FC-2d until 2026-09-21

`hash_size=8` keeps only the top-left 8x8 block of the DCT — 64 bits
describing an image's coarse light/dark layout and nothing else. Variant
artwork that shares a composition (same pose and framing, a different
outfit / expression / overlay) collided OUTRIGHT at that size, so the
operator's near-duplicate dial could not separate variants from rescales
even at its floor: `phash_threshold=0` means "the same 64 bits", not "the
same image", and packs of 15 variants were landing as 3 records
(issue #4223). IR always used 16; FC's deviation to 8 was made to fit the
old String(32) column without a migration — a schema convenience that cost
the operator artwork.

## The hash no longer decides a merge on its own

Dropping a file or superseding one is destructive, so `find_similar` now
runs three gates, cheapest first:

  1. Hamming distance within the operator's `phash_threshold` — the cheap
     indexed pre-filter that PROPOSES candidates.
  2. Aspect ratio within ASPECT_TOL. A crop or a re-canvas is not a
     rescale, and this is the same identity test the tier-1 video near-dup
     path already uses (`_VIDEO_DUP_ASPECT_TOL`).
  3. A pixel-level confirm on the candidate's actual file, supplied by the
     caller (the importer opens the files; this module stays I/O-free apart
     from `fingerprint_path`).

Because the confirm is what ACCEPTS a match, the threshold can stay
generous enough to tolerate a re-encode without putting variants at risk.

Every gate fails CLOSED: unknown dimensions, an unreadable candidate, a
hash that won't parse — all mean "not a duplicate". The two failure
directions are not symmetrical. Too strict keeps a redundant lower-res copy,
which the operator can see and delete; too loose deletes artwork that only
a re-walk of the source can bring back.
"""

import imagehash
from PIL import Image, ImageChops, ImageStat

HASH_SIZE = 16

# Gate 2. Matching `importer._VIDEO_DUP_ASPECT_TOL` — a rescale preserves
# aspect ratio to within rounding, so this only has to absorb off-by-one
# pixel dimensions.
ASPECT_TOL = 0.02

# Gate 3. Both images are reduced to one FINGERPRINT_SIZE-square grayscale
# thumbnail and compared directly, which is the question actually being
# asked: "are these the same picture at different resolutions?"
#
# Two criteria, because they catch different things. MEAN absolute
# difference catches a global change (a recolour, a filter, a different
# shading pass) that leaves the composition intact. The CHANGED-PIXEL
# fraction catches a LOCAL one — an added overlay, a different expression,
# an alternate outfit on part of the figure — which a mean over 4096 pixels
# would otherwise dilute into noise.
#
# Tuned to fail closed (see the module docstring). A true rescale lands near
# zero on both; these ceilings sit well above that and well below a variant.
FINGERPRINT_SIZE = 64
FINGERPRINT_MAX_MEAN_DIFF = 6.0
FINGERPRINT_CHANGED_LEVEL = 64
FINGERPRINT_MAX_CHANGED_FRACTION = 0.02


def _seek_first_frame(pil_image) -> None:
    """Animated images (multi-frame WebP/GIF/APNG) are hashed and
    fingerprinted on frame 0 — the conventional choice for animated
    content, and the one that keeps PIL from iterating every frame."""
    if getattr(pil_image, "is_animated", False):
        try:
            pil_image.seek(0)
        except Exception:
            pass


def hash_bits(hex_str: str | None) -> int | None:
    """A stored pHash hex string as an integer, or None if it is missing or
    unparseable. Fails CLOSED, like every other gate in this module.

    Parsed to an int rather than an imagehash object because the caller that
    needs this compares one image against many: `int.bit_count()` on an XOR is
    a machine instruction, where rebuilding a 16x16 boolean array per
    comparison is not.
    """
    if not hex_str:
        return None
    try:
        return int(hex_str, 16)
    except (TypeError, ValueError):
        return None


def hamming(a: int | None, b: int | None) -> int | None:
    """Bits differing between two parsed hashes, or None if either is absent.

    Out of 256 at HASH_SIZE 16.
    """
    if a is None or b is None:
        return None
    return (a ^ b).bit_count()


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).

    Frame 0 for animated images: without the seek, 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).
    """
    try:
        _seek_first_frame(pil_image)
        return str(imagehash.phash(pil_image, hash_size=HASH_SIZE))
    except Exception:
        return None


def fingerprint(pil_image):
    """A small grayscale thumbnail of an opened PIL image, for gate 3.

    Returns a detached PIL image (so the caller may close the original) or
    None on any failure. PIL-only on purpose: numpy is an imagehash
    transitive dependency, not a declared one for this path.
    """
    try:
        _seek_first_frame(pil_image)
        return pil_image.convert("L").resize(
            (FINGERPRINT_SIZE, FINGERPRINT_SIZE), Image.LANCZOS
        )
    except Exception:
        return None


def fingerprint_path(path) -> Image.Image | None:
    """`fingerprint` for a file on disk. None if it cannot be read — which
    the gate treats as "not a duplicate"."""
    try:
        with Image.open(path) as im:
            return fingerprint(im)
    except Exception:
        return None


def fingerprint_diff(
    a, b, *, changed_level: int = FINGERPRINT_CHANGED_LEVEL,
) -> tuple[float, float] | None:
    """(mean absolute difference, fraction of pixels past `changed_level`) for
    two fingerprints. None if either is missing or the comparison fails.

    This is the MEASUREMENT behind `fingerprints_match`, split out so the
    calibration report (scripts/phash_gate_report.py) can show how far a pair
    sat from the limits instead of only which side of them it fell on. The
    constants were chosen without a real-library sample; the numbers this
    returns are what moves them.
    """
    if a is None or b is None:
        return None
    try:
        diff = ImageChops.difference(a, b)
        hist = diff.histogram()
        total = sum(hist)
        if not total:
            return None
        return (ImageStat.Stat(diff).mean[0], sum(hist[changed_level:]) / total)
    except Exception:
        return None


def fingerprints_match(
    a, b,
    *,
    max_mean_diff: float = FINGERPRINT_MAX_MEAN_DIFF,
    changed_level: int = FINGERPRINT_CHANGED_LEVEL,
    max_changed_fraction: float = FINGERPRINT_MAX_CHANGED_FRACTION,
) -> bool:
    """True when two fingerprints are the same picture: no large global
    drift AND no meaningful local region that differs. False on any
    failure."""
    measured = fingerprint_diff(a, b, changed_level=changed_level)
    if measured is None:
        return False
    mean, changed_fraction = measured
    return mean <= max_mean_diff and changed_fraction <= max_changed_fraction


def aspect_matches(
    width: int | None, height: int | None,
    cand_width: int | None, cand_height: int | None,
    *, tol: float = ASPECT_TOL,
) -> bool:
    """Gate 2. False when either side's dimensions are unknown — an
    unmeasurable candidate is not a proven duplicate."""
    if not width or not height or not cand_width or not cand_height:
        return False
    a, b = width / height, cand_width / cand_height
    if a <= 0 or b <= 0:
        return False
    return abs(a - b) / max(a, b) <= tol


def find_similar(
    phash_hex: str,
    width: int,
    height: int,
    candidates: list[tuple[str, int, int, int]],
    threshold: int,
    *,
    confirm=None,
) -> tuple[str, int | None]:
    """candidates: (phash_hex, width, height, image_id). Returns one of
    ("none", None) / ("larger_exists", id) / ("smaller_exists", id).
    First candidate to pass EVERY gate wins (IR loop order).

    `confirm` is gate 3: an optional callable(image_id) -> bool, called only
    for a candidate that already passed the hash and aspect gates, and
    expected to compare the two files' pixels. A rejected candidate does not
    end the search — the loop moves on, so a false pre-filter hit cannot
    mask a real duplicate further down the list. Omitting it leaves the
    hash+aspect behaviour, which is what the unit tests exercise.
    """
    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:
            # Includes the mismatched-length case while a library re-hash
            # (migration 0098) is still in flight: an old 64-bit hash cannot
            # be compared to a new 256-bit one, and skipping it degrades to
            # "no dedup yet" rather than to a wrong merge.
            continue
        if dist > threshold:
            continue
        if not aspect_matches(width, height, cw, ch):
            continue
        if confirm is not None and not confirm(cid):
            continue
        if cw >= width and ch >= height:
            return ("larger_exists", cid)
        if width > cw or height > ch:
            return ("smaller_exists", cid)
    return ("none", None)