From 691d5a56516ce2b3af0f47780054714c0e509234 Mon Sep 17 00:00:00 2001 From: domverse Date: Wed, 15 Jul 2026 17:14:24 +0200 Subject: [PATCH] US01-04: Detect and Decide Duplicates (#49) --- migrations/versions/0002_duplicate_engine.py | 99 ++++ photo_pipeline/models/__init__.py | 13 +- photo_pipeline/models/assets.py | 3 + photo_pipeline/models/duplicates.py | 62 ++ photo_pipeline/services/duplicates.py | 568 +++++++++++++++++++ photo_pipeline/services/hashing.py | 37 ++ tests/integration/test_duplicate_engine.py | 291 ++++++++++ tests/integration/test_migrations.py | 16 +- tests/unit/test_phash.py | 52 ++ 9 files changed, 1137 insertions(+), 4 deletions(-) create mode 100644 migrations/versions/0002_duplicate_engine.py create mode 100644 photo_pipeline/models/duplicates.py create mode 100644 photo_pipeline/services/duplicates.py create mode 100644 tests/integration/test_duplicate_engine.py create mode 100644 tests/unit/test_phash.py diff --git a/migrations/versions/0002_duplicate_engine.py b/migrations/versions/0002_duplicate_engine.py new file mode 100644 index 0000000..39cef55 --- /dev/null +++ b/migrations/versions/0002_duplicate_engine.py @@ -0,0 +1,99 @@ +"""Duplicate engine: canonical link, phash version, clusters, members, negative links. + +Revision ID: 0002_duplicate_engine +Revises: 0001_initial_identity +Create Date: 2026-07-15 +""" + +import sqlalchemy as sa +from alembic import op + +revision = "0002_duplicate_engine" +down_revision = "0001_initial_identity" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + # Plain ADD COLUMN: SQLite cannot ALTER-ADD a foreign key, and the ORM model + # keeps the ForeignKey for relationship metadata. canonical_asset_id is a soft + # self-link validated in the service (no cycles), not DB-enforced. + op.add_column("assets", sa.Column("phash_version", sa.Integer(), nullable=True)) + op.add_column("assets", sa.Column("canonical_asset_id", sa.String(), nullable=True)) + op.create_index("ix_assets_canonical_asset_id", "assets", ["canonical_asset_id"]) + + op.create_table( + "duplicate_clusters", + sa.Column("id", sa.String(), primary_key=True), + sa.Column("method", sa.String(), nullable=False), + sa.Column("confidence", sa.String(), nullable=False), + sa.Column("state", sa.String(), nullable=False), + sa.Column("decision", sa.String(), nullable=True), + sa.Column( + "canonical_asset_id", sa.String(), sa.ForeignKey("assets.id"), nullable=True + ), + sa.Column("version", sa.Integer(), nullable=False, server_default="1"), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("CURRENT_TIMESTAMP"), + ), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("CURRENT_TIMESTAMP"), + ), + sa.Column("decided_at", sa.DateTime(timezone=True), nullable=True), + ) + op.create_index("ix_duplicate_clusters_state", "duplicate_clusters", ["state"]) + + op.create_table( + "duplicate_members", + sa.Column( + "cluster_id", + sa.String(), + sa.ForeignKey("duplicate_clusters.id"), + primary_key=True, + ), + sa.Column( + "asset_id", sa.String(), sa.ForeignKey("assets.id"), primary_key=True + ), + sa.Column("role", sa.String(), nullable=False, server_default="member"), + sa.Column("distance", sa.Integer(), nullable=True), + sa.Column("evidence", sa.String(), nullable=True), + sa.Column( + "added_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("CURRENT_TIMESTAMP"), + ), + ) + op.create_index("ix_duplicate_members_asset_id", "duplicate_members", ["asset_id"]) + + op.create_table( + "duplicate_negative_links", + sa.Column( + "asset_a", sa.String(), sa.ForeignKey("assets.id"), primary_key=True + ), + sa.Column( + "asset_b", sa.String(), sa.ForeignKey("assets.id"), primary_key=True + ), + sa.Column("reason", sa.String(), nullable=True), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("CURRENT_TIMESTAMP"), + ), + ) + + +def downgrade() -> None: + op.drop_table("duplicate_negative_links") + op.drop_table("duplicate_members") + op.drop_table("duplicate_clusters") + op.drop_index("ix_assets_canonical_asset_id", "assets") + op.drop_column("assets", "canonical_asset_id") + op.drop_column("assets", "phash_version") diff --git a/photo_pipeline/models/__init__.py b/photo_pipeline/models/__init__.py index 7fac3c6..a825d0b 100644 --- a/photo_pipeline/models/__init__.py +++ b/photo_pipeline/models/__init__.py @@ -5,5 +5,16 @@ Alembic environment relies on. """ from photo_pipeline.models.assets import Asset, AssetPath +from photo_pipeline.models.duplicates import ( + DuplicateCluster, + DuplicateMember, + DuplicateNegativeLink, +) -__all__ = ["Asset", "AssetPath"] +__all__ = [ + "Asset", + "AssetPath", + "DuplicateCluster", + "DuplicateMember", + "DuplicateNegativeLink", +] diff --git a/photo_pipeline/models/assets.py b/photo_pipeline/models/assets.py index 53fc547..a2601d7 100644 --- a/photo_pipeline/models/assets.py +++ b/photo_pipeline/models/assets.py @@ -30,11 +30,14 @@ class Asset(Base): pixel_sha256: Mapped[str | None] = mapped_column(String) phash: Mapped[str | None] = mapped_column(String) hash_version: Mapped[int] = mapped_column(Integer, nullable=False, default=1) + phash_version: Mapped[int | None] = mapped_column(Integer) byte_size: Mapped[int | None] = mapped_column(Integer) discovered_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) missing_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) availability_state: Mapped[str] = mapped_column(String, nullable=False, default="active") + # Duplicate canonical link: NULL when the asset is itself canonical or undecided. + canonical_asset_id: Mapped[str | None] = mapped_column(ForeignKey("assets.id")) state_version: Mapped[int] = mapped_column(Integer, nullable=False, default=1) created_at: Mapped[datetime] = mapped_column( diff --git a/photo_pipeline/models/duplicates.py b/photo_pipeline/models/duplicates.py new file mode 100644 index 0000000..64f3d3c --- /dev/null +++ b/photo_pipeline/models/duplicates.py @@ -0,0 +1,62 @@ +"""Duplicate clusters, their members, and not-duplicate negative links. + +A cluster is a reviewable grouping produced by a detection ``method`` (exact, +pixel, or perceptual) at a ``confidence`` band. Exact/pixel clusters can be +decided automatically; perceptual clusters stay open for human review. Every +member carries the evidence behind its inclusion. Negative links record +``not_duplicate`` decisions so rescans do not re-suggest a rejected pair. +""" + +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy import DateTime, ForeignKey, Integer, String, func +from sqlalchemy.orm import Mapped, mapped_column + +from photo_pipeline.db import Base + + +class DuplicateCluster(Base): + __tablename__ = "duplicate_clusters" + + id: Mapped[str] = mapped_column(String, primary_key=True) + method: Mapped[str] = mapped_column(String, nullable=False) # exact|pixel|perceptual + confidence: Mapped[str] = mapped_column(String, nullable=False) # band name + state: Mapped[str] = mapped_column(String, nullable=False) # open|decided|dismissed|reopened + decision: Mapped[str | None] = mapped_column(String) # canonical|not_duplicate|deferred + canonical_asset_id: Mapped[str | None] = mapped_column(ForeignKey("assets.id")) + version: Mapped[int] = mapped_column(Integer, nullable=False, default=1) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now() + ) + decided_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + + +class DuplicateMember(Base): + __tablename__ = "duplicate_members" + + cluster_id: Mapped[str] = mapped_column( + ForeignKey("duplicate_clusters.id"), primary_key=True + ) + asset_id: Mapped[str] = mapped_column(ForeignKey("assets.id"), primary_key=True) + role: Mapped[str] = mapped_column(String, nullable=False, default="member") # member|canonical|variant + distance: Mapped[int | None] = mapped_column(Integer) # phash distance to representative + evidence: Mapped[str | None] = mapped_column(String) # JSON blob explaining inclusion + added_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + + +class DuplicateNegativeLink(Base): + __tablename__ = "duplicate_negative_links" + + asset_a: Mapped[str] = mapped_column(ForeignKey("assets.id"), primary_key=True) + asset_b: Mapped[str] = mapped_column(ForeignKey("assets.id"), primary_key=True) + reason: Mapped[str | None] = mapped_column(String) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) diff --git a/photo_pipeline/services/duplicates.py b/photo_pipeline/services/duplicates.py new file mode 100644 index 0000000..bdb59e9 --- /dev/null +++ b/photo_pipeline/services/duplicates.py @@ -0,0 +1,568 @@ +"""DuplicateService — explainable clusters and reversible decisions. + +Detection runs in two categories: + +- **Content-identical** clusters join assets that share exact bytes (``sha256``) or + normalized pixels (``pixel_sha256``). These are high confidence and are decided + automatically (a recommended canonical, the rest variants) — unless a member pair + carries a ``not_duplicate`` negative link, which is a contradiction that reopens + the cluster for review instead of silently overriding the user. +- **Perceptual** clusters join assets whose perceptual hashes are within a distance + band (NEAR/SIMILAR). These are review candidates: never decided automatically, and + negative-linked pairs are suppressed so a rejected pair is not re-suggested. + +Decisions (``canonical`` / ``not_duplicate`` / ``deferred``) persist with evidence, +use optimistic version checks, are reversible, and can never form a canonical cycle. +A new content-identical member of an already-decided cluster inherits the established +canonical relationship rather than reopening it. + +Extracts photo_analyzer.cluster_duplicates/mark_duplicates (union-find over the +Hamming graph); path-keyed identity and automatic largest-file selection are +replaced by stable asset IDs and reviewable decisions (donor_ledger.yaml: pa-dupes). +""" + +from __future__ import annotations + +import json +import uuid +from collections import Counter +from dataclasses import dataclass, field +from datetime import datetime, timezone +from enum import Enum + +from sqlalchemy import select +from sqlalchemy.orm import sessionmaker + +from photo_pipeline.models import ( + Asset, + DuplicateCluster, + DuplicateMember, + DuplicateNegativeLink, +) +from photo_pipeline.services import hashing + +NEAR_MAX = 5 +SIMILAR_MAX = 10 + + +class Method(str, Enum): + EXACT = "exact" + PIXEL = "pixel" + PERCEPTUAL = "perceptual" + + +class Confidence(str, Enum): + EXACT = "exact" + PIXEL = "pixel" + NEAR = "near" + SIMILAR = "similar" + + +class ClusterState(str, Enum): + OPEN = "open" + DECIDED = "decided" + DISMISSED = "dismissed" + REOPENED = "reopened" + DEFERRED = "deferred" + + +class Decision(str, Enum): + CANONICAL = "canonical" + NOT_DUPLICATE = "not_duplicate" + DEFERRED = "deferred" + + +class Role(str, Enum): + MEMBER = "member" + CANONICAL = "canonical" + VARIANT = "variant" + + +class DuplicateError(RuntimeError): + """Invalid decision (unknown cluster, non-member canonical, would cycle).""" + + +class ConflictError(RuntimeError): + """Optimistic version check failed; the cluster changed since it was read.""" + + +@dataclass +class DetectionReport: + clusters: list[dict] = field(default_factory=list) + + @property + def counts(self) -> dict[str, int]: + return dict(Counter(c["state"] for c in self.clusters)) + + +def _pair(a: str, b: str) -> tuple[str, str]: + return (a, b) if a <= b else (b, a) + + +class _UnionFind: + def __init__(self, items): + self.parent = {x: x for x in items} + + def find(self, x): + while self.parent[x] != x: + self.parent[x] = self.parent[self.parent[x]] + x = self.parent[x] + return x + + def union(self, a, b): + ra, rb = self.find(a), self.find(b) + if ra != rb: + self.parent[rb] = ra + + def groups(self) -> list[list[str]]: + out: dict[str, list[str]] = {} + for x in self.parent: + out.setdefault(self.find(x), []).append(x) + return [sorted(g) for g in out.values()] + + +class DuplicateService: + def __init__(self, session_factory: sessionmaker) -> None: + self._session_factory = session_factory + + # ── perceptual hash backfill ─────────────────────────────────────────── + def ensure_phashes(self) -> int: + updated = 0 + with self._session_factory() as session: + assets = session.execute( + select(Asset).where( + Asset.availability_state == "active", + Asset.current_path.isnot(None), + ) + ).scalars() + for asset in assets: + if asset.phash is not None and asset.phash_version == hashing.PHASH_VERSION: + continue + value = hashing.safe_phash(asset.current_path) + if value is not None: + asset.phash = value + asset.phash_version = hashing.PHASH_VERSION + updated += 1 + session.commit() + return updated + + # ── detection ────────────────────────────────────────────────────────── + def detect(self) -> DetectionReport: + self.ensure_phashes() + now = datetime.now(timezone.utc) + report = DetectionReport() + with self._session_factory() as session: + assets = list( + session.execute( + select(Asset).where( + Asset.availability_state == "active", + Asset.current_path.isnot(None), + ) + ).scalars() + ) + by_id = {a.id: a for a in assets} + negatives = { + _pair(link.asset_a, link.asset_b) + for link in session.execute(select(DuplicateNegativeLink)).scalars() + } + clusters = list(session.execute(select(DuplicateCluster)).scalars()) + members = { + c.id: list( + session.execute( + select(DuplicateMember).where(DuplicateMember.cluster_id == c.id) + ).scalars() + ) + for c in clusters + } + + content_groups, content_pairs = self._content_groups(assets) + perceptual_groups = self._perceptual_groups(assets, content_pairs, negatives) + + touched: list[DuplicateCluster] = [] + for method, ids, distances in content_groups: + touched.append( + self._sync_content_cluster( + session, method, ids, by_id, negatives, clusters, members, now + ) + ) + for ids, distances in perceptual_groups: + touched.append( + self._sync_perceptual_cluster( + session, ids, distances, by_id, clusters, members, now + ) + ) + session.commit() + report.clusters = [self._snapshot(session, c.id) for c in touched] + return report + + def _content_groups(self, assets): + """Union-find over exact-byte and same-pixel edges. Returns groups of >=2 + with method EXACT (all same bytes) or PIXEL, plus the set of joined pairs.""" + uf = _UnionFind([a.id for a in assets]) + pairs: set[tuple[str, str]] = set() + for key in ("current_sha256", "pixel_sha256"): + buckets: dict[str, list[str]] = {} + for a in assets: + value = getattr(a, key) + if value: + buckets.setdefault(value, []).append(a.id) + for group in buckets.values(): + for other in group[1:]: + uf.union(group[0], other) + for i in range(len(group)): + for j in range(i + 1, len(group)): + pairs.add(_pair(group[i], group[j])) + by_id = {a.id: a for a in assets} + result = [] + for group in uf.groups(): + if len(group) < 2: + continue + shas = {by_id[i].current_sha256 for i in group} + method = Method.EXACT if len(shas) == 1 and None not in shas else Method.PIXEL + result.append((method, sorted(group), {})) + return result, pairs + + def _perceptual_groups(self, assets, content_pairs, negatives): + """Union-find over phash distance edges (<= SIMILAR_MAX), excluding pairs + that are already content-identical or negatively linked.""" + hashed = [a for a in assets if a.phash] + uf = _UnionFind([a.id for a in hashed]) + distances: dict[tuple[str, str], int] = {} + for i in range(len(hashed)): + for j in range(i + 1, len(hashed)): + a, b = hashed[i], hashed[j] + pair = _pair(a.id, b.id) + if pair in content_pairs or pair in negatives: + continue + dist = hashing.phash_distance(a.phash, b.phash) + if dist <= SIMILAR_MAX: + uf.union(a.id, b.id) + distances[pair] = dist + result = [] + for group in uf.groups(): + if len(group) < 2: + continue + group_dists = { + _pair(x, y): distances[_pair(x, y)] + for x in group + for y in group + if x < y and _pair(x, y) in distances + } + result.append((sorted(group), group_dists)) + return result + + # ── cluster upserts ────────────────────────────────────────────────────── + def _match_existing(self, ids, clusters, members, categories): + best = None + best_overlap = 0 + for cluster in clusters: + if cluster.method not in categories: + continue + overlap = len(set(ids) & {m.asset_id for m in members[cluster.id]}) + if overlap > best_overlap: + best, best_overlap = cluster, overlap + return best + + def _sync_content_cluster( + self, session, method, ids, by_id, negatives, clusters, members, now + ): + contradiction = any( + _pair(ids[i], ids[j]) in negatives + for i in range(len(ids)) + for j in range(i + 1, len(ids)) + ) + confidence = Confidence.EXACT if method == Method.EXACT else Confidence.PIXEL + # Also consider a prior perceptual cluster over the same assets: if a member + # pair became byte-identical after a not_duplicate decision, reopen that same + # reviewed cluster rather than spawning a parallel one. + existing = self._match_existing( + ids, + clusters, + members, + {Method.EXACT.value, Method.PIXEL.value, Method.PERCEPTUAL.value}, + ) + + if existing is None: + cluster = DuplicateCluster( + id=str(uuid.uuid4()), + method=method.value, + confidence=confidence.value, + state=ClusterState.OPEN.value, + version=1, + ) + session.add(cluster) + session.flush() + self._replace_members(session, cluster, ids, by_id, distances={}) + if contradiction: + cluster.state = ClusterState.REOPENED.value + else: + self._auto_decide(session, cluster, ids, by_id, now) + clusters.append(cluster) + members[cluster.id] = list( + session.execute( + select(DuplicateMember).where(DuplicateMember.cluster_id == cluster.id) + ).scalars() + ) + return cluster + + cluster = existing + cluster.method = method.value + cluster.confidence = confidence.value + prior_members = {m.asset_id for m in members[cluster.id]} + added = set(ids) - prior_members + self._replace_members(session, cluster, ids, by_id, distances={}) + + if contradiction: + cluster.state = ClusterState.REOPENED.value + cluster.decision = None + cluster.canonical_asset_id = None + self._clear_asset_links(session, ids) + elif ( + cluster.decision == Decision.CANONICAL.value + and cluster.canonical_asset_id in ids + ): + # Consistent growth: new members inherit the established canonical. + self._apply_canonical(session, cluster, ids, cluster.canonical_asset_id) + cluster.state = ClusterState.DECIDED.value + elif cluster.state in (ClusterState.DISMISSED.value,) and added: + cluster.state = ClusterState.REOPENED.value + elif cluster.state in (ClusterState.OPEN.value, ClusterState.REOPENED.value): + self._auto_decide(session, cluster, ids, by_id, now) + cluster.version += 1 + cluster.updated_at = now + members[cluster.id] = list( + session.execute( + select(DuplicateMember).where(DuplicateMember.cluster_id == cluster.id) + ).scalars() + ) + return cluster + + def _sync_perceptual_cluster(self, session, ids, distances, by_id, clusters, members, now): + confidence = ( + Confidence.NEAR + if distances and max(distances.values()) <= NEAR_MAX + else Confidence.SIMILAR + ) + existing = self._match_existing(ids, clusters, members, {Method.PERCEPTUAL.value}) + if existing is None: + cluster = DuplicateCluster( + id=str(uuid.uuid4()), + method=Method.PERCEPTUAL.value, + confidence=confidence.value, + state=ClusterState.OPEN.value, + version=1, + ) + session.add(cluster) + session.flush() + self._replace_members(session, cluster, ids, by_id, distances) + clusters.append(cluster) + members[cluster.id] = list( + session.execute( + select(DuplicateMember).where(DuplicateMember.cluster_id == cluster.id) + ).scalars() + ) + return cluster + + cluster = existing + prior = {m.asset_id for m in members[cluster.id]} + added = set(ids) - prior + cluster.confidence = confidence.value + self._replace_members(session, cluster, ids, by_id, distances) + if added and cluster.state in ( + ClusterState.DECIDED.value, + ClusterState.DISMISSED.value, + ClusterState.DEFERRED.value, + ): + cluster.state = ClusterState.REOPENED.value + cluster.version += 1 + cluster.updated_at = now + members[cluster.id] = list( + session.execute( + select(DuplicateMember).where(DuplicateMember.cluster_id == cluster.id) + ).scalars() + ) + return cluster + + def _replace_members(self, session, cluster, ids, by_id, distances): + current = { + m.asset_id: m + for m in session.execute( + select(DuplicateMember).where(DuplicateMember.cluster_id == cluster.id) + ).scalars() + } + wanted = set(ids) + for asset_id, member in current.items(): + if asset_id not in wanted: + session.delete(member) + for asset_id in ids: + asset = by_id[asset_id] + evidence = json.dumps( + { + "sha256": asset.current_sha256, + "pixel_sha256": asset.pixel_sha256, + "phash": asset.phash, + } + ) + distance = next( + (d for p, d in distances.items() if asset_id in p), None + ) + if asset_id in current: + current[asset_id].evidence = evidence + current[asset_id].distance = distance + else: + session.add( + DuplicateMember( + cluster_id=cluster.id, + asset_id=asset_id, + role=Role.MEMBER.value, + distance=distance, + evidence=evidence, + ) + ) + session.flush() + + def _auto_decide(self, session, cluster, ids, by_id, now): + canonical = self._recommend_canonical(ids, by_id) + self._apply_canonical(session, cluster, ids, canonical) + cluster.decision = Decision.CANONICAL.value + cluster.canonical_asset_id = canonical + cluster.state = ClusterState.DECIDED.value + cluster.decided_at = now + + @staticmethod + def _recommend_canonical(ids, by_id) -> str: + # ponytail: largest file, path as deterministic tie-break. The concept's + # richer policy (resolution, least recompression, metadata richness) lands + # with the review UI story. + return max(ids, key=lambda i: (by_id[i].byte_size or 0, by_id[i].current_path or "")) + + def _apply_canonical(self, session, cluster, ids, canonical_id): + for member in session.execute( + select(DuplicateMember).where(DuplicateMember.cluster_id == cluster.id) + ).scalars(): + asset = session.get(Asset, member.asset_id) + if member.asset_id == canonical_id: + member.role = Role.CANONICAL.value + asset.canonical_asset_id = None + else: + member.role = Role.VARIANT.value + asset.canonical_asset_id = canonical_id + self._assert_acyclic(session, ids) + + @staticmethod + def _clear_asset_links(session, ids): + for asset_id in ids: + session.get(Asset, asset_id).canonical_asset_id = None + + @staticmethod + def _assert_acyclic(session, ids): + for start in ids: + seen = set() + cur = start + while cur is not None: + if cur in seen: + raise DuplicateError(f"canonical link cycle at {cur}") + seen.add(cur) + asset = session.get(Asset, cur) + cur = asset.canonical_asset_id if asset else None + + # ── decisions ──────────────────────────────────────────────────────────── + def decide( + self, + cluster_id: str, + decision: Decision | str, + *, + canonical_asset_id: str | None = None, + expected_version: int | None = None, + reason: str | None = None, + ) -> dict: + decision = Decision(decision) + now = datetime.now(timezone.utc) + with self._session_factory() as session: + cluster = session.get(DuplicateCluster, cluster_id) + if cluster is None: + raise DuplicateError(f"unknown cluster {cluster_id}") + if expected_version is not None and cluster.version != expected_version: + raise ConflictError( + f"cluster {cluster_id} is at version {cluster.version}, " + f"expected {expected_version}" + ) + member_ids = [ + m.asset_id + for m in session.execute( + select(DuplicateMember).where(DuplicateMember.cluster_id == cluster_id) + ).scalars() + ] + self._reset_members(session, cluster, member_ids) + self._remove_negative_links(session, member_ids) + + if decision == Decision.CANONICAL: + if canonical_asset_id not in member_ids: + raise DuplicateError("canonical_asset_id must be a cluster member") + self._apply_canonical(session, cluster, member_ids, canonical_asset_id) + cluster.decision = Decision.CANONICAL.value + cluster.canonical_asset_id = canonical_asset_id + cluster.state = ClusterState.DECIDED.value + elif decision == Decision.NOT_DUPLICATE: + self._add_negative_links(session, member_ids, reason) + cluster.decision = Decision.NOT_DUPLICATE.value + cluster.canonical_asset_id = None + cluster.state = ClusterState.DISMISSED.value + else: # DEFERRED + cluster.decision = Decision.DEFERRED.value + cluster.canonical_asset_id = None + cluster.state = ClusterState.DEFERRED.value + + cluster.decided_at = now + cluster.updated_at = now + cluster.version += 1 + session.commit() + return self._snapshot(session, cluster_id) + + def _reset_members(self, session, cluster, member_ids): + for member in session.execute( + select(DuplicateMember).where(DuplicateMember.cluster_id == cluster.id) + ).scalars(): + member.role = Role.MEMBER.value + for asset_id in member_ids: + session.get(Asset, asset_id).canonical_asset_id = None + + @staticmethod + def _add_negative_links(session, member_ids, reason): + existing = { + _pair(link.asset_a, link.asset_b) + for link in session.execute(select(DuplicateNegativeLink)).scalars() + } + for i in range(len(member_ids)): + for j in range(i + 1, len(member_ids)): + a, b = _pair(member_ids[i], member_ids[j]) + if (a, b) not in existing: + session.add( + DuplicateNegativeLink(asset_a=a, asset_b=b, reason=reason) + ) + + @staticmethod + def _remove_negative_links(session, member_ids): + member_set = set(member_ids) + for link in session.execute(select(DuplicateNegativeLink)).scalars(): + if link.asset_a in member_set and link.asset_b in member_set: + session.delete(link) + + def _snapshot(self, session, cluster_id) -> dict: + cluster = session.get(DuplicateCluster, cluster_id) + members = session.execute( + select(DuplicateMember).where(DuplicateMember.cluster_id == cluster_id) + ).scalars() + return { + "id": cluster.id, + "method": cluster.method, + "confidence": cluster.confidence, + "state": cluster.state, + "decision": cluster.decision, + "canonical_asset_id": cluster.canonical_asset_id, + "version": cluster.version, + "members": sorted( + ({"asset_id": m.asset_id, "role": m.role, "distance": m.distance} for m in members), + key=lambda m: m["asset_id"], + ), + } diff --git a/photo_pipeline/services/hashing.py b/photo_pipeline/services/hashing.py index 7af9883..79526c5 100644 --- a/photo_pipeline/services/hashing.py +++ b/photo_pipeline/services/hashing.py @@ -19,6 +19,7 @@ from pathlib import Path from PIL import Image, ImageOps PIXEL_HASH_VERSION = 1 +PHASH_VERSION = 1 _CHUNK = 1 << 20 @@ -45,3 +46,39 @@ def safe_pixel_sha256(path: Path | str) -> str | None: return pixel_sha256(path) except Exception: return None + + +def phash(path: Path | str) -> str: + """DCT perceptual hash → 16 hex chars (64 bits). + + Standard imagehash.phash recipe (extracted verbatim from + photo_analyzer._phash_image): grayscale → 32x32 → 2D DCT-II → keep the top-left + 8x8 low-frequency block → bit = coefficient > median. Resilient to resize and + recompression; used only as evidence for review, never for automatic exclusion. + """ + import numpy as np + from scipy.fftpack import dct + + with Image.open(path) as image: + small = image.convert("L").resize((32, 32), Image.LANCZOS) + matrix = np.asarray(small, dtype=np.float64) + transformed = dct(dct(matrix, axis=0), axis=1) + low = transformed[:8, :8] + bits = (low > np.median(low)).flatten() + value = 0 + for bit in bits: + value = (value << 1) | int(bit) + return f"{value:016x}" + + +def safe_phash(path: Path | str) -> str | None: + """``phash`` but returns None for undecodable images instead of raising.""" + try: + return phash(path) + except Exception: + return None + + +def phash_distance(a_hex: str, b_hex: str) -> int: + """Hamming distance between two 64-bit hex perceptual hashes.""" + return bin(int(a_hex, 16) ^ int(b_hex, 16)).count("1") diff --git a/tests/integration/test_duplicate_engine.py b/tests/integration/test_duplicate_engine.py new file mode 100644 index 0000000..03558cd --- /dev/null +++ b/tests/integration/test_duplicate_engine.py @@ -0,0 +1,291 @@ +"""Duplicate engine: banded clustering, reversible decisions, negative links, +reopening on contradiction, no canonical cycles, and durability across restart. + +Fixtures are inlined (not in a sibling conftest.py) to avoid shadowing the +characterization suite's bare ``conftest`` import under pytest's prepend mode. +""" + +import shutil + +import numpy as np +import pytest +from PIL import Image +from sqlalchemy import select + +from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations +from photo_pipeline.models import Asset, DuplicateNegativeLink +from photo_pipeline.services.duplicates import ( + ClusterState, + ConflictError, + Decision, + DuplicateError, + DuplicateService, + Method, +) +from photo_pipeline.services.inventory import InventoryService + + +@pytest.fixture +def db_url(tmp_path): + url = f"sqlite:///{tmp_path / 'dup.db'}" + run_migrations(url) + return url + + +@pytest.fixture +def make_factory(): + engines = [] + + def _make(url): + engine = create_db_engine(url) + engines.append(engine) + return create_session_factory(engine) + + yield _make + for engine in engines: + engine.dispose() + + +@pytest.fixture +def factory(db_url, make_factory): + return make_factory(db_url) + + +@pytest.fixture +def inventory(factory): + return InventoryService(factory) + + +@pytest.fixture +def duplicates(factory): + return DuplicateService(factory) + + +def structured(path, seed, size=(256, 192)): + path.parent.mkdir(parents=True, exist_ok=True) + rng = np.random.default_rng(seed) + w, h = size + base = np.zeros((h, w, 3), dtype=np.uint8) + for _ in range(6): + x0 = int(rng.integers(0, w - 60)) + y0 = int(rng.integers(0, h - 60)) + base[y0 : y0 + 60, x0 : x0 + 60] = rng.integers(0, 256, 3) + grad = np.linspace(0, 120, w, dtype=np.uint8) + base[:, :, 0] = np.clip(base[:, :, 0].astype(int) + grad[None, :], 0, 255) + Image.fromarray(base).save(path, quality=95) + return path + + +def resized_copy(src, dst, scale=0.5): + with Image.open(src) as image: + image.resize( + (int(image.width * scale), int(image.height * scale)), Image.LANCZOS + ).save(dst, quality=95) + return dst + + +def only(clusters, method): + return [c for c in clusters if c["method"] == method] + + +def test_exact_copies_form_auto_decided_cluster(tmp_path, inventory, duplicates, factory): + lib = tmp_path / "lib" + a = structured(lib / "a.jpg", 1) + shutil.copy2(a, lib / "a_copy.jpg") + inventory.scan(lib) + + clusters = duplicates.detect().clusters + exact = only(clusters, Method.EXACT.value) + assert len(exact) == 1 + cluster = exact[0] + assert cluster["state"] == ClusterState.DECIDED.value + assert cluster["decision"] == Decision.CANONICAL.value + roles = {m["role"] for m in cluster["members"]} + assert roles == {"canonical", "variant"} + + # The variant asset points at the canonical; the canonical points at nothing. + assets = {a.id: a for a in factory().execute(select(Asset)).scalars()} + canonical = cluster["canonical_asset_id"] + for m in cluster["members"]: + if m["role"] == "variant": + assert assets[m["asset_id"]].canonical_asset_id == canonical + else: + assert assets[m["asset_id"]].canonical_asset_id is None + + +def test_same_pixels_different_bytes_form_pixel_cluster(tmp_path, inventory, duplicates): + lib = tmp_path / "lib" + a = structured(lib / "a.jpg", 1) + twin = lib / "a_meta.jpg" + shutil.copy2(a, twin) + with open(twin, "ab") as handle: # trailing bytes: same pixels, different bytes + handle.write(b"\xff\xfe metadata") + inventory.scan(lib) + + clusters = duplicates.detect().clusters + pixel = only(clusters, Method.PIXEL.value) + assert len(pixel) == 1 + assert pixel[0]["confidence"] == "pixel" + assert pixel[0]["state"] == ClusterState.DECIDED.value + + +def test_perceptual_variant_is_review_only(tmp_path, inventory, duplicates): + lib = tmp_path / "lib" + a = structured(lib / "a.jpg", 1) + resized_copy(a, lib / "a_small.jpg") + inventory.scan(lib) + + clusters = duplicates.detect().clusters + perceptual = only(clusters, Method.PERCEPTUAL.value) + assert len(perceptual) == 1 + assert perceptual[0]["state"] == ClusterState.OPEN.value # never auto-decided + assert perceptual[0]["decision"] is None + assert perceptual[0]["confidence"] in ("near", "similar") + + +def test_distinct_images_do_not_cluster(tmp_path, inventory, duplicates): + lib = tmp_path / "lib" + structured(lib / "a.jpg", 1) + structured(lib / "b.jpg", 2) + inventory.scan(lib) + assert duplicates.detect().clusters == [] + + +def test_not_duplicate_decision_persists_negative_link_and_suppresses( + tmp_path, inventory, duplicates, factory +): + lib = tmp_path / "lib" + a = structured(lib / "a.jpg", 1) + resized_copy(a, lib / "a_small.jpg") + inventory.scan(lib) + cluster = only(duplicates.detect().clusters, Method.PERCEPTUAL.value)[0] + + duplicates.decide(cluster["id"], Decision.NOT_DUPLICATE) + + links = list(factory().execute(select(DuplicateNegativeLink)).scalars()) + assert len(links) == 1 + + # Re-detecting must not re-suggest the rejected pair. + again = only(duplicates.detect().clusters, Method.PERCEPTUAL.value) + assert again == [] + + +def test_decision_is_reversible(tmp_path, inventory, duplicates, factory): + lib = tmp_path / "lib" + a = structured(lib / "a.jpg", 1) + resized_copy(a, lib / "a_small.jpg") + inventory.scan(lib) + cluster = only(duplicates.detect().clusters, Method.PERCEPTUAL.value)[0] + members = [m["asset_id"] for m in cluster["members"]] + + duplicates.decide(cluster["id"], Decision.NOT_DUPLICATE) + assert factory().execute(select(DuplicateNegativeLink)).scalars().all() + + # Reverse: choose a canonical instead. + reversed_snapshot = duplicates.decide( + cluster["id"], Decision.CANONICAL, canonical_asset_id=members[0] + ) + assert reversed_snapshot["state"] == ClusterState.DECIDED.value + assert not factory().execute(select(DuplicateNegativeLink)).scalars().all() + + +def test_version_conflict_is_rejected(tmp_path, inventory, duplicates): + lib = tmp_path / "lib" + a = structured(lib / "a.jpg", 1) + resized_copy(a, lib / "a_small.jpg") + inventory.scan(lib) + cluster = only(duplicates.detect().clusters, Method.PERCEPTUAL.value)[0] + + with pytest.raises(ConflictError): + duplicates.decide( + cluster["id"], Decision.DEFERRED, expected_version=cluster["version"] + 5 + ) + + +def test_canonical_must_be_a_member(tmp_path, inventory, duplicates): + lib = tmp_path / "lib" + a = structured(lib / "a.jpg", 1) + resized_copy(a, lib / "a_small.jpg") + inventory.scan(lib) + cluster = only(duplicates.detect().clusters, Method.PERCEPTUAL.value)[0] + + with pytest.raises(DuplicateError): + duplicates.decide(cluster["id"], Decision.CANONICAL, canonical_asset_id="ghost") + + +def test_no_canonical_cycles(tmp_path, inventory, duplicates, factory): + lib = tmp_path / "lib" + a = structured(lib / "a.jpg", 1) + shutil.copy2(a, lib / "a_copy.jpg") + inventory.scan(lib) + duplicates.detect() + + # Every asset's canonical chain terminates (no loop). + assets = {a.id: a for a in factory().execute(select(Asset)).scalars()} + for start in assets: + seen, cur = set(), start + while cur is not None: + assert cur not in seen + seen.add(cur) + cur = assets[cur].canonical_asset_id + + +def test_new_exact_member_inherits_decision(tmp_path, inventory, duplicates, factory): + lib = tmp_path / "lib" + a = structured(lib / "a.jpg", 1) + shutil.copy2(a, lib / "a2.jpg") + inventory.scan(lib) + cluster = only(duplicates.detect().clusters, Method.EXACT.value)[0] + canonical = cluster["canonical_asset_id"] + + # A third exact copy appears later. + shutil.copy2(a, lib / "a3.jpg") + inventory.scan(lib) + cluster2 = only(duplicates.detect().clusters, Method.EXACT.value)[0] + + assert cluster2["id"] == cluster["id"] + assert cluster2["state"] == ClusterState.DECIDED.value + assert len(cluster2["members"]) == 3 + assets = {a.id: a for a in factory().execute(select(Asset)).scalars()} + variants = [m for m in cluster2["members"] if m["role"] == "variant"] + assert len(variants) == 2 + for m in variants: + assert assets[m["asset_id"]].canonical_asset_id == canonical + + +def test_contradictory_member_reopens_reviewed_cluster( + tmp_path, inventory, duplicates, db_url, make_factory +): + lib = tmp_path / "lib" + a = structured(lib / "a.jpg", 1) + b = resized_copy(a, lib / "b.jpg") + inventory.scan(lib) + cluster = only(duplicates.detect().clusters, Method.PERCEPTUAL.value)[0] + duplicates.decide(cluster["id"], Decision.NOT_DUPLICATE) # dismissed + negative link + + # b is replaced by an exact copy of a — now byte-identical to a "not duplicate". + shutil.copy2(a, b) + inventory.scan(lib) + clusters = duplicates.detect().clusters + + reopened = [c for c in clusters if c["state"] == ClusterState.REOPENED.value] + assert reopened, "byte-identical contradiction of a not_duplicate must reopen review" + + +def test_decisions_durable_across_restart(tmp_path, db_url, make_factory): + lib = tmp_path / "lib" + a = structured(lib / "a.jpg", 1) + shutil.copy2(a, lib / "a_copy.jpg") + + InventoryService(make_factory(db_url)).scan(lib) + cluster = only( + DuplicateService(make_factory(db_url)).detect().clusters, Method.EXACT.value + )[0] + + # Fresh factory against the same database: the decision persisted. + from photo_pipeline.models import DuplicateCluster + + with make_factory(db_url)() as session: + persisted = session.get(DuplicateCluster, cluster["id"]) + assert persisted.state == ClusterState.DECIDED.value + assert persisted.decision == Decision.CANONICAL.value diff --git a/tests/integration/test_migrations.py b/tests/integration/test_migrations.py index c9a89db..fe3285e 100644 --- a/tests/integration/test_migrations.py +++ b/tests/integration/test_migrations.py @@ -7,6 +7,16 @@ import pytest from photo_pipeline.db import run_migrations + +def _alembic_head() -> str: + from alembic.config import Config as AlembicConfig + from alembic.script import ScriptDirectory + + repo = Path(__file__).resolve().parents[2] + cfg = AlembicConfig(str(repo / "alembic.ini")) + cfg.set_main_option("script_location", str(repo / "migrations")) + return ScriptDirectory.from_config(cfg).get_current_head() + # Faithful snapshot of the donor photo_analyzer.py schema (photos + FTS + triggers). LEGACY_SCHEMA = """ CREATE TABLE photos ( @@ -50,7 +60,7 @@ def test_migrations_from_empty_database(tmp_path): tables = _tables(db) assert {"assets", "asset_paths"} <= tables - assert _head_revision(db) == "0001_initial_identity" + assert _head_revision(db) == _alembic_head() def test_migrations_from_legacy_snapshot_preserve_existing_data(tmp_path): @@ -75,7 +85,7 @@ def test_migrations_from_legacy_snapshot_preserve_existing_data(tmp_path): finally: conn.close() assert row == ("album/one.jpg", "analyzed", "a red block") - assert _head_revision(db) == "0001_initial_identity" + assert _head_revision(db) == _alembic_head() def test_migrations_are_idempotent(tmp_path): @@ -83,7 +93,7 @@ def test_migrations_are_idempotent(tmp_path): url = f"sqlite:///{db}" run_migrations(url) run_migrations(url) # second run is a no-op at head - assert _head_revision(db) == "0001_initial_identity" + assert _head_revision(db) == _alembic_head() @pytest.mark.parametrize("expected", ["assets", "asset_paths"]) diff --git a/tests/unit/test_phash.py b/tests/unit/test_phash.py new file mode 100644 index 0000000..079b04d --- /dev/null +++ b/tests/unit/test_phash.py @@ -0,0 +1,52 @@ +"""Golden perceptual-hash relationships and confidence bands.""" + +import numpy as np +from PIL import Image + +from photo_pipeline.services import hashing +from photo_pipeline.services.duplicates import NEAR_MAX, SIMILAR_MAX + + +def _structured(path, seed, size=(256, 192)): + rng = np.random.default_rng(seed) + w, h = size + base = np.zeros((h, w, 3), dtype=np.uint8) + for _ in range(6): + x0 = int(rng.integers(0, w - 60)) + y0 = int(rng.integers(0, h - 60)) + base[y0 : y0 + 60, x0 : x0 + 60] = rng.integers(0, 256, 3) + grad = np.linspace(0, 120, w, dtype=np.uint8) + base[:, :, 0] = np.clip(base[:, :, 0].astype(int) + grad[None, :], 0, 255) + Image.fromarray(base).save(path, quality=95) + return path + + +def test_phash_is_16_hex_chars(tmp_path): + value = hashing.phash(_structured(tmp_path / "a.jpg", 1)) + assert len(value) == 16 + int(value, 16) # parses as hex + + +def test_identical_pixels_have_zero_distance(tmp_path): + a = _structured(tmp_path / "a.jpg", 1) + assert hashing.phash_distance(hashing.phash(a), hashing.phash(a)) == 0 + + +def test_resized_copy_stays_within_near_band(tmp_path): + a = _structured(tmp_path / "a.jpg", 1) + small = tmp_path / "a_small.jpg" + with Image.open(a) as image: + image.resize((image.width // 2, image.height // 2), Image.LANCZOS).save(small, quality=95) + distance = hashing.phash_distance(hashing.phash(a), hashing.phash(small)) + assert distance <= NEAR_MAX + + +def test_distinct_images_exceed_similar_band(tmp_path): + a = _structured(tmp_path / "a.jpg", 1) + b = _structured(tmp_path / "b.jpg", 2) + distance = hashing.phash_distance(hashing.phash(a), hashing.phash(b)) + assert distance > SIMILAR_MAX + + +def test_phash_is_versioned(): + assert hashing.PHASH_VERSION >= 1