Dashboard: pipeline status view (Prefect flow runs + recent review actions, 30s auto-refresh); reviewed_at column

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','?')}")