web: persist song input as a sidecar and reload jobs on startup

2026-07-28 02:38:55 -07:00
parent 7b8fce5b6e
commit c8c3d941ca
2 changed files with 99 additions and 7 deletions
@@ -1,3 +1,4 @@
+import json
 import math
 import queue
 import re
@@ -5,7 +6,7 @@ import shutil
 import threading
 import time
 import uuid
-from dataclasses import dataclass, field
+from dataclasses import asdict, dataclass, field
 from pathlib import Path

 from fastapi import FastAPI, File, Form, HTTPException, UploadFile
@@ -15,8 +16,8 @@ from fastapi.staticfiles import StaticFiles

 from . import rebuild as rebuild_policy
 from .config import Config
-from .pipeline import (STAGE_ORDER, PipelineError, SongInput, run_pipeline,
-                       song_slug)
+from .pipeline import (STAGE_ORDER, PipelineError, SongInput, read_warnings,
+                       run_pipeline, song_slug)

 STATIC = Path(__file__).parent / "static"
 STAGE_LINE = re.compile(r"^\[(\w+)\] (?:running|cached, skipping)")
@@ -92,6 +93,34 @@ def _cleanup_uploads(cfg: Config, job_id: str, video_path: Path | None) -> None:
            pass


+def _discover_jobs(cfg: Config) -> list[Job]:
+    discovered = []
+    if not cfg.work_dir.exists():
+        return discovered
+    for sidecar in cfg.work_dir.glob("*/input.json"):
+        try:
+            data = json.loads(sidecar.read_text())
+            for key in ("video_path", "cover_path"):
+                if data.get(key) is not None:
+                    data[key] = Path(data[key])
+            inp = SongInput(**data)
+        except (OSError, ValueError, TypeError):
+            continue
+        slug = song_slug(inp)
+        out = cfg.output_dir / slug
+        done = out.exists() and any(out.iterdir())
+        mtime = sidecar.stat().st_mtime
+        discovered.append(Job(
+            id=uuid.uuid4().hex, title=inp.title, artist=inp.artist, inp=inp,
+            status="done" if done else "failed",
+            output_dir=str(out) if done else None,
+            warnings=read_warnings(sidecar.parent) if done else [],
+            error=None if done else "interrupted before completion",
+            created_at=mtime, finished_at=mtime))
+    discovered.sort(key=lambda j: j.created_at)
+    return discovered
+
+
 def create_app(cfg: Config, runner=run_pipeline) -> FastAPI:
    app = FastAPI()
    app.mount("/static", StaticFiles(directory=STATIC), name="static")
@@ -100,6 +129,10 @@ def create_app(cfg: Config, runner=run_pipeline) -> FastAPI:
    lock = threading.Lock()
    todo: queue.Queue[str] = queue.Queue()

+    for job in _discover_jobs(cfg):
+        jobs[job.id] = job
+        order.append(job.id)
+
    def _prepare_rebuild(job: Job, log) -> str | None:
        """Relocate cached work and report the stage this rebuild restarts from."""
        if job.force_stage:
@@ -132,6 +165,15 @@ def create_app(cfg: Config, runner=run_pipeline) -> FastAPI:
        shutil.rmtree(old, ignore_errors=True)
        log(f"[rebuild] removed superseded {old.name}")

+    def _write_sidecar(job: Job) -> None:
+        work = cfg.work_dir / song_slug(job.inp)
+        work.mkdir(parents=True, exist_ok=True)
+        data = asdict(job.inp)
+        for key in ("video_path", "cover_path"):
+            if data[key] is not None:
+                data[key] = str(data[key])
+        (work / "input.json").write_text(json.dumps(data))
+
    def worker():
        while True:
            job_id = todo.get()
@@ -148,6 +190,7 @@ def create_app(cfg: Config, runner=run_pipeline) -> FastAPI:

            try:
                force_from = _prepare_rebuild(job, log)
+                _write_sidecar(job)
                result = runner(job.inp, cfg, log=log, force_from=force_from)
                with lock:
                    job.output_dir = str(result.output_dir)
@@ -1,5 +1,6 @@
 import io
 import re
+import shutil
 import threading
 import time

@@ -776,7 +777,7 @@ def test_rebuild_renames_the_work_dir_on_a_retitle(tmp_path):
    seen = []
    client, cfg = make_client(tmp_path, capture_runner_factory(seen))
    parent = make_done_job(client)
-    (cfg.work_dir / "A - T").mkdir(parents=True)
+    (cfg.work_dir / "A - T").mkdir(parents=True, exist_ok=True)
    (cfg.work_dir / "A - T" / "audio.wav").write_bytes(b"x")
    child = client.post(f"/api/jobs/{parent}/rebuild",
                        data=rebuild_form(title="T2")).json()["id"]
@@ -789,7 +790,7 @@ def test_rebuild_renames_the_work_dir_on_a_retitle(tmp_path):
 def test_rebuild_fails_the_job_on_a_work_dir_collision(tmp_path):
    client, cfg = make_client(tmp_path, ok_runner)
    parent = make_done_job(client)
-    (cfg.work_dir / "A - T").mkdir(parents=True)
+    (cfg.work_dir / "A - T").mkdir(parents=True, exist_ok=True)
    (cfg.work_dir / "A - T2").mkdir(parents=True)
    child = client.post(f"/api/jobs/{parent}/rebuild",
                        data=rebuild_form(title="T2")).json()["id"]
@@ -817,7 +818,7 @@ def test_rebuild_inherits_the_parents_source(tmp_path):
                        files={"video": ("v.mp4", io.BytesIO(b"data"), "video/mp4")})
    parent = resp.json()["id"]
    wait_status(client, parent, "done")
-    (cfg.work_dir / "A - T").mkdir(parents=True)
+    (cfg.work_dir / "A - T").mkdir(parents=True, exist_ok=True)
    for name in ("meta.json", "audio.wav"):
        (cfg.work_dir / "A - T" / name).write_bytes(b"x")
    child = client.post(f"/api/jobs/{parent}/rebuild",
@@ -1007,7 +1008,7 @@ def test_rebuild_of_a_failed_parent_removes_nothing(tmp_path):
 def test_delete_removes_work_and_output_and_the_job(tmp_path):
    client, cfg = make_client(tmp_path, slug_runner)
    job_id = make_done_job(client)
-    (cfg.work_dir / "A - T").mkdir(parents=True)
+    (cfg.work_dir / "A - T").mkdir(parents=True, exist_ok=True)
    (cfg.work_dir / "A - T" / "audio.wav").write_bytes(b"x")
    assert (cfg.output_dir / "A - T").exists()

@@ -1057,3 +1058,51 @@ def test_a_plain_job_never_removes_anything(tmp_path):
    stray.mkdir(parents=True)
    make_done_job(client, title="T2")
    assert stray.exists()
+
+
+def test_discover_jobs_ignores_a_work_dir_without_a_sidecar(tmp_path):
+    from songtrace.web import _discover_jobs
+    cfg = Config(whisper_url="u", work_dir=tmp_path / "work", output_dir=tmp_path / "out")
+    (cfg.work_dir / "A - T").mkdir(parents=True)
+    assert _discover_jobs(cfg) == []
+
+
+def test_discover_jobs_reconstructs_a_completed_song(tmp_path):
+    from songtrace.web import _discover_jobs
+    cfg = Config(whisper_url="u", work_dir=tmp_path / "work", output_dir=tmp_path / "out")
+    client = TestClient(create_app(cfg, runner=slug_runner))
+    make_done_job(client)
+
+    discovered = _discover_jobs(cfg)
+
+    assert len(discovered) == 1
+    job = discovered[0]
+    assert job.title == "T" and job.artist == "A" and job.status == "done"
+    assert job.output_dir == str(cfg.output_dir / "A - T")
+    assert job.error is None
+
+
+def test_discover_jobs_reconstructs_a_song_whose_output_is_missing_as_failed(tmp_path):
+    from songtrace.web import _discover_jobs
+    cfg = Config(whisper_url="u", work_dir=tmp_path / "work", output_dir=tmp_path / "out")
+    client = TestClient(create_app(cfg, runner=slug_runner))
+    make_done_job(client)
+    shutil.rmtree(cfg.output_dir / "A - T")
+
+    discovered = _discover_jobs(cfg)
+
+    assert len(discovered) == 1
+    assert discovered[0].status == "failed"
+    assert discovered[0].error == "interrupted before completion"
+    assert discovered[0].output_dir is None
+
+
+def test_discovered_jobs_are_visible_from_a_fresh_app(tmp_path):
+    cfg = Config(whisper_url="u", work_dir=tmp_path / "work", output_dir=tmp_path / "out")
+    client1 = TestClient(create_app(cfg, runner=slug_runner))
+    make_done_job(client1)
+
+    client2 = TestClient(create_app(cfg, runner=slug_runner))
+    jobs = client2.get("/api/jobs").json()
+    assert len(jobs) == 1
+    assert jobs[0]["title"] == "T" and jobs[0]["status"] == "done"