From a15ee5728ddc007893c9338a88f9ba21f95f02eb Mon Sep 17 00:00:00 2001 From: Sam Rolfe Date: Sat, 8 Aug 2026 10:39:10 +1000 Subject: [PATCH] Fix dashboard review speed: DB-only decisions (0.03s vs 11.6s); add process-staging flow for moves --- dashboard/app.py | 86 ++++++++++----------------------- dashboard/templates/_card.html | 4 +- dashboard/templates/review.html | 66 +++++++++++++++++++------ process_staging.py | 85 ++++++++++++++++++++++++++++++++ 4 files changed, 163 insertions(+), 78 deletions(-) create mode 100644 process_staging.py diff --git a/dashboard/app.py b/dashboard/app.py index ee700a8..7e3343b 100644 --- a/dashboard/app.py +++ b/dashboard/app.py @@ -5,10 +5,13 @@ Reads photo_pipeline.db + staging dirs on .13. Serves: /review — thumbnail review grid with keep/reject/reset (htmx) /file/{sha} — full-size original image /thumbs/... — generated thumbnails - /stats — JSON stats (for future dashboard widgets) + /stats — JSON stats + +Review actions update the DB ONLY (instant). File moves happen at import time +(via the immich-import / process-staging flow) — this keeps review responsive +even for large batches; cross-filesystem moves are slow. """ -import shutil import sqlite3 from pathlib import Path @@ -82,7 +85,6 @@ def index(request: Request): ).fetchall() total = conn.execute("SELECT COUNT(*) FROM image_hashes").fetchone()[0] conn.close() - # folder targets (staging dirs) folders = {} for name in ("01_keep", "02_review", "03_delete"): d = STAGING / name @@ -104,6 +106,8 @@ def review(request: Request, source: str = None, status: str = None): if status: q += " AND status=?" params.append(status) + else: + q += " AND status IN ('scanned','review')" q += " ORDER BY added_at DESC LIMIT 300" rows = conn.execute(q, params).fetchall() conn.close() @@ -114,9 +118,8 @@ def review(request: Request, source: str = None, status: str = None): ) -@app.get("/stats", response_class=FileResponse) +@app.get("/stats") def stats(): - """JSON stats for widgets.""" import json conn = _conn() @@ -126,64 +129,43 @@ def stats(): by_source = dict(conn.execute( "SELECT source, COUNT(*) FROM image_hashes GROUP BY source").fetchall()) conn.close() - return FileResponse(path=None, content=json.dumps( - {"total": total, "by_status": by_status, "by_source": by_source})) + return json.dumps({"total": total, "by_status": by_status, "by_source": by_source}) -def _set_status(sha: str, status: str, move_to: str | None, request: Request): - """Update DB status AND (optionally) move the file into a staging folder.""" +def _set_status(sha: str, status: str, request: Request) -> HTMLResponse: + """Update DB status ONLY — instant. File moves happen at import time.""" conn = _conn() row = conn.execute("SELECT * FROM image_hashes WHERE sha256=?", (sha,)).fetchone() if not row: conn.close() raise HTTPException(404) - src = Path(row["path"]) - # move file if requested and source exists - if move_to and src.exists(): - dest_dir = STAGING / move_to - dest_dir.mkdir(parents=True, exist_ok=True) - dest = dest_dir / src.name - try: - shutil.move(str(src), str(dest)) - conn.execute( - "UPDATE image_hashes SET path=?, status=? WHERE sha256=?", - (str(dest), status, sha)) - except Exception as e: - conn.close() - raise HTTPException(500, detail=f"move failed: {e}") - else: - conn.execute("UPDATE image_hashes SET status=? WHERE sha256=?", (status, sha)) + conn.execute("UPDATE image_hashes SET status=? WHERE sha256=?", (status, sha)) conn.commit() + row = conn.execute("SELECT * FROM image_hashes WHERE sha256=?", (sha,)).fetchone() conn.close() return templates.TemplateResponse( - request, "_card.html", {"item": _load_item(row)}) + request, "_card.html", {"item": _load_item(row)} + ) @app.post("/review/{sha}/approve") def approve(sha: str, request: Request): - # approved → move to 01_keep (ready for Immich import) - return _set_status(sha, "approved", "01_keep", request) + return _set_status(sha, "approved", request) @app.post("/review/{sha}/reject") def reject(sha: str, request: Request): - # rejected → move to 03_delete (holding, never auto-deleted) - return _set_status(sha, "rejected", "03_delete", request) + return _set_status(sha, "rejected", request) @app.post("/review/{sha}/reset") def reset_status(sha: str, request: Request): - # undo: move back to a neutral status without moving the file - return _set_status(sha, "scanned", None, request) + return _set_status(sha, "scanned", request) @app.post("/review/bulk") async def bulk_action(request: Request): - """Bulk keep/reject/reset for a list of sha256 values. - - Body: JSON {"action": "approve"|"reject"|"reset", "shas": ["..."]} - Moves files accordingly; returns counts. - """ + """Bulk status update — DB only, instant. Body: {"action", "shas": []}""" import json body = json.loads(await request.body()) @@ -191,39 +173,21 @@ async def bulk_action(request: Request): shas = body.get("shas", []) if action not in ("approve", "reject", "reset"): raise HTTPException(400, "action must be approve/reject/reset") - - move_map = {"approve": "01_keep", "reject": "03_delete", "reset": None} status_map = {"approve": "approved", "reject": "rejected", "reset": "scanned"} - move_to = move_map[action] status = status_map[action] conn = _conn() - moved = 0 + updated = 0 missing = 0 for sha in shas: - row = conn.execute("SELECT path FROM image_hashes WHERE sha256=?", (sha,)).fetchone() - if not row: + cur = conn.execute("UPDATE image_hashes SET status=? WHERE sha256=?", (status, sha)) + if cur.rowcount: + updated += 1 + else: missing += 1 - continue - src = Path(row["path"]) - if move_to and src.exists(): - dest_dir = STAGING / move_to - dest_dir.mkdir(parents=True, exist_ok=True) - dest = dest_dir / src.name - try: - shutil.move(str(src), str(dest)) - conn.execute( - "UPDATE image_hashes SET path=?, status=? WHERE sha256=?", - (str(dest), status, sha)) - moved += 1 - continue - except Exception: - pass - conn.execute("UPDATE image_hashes SET status=? WHERE sha256=?", (status, sha)) - moved += 1 conn.commit() conn.close() - return {"action": action, "moved": moved, "missing": missing, "total": len(shas)} + return {"action": action, "updated": updated, "missing": missing, "total": len(shas)} @app.get("/file/{sha}") diff --git a/dashboard/templates/_card.html b/dashboard/templates/_card.html index c2d9558..386b8d7 100644 --- a/dashboard/templates/_card.html +++ b/dashboard/templates/_card.html @@ -17,8 +17,8 @@ {% if item["size_mb"] %} · {{ item["size_mb"] }}MB{% endif %}
- - + +
diff --git a/dashboard/templates/review.html b/dashboard/templates/review.html index e493eab..619151a 100644 --- a/dashboard/templates/review.html +++ b/dashboard/templates/review.html @@ -40,6 +40,7 @@ .reset { background: #444; color: #eee; } .none { color: #666; font-style: italic; padding: 2rem; text-align: center; } .selall { display: inline-flex; align-items: center; gap: .35rem; background: #222; border: 1px solid #444; border-radius: 6px; padding: .4rem .8rem; cursor: pointer; } + .toast { position: fixed; bottom: 1.2rem; right: 1.2rem; background: #1d4; color: #031; padding: .7rem 1rem; border-radius: 8px; display: none; z-index: 50; font-size: .9rem; box-shadow: 0 2px 12px rgba(0,0,0,.5); } @@ -70,6 +71,7 @@ +

{{ items|length }} images

@@ -120,22 +122,56 @@ async function bulk(action) { if (selected.size === 0) return; const shas = [...selected]; - const resp = await fetch('/review/bulk', { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ action, shas }) - }); - const data = await resp.json(); - alert(`Bulk ${action}: ${data.moved} processed, ${data.missing} missing`); - // refresh: remove done cards - shas.forEach(s => { - const card = document.querySelector(`[data-name="${s}"]`)?.closest('.card'); - if (card) card.remove(); - }); - selected.clear(); - refresh(); - location.reload(); + // disable buttons during the operation (cross-filesystem moves can be slow) + setBulkBusy(true); + try { + const resp = await fetch('/review/bulk', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ action, shas }) + }); + const data = await resp.json(); + // remove processed cards from DOM immediately (match by sha in value attr) + shas.forEach(s => { + const cb = document.querySelector(`.item-check[value="${s}"]`); + if (cb) cb.closest('.card')?.remove(); + }); + selected.clear(); + refresh(); + updateCount(); + showToast(`Bulk ${action}: ${data.moved} done, ${data.missing} missing`); + } catch (e) { + showToast('Bulk action failed: ' + e, true); + } finally { + setBulkBusy(false); + } } + + function setBulkBusy(busy) { + ['bulk-keep','bulk-reject','bulk-reset'].forEach(id => { + const b = document.getElementById(id); + if (b) { b.disabled = busy; b.textContent = busy ? 'Working…' : b.dataset.orig; } + }); + } + + function updateCount() { + const el = document.querySelector('.count'); + const n = document.querySelectorAll('.card').length; + if (el) el.textContent = n + ' images'; + } + + function showToast(msg, isErr) { + const t = document.getElementById('toast'); + t.textContent = msg; + t.style.background = isErr ? '#d43' : '#1d4'; + t.style.color = isErr ? '#fff' : '#031'; + t.style.display = 'block'; + setTimeout(() => { t.style.display = 'none'; }, 3000); + } + + // stash original labels for busy-state restore + document.querySelectorAll('.bulk-bar button').forEach(b => { b.dataset.orig = b.textContent; }); + document.getElementById('bulk-keep').addEventListener('click', () => bulk('approve')); document.getElementById('bulk-reject').addEventListener('click', () => bulk('reject')); document.getElementById('bulk-reset').addEventListener('click', () => bulk('reset')); diff --git a/process_staging.py b/process_staging.py new file mode 100644 index 0000000..62db881 --- /dev/null +++ b/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()