US01-04: Detect and Decide Duplicates (#49)

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",
+]

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(

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()
+    )

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"],
+            ),
+        }

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