Fix dashboard review speed: DB-only decisions (0.03s vs 11.6s); add process-staging flow for moves

process_staging.py

@@ -0,0 +1,85 @@
+"""photo-pipeline: process-staging flow.
+
+After review (dashboard = DB-only decisions), this flow materializes the
+decisions: moves approved files to /mnt/data/01_keep, rejected to 03_delete.
+
+Separating decisions (instant, in dashboard) from moves (here) keeps review
+fast — cross-filesystem moves are the slow part and happen in one efficient pass.
+"""
+
+import shutil
+from pathlib import Path
+
+from prefect import flow, task
+
+import photo_db as db
+
+STAGING = Path("/mnt/data")
+STATUS_TO_DIR = {
+    "approved": "01_keep",
+    "rejected": "03_delete",
+}
+
+
+@task
+def materialize_decisions(status: str = None) -> dict:
+    """Move files according to their DB status. Idempotent (skips already-moved)."""
+    db.init_db()
+    conn = db.get_db()
+    if status:
+        rows = conn.execute(
+            "SELECT sha256, path, status FROM image_hashes WHERE status=?", (status,)
+        ).fetchall()
+    else:
+        rows = conn.execute(
+            "SELECT sha256, path, status FROM image_hashes WHERE status IN ('approved','rejected')"
+        ).fetchall()
+    conn.close()
+
+    moved = 0
+    already = 0
+    errors = []
+    for sha, path, st in rows:
+        target_dir = STATUS_TO_DIR.get(st)
+        if not target_dir:
+            continue
+        src = Path(path)
+        dest_dir = STAGING / target_dir
+        dest_dir.mkdir(parents=True, exist_ok=True)
+        dest = dest_dir / src.name
+
+        # already in the right place?
+        if src.parent == dest_dir:
+            already += 1
+            continue
+        if not src.exists():
+            errors.append(f"{path}: missing source")
+            continue
+        # avoid collision: append suffix if dest exists
+        if dest.exists():
+            dest = dest_dir / f"{src.stem}_{sha[:8]}{src.suffix}"
+        try:
+            shutil.move(str(src), str(dest))
+            conn = db.get_db()
+            conn.execute(
+                "UPDATE image_hashes SET path=?, status=? WHERE sha256=?", (str(dest), st, sha)
+            )
+            conn.commit()
+            conn.close()
+            moved += 1
+        except Exception as e:
+            errors.append(f"{path}: {e}")
+
+    return {"moved": moved, "already": already, "errors": errors}
+
+
+@flow(name="process-staging")
+def process_staging(status: str = None):
+    """Materialize review decisions: move approved/rejected files to staging."""
+    result = materialize_decisions(status)
+    print(f"process-staging: {result}")
+    return result
+
+
+if __name__ == "__main__":
+    process_staging()