US01-06: Review Inventory and Duplicates in the Browser (#51)

photo_pipeline/services/duplicates.py

@@ -30,7 +30,7 @@ from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from enum import Enum
 
-from sqlalchemy import select
+from sqlalchemy import func, select
 from sqlalchemy.orm import sessionmaker
 
 from photo_pipeline.models import (
@@ -466,6 +466,62 @@ class DuplicateService:
                 asset = session.get(Asset, cur)
                 cur = asset.canonical_asset_id if asset else None
 
+    # ── reads for the review UI ────────────────────────────────────────────────
+    def list_clusters(
+        self, *, state: str | None = None, limit: int = 50, offset: int = 0
+    ) -> dict:
+        limit = max(1, min(limit, 200))
+        offset = max(0, offset)
+        with self._session_factory() as session:
+            stmt = select(DuplicateCluster)
+            if state:
+                stmt = stmt.where(DuplicateCluster.state == state)
+            total = session.scalar(select(func.count()).select_from(stmt.subquery()))
+            rows = session.execute(
+                stmt.order_by(DuplicateCluster.created_at).limit(limit).offset(offset)
+            ).scalars()
+            items = [self._snapshot(session, c.id) for c in rows]
+        return {"items": items, "total": int(total or 0), "limit": limit, "offset": offset}
+
+    def get_cluster(self, cluster_id: str) -> dict | None:
+        """Cluster detail enriched with per-member asset evidence for comparison."""
+        with self._session_factory() as session:
+            cluster = session.get(DuplicateCluster, cluster_id)
+            if cluster is None:
+                return None
+            members = []
+            for member in session.execute(
+                select(DuplicateMember).where(DuplicateMember.cluster_id == cluster_id)
+            ).scalars():
+                asset = session.get(Asset, member.asset_id)
+                try:
+                    evidence = json.loads(member.evidence) if member.evidence else {}
+                except json.JSONDecodeError:
+                    evidence = {}
+                members.append(
+                    {
+                        "asset_id": member.asset_id,
+                        "role": member.role,
+                        "distance": member.distance,
+                        "evidence": evidence,
+                        "current_path": asset.current_path if asset else None,
+                        "byte_size": asset.byte_size if asset else None,
+                        "phash": asset.phash if asset else None,
+                    }
+                )
+            members.sort(key=lambda m: m["asset_id"])
+            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,
+                "requires_confirmation": cluster.method == Method.PERCEPTUAL.value,
+                "members": members,
+            }
+
     # ── decisions ────────────────────────────────────────────────────────────
     def decide(
         self,

photo_pipeline/services/inventory.py

@@ -30,7 +30,7 @@ from enum import Enum
 from pathlib import Path
 from typing import Iterable
 
-from sqlalchemy import select
+from sqlalchemy import func, select
 from sqlalchemy.orm import Session, sessionmaker
 
 from photo_pipeline import path_policy
@@ -55,6 +55,21 @@ class ScanResult:
     asset_ids: dict[str, str] = field(default_factory=dict)  # path -> asset id
 
 
+def _asset_dict(asset: Asset) -> dict:
+    return {
+        "id": asset.id,
+        "current_path": asset.current_path,
+        "availability_state": asset.availability_state,
+        "byte_size": asset.byte_size,
+        "current_sha256": asset.current_sha256,
+        "pixel_sha256": asset.pixel_sha256,
+        "phash": asset.phash,
+        "canonical_asset_id": asset.canonical_asset_id,
+        "missing": asset.missing_at is not None,
+        "state_version": asset.state_version,
+    }
+
+
 class InventoryService:
     def __init__(self, session_factory: sessionmaker) -> None:
         self._session_factory = session_factory
@@ -101,6 +116,30 @@ class InventoryService:
         result.counts = dict(Counter(result.occurrences.values()))
         return result
 
+    def list_assets(
+        self,
+        *,
+        limit: int = 50,
+        offset: int = 0,
+        availability: str | None = None,
+        query: str | None = None,
+    ) -> dict:
+        """Paged, filterable asset listing for the inventory UI (read-only)."""
+        limit = max(1, min(limit, 200))
+        offset = max(0, offset)
+        with self._session_factory() as session:
+            stmt = select(Asset)
+            if availability:
+                stmt = stmt.where(Asset.availability_state == availability)
+            if query:
+                stmt = stmt.where(Asset.current_path.like(f"%{query}%"))
+            total = session.scalar(select(func.count()).select_from(stmt.subquery()))
+            rows = session.execute(
+                stmt.order_by(Asset.current_path).limit(limit).offset(offset)
+            ).scalars()
+            items = [_asset_dict(a) for a in rows]
+        return {"items": items, "total": int(total or 0), "limit": limit, "offset": offset}
+
     def _reconcile_one(
         self,
         session: Session,