diff --git a/dashboard/app.py b/dashboard/app.py
index 7e3343b..5d6de60 100644
--- a/dashboard/app.py
+++ b/dashboard/app.py
@@ -132,6 +132,22 @@ def stats():
return json.dumps({"total": total, "by_status": by_status, "by_source": by_source})
+@app.get("/pipeline", response_class=HTMLResponse)
+def pipeline(request: Request):
+ """Pipeline status: recent Prefect flow runs + recent review actions."""
+ import sys
+ if str(BASE.parent) not in sys.path:
+ sys.path.insert(0, str(BASE.parent))
+ import status_helper
+
+ runs = status_helper.get_flow_runs(limit=15)
+ actions = status_helper.get_recent_actions(limit=20)
+ return templates.TemplateResponse(
+ request, "pipeline.html",
+ {"runs": runs, "actions": actions},
+ )
+
+
def _set_status(sha: str, status: str, request: Request) -> HTMLResponse:
"""Update DB status ONLY — instant. File moves happen at import time."""
conn = _conn()
@@ -139,7 +155,9 @@ def _set_status(sha: str, status: str, request: Request) -> HTMLResponse:
if not row:
conn.close()
raise HTTPException(404)
- conn.execute("UPDATE image_hashes SET status=? WHERE sha256=?", (status, sha))
+ conn.execute(
+ "UPDATE image_hashes SET status=?, reviewed_at=datetime('now') WHERE sha256=?",
+ (status, sha))
conn.commit()
row = conn.execute("SELECT * FROM image_hashes WHERE sha256=?", (sha,)).fetchone()
conn.close()
@@ -180,7 +198,9 @@ async def bulk_action(request: Request):
updated = 0
missing = 0
for sha in shas:
- cur = conn.execute("UPDATE image_hashes SET status=? WHERE sha256=?", (status, sha))
+ cur = conn.execute(
+ "UPDATE image_hashes SET status=?, reviewed_at=datetime('now') WHERE sha256=?",
+ (status, sha))
if cur.rowcount:
updated += 1
else:
diff --git a/dashboard/templates/pipeline.html b/dashboard/templates/pipeline.html
new file mode 100644
index 0000000..0bce04c
--- /dev/null
+++ b/dashboard/templates/pipeline.html
@@ -0,0 +1,73 @@
+
+
+
+
+
+ Pipeline — photo-pipeline
+
+
+
+
+
+
+ Recent flow runs (Prefect)
+
+ | Flow | State | Run | Started | Duration |
|---|
+ {% for r in runs %}
+
+ | {{ r["flow"] }} |
+ {{ r["state"] }} |
+ {{ r["name"] }} |
+ {{ r["start"] }} |
+ {{ r["duration"] }} |
+
+ {% else %}
+ | No flow runs yet |
+ {% endfor %}
+
+
+ Recent review actions
+
+ | File | Decision | Source | When |
|---|
+ {% for a in actions %}
+
+ | {{ a["name"] }} |
+ {{ a["status"] }} |
+ {{ a["source"] }} |
+ {{ a["when"] }} |
+
+ {% else %}
+ | No review actions yet — decisions you make appear here |
+ {% endfor %}
+
+ Auto-refreshes every 30s.
+
+
+
diff --git a/status_helper.py b/status_helper.py
new file mode 100644
index 0000000..ae313c4
--- /dev/null
+++ b/status_helper.py
@@ -0,0 +1,105 @@
+"""photo-pipeline: Prefect status helper for the dashboard.
+
+Queries the Prefect API (via the client) for recent flow runs and their state,
+plus recent review actions from the fingerprint DB.
+"""
+
+import asyncio
+import os
+import sqlite3
+from datetime import datetime, timezone
+from pathlib import Path
+
+os.environ.setdefault("PREFECT_API_URL", "http://localhost:4200/api")
+
+BASE = Path(__file__).parent
+DB_PATH = BASE / "photo_pipeline.db"
+
+
+def _fmt_time(dt):
+ if not dt:
+ return "—"
+ try:
+ return dt.astimezone().strftime("%Y-%m-%d %H:%M")
+ except Exception:
+ return str(dt)[:16]
+
+
+def get_flow_runs(limit: int = 15) -> list[dict]:
+ """Recent flow runs from Prefect, newest first."""
+ from prefect.client.orchestration import get_client
+
+ async def _fetch():
+ async with get_client() as client:
+ runs = await client.read_flow_runs(limit=limit)
+ # fetch flow names
+ flow_names = {}
+ for r in runs:
+ if r.flow_id not in flow_names:
+ try:
+ f = await client.read_flow(r.flow_id)
+ flow_names[r.flow_id] = f.name
+ except Exception:
+ flow_names[r.flow_id] = "?"
+ out = []
+ for r in runs:
+ out.append({
+ "name": r.name or "?",
+ "flow": flow_names.get(r.flow_id, "?"),
+ "state": (r.state_type or "?").replace("StateType.", ""),
+ "state_name": r.state_name or "",
+ "start": _fmt_time(r.start_time),
+ "end": _fmt_time(r.end_time),
+ "duration": _duration(r.start_time, r.end_time),
+ })
+ return out
+
+ try:
+ return asyncio.run(_fetch())
+ except Exception as e:
+ return [{"error": str(e)}]
+
+
+def _duration(start, end):
+ if not start or not end:
+ return ""
+ try:
+ s = (end - start).total_seconds()
+ if s < 60:
+ return f"{s:.0f}s"
+ if s < 3600:
+ return f"{s/60:.1f}m"
+ return f"{s/3600:.1f}h"
+ except Exception:
+ return ""
+
+
+def get_recent_actions(limit: int = 20) -> list[dict]:
+ """Recent review decisions from the DB (by reviewed_at timestamp)."""
+ try:
+ conn = sqlite3.connect(DB_PATH)
+ conn.row_factory = sqlite3.Row
+ rows = conn.execute(
+ "SELECT sha256, path, status, source, reviewed_at FROM image_hashes "
+ "WHERE reviewed_at IS NOT NULL "
+ "ORDER BY reviewed_at DESC LIMIT ?", (limit,)
+ ).fetchall()
+ conn.close()
+ return [{
+ "name": Path(r["path"]).name,
+ "status": r["status"],
+ "source": r["source"],
+ "when": r["reviewed_at"] or "—",
+ } for r in rows]
+ except Exception as e:
+ return [{"error": str(e)}]
+
+
+if __name__ == "__main__":
+ import json
+ print("=== flow runs ===")
+ for r in get_flow_runs(6):
+ print(f" {r.get('flow','?'):25s} {r.get('state','?'):12s} {r.get('start','')}")
+ print("=== recent actions ===")
+ for a in get_recent_actions(6):
+ print(f" {a.get('status','?'):10s} {a.get('name','?')}")