""" Duplicate detection: group photos by perceptual-hash similarity. Strategy -------- Each photo carries a 16-char hex perceptual hash (`Photo.phash`) computed by the thumbnail worker from the original-resolution decoded frame (`app.tasks.thumbs._generate_thumbnails_async`). pHash is robust to resize / re-encoding / mild edits, so two photos that are visually the "same shot" land at small Hamming distance even when their bytes are completely different. This module turns those per-photo hashes into explicit *groups*. The result is persisted in two columns: * `Photo.duplicate_group_id` — shared by every member of a group * `Photo.is_duplicate` — derived: True iff group_id IS NOT NULL (kept as a column so the existing PhotoThumbnail badge and /library/stats count don't have to change). The grouping is recomputed in batches by `regroup_duplicates`, NOT on individual writes — that lets us use a single in-memory pass instead of maintaining a per-row similarity index. Triggered automatically after each scan and on demand from the Settings panel. Complexity ---------- Pairwise O(N²) over photos with a non-null phash. At ~5 µs per Hamming distance in CPython this is roughly: 1k photos → ~5 s 5k photos → ~125 s 10k photos → ~500 s That's the wrong shape for libraries past ~5k. The drop-in replacement is a BK-tree (e.g. `pybktree`) which gives O(log N) lookups for a fixed Hamming threshold; swap it in here when someone trips the limit. The public function signature stays the same. Out of scope (deferred) ----------------------- * Dismissing a group / "intentional duplicates" — would need a per-group or per-pair flag plus a skip-set in this function so re-grouping doesn't bring them back. Add when there's a real user need. * Incremental updates on individual photo writes — currently we just re-run the whole job after each scan, which is fine while the cost is bounded. """ from __future__ import annotations import logging import uuid from typing import Optional from sqlalchemy import select, update from app.database import AsyncSessionLocal from app.models.photos import Photo logger = logging.getLogger(__name__) # Hamming distance threshold under which two phashes are considered # "the same image". 6 bits out of 64 is the rule-of-thumb sweet spot for # pHash — tight enough to avoid false positives between unrelated photos, # loose enough to catch JPEG re-encodes, slight crops, and a screenshot # of a screenshot. DEFAULT_THRESHOLD = 6 def _hex_to_int(h: str) -> int: """Parse a 16-char hex pHash to a Python int. Returns -1 on bad input so the pairwise loop can skip the row without raising.""" try: return int(h, 16) except (TypeError, ValueError): return -1 def _hamming(a: int, b: int) -> int: """Population count of XOR — the canonical hash distance metric. `int.bit_count()` is C-implemented in CPython 3.10+ and is by far the fastest path; the `bin(...).count('1')` fallback is here only so the function still works on older interpreters.""" x = a ^ b try: return x.bit_count() # type: ignore[attr-defined] except AttributeError: return bin(x).count('1') class _UnionFind: """Tiny union-find / disjoint-set used to merge similar phashes into connected components. Inlined here (rather than pulled from a dep) because it's ~15 lines and we don't need anything fancy.""" def __init__(self, n: int) -> None: self.parent = list(range(n)) self.rank = [0] * n def find(self, x: int) -> int: # Path compression — flattens the tree on lookup so subsequent # finds are amortized O(α(N)) ≈ O(1). while self.parent[x] != x: self.parent[x] = self.parent[self.parent[x]] x = self.parent[x] return x def union(self, a: int, b: int) -> None: ra, rb = self.find(a), self.find(b) if ra == rb: return if self.rank[ra] < self.rank[rb]: ra, rb = rb, ra self.parent[rb] = ra if self.rank[ra] == self.rank[rb]: self.rank[ra] += 1 async def regroup_duplicates(threshold: int = DEFAULT_THRESHOLD) -> dict: """Recompute every photo's duplicate_group_id from current phashes. Idempotent — safe to call as often as you like. Returns a small summary dict the maintenance endpoint surfaces back to the UI. Photos that end up alone in a component (size 1) get `duplicate_group_id=NULL` and `is_duplicate=False`. This is what cleans up "dead" groups after the user discards N-1 members from one. """ async with AsyncSessionLocal() as session: # Pull (id, phash) for every non-discarded photo with a hash. # Discarded photos are excluded so we don't keep showing groups # made up of trashed copies. rows = ( await session.execute( select(Photo.id, Photo.phash) .where(Photo.phash.is_not(None)) .where(Photo.is_discarded.is_(False)) ) ).all() n = len(rows) if n == 0: # Still need to clear stale group_ids in case the user just # discarded the last surviving member of every group. await _clear_all_groups(session) await session.commit() return { 'photos_considered': 0, 'groups': 0, 'members': 0, } ids: list[str] = [row[0] for row in rows] hashes: list[int] = [_hex_to_int(row[1]) for row in rows] uf = _UnionFind(n) # O(N²) pairwise comparison. See module docstring for the # scaling analysis and the BK-tree upgrade path. for i in range(n): hi = hashes[i] if hi < 0: continue for j in range(i + 1, n): hj = hashes[j] if hj < 0: continue if _hamming(hi, hj) <= threshold: uf.union(i, j) # Collect components. Each connected component of size >= 2 gets # a fresh group id; size-1 components are intentionally dropped. components: dict[int, list[int]] = {} for i in range(n): root = uf.find(i) components.setdefault(root, []).append(i) # First pass: clear EVERY photo's group_id so survivors of an # earlier grouping that no longer match anyone end up clean. This # is one bulk UPDATE rather than per-photo to keep the cost low # even on big libraries. await _clear_all_groups(session) # Second pass: write the new group ids for components of size 2+. groups_created = 0 members_total = 0 for members in components.values(): if len(members) < 2: continue group_id = str(uuid.uuid4()) member_ids = [ids[i] for i in members] await session.execute( update(Photo) .where(Photo.id.in_(member_ids)) .values( duplicate_group_id=group_id, is_duplicate=True, ) ) groups_created += 1 members_total += len(member_ids) await session.commit() logger.info( f"regroup_duplicates: considered {n} photos, " f"created {groups_created} group(s) covering {members_total} member(s)" ) return { 'photos_considered': n, 'groups': groups_created, 'members': members_total, } async def _clear_all_groups(session) -> None: """Reset duplicate_group_id / is_duplicate on every photo. Used as the first half of a regroup pass so photos that no longer cluster with anyone end up clean instead of carrying a stale group id.""" await session.execute( update(Photo).values(duplicate_group_id=None, is_duplicate=False) )