auto-commit for b5490b7b-d57f-4f53-9a99-bd8ed84027db

backend/server.py

@@ -23,6 +23,7 @@ from models import (
     RequestCreate, RequestUpdate, MovieRequest,
     AppSettings, AppSettingsPublic,
     TMDBSearchResult, RadarrMovieDTO,
+    TranscodeJob, QueuePauseToggle,
 )
 from auth import hash_password, verify_password, create_token, decode_token
 from seed import SAMPLE_MOVIES
@@ -158,6 +159,16 @@ async def on_startup():
     except Exception as e:
         logger.warning(f"progress unique index skipped: {e}")
 
+    # Reset any stuck "running" jobs from a previous backend crash
+    await db.transcode_queue.update_many(
+        {"status": "running"},
+        {"$set": {"status": "failed", "error": "Backend restarted while running"}},
+    )
+    await db.transcode_queue.create_index([("status", 1), ("created_at", 1)])
+
+    # Start the background transcode worker
+    asyncio.create_task(_transcode_worker())
+
 
 @app.on_event("shutdown")
 async def on_shutdown():
@@ -394,6 +405,12 @@ async def upload_video(
     )
     doc = movie.model_dump()
     await db.movies.insert_one(doc)
+
+    # Auto-enqueue transcode if enabled
+    s = await get_settings()
+    auto = s.get("auto_transcode", "off")
+    if auto in ("quick", "abr"):
+        await _enqueue_transcode(doc["id"], auto, triggered_by="auto")
     return _strip(doc)
 
 
@@ -469,27 +486,107 @@ async def _set_hls_status(movie_id: str, status: str, path: Optional[str] = None
         logger.warning(f"HLS {movie_id}: {error}")
 
 
-async def _run_transcode(movie_id: str, source: Path, quality: str):
-    out_dir = HLS_DIR / movie_id
-    # Clear any prior output before re-encoding
+async def _process_job(job: dict):
+    """Run a transcode job. Updates both movie.hls_status and the job row."""
+    movie = await db.movies.find_one({"id": job["movie_id"]}, {"_id": 0})
+    if not movie:
+        await db.transcode_queue.update_one(
+            {"id": job["id"]},
+            {"$set": {"status": "failed", "error": "Movie not found", "finished_at": datetime.now(timezone.utc).isoformat()}},
+        )
+        return
+    if movie.get("storage_type") not in ("local", "radarr") or not movie.get("storage_path"):
+        await db.transcode_queue.update_one(
+            {"id": job["id"]},
+            {"$set": {"status": "failed", "error": "Not a local/radarr movie", "finished_at": datetime.now(timezone.utc).isoformat()}},
+        )
+        return
+
+    source = Path(movie["storage_path"]) if movie["storage_type"] == "radarr" else VIDEOS_DIR / movie["storage_path"]
+    out_dir = HLS_DIR / job["movie_id"]
     import shutil as _sh
     _sh.rmtree(out_dir, ignore_errors=True)
 
+    last_error = {"msg": ""}
+
     async def cb(status, entry: Optional[str] = None, error: Optional[str] = None):
         if status == "done" and entry:
-            await _set_hls_status(movie_id, "done", path=entry)
+            await _set_hls_status(job["movie_id"], "done", path=entry)
         else:
-            await _set_hls_status(movie_id, status, error=error)
+            await _set_hls_status(job["movie_id"], status, error=error)
+        if error:
+            last_error["msg"] = error
 
-    if quality == "abr":
-        await transcode_abr(source, out_dir, cb)
-    else:
-        await transcode_quick(source, out_dir, cb)
+    try:
+        if job["quality"] == "abr":
+            await transcode_abr(source, out_dir, cb)
+        else:
+            await transcode_quick(source, out_dir, cb)
+    except Exception as e:
+        logger.exception("transcode crashed")
+        last_error["msg"] = str(e)
+        await _set_hls_status(job["movie_id"], "failed", error=str(e))
+
+    movie_after = await db.movies.find_one({"id": job["movie_id"]}, {"_id": 0, "hls_status": 1})
+    final_status = "done" if movie_after and movie_after.get("hls_status") == "done" else "failed"
+    await db.transcode_queue.update_one(
+        {"id": job["id"]},
+        {"$set": {
+            "status": final_status,
+            "error": last_error["msg"],
+            "finished_at": datetime.now(timezone.utc).isoformat(),
+        }},
+    )
+
+
+async def _transcode_worker():
+    """Single FIFO worker. Polls for pending jobs and runs them serially."""
+    logger.info("Transcode worker started")
+    while True:
+        try:
+            settings = await get_settings()
+            if settings.get("queue_paused", False):
+                await asyncio.sleep(15)
+                continue
+            job = await db.transcode_queue.find_one_and_update(
+                {"status": "pending"},
+                {"$set": {"status": "running", "started_at": datetime.now(timezone.utc).isoformat()}},
+                sort=[("created_at", 1)],
+                projection={"_id": 0},
+                return_document=True,
+            )
+            if not job:
+                await asyncio.sleep(8)
+                continue
+            await _set_hls_status(job["movie_id"], "running")
+            await _process_job(job)
+        except Exception:
+            logger.exception("Transcode worker error")
+            await asyncio.sleep(10)
+
+
+async def _enqueue_transcode(movie_id: str, quality: str, triggered_by: str = "manual") -> dict:
+    """Add a job to the queue. Returns the job document."""
+    movie = await db.movies.find_one({"id": movie_id}, {"_id": 0, "title": 1})
+    title = movie.get("title", "") if movie else ""
+    # Skip if a non-finished job for this movie already exists
+    existing = await db.transcode_queue.find_one(
+        {"movie_id": movie_id, "status": {"$in": ["pending", "running"]}},
+        {"_id": 0},
+    )
+    if existing:
+        return existing
+    job = TranscodeJob(
+        movie_id=movie_id, movie_title=title, quality=quality, triggered_by=triggered_by,
+    ).model_dump()
+    await db.transcode_queue.insert_one(job)
+    await _set_hls_status(movie_id, "pending")
+    return job
 
 
 @api.post("/movies/{movie_id}/transcode")
 async def trigger_transcode(movie_id: str, payload: Optional[dict] = None, user: dict = Depends(require_admin)):
-    """Body: {"quality": "quick"|"abr"} — default "quick"."""
+    """Body: {"quality": "quick"|"abr"} — default "quick". Adds to background queue."""
     quality = (payload or {}).get("quality", "quick")
     if quality not in ("quick", "abr"):
         raise HTTPException(status_code=400, detail="quality must be 'quick' or 'abr'")
@@ -498,11 +595,66 @@ async def trigger_transcode(movie_id: str, payload: Optional[dict] = None, user:
     if movie.get("storage_type") not in ("local", "radarr") or not movie.get("storage_path"):
         raise HTTPException(status_code=400, detail="Only local/radarr movies can be transcoded")
     if movie.get("hls_status") in ("running", "pending"):
-        raise HTTPException(status_code=409, detail="Already transcoding")
-    source = Path(movie["storage_path"]) if movie["storage_type"] == "radarr" else VIDEOS_DIR / movie["storage_path"]
-    await _set_hls_status(movie_id, "pending")
-    asyncio.create_task(_run_transcode(movie_id, source, quality))
-    return {"ok": True, "status": "pending", "quality": quality}
+        raise HTTPException(status_code=409, detail="Already transcoding or queued")
+    job = await _enqueue_transcode(movie_id, quality, triggered_by="manual")
+    return {"ok": True, "status": "pending", "quality": quality, "job_id": job["id"]}
+
+
+# ============ Transcode Queue endpoints ============
+@api.get("/transcode/queue", response_model=List[TranscodeJob])
+async def list_queue(status: Optional[str] = None, user: dict = Depends(require_admin)):
+    q: dict = {}
+    if status:
+        q["status"] = {"$in": status.split(",")}
+    rows = await db.transcode_queue.find(q, {"_id": 0}).sort("created_at", -1).to_list(500)
+    return rows
+
+
+@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}
+    async for row in db.transcode_queue.aggregate(pipeline):
+        out[row["_id"]] = row["n"]
+    s = await get_settings()
+    out["paused"] = bool(s.get("queue_paused", False))
+    out["auto_transcode"] = s.get("auto_transcode", "off")
+    return out
+
+
+@api.delete("/transcode/queue/{job_id}")
+async def cancel_job(job_id: str, user: dict = Depends(require_admin)):
+    job = await db.transcode_queue.find_one({"id": job_id}, {"_id": 0})
+    if not job: raise HTTPException(status_code=404, detail="Job not found")
+    if job["status"] == "running":
+        raise HTTPException(status_code=400, detail="Cannot cancel a running job")
+    await db.transcode_queue.update_one({"id": job_id}, {"$set": {"status": "cancelled", "finished_at": datetime.now(timezone.utc).isoformat()}})
+    if job["status"] == "pending":
+        # Reset movie hls_status so a new transcode can be triggered
+        await db.movies.update_one({"id": job["movie_id"]}, {"$set": {"hls_status": None}})
+    return {"ok": True}
+
+
+@api.post("/transcode/queue/{job_id}/retry")
+async def retry_job(job_id: str, user: dict = Depends(require_admin)):
+    job = await db.transcode_queue.find_one({"id": job_id}, {"_id": 0})
+    if not job: raise HTTPException(status_code=404, detail="Job not found")
+    if job["status"] not in ("failed", "cancelled"):
+        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"))
+    return {"ok": True, "job_id": new_job["id"]}
+
+
+@api.post("/transcode/queue/clear")
+async def clear_finished(user: dict = Depends(require_admin)):
+    res = await db.transcode_queue.delete_many({"status": {"$in": ["done", "failed", "cancelled"]}})
+    return {"ok": True, "deleted": res.deleted_count}
+
+
+@api.post("/transcode/queue/pause")
+async def pause_queue(payload: QueuePauseToggle, user: dict = Depends(require_admin)):
+    await db.settings.update_one({"id": "app"}, {"$set": {"queue_paused": payload.paused}}, upsert=True)
+    return {"ok": True, "paused": payload.paused}
 
 
 @api.get("/movies/{movie_id}/hls/{filename:path}")
@@ -706,6 +858,8 @@ async def read_settings(user: dict = Depends(require_admin)):
         tmdb_api_key=s.get("tmdb_api_key", ""),
         radarr_url=s.get("radarr_url", ""),
         radarr_api_key=s.get("radarr_api_key", ""),
+        auto_transcode=s.get("auto_transcode", "off"),
+        queue_paused=bool(s.get("queue_paused", False)),
     )
 
 
@@ -720,6 +874,8 @@ async def update_settings(payload: AppSettings, user: dict = Depends(require_adm
         tmdb_api_key=doc.get("tmdb_api_key", ""),
         radarr_url=doc.get("radarr_url", ""),
         radarr_api_key=doc.get("radarr_api_key", ""),
+        auto_transcode=doc.get("auto_transcode", "off"),
+        queue_paused=bool(doc.get("queue_paused", False)),
     )
 
 
@@ -780,10 +936,10 @@ async def radarr_import(payload: dict, user: dict = Depends(require_admin)):
     movies = await asyncio.to_thread(radarr_client.list_movies, s["radarr_url"], s["radarr_api_key"])
     by_id = {m["radarr_id"]: m for m in movies}
     created = 0
+    auto = s.get("auto_transcode", "off")
     for rid in radarr_ids:
         m = by_id.get(int(rid))
         if not m or not m.get("has_file") or not m.get("file_path"): continue
-        # Skip if already imported
         if await db.movies.find_one({"radarr_id": m["radarr_id"]}, {"_id": 1}): continue
         movie = Movie(
             title=m["title"], description=m.get("overview", ""),
@@ -792,10 +948,12 @@ async def radarr_import(payload: dict, user: dict = Depends(require_admin)):
             poster_url=m.get("poster_url", ""),
             backdrop_url=m.get("poster_url", ""),
             video_url="", storage_type="radarr",
-            storage_path=m["file_path"],  # absolute path on Radarr-mounted storage
+            storage_path=m["file_path"],
             radarr_id=m["radarr_id"], tmdb_id=m.get("tmdb_id"),
         ).model_dump()
         await db.movies.insert_one(movie)
+        if auto in ("quick", "abr"):
+            await _enqueue_transcode(movie["id"], auto, triggered_by="auto")
         created += 1
     return {"ok": True, "imported": created}