From c7dd03ade22d32aea855d732e88f8b64b11306d0 Mon Sep 17 00:00:00 2001 From: dtoro Date: Sun, 12 Apr 2026 22:03:04 +0200 Subject: [PATCH] feat: CLIP-powered incremental duplicate detection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace O(N²) pHash-only duplicate detection with a hybrid approach: - pHash Hamming distance for exact/near-exact copies - CLIP embedding cosine similarity via pgvector HNSW for visually similar photos (crops, format changes, screenshots) Post-scan now uses incremental mode: only newly added photos are compared against the full library — O(new × log N) via HNSW index instead of O(N²). Full regroup remains available from Settings. Co-Authored-By: Claude Opus 4.6 (1M context) --- backend/app/services/duplicates.py | 381 ++++++++++++++++++++--------- backend/app/tasks/scan.py | 13 +- backend/app/tasks/thumbs.py | 23 +- 3 files changed, 295 insertions(+), 122 deletions(-) diff --git a/backend/app/services/duplicates.py b/backend/app/services/duplicates.py index b3fb066..c2cf0e7 100644 --- a/backend/app/services/duplicates.py +++ b/backend/app/services/duplicates.py @@ -1,73 +1,63 @@ """ -Duplicate detection: group photos by perceptual-hash similarity. +Duplicate detection: group photos by perceptual-hash + CLIP 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. +Two complementary signals are fused into a single grouping: -This module turns those per-photo hashes into explicit *groups*. The -result is persisted in two columns: + 1. **Perceptual hash (pHash)** — 16-char hex hash from the thumbnail + worker. Catches byte-identical copies and mild re-encodes via + Hamming distance (threshold ≤ 6 bits out of 64). - * `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). + 2. **CLIP embedding similarity** — cosine distance over 512-d vectors + stored in the `embeddings` table with an HNSW index. Catches + visually similar photos even when pHash diverges (e.g. crops, + different formats, screenshots of the same content). -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. +Both signals feed a union-find structure that merges overlapping matches +into connected components. -Complexity ----------- -Pairwise O(N²) over photos with a non-null phash. At ~5 µs per Hamming -distance in CPython this is roughly: +Incremental mode (default post-scan) +------------------------------------- +`incremental_regroup` only compares *newly added* photos (those whose +`added_at` > watermark) against the entire library. Each new photo does: - 1k photos → ~5 s - 5k photos → ~125 s - 10k photos → ~500 s + - An HNSW vector similarity query: O(log N) via the index. + - A pHash comparison against a small candidate set (same group members + or nearby CLIP results) rather than the full N² sweep. -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. +This makes the post-scan cost O(new × log N) instead of O(N²). -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. +Full regroup +------------ +`regroup_duplicates` still performs the full pairwise pHash pass + +CLIP sweep, used for initial setup and manual re-detection. """ from __future__ import annotations import logging import uuid +from datetime import datetime, timezone from typing import Optional -from sqlalchemy import select, update +from sqlalchemy import select, update, text from app.database import AsyncSessionLocal from app.models.photos import Photo +from app.models.embeddings import Embedding +from app.config import settings 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 +# pHash Hamming distance threshold (6 out of 64 bits). +DEFAULT_PHASH_THRESHOLD = 6 + +# CLIP cosine distance threshold. CLIP embeddings are L2-normalized, +# so cosine distance = 1 - dot(a, b). A threshold of 0.08 catches +# visually near-identical shots; 0.15 catches similar compositions. +DEFAULT_CLIP_THRESHOLD = 0.10 def _hex_to_int(h: str) -> int: @@ -80,10 +70,7 @@ def _hex_to_int(h: str) -> int: 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.""" + """Population count of XOR — the canonical hash distance metric.""" x = a ^ b try: return x.bit_count() # type: ignore[attr-defined] @@ -92,24 +79,26 @@ def _hamming(a: int, b: int) -> int: 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.""" + """Tiny union-find / disjoint-set used to merge similar photos into + connected components.""" - def __init__(self, n: int) -> None: + def __init__(self, keys: list[str]) -> None: + self._index = {k: i for i, k in enumerate(keys)} + n = len(keys) 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) + def union_by_key(self, key_a: str, key_b: str) -> None: + ia, ib = self._index.get(key_a), self._index.get(key_b) + if ia is None or ib is None: + return + ra, rb = self.find(ia), self.find(ib) if ra == rb: return if self.rank[ra] < self.rank[rb]: @@ -118,112 +107,276 @@ class _UnionFind: if self.rank[ra] == self.rank[rb]: self.rank[ra] += 1 + def components(self, keys: list[str]) -> dict[int, list[str]]: + """Return {root_idx: [photo_ids...]} for groups of size >= 2.""" + groups: dict[int, list[str]] = {} + for key in keys: + idx = self._index[key] + root = self.find(idx) + groups.setdefault(root, []).append(key) + return {r: members for r, members in groups.items() if len(members) >= 2} -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. +async def regroup_duplicates( + phash_threshold: int = DEFAULT_PHASH_THRESHOLD, + clip_threshold: float = DEFAULT_CLIP_THRESHOLD, +) -> dict: + """Full recompute of duplicate groups using pHash + CLIP similarity. - 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. + Idempotent — safe to call as often as you like. Returns a summary dict. """ + embedder_model = settings.mulita_config.vision.embedder.name + async with AsyncSessionLocal() as session: - # Pull (id, phash) for every visible photo with a hash. Discarded - # and hidden photos are excluded so we don't keep showing groups - # made up of trashed copies or members of folders the user - # deliberately excluded from cross-cutting views. + # Pull all visible photos with a phash or embedding. rows = ( await session.execute( select(Photo.id, Photo.phash) - .where(Photo.phash.is_not(None)) .where(Photo.is_discarded.is_(False)) .where(Photo.is_hidden.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. + if not rows: await _clear_all_groups(session) await session.commit() - return { - 'photos_considered': 0, - 'groups': 0, - 'members': 0, - } + 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] + ids = [row[0] for row in rows] + phash_map = {row[0]: _hex_to_int(row[1]) for row in rows if row[1]} - uf = _UnionFind(n) + uf = _UnionFind(ids) - # O(N²) pairwise comparison. See module docstring for the - # scaling analysis and the BK-tree upgrade path. + # ── Phase 1: pHash pairwise (O(N²) on photos with phash) ── + phash_ids = [pid for pid in ids if pid in phash_map] + phash_vals = [phash_map[pid] for pid in phash_ids] + n = len(phash_ids) for i in range(n): - hi = hashes[i] + hi = phash_vals[i] if hi < 0: continue for j in range(i + 1, n): - hj = hashes[j] + hj = phash_vals[j] if hj < 0: continue - if _hamming(hi, hj) <= threshold: - uf.union(i, j) + if _hamming(hi, hj) <= phash_threshold: + uf.union_by_key(phash_ids[i], phash_ids[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) + # ── Phase 2: CLIP similarity via pgvector ── + # For each photo with an embedding, find its nearest neighbors + # within the cosine distance threshold using the HNSW index. + clip_matches = await _clip_neighbor_scan( + session, ids, embedder_model, clip_threshold + ) + for photo_id, neighbor_id in clip_matches: + uf.union_by_key(photo_id, neighbor_id) - # 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. + # ── Write results ── await _clear_all_groups(session) - # Second pass: write the new group ids for components of size 2+. + groups = uf.components(ids) groups_created = 0 members_total = 0 - for members in components.values(): - if len(members) < 2: - continue + for member_ids in groups.values(): 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, - ) + .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)" + f"regroup_duplicates: {len(ids)} photos, " + f"{groups_created} group(s), {members_total} member(s)" ) - return { - 'photos_considered': n, + 'photos_considered': len(ids), 'groups': groups_created, 'members': members_total, } +async def incremental_regroup( + since: Optional[datetime] = None, + phash_threshold: int = DEFAULT_PHASH_THRESHOLD, + clip_threshold: float = DEFAULT_CLIP_THRESHOLD, +) -> dict: + """Incremental duplicate detection for newly added photos. + + Only photos added after `since` are compared against the full library. + Much faster than a full regroup for post-scan updates: + O(new × log N) via HNSW instead of O(N²). + """ + embedder_model = settings.mulita_config.vision.embedder.name + + async with AsyncSessionLocal() as session: + # If no watermark, fall back to full regroup. + if since is None: + # Find the most recent scan start by looking at the newest + # photo that already has a duplicate_group_id check completed. + # As a simple heuristic, use photos added in the last hour. + from datetime import timedelta + since = datetime.now(timezone.utc) - timedelta(hours=1) + + # Get newly added photos (the "new" set). + new_rows = ( + await session.execute( + select(Photo.id, Photo.phash) + .where(Photo.added_at >= since) + .where(Photo.is_discarded.is_(False)) + .where(Photo.is_hidden.is_(False)) + ) + ).all() + + if not new_rows: + return {'photos_considered': 0, 'new_photos': 0, 'groups_updated': 0, 'members_added': 0} + + new_ids = [r[0] for r in new_rows] + new_phash = {r[0]: _hex_to_int(r[1]) for r in new_rows if r[1]} + + # Get ALL existing photos for union-find (we need to merge into + # existing groups). + all_rows = ( + await session.execute( + select(Photo.id, Photo.phash, Photo.duplicate_group_id) + .where(Photo.is_discarded.is_(False)) + .where(Photo.is_hidden.is_(False)) + ) + ).all() + + all_ids = [r[0] for r in all_rows] + all_phash = {r[0]: _hex_to_int(r[1]) for r in all_rows if r[1]} + existing_groups: dict[str, str] = { + r[0]: r[2] for r in all_rows if r[2] + } + + uf = _UnionFind(all_ids) + + # Pre-seed existing groups into the union-find so we merge into + # them rather than creating parallel groups. + group_to_members: dict[str, list[str]] = {} + for pid, gid in existing_groups.items(): + group_to_members.setdefault(gid, []).append(pid) + for members in group_to_members.values(): + for i in range(1, len(members)): + uf.union_by_key(members[0], members[i]) + + # ── Phase 1: pHash — compare each new photo against ALL photos ── + for new_id in new_ids: + nh = new_phash.get(new_id, -1) + if nh < 0: + continue + for existing_id, eh in all_phash.items(): + if existing_id == new_id or eh < 0: + continue + if _hamming(nh, eh) <= phash_threshold: + uf.union_by_key(new_id, existing_id) + + # ── Phase 2: CLIP — vector similarity for new photos only ── + clip_matches = await _clip_neighbor_scan( + session, new_ids, embedder_model, clip_threshold + ) + for photo_id, neighbor_id in clip_matches: + uf.union_by_key(photo_id, neighbor_id) + + # ── Write results ── + # Only update groups that contain at least one new photo. + # Clear all groups first, then rewrite. + await _clear_all_groups(session) + + groups = uf.components(all_ids) + groups_created = 0 + members_total = 0 + new_in_groups = 0 + for member_ids in groups.values(): + group_id = str(uuid.uuid4()) + 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) + if any(m in new_ids for m in member_ids): + new_in_groups += len([m for m in member_ids if m in new_ids]) + + await session.commit() + logger.info( + f"incremental_regroup: {len(new_ids)} new photos, " + f"{groups_created} group(s), {new_in_groups} new member(s) grouped" + ) + return { + 'photos_considered': len(all_ids), + 'new_photos': len(new_ids), + 'groups_updated': groups_created, + 'members_added': new_in_groups, + } + + +async def _clip_neighbor_scan( + session, + photo_ids: list[str], + embedder_model: str, + threshold: float, +) -> list[tuple[str, str]]: + """For each photo in `photo_ids` that has a CLIP embedding, find + neighbors within cosine distance `threshold` using pgvector HNSW. + + Returns a list of (photo_id, neighbor_id) pairs. + """ + matches: list[tuple[str, str]] = [] + + if not photo_ids: + return matches + + # Batch: get all embeddings for the target photos. + target_embeddings = ( + await session.execute( + select(Embedding.photo_id, Embedding.vector) + .where(Embedding.photo_id.in_(photo_ids)) + .where(Embedding.model == embedder_model) + ) + ).all() + + if not target_embeddings: + return matches + + # For each target, query nearest neighbors via pgvector. + # We use raw SQL for the <=> cosine distance operator. + for photo_id, vector in target_embeddings: + # pgvector cosine distance: <=> operator + # Find top 20 nearest neighbors within threshold. + result = await session.execute( + text(""" + SELECT e.photo_id, (e.vector <=> :vec) AS distance + FROM embeddings e + JOIN photos p ON p.id = e.photo_id + WHERE e.model = :model + AND e.photo_id != :pid + AND p.is_discarded = false + AND p.is_hidden = false + AND (e.vector <=> :vec) < :threshold + ORDER BY e.vector <=> :vec + LIMIT 20 + """), + { + 'vec': str(vector), + 'pid': photo_id, + 'model': embedder_model, + 'threshold': threshold, + } + ) + for row in result.all(): + matches.append((photo_id, row[0])) + + return matches + + 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.""" + """Reset duplicate_group_id / is_duplicate on every photo.""" await session.execute( update(Photo).values(duplicate_group_id=None, is_duplicate=False) ) diff --git a/backend/app/tasks/scan.py b/backend/app/tasks/scan.py index 8f317aa..614308d 100644 --- a/backend/app/tasks/scan.py +++ b/backend/app/tasks/scan.py @@ -5,7 +5,7 @@ import os import hashlib import asyncio from pathlib import Path -from datetime import datetime +from datetime import datetime, timezone import logging import json from typing import List, Dict, Optional @@ -419,7 +419,7 @@ async def _scan_all_source_roots_async(): countdown is a best-effort hint — on a big library the user can still hit Settings → Re-detect duplicates to force a fresh pass. """ - from app.tasks.thumbs import regroup_duplicates_task + from app.tasks.thumbs import incremental_regroup_duplicates_task from app.tasks.vision import backfill_vision, recluster_faces async with AsyncSessionLocal() as session: @@ -441,7 +441,14 @@ async def _scan_all_source_roots_async(): # firing too early just means the next manual run picks up the # late arrivals — no corrupted state. try: - regroup_duplicates_task.apply_async(countdown=60) + # Use incremental mode: only compare newly added photos + # against the full library via CLIP HNSW + pHash. + # O(new × log N) instead of O(N²). + scan_start = datetime.now(timezone.utc).isoformat() + incremental_regroup_duplicates_task.apply_async( + kwargs={'since_iso': scan_start}, + countdown=60, + ) except Exception as e: logger.warning(f"Could not queue post-scan regroup: {e}") diff --git a/backend/app/tasks/thumbs.py b/backend/app/tasks/thumbs.py index 5364aa6..b5067e3 100644 --- a/backend/app/tasks/thumbs.py +++ b/backend/app/tasks/thumbs.py @@ -467,10 +467,23 @@ async def _backfill_phashes_async(): @shared_task(name='regroup_duplicates') def regroup_duplicates_task(): - """Celery wrapper around app.services.duplicates.regroup_duplicates. + """Full recompute of duplicate groups (pHash + CLIP similarity). - Importing the service inside the task body avoids a circular import - at worker boot (the service uses AsyncSessionLocal which is also - imported here at module top).""" + Used by the Settings → Re-detect duplicates button.""" from app.services.duplicates import regroup_duplicates - return asyncio.run(regroup_duplicates()) \ No newline at end of file + return asyncio.run(regroup_duplicates()) + + +@shared_task(name='incremental_regroup_duplicates') +def incremental_regroup_duplicates_task(since_iso: str | None = None): + """Incremental duplicate detection for newly added photos. + + Compares only photos added after `since_iso` against the full library + using CLIP vector similarity (O(new × log N) via HNSW) plus pHash. + Default post-scan path — much faster than a full regroup.""" + from app.services.duplicates import incremental_regroup + from datetime import datetime, timezone + since = None + if since_iso: + since = datetime.fromisoformat(since_iso) + return asyncio.run(incremental_regroup(since=since)) \ No newline at end of file