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 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 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."""
    if a is None or b is None:
        return False
    try:
        diff = ImageChops.difference(a, b)
        if ImageStat.Stat(diff).mean[0] > max_mean_diff:
            return False
        hist = diff.histogram()
        total = sum(hist)
        if not total:
            return False
        changed = sum(hist[changed_level:])
        return (changed / total) <= max_changed_fraction
    except Exception:
        return False


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)