From 71e731124a402d2d7b89692e6432a54248998d77 Mon Sep 17 00:00:00 2001 From: Myron Blair Date: Mon, 27 Jul 2026 17:38:53 -0500 Subject: [PATCH] Add live transcode progress with a dedicated now-processing panel - ffmpeg progress streamed via -progress pipe:1, persisted on the job as a percent - retry/retry-all now mark the old row as superseded instead of deleting it, preserving the failed/cancelled audit trail - frontend shows the running job in its own panel above the list with a live progress bar instead of burying it as just another row --- backend/models.py | 2 + backend/server.py | 12 +++-- backend/transcode.py | 69 ++++++++++++++++++++++++--- frontend/src/pages/TranscodeQueue.jsx | 49 ++++++++++++++----- 4 files changed, 109 insertions(+), 23 deletions(-) diff --git a/backend/models.py b/backend/models.py index 39d8fe2..479125d 100644 --- a/backend/models.py +++ b/backend/models.py @@ -366,6 +366,8 @@ class TranscodeJob(BaseModel): status: str = "pending" error: str = "" triggered_by: str = "manual" + progress: float = 0.0 + superseded_by: Optional[str] = None created_at: str = Field(default_factory=_now_iso) started_at: Optional[str] = None finished_at: Optional[str] = None diff --git a/backend/server.py b/backend/server.py index 2dca616..6f54ccd 100644 --- a/backend/server.py +++ b/backend/server.py @@ -583,11 +583,13 @@ async def _process_job(job: dict): last_error = {"msg": ""} - async def cb(status, entry: Optional[str] = None, error: Optional[str] = None): + async def cb(status, entry: Optional[str] = None, error: Optional[str] = None, progress: Optional[float] = None): update = {"hls_status": status} if status == "done" and entry: update["hls_path"] = entry await coll.update_one({"id": job["movie_id"]}, {"$set": update}) if error: last_error["msg"] = error + if progress is not None: + await db.transcode_queue.update_one({"id": job["id"]}, {"$set": {"progress": progress}}) try: if job["quality"] == "abr": @@ -684,7 +686,7 @@ async def list_queue(status: Optional[str] = None, user: dict = Depends(require_ @api.get("/transcode/queue/stats") async def queue_stats(user: dict = Depends(require_admin)): pipeline = [{"$group": {"_id": "$status", "n": {"$sum": 1}}}] - out = {"pending": 0, "running": 0, "done": 0, "failed": 0, "cancelled": 0} + out = {"pending": 0, "running": 0, "done": 0, "failed": 0, "cancelled": 0, "superseded": 0} async for row in db.transcode_queue.aggregate(pipeline): out[row["_id"]] = row["n"] s = await get_settings() @@ -715,7 +717,9 @@ async def retry_job(job_id: str, user: dict = Depends(require_admin)): raise HTTPException(status_code=400, detail="Only failed/cancelled jobs can be retried") new_job = await _enqueue_transcode(job["movie_id"], job["quality"], triggered_by=job.get("triggered_by", "manual"), content_type=job.get("content_type", "movie")) if new_job["id"] != job["id"]: - await db.transcode_queue.delete_one({"id": job["id"]}) + # Keep the original row for the audit trail — mark it superseded rather than deleting, + # so cancelled/failed history never silently disappears from the list. + await db.transcode_queue.update_one({"id": job["id"]}, {"$set": {"status": "superseded", "superseded_by": new_job["id"]}}) return {"ok": True, "job_id": new_job["id"]} @@ -726,7 +730,7 @@ async def retry_all_failed(user: dict = Depends(require_admin)): for job in jobs: new_job = await _enqueue_transcode(job["movie_id"], job["quality"], triggered_by=job.get("triggered_by", "manual"), content_type=job.get("content_type", "movie")) if new_job["id"] != job["id"]: - await db.transcode_queue.delete_one({"id": job["id"]}) + await db.transcode_queue.update_one({"id": job["id"]}, {"$set": {"status": "superseded", "superseded_by": new_job["id"]}}) retried += 1 return {"ok": True, "retried": retried} diff --git a/backend/transcode.py b/backend/transcode.py index 3734344..efb2e71 100644 --- a/backend/transcode.py +++ b/backend/transcode.py @@ -37,6 +37,23 @@ async def probe_video(source: Path) -> Tuple[int, int]: return (0, 0) +async def probe_duration(source: Path) -> float: + """Return source duration in seconds using ffprobe; 0 on failure.""" + try: + proc = await asyncio.create_subprocess_exec( + "ffprobe", "-v", "error", "-show_entries", "format=duration", + "-of", "json", str(source), + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + ) + stdout, _ = await proc.communicate() + if proc.returncode != 0: + return 0.0 + data = json.loads(stdout.decode("utf-8", errors="ignore") or "{}") + return float((data.get("format") or {}).get("duration") or 0.0) + except Exception: + return 0.0 + + # (height, video_bitrate, max_bitrate, buffer, audio_bitrate) ALL_VARIANTS = [ (1080, "5000k", "5500k", "7500k", "192k"), @@ -75,7 +92,7 @@ async def transcode_audio_fix(source: Path, out_dir: Path, on_status: Callable[. "-hls_segment_filename", str(out_dir / "seg_%04d.ts"), str(out_dir / "playlist.m3u8"), ] - await _run(cmd, on_status, out_dir, "playlist.m3u8") + await _run(cmd, on_status, out_dir, "playlist.m3u8", source) async def transcode_quick(source: Path, out_dir: Path, on_status: Callable[..., Awaitable]) -> None: @@ -94,7 +111,7 @@ async def transcode_quick(source: Path, out_dir: Path, on_status: Callable[..., "-hls_segment_filename", str(out_dir / "seg_%04d.ts"), str(out_dir / "playlist.m3u8"), ] - await _run(cmd, on_status, out_dir, "playlist.m3u8") + await _run(cmd, on_status, out_dir, "playlist.m3u8", source) async def transcode_abr(source: Path, out_dir: Path, on_status: Callable[..., Awaitable]) -> None: @@ -158,23 +175,61 @@ async def transcode_abr(source: Path, out_dir: Path, on_status: Callable[..., Aw for i in range(n): (out_dir / f"v{i}").mkdir(exist_ok=True) - await _run(cmd, on_status, out_dir, "master.m3u8") + await _run(cmd, on_status, out_dir, "master.m3u8", source) -async def _run(cmd: List[str], on_status, out_dir: Path, entry_filename: str) -> None: - await on_status("running") +def _parse_out_time_seconds(line: str) -> Optional[float]: + """Parse an `out_time=HH:MM:SS.microseconds` line from ffmpeg's -progress output.""" + _, _, value = line.partition("=") + value = value.strip() + if not value or value == "N/A": + return None + try: + h, m, s = value.split(":") + return int(h) * 3600 + int(m) * 60 + float(s) + except ValueError: + return None + + +async def _run(cmd: List[str], on_status, out_dir: Path, entry_filename: str, source: Optional[Path] = None) -> None: + duration = await probe_duration(source) if source else 0.0 + await on_status("running", progress=0.0) + # Machine-readable progress on stdout, separate from ffmpeg's normal stderr logging. + cmd = [cmd[0], "-progress", "pipe:1", "-nostats"] + cmd[1:] try: proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) - _stdout, stderr = await proc.communicate() + + stderr_chunks: List[bytes] = [] + + async def _drain_stderr(): + async for line in proc.stderr: + stderr_chunks.append(line) + + stderr_task = asyncio.create_task(_drain_stderr()) + + last_reported = -1.0 + async for raw in proc.stdout: + line = raw.decode("utf-8", errors="ignore").strip() + if line.startswith("out_time=") and duration > 0: + seconds = _parse_out_time_seconds(line) + if seconds is not None: + pct = round(min(99.0, seconds / duration * 100), 1) + if pct != last_reported: + last_reported = pct + await on_status("running", progress=pct) + + await proc.wait() + await stderr_task + stderr = b"".join(stderr_chunks) if proc.returncode != 0: err = stderr.decode("utf-8", errors="ignore")[-500:] logger.error(f"ffmpeg failed: {err}") await on_status("failed", error=err[:200]) shutil.rmtree(out_dir, ignore_errors=True) return - await on_status("done", entry=entry_filename) + await on_status("done", entry=entry_filename, progress=100.0) except FileNotFoundError: await on_status("failed", error="ffmpeg not installed") except Exception as e: diff --git a/frontend/src/pages/TranscodeQueue.jsx b/frontend/src/pages/TranscodeQueue.jsx index edb25c9..efb9357 100644 --- a/frontend/src/pages/TranscodeQueue.jsx +++ b/frontend/src/pages/TranscodeQueue.jsx @@ -9,6 +9,7 @@ const STATUS_COLORS = { done: "text-[#86efac]", failed: "text-[#fca5a5]", cancelled: "text-[#8A8A8A]", + superseded: "text-[#8A8A8A]", }; export default function TranscodeQueue() { @@ -26,14 +27,19 @@ export default function TranscodeQueue() { }; useEffect(() => { load(); }, []); - // Poll while jobs are active + // Poll while jobs are active — faster while something is actually running so the + // progress bar feels live, slower while only pending jobs are waiting. useEffect(() => { - const active = jobs.some((j) => j.status === "running" || j.status === "pending"); - if (!active) return; - const t = setInterval(load, 4000); + const running = jobs.some((j) => j.status === "running"); + const pending = jobs.some((j) => j.status === "pending"); + if (!running && !pending) return; + const t = setInterval(load, running ? 2000 : 4000); return () => clearInterval(t); }, [jobs]); + const runningJob = jobs.find((j) => j.status === "running"); + const listedJobs = jobs.filter((j) => j.status !== "running"); + const cancel = async (id) => { try { await api.delete(`/transcode/queue/${id}`); toast.success("Job cancelled"); load(); } catch (err) { toast.error(err.response?.data?.detail || "Could not cancel"); } @@ -116,7 +122,31 @@ export default function TranscodeQueue() { -
+ {runningJob && ( +
+
+ + + Now Processing + + {(runningJob.progress || 0).toFixed(1)}% +
+
+ {runningJob.movie_title || runningJob.movie_id.slice(0, 8)} +
+
+ {runningJob.quality} · {runningJob.triggered_by} +
+
+
+
+
+ )} + +
Movie Quality @@ -125,9 +155,9 @@ export default function TranscodeQueue() { Created Actions
- {jobs.length === 0 ? ( + {listedJobs.length === 0 ? (
No jobs yet.
- ) : jobs.map((j) => ( + ) : listedJobs.map((j) => (
{j.movie_title || j.movie_id.slice(0, 8)}
{j.quality} @@ -138,11 +168,6 @@ export default function TranscodeQueue() { {new Date(j.created_at).toLocaleString()}
- {j.status === "running" && ( - - )} {j.status === "pending" && (