diff --git a/backend/VERSION b/backend/VERSION index 3eefcb9..1cc5f65 100644 --- a/backend/VERSION +++ b/backend/VERSION @@ -1 +1 @@ -1.0.0 +1.1.0 \ No newline at end of file diff --git a/backend/backfill_transcode.py b/backend/backfill_transcode.py new file mode 100644 index 0000000..f388b84 --- /dev/null +++ b/backend/backfill_transcode.py @@ -0,0 +1,53 @@ +import asyncio +import os +import uuid +from datetime import datetime, timezone +from pathlib import Path + +from motor.motor_asyncio import AsyncIOMotorClient +from transcode import classify_source + + +async def main(): + client = AsyncIOMotorClient(os.environ["MONGO_URL"]) + db = client[os.environ.get("DB_NAME", "kino")] + videos_dir = Path(os.environ.get("MEDIA_ROOT", "/media")) / "videos" + + cancel_res = await db.transcode_queue.update_many( + {"status": "pending"}, + {"$set": {"status": "cancelled", "finished_at": datetime.now(timezone.utc).isoformat()}}, + ) + print(f"cancelled stale pending jobs: {cancel_res.modified_count}") + + stats = {"native": 0, "audio": 0, "abr": 0, "no_source": 0} + for coll_name, content_type in (("movies", "movie"), ("episodes", "episode")): + coll = db[coll_name] + async for doc in coll.find({"hls_status": {"$ne": "done"}}, {"_id": 0}): + storage_type = doc.get("storage_type") + storage_path = doc.get("storage_path") + if storage_type not in ("local", "radarr", "sonarr", "scan") or not storage_path: + stats["no_source"] += 1 + continue + source = Path(storage_path) if storage_type in ("radarr", "sonarr", "scan") else videos_dir / storage_path + + classification = await classify_source(source) + if classification == "native": + await coll.update_one({"id": doc["id"]}, {"$set": {"hls_status": "not_needed"}}) + stats["native"] += 1 + continue + + quality = "abr" if classification == "abr" else "audio" + stats[quality] += 1 + job = { + "id": str(uuid.uuid4()), "movie_id": doc["id"], "movie_title": doc.get("title", ""), + "content_type": content_type, "quality": quality, "status": "pending", "error": "", + "triggered_by": "backfill", "progress": 0.0, "superseded_by": None, + "created_at": datetime.now(timezone.utc).isoformat(), "started_at": None, "finished_at": None, + } + await db.transcode_queue.insert_one(job) + await coll.update_one({"id": doc["id"]}, {"$set": {"hls_status": "pending"}}) + + print(stats) + + +asyncio.run(main()) diff --git a/backend/models.py b/backend/models.py index 9c85978..4385445 100644 --- a/backend/models.py +++ b/backend/models.py @@ -193,6 +193,8 @@ class RequestUpdate(BaseModel): class RequestApprove(BaseModel): content_type: str # "movie" | "tv" + tmdb_id: Optional[int] = None # movie: explicit pick from /requests/{id}/candidates + tvdb_id: Optional[int] = None # tv: explicit pick from /requests/{id}/candidates class MovieRequest(BaseModel): @@ -380,6 +382,26 @@ class TranscodeJob(BaseModel): finished_at: Optional[str] = None +class CodecCheckJob(BaseModel): + """Step 1 of the pipeline, ahead of TranscodeJob (step 2): probes the source file's actual + video/audio codecs and decides whether it needs a transcode at all. 'native' sources skip + straight to the library with no transcode job ever created; 'audio'/'abr' sources get + handed off to a TranscodeJob.""" + model_config = ConfigDict(extra="ignore") + id: str = Field(default_factory=lambda: str(uuid.uuid4())) + movie_id: str # generic content id (movie_id OR episode_id, depending on content_type) + movie_title: str = "" + content_type: str = "movie" # movie | episode + status: str = "pending" # pending | running | done | failed | cancelled | superseded + classification: Optional[str] = None # native | audio | abr, set once done + error: str = "" + triggered_by: str = "manual" + superseded_by: Optional[str] = None + created_at: str = Field(default_factory=_now_iso) + started_at: Optional[str] = None + finished_at: Optional[str] = None + + class QueuePauseToggle(BaseModel): paused: bool diff --git a/backend/radarr.py b/backend/radarr.py index 928839e..ff015fe 100644 --- a/backend/radarr.py +++ b/backend/radarr.py @@ -54,7 +54,10 @@ def search_movie(base_url: str, api_key: str, term: str) -> List[Dict]: def add_movie(base_url: str, api_key: str, tmdb_id: int, title: str, year: Optional[int]) -> Dict: """Adds a movie by tmdb_id with monitoring + an immediate search enabled, using the first - configured quality profile and root folder — same defaults Radarr's own UI would pre-fill.""" + configured quality profile and root folder — same defaults Radarr's own UI would pre-fill. + If the movie is already in Radarr's library, Radarr rejects the add with a 400 + MovieExistsValidator — in that case, reuse the existing movie and just kick off a search + instead of surfacing the add as a failure.""" base_url = base_url.rstrip("/") h = _h(api_key) profiles = requests.get(f"{base_url}/api/v3/qualityprofile", headers=h, timeout=15).json() @@ -67,6 +70,14 @@ def add_movie(base_url: str, api_key: str, tmdb_id: int, title: str, year: Optio "monitored": True, "addOptions": {"searchForMovie": True}, } r = requests.post(f"{base_url}/api/v3/movie", json=payload, headers=h, timeout=20) + if r.status_code == 400 and any(e.get("errorCode") == "MovieExistsValidator" for e in (r.json() or [])): + existing = requests.get(f"{base_url}/api/v3/movie", params={"tmdbId": tmdb_id}, headers=h, timeout=15).json() + if not existing: + raise RuntimeError("Radarr reported movie already exists but it could not be found") + movie = existing[0] + requests.put(f"{base_url}/api/v3/movie/{movie['id']}", json={**movie, "monitored": True}, headers=h, timeout=15) + requests.post(f"{base_url}/api/v3/command", json={"name": "MoviesSearch", "movieIds": [movie["id"]]}, headers=h, timeout=15) + return {"radarr_id": movie["id"], "title": movie.get("title"), "year": movie.get("year")} r.raise_for_status() added = r.json() return {"radarr_id": added["id"], "title": added.get("title"), "year": added.get("year")} diff --git a/backend/server.py b/backend/server.py index 7a69c5d..e92ea6a 100644 --- a/backend/server.py +++ b/backend/server.py @@ -24,7 +24,7 @@ from models import ( RequestCreate, RequestUpdate, RequestApprove, MovieRequest, AppSettings, AppSettingsPublic, TMDBSearchResult, RadarrMovieDTO, - TranscodeJob, QueuePauseToggle, + TranscodeJob, CodecCheckJob, QueuePauseToggle, Show, ShowCreate, ShowUpdate, Episode, EpisodeCreate, EpisodeProgressUpsert, SonarrSeriesDTO, TraktDeviceCodeResponse, TraktStatus, OpenSubsResult, Track, TrackUpdate, @@ -42,7 +42,7 @@ import itunes as itunes_client import acoustid as acoustid_client import trakt as trakt_client import opensubtitles as opensubs_client -from transcode import transcode_quick, transcode_abr, transcode_audio_fix, srt_to_vtt +from transcode import transcode_quick, transcode_abr, transcode_audio_fix, srt_to_vtt, classify_source ROOT_DIR = Path(__file__).parent @@ -185,8 +185,14 @@ async def on_startup(): {"$set": {"status": "failed", "error": "Backend restarted while running"}}, ) await db.transcode_queue.create_index([("status", 1), ("created_at", 1)]) + await db.codec_check_queue.update_many( + {"status": "running"}, + {"$set": {"status": "failed", "error": "Backend restarted while running"}}, + ) + await db.codec_check_queue.create_index([("status", 1), ("created_at", 1)]) - # Start the background transcode worker + # Start the background workers: codec check (step 1) feeds the transcode queue (step 2) + asyncio.create_task(_codec_check_worker()) asyncio.create_task(_transcode_worker()) @@ -477,11 +483,11 @@ async def upload_video( doc = movie.model_dump() await db.movies.insert_one(doc) - # Auto-enqueue transcode if enabled + # Auto-enqueue codec check (step 1) 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") + await _enqueue_codec_check(doc["id"], content_type="movie", triggered_by="auto") return _strip(doc) @@ -613,6 +619,93 @@ async def _process_job(job: dict): ) +def _source_path_for(doc: dict) -> Optional[Path]: + """Resolve a movie/episode doc's on-disk source path, or None if it has no local file.""" + storage_type = doc.get("storage_type") + storage_path = doc.get("storage_path") + if not storage_path: + return None + if storage_type in ("radarr", "sonarr", "scan"): + return Path(storage_path) + if storage_type == "local": + return VIDEOS_DIR / storage_path + return None + + +async def _enqueue_codec_check(content_id: str, content_type: str = "movie", triggered_by: str = "auto") -> dict: + """Step 1: queue a codec probe for newly-imported content. The worker below decides, + per file, whether it needs a transcode at all (see classify_source) — 'native' sources go + straight to the library, everything else gets handed to _enqueue_transcode (step 2).""" + coll = db.movies if content_type == "movie" else db.episodes + doc = await coll.find_one({"id": content_id}, {"_id": 0, "title": 1}) + title = doc.get("title", "") if doc else "" + existing = await db.codec_check_queue.find_one( + {"movie_id": content_id, "status": {"$in": ["pending", "running"]}}, {"_id": 0}, + ) + if existing: + return existing + job = CodecCheckJob( + movie_id=content_id, movie_title=title, content_type=content_type, triggered_by=triggered_by, + ).model_dump() + await db.codec_check_queue.insert_one(job) + return job + + +async def _process_codec_check(job: dict): + content_type = job.get("content_type", "movie") + coll = db.movies if content_type == "movie" else db.episodes + doc = await coll.find_one({"id": job["movie_id"]}, {"_id": 0}) + if not doc: + await db.codec_check_queue.update_one( + {"id": job["id"]}, + {"$set": {"status": "failed", "error": f"{content_type} not found", "finished_at": datetime.now(timezone.utc).isoformat()}}, + ) + return + source = _source_path_for(doc) + if source is None: + await db.codec_check_queue.update_one( + {"id": job["id"]}, + {"$set": {"status": "failed", "error": "Not a local file", "finished_at": datetime.now(timezone.utc).isoformat()}}, + ) + return + + classification = await classify_source(source) + await db.codec_check_queue.update_one( + {"id": job["id"]}, + {"$set": {"status": "done", "classification": classification, "finished_at": datetime.now(timezone.utc).isoformat()}}, + ) + if classification == "native": + await coll.update_one({"id": job["movie_id"]}, {"$set": {"hls_status": "not_needed"}}) + else: + quality = "abr" if classification == "abr" else "audio" + await _enqueue_transcode(job["movie_id"], quality, triggered_by="codec_check", content_type=content_type) + + +async def _codec_check_worker(): + """Single FIFO worker for step 1, mirroring the transcode worker below.""" + logger.info("Codec check worker started") + while True: + try: + settings = await get_settings() + if settings.get("codec_check_paused", False): + await asyncio.sleep(15) + continue + job = await db.codec_check_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 _process_codec_check(job) + except Exception: + logger.exception("Codec check worker error") + await asyncio.sleep(10) + + async def _transcode_worker(): """Single FIFO worker. Polls for pending jobs and runs them serially.""" logger.info("Transcode worker started") @@ -640,9 +733,15 @@ async def _transcode_worker(): async def _enqueue_transcode(content_id: str, quality: str, triggered_by: str = "manual", content_type: str = "movie") -> dict: - """Add a job to the queue. Returns the job document. Works for movies + episodes.""" + """Add a job to the queue. Returns the job document. Works for movies + episodes. + Before queuing, probes the actual source codec and lets that override the requested + quality: already-browser-native sources (H.264 + AAC/MP3) skip transcoding entirely + (hls_status becomes "not_needed" and playback falls back to the raw-file stream endpoint, + which already handles this case), and non-H.264 sources are forced to "abr" regardless of + what was requested — "quick"/"audio" apply a bitstream filter that assumes H.264 input and + crashes ffmpeg outright otherwise (this is what caused the recent fix-audio-all failures).""" coll = db.movies if content_type == "movie" else db.episodes - doc = await coll.find_one({"id": content_id}, {"_id": 0, "title": 1}) + doc = await coll.find_one({"id": content_id}, {"_id": 0}) title = doc.get("title", "") if doc else "" existing = await db.transcode_queue.find_one( {"movie_id": content_id, "status": {"$in": ["pending", "running"]}}, @@ -650,6 +749,24 @@ async def _enqueue_transcode(content_id: str, quality: str, triggered_by: str = ) if existing: return existing + + source = None + if doc and doc.get("storage_path"): + if doc.get("storage_type") in ("radarr", "sonarr", "scan"): + source = Path(doc["storage_path"]) + elif doc.get("storage_type") == "local": + source = VIDEOS_DIR / doc["storage_path"] + + if source is not None: + classification = await classify_source(source) + if classification == "native": + await coll.update_one({"id": content_id}, {"$set": {"hls_status": "not_needed"}}) + return {"skipped": True, "reason": "native", "movie_id": content_id, "movie_title": title} + if classification == "abr": + quality = "abr" + elif classification == "audio" and quality == "quick": + quality = "audio" + job = TranscodeJob( movie_id=content_id, movie_title=title, quality=quality, triggered_by=triggered_by, content_type=content_type, @@ -672,9 +789,83 @@ async def trigger_transcode(movie_id: str, payload: Optional[dict] = None, user: if movie.get("hls_status") in ("running", "pending"): raise HTTPException(status_code=409, detail="Already transcoding or queued") job = await _enqueue_transcode(movie_id, quality, triggered_by="manual") + if job.get("skipped"): + return {"ok": True, "status": "not_needed", "quality": quality, "job_id": None} return {"ok": True, "status": "pending", "quality": quality, "job_id": job["id"]} +# ============ Codec Check Queue endpoints (step 1, ahead of the Transcode Queue) ============ +@api.get("/codec-check/queue", response_model=List[CodecCheckJob]) +async def list_codec_check_queue(status: Optional[str] = None, user: dict = Depends(require_admin)): + q: dict = {} + if status: + q["status"] = {"$in": status.split(",")} + rows = await db.codec_check_queue.find(q, {"_id": 0}).sort("created_at", -1).to_list(500) + return rows + + +@api.get("/codec-check/queue/running", response_model=Optional[CodecCheckJob]) +async def get_running_codec_check_job(user: dict = Depends(require_admin)): + return await db.codec_check_queue.find_one({"status": "running"}, {"_id": 0}) + + +@api.get("/codec-check/queue/stats") +async def codec_check_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, "superseded": 0} + async for row in db.codec_check_queue.aggregate(pipeline): + out[row["_id"]] = row["n"] + s = await get_settings() + out["paused"] = bool(s.get("codec_check_paused", False)) + return out + + +@api.delete("/codec-check/queue/{job_id}") +async def cancel_codec_check_job(job_id: str, user: dict = Depends(require_admin)): + job = await db.codec_check_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.codec_check_queue.update_one({"id": job_id}, {"$set": {"status": "cancelled", "finished_at": datetime.now(timezone.utc).isoformat()}}) + return {"ok": True} + + +@api.post("/codec-check/queue/{job_id}/retry") +async def retry_codec_check_job(job_id: str, user: dict = Depends(require_admin)): + job = await db.codec_check_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_codec_check(job["movie_id"], content_type=job.get("content_type", "movie"), triggered_by=job.get("triggered_by", "manual")) + if new_job["id"] != job["id"]: + await db.codec_check_queue.update_one({"id": job["id"]}, {"$set": {"status": "superseded", "superseded_by": new_job["id"]}}) + return {"ok": True, "job_id": new_job["id"]} + + +@api.post("/codec-check/queue/retry-failed") +async def retry_all_failed_codec_check(user: dict = Depends(require_admin)): + jobs = await db.codec_check_queue.find({"status": {"$in": ["failed", "cancelled"]}}, {"_id": 0}).to_list(None) + retried = 0 + for job in jobs: + new_job = await _enqueue_codec_check(job["movie_id"], content_type=job.get("content_type", "movie"), triggered_by=job.get("triggered_by", "manual")) + if new_job["id"] != job["id"]: + await db.codec_check_queue.update_one({"id": job["id"]}, {"$set": {"status": "superseded", "superseded_by": new_job["id"]}}) + retried += 1 + return {"ok": True, "retried": retried} + + +@api.post("/codec-check/queue/clear") +async def clear_finished_codec_check(user: dict = Depends(require_admin)): + res = await db.codec_check_queue.delete_many({"status": {"$in": ["done", "failed", "cancelled", "superseded"]}}) + return {"ok": True, "deleted": res.deleted_count} + + +@api.post("/codec-check/queue/pause") +async def pause_codec_check_queue(payload: QueuePauseToggle, user: dict = Depends(require_admin)): + await db.settings.update_one({"id": "app"}, {"$set": {"codec_check_paused": payload.paused}}, upsert=True) + return {"ok": True, "paused": payload.paused} + + # ============ Transcode Queue endpoints ============ @api.get("/transcode/queue", response_model=List[TranscodeJob]) async def list_queue(status: Optional[str] = None, user: dict = Depends(require_admin)): @@ -726,6 +917,10 @@ async def retry_job(job_id: str, user: dict = Depends(require_admin)): 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"), content_type=job.get("content_type", "movie")) + if new_job.get("skipped"): + # Turned out to be browser-native on re-check — nothing to queue, just close out the old job. + await db.transcode_queue.update_one({"id": job["id"]}, {"$set": {"status": "superseded", "superseded_by": None}}) + return {"ok": True, "job_id": None, "status": "not_needed"} if new_job["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. @@ -739,7 +934,9 @@ async def retry_all_failed(user: dict = Depends(require_admin)): retried = 0 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"]: + if new_job.get("skipped"): + await db.transcode_queue.update_one({"id": job["id"]}, {"$set": {"status": "superseded", "superseded_by": None}}) + elif new_job["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} @@ -982,11 +1179,39 @@ async def update_request(request_id: str, payload: RequestUpdate, user: dict = D return res +@api.get("/requests/{request_id}/candidates") +async def request_candidates(request_id: str, content_type: str, user: dict = Depends(require_admin)): + """Returns Radarr/Sonarr's raw title-search results so an admin can see every match + (title, year, poster) and pick the right one before anything gets added/grabbed — + instead of approve blindly taking whatever came back first.""" + if content_type not in ("movie", "tv"): + raise HTTPException(status_code=400, detail="content_type must be movie or tv") + req = await db.requests.find_one({"id": request_id}, {"_id": 0}) + if not req: raise HTTPException(status_code=404, detail="Request not found") + s = await get_settings() + try: + if content_type == "movie": + if not s.get("radarr_url") or not s.get("radarr_api_key"): + raise HTTPException(status_code=400, detail="Radarr not configured") + return await asyncio.to_thread(radarr_client.search_movie, s["radarr_url"], s["radarr_api_key"], req["title"]) + else: + if not s.get("sonarr_url") or not s.get("sonarr_api_key"): + raise HTTPException(status_code=400, detail="Sonarr not configured") + return await asyncio.to_thread(sonarr_client.search_series, s["sonarr_url"], s["sonarr_api_key"], req["title"]) + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=502, detail=f"Lookup error: {e}") + + @api.post("/requests/{request_id}/approve", response_model=MovieRequest) async def approve_request(request_id: str, payload: RequestApprove, user: dict = Depends(require_admin)): - """Looks up the request's title in Radarr/Sonarr, adds the best match with an immediate - search enabled, and marks the request approved. Content actually lands via the normal - Radarr/Sonarr download pipeline — this only kicks that off.""" + """Looks up the request's title in Radarr/Sonarr and adds a match with an immediate + search enabled, then marks the request approved. Content actually lands via the normal + Radarr/Sonarr download pipeline — this only kicks that off. + If payload.tmdb_id/tvdb_id is set (admin picked a specific result from + GET /requests/{id}/candidates), that exact candidate is used. Otherwise falls back to + the old best-effort match (year match, else first search result) for backward compat.""" if payload.content_type not in ("movie", "tv"): raise HTTPException(status_code=400, detail="content_type must be movie or tv") # Atomically claim the request so two concurrent approve calls (double-click, two admins) @@ -1009,7 +1234,12 @@ async def approve_request(request_id: str, payload: RequestApprove, user: dict = candidates = await asyncio.to_thread(radarr_client.search_movie, s["radarr_url"], s["radarr_api_key"], req["title"]) except Exception as e: raise HTTPException(status_code=502, detail=f"Radarr lookup error: {e}") - match = next((c for c in candidates if req.get("year") and c.get("year") == req["year"]), candidates[0] if candidates else None) + if payload.tmdb_id is not None: + match = next((c for c in candidates if c.get("tmdb_id") == payload.tmdb_id), None) + if not match: + raise HTTPException(status_code=404, detail="Selected match no longer in Radarr's search results") + else: + match = next((c for c in candidates if req.get("year") and c.get("year") == req["year"]), candidates[0] if candidates else None) if not match: raise HTTPException(status_code=404, detail="No match found in Radarr") try: @@ -1025,7 +1255,12 @@ async def approve_request(request_id: str, payload: RequestApprove, user: dict = candidates = await asyncio.to_thread(sonarr_client.search_series, s["sonarr_url"], s["sonarr_api_key"], req["title"]) except Exception as e: raise HTTPException(status_code=502, detail=f"Sonarr lookup error: {e}") - match = next((c for c in candidates if req.get("year") and c.get("year") == req["year"]), candidates[0] if candidates else None) + if payload.tvdb_id is not None: + match = next((c for c in candidates if c.get("tvdb_id") == payload.tvdb_id), None) + if not match: + raise HTTPException(status_code=404, detail="Selected match no longer in Sonarr's search results") + else: + match = next((c for c in candidates if req.get("year") and c.get("year") == req["year"]), candidates[0] if candidates else None) if not match: raise HTTPException(status_code=404, detail="No match found in Sonarr") try: @@ -1190,7 +1425,7 @@ async def radarr_import(payload: dict, user: dict = Depends(require_admin)): ).model_dump() await db.movies.insert_one(movie) if auto in ("quick", "abr"): - await _enqueue_transcode(movie["id"], auto, triggered_by="auto") + await _enqueue_codec_check(movie["id"], content_type="movie", triggered_by="auto") # Auto-fetch English subtitle (fire and forget) asyncio.create_task(_auto_fetch_subtitle("movie", movie["id"], movie.get("tmdb_id"))) created += 1 @@ -1323,6 +1558,8 @@ async def transcode_episode(episode_id: str, payload: Optional[dict] = None, use if ep.get("hls_status") in ("running", "pending"): raise HTTPException(status_code=409, detail="Already transcoding or queued") job = await _enqueue_transcode(episode_id, quality, triggered_by="manual", content_type="episode") + if job.get("skipped"): + return {"ok": True, "status": "not_needed", "job_id": None} return {"ok": True, "job_id": job["id"]} @@ -1477,7 +1714,7 @@ async def sonarr_import(payload: dict, user: dict = Depends(require_admin)): ).model_dump() await db.episodes.insert_one(edoc); created_eps += 1 if auto in ("quick", "abr"): - await _enqueue_transcode(edoc["id"], auto, triggered_by="auto", content_type="episode") + await _enqueue_codec_check(edoc["id"], content_type="episode", triggered_by="auto") # Auto-fetch English subtitle for this episode (needs show's tmdb_id + season/episode numbers) show_doc = await db.shows.find_one({"id": show_id}, {"_id": 0, "tmdb_id": 1}) if show_doc and show_doc.get("tmdb_id"): @@ -1544,7 +1781,7 @@ async def scan_import(payload: dict, user: dict = Depends(require_admin)): ).model_dump() await db.movies.insert_one(movie) if auto in ("quick", "abr"): - await _enqueue_transcode(movie["id"], auto, triggered_by="auto") + await _enqueue_codec_check(movie["id"], content_type="movie", triggered_by="auto") created += 1 elif kind == "tv": @@ -1570,7 +1807,7 @@ async def scan_import(payload: dict, user: dict = Depends(require_admin)): ).model_dump() await db.episodes.insert_one(edoc) if auto in ("quick", "abr"): - await _enqueue_transcode(edoc["id"], auto, triggered_by="auto", content_type="episode") + await _enqueue_codec_check(edoc["id"], content_type="episode", triggered_by="auto") created += 1 for show_id in set(shows_by_title.values()): seasons = await db.episodes.distinct("season_number", {"show_id": show_id}) @@ -2067,13 +2304,46 @@ async def _auto_fetch_subtitle(content_type: str, content_id: str, tmdb_id: Opti # ============ SERVICES (admin) ============ +WORKER_SERVICES = [ + # (key, label, queue collection, stall threshold in seconds) + ("codec_check_worker", "Codec Check Worker", "codec_check_queue", 600), # ffprobe is fast — 10min stuck means hung + ("transcode_worker", "Transcode Worker", "transcode_queue", 4 * 60 * 60), # abr encodes can legitimately run hours +] + + +async def _worker_service_statuses() -> List[dict]: + out = [] + for key, label, coll_name, stall_seconds in WORKER_SERVICES: + running_job = await db[coll_name].find_one({"status": "running"}, {"_id": 0, "started_at": 1}) + stalled = False + if running_job and running_job.get("started_at"): + try: + started = datetime.fromisoformat(running_job["started_at"]) + if (datetime.now(timezone.utc) - started).total_seconds() > stall_seconds: + stalled = True + except ValueError: + pass + out.append({"key": key, "label": label, "running": not stalled}) + return out + + @api.get("/services") async def services_status(user: dict = Depends(require_admin)): - return await asyncio.to_thread(services_client.list_service_status) + docker_services = await asyncio.to_thread(services_client.list_service_status) + return docker_services + await _worker_service_statuses() @api.post("/services/{key}/restart") async def services_restart(key: str, user: dict = Depends(require_admin)): + if key in ("codec_check_worker", "transcode_worker"): + # Both workers live in the backend process — restarting one restarts the whole + # container (and briefly the API with it). The stuck "running" job gets marked + # failed automatically by the startup crash-recovery logic, so it won't just hang + # again immediately; retry it from the relevant queue page afterward. + ok = await asyncio.to_thread(services_client.restart_backend) + if not ok: + raise HTTPException(status_code=500, detail="Could not restart backend") + return {"ok": True} ok = await asyncio.to_thread(services_client.restart_service, key) if not ok: raise HTTPException(status_code=404, detail="Unknown service") diff --git a/backend/services.py b/backend/services.py index 5972ebd..a163b1e 100644 --- a/backend/services.py +++ b/backend/services.py @@ -44,3 +44,18 @@ def restart_service(key: str) -> bool: c = client.containers.get(svc["container"]) c.restart(timeout=15) return True + + +# The codec-check and transcode workers are asyncio background tasks inside the backend +# process itself, not separate containers — there's no way to bounce just one of them, so +# "restart" for either really means restarting the whole streamhoard-backend container. That +# briefly takes the API down too, not just the stalled queue; the caller (server.py) surfaces +# both worker "services" as needing this same restart target. +BACKEND_CONTAINER = "streamhoard-backend" + + +def restart_backend() -> bool: + client = _get_client() + c = client.containers.get(BACKEND_CONTAINER) + c.restart(timeout=15) + return True diff --git a/backend/sonarr.py b/backend/sonarr.py index 2e96ec0..766e8af 100644 --- a/backend/sonarr.py +++ b/backend/sonarr.py @@ -28,7 +28,11 @@ def search_series(base_url: str, api_key: str, term: str) -> List[Dict]: def add_series(base_url: str, api_key: str, tvdb_id: int, title: str, year) -> Dict: - """Adds a series by tvdb_id with monitoring + an immediate missing-episode search enabled.""" + """Adds a series by tvdb_id with monitoring + an immediate missing-episode search enabled. + If the series is already in Sonarr's library (common for older/library titles carried over + from a past migration), Sonarr rejects the add with a 400 SeriesExistsValidator — in that + case, reuse the existing series and just kick off a missing-episode search instead of + surfacing the add as a failure.""" base_url = base_url.rstrip("/") h = _h(api_key) profiles = requests.get(f"{base_url}/api/v3/qualityprofile", headers=h, timeout=15).json() @@ -42,6 +46,16 @@ def add_series(base_url: str, api_key: str, tvdb_id: int, title: str, year) -> D "addOptions": {"searchForMissingEpisodes": True, "monitor": "all"}, } r = requests.post(f"{base_url}/api/v3/series", json=payload, headers=h, timeout=20) + if r.status_code == 400 and any(e.get("errorCode") == "SeriesExistsValidator" for e in (r.json() or [])): + existing = requests.get(f"{base_url}/api/v3/series", params={"tvdbId": tvdb_id}, headers=h, timeout=15).json() + if not existing: + raise RuntimeError("Sonarr reported series already exists but it could not be found") + series = existing[0] + requests.post(f"{base_url}/api/v3/series/editor", json={ + "seriesIds": [series["id"]], "monitored": True, + }, headers=h, timeout=15) + requests.post(f"{base_url}/api/v3/command", json={"name": "SeriesSearch", "seriesId": series["id"]}, headers=h, timeout=15) + return {"sonarr_id": series["id"], "title": series.get("title"), "year": series.get("year")} r.raise_for_status() added = r.json() return {"sonarr_id": added["id"], "title": added.get("title"), "year": added.get("year")} diff --git a/backend/transcode.py b/backend/transcode.py index 79a5668..7f2d65e 100644 --- a/backend/transcode.py +++ b/backend/transcode.py @@ -54,6 +54,53 @@ async def probe_duration(source: Path) -> float: return 0.0 +async def probe_codecs(source: Path) -> Tuple[str, str]: + """Return (video_codec, audio_codec), lowercased, via ffprobe; ('', '') on failure.""" + try: + proc = await asyncio.create_subprocess_exec( + "ffprobe", "-v", "error", "-show_entries", "stream=codec_type,codec_name", + "-of", "json", str(source), + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + ) + stdout, _ = await proc.communicate() + if proc.returncode != 0: + return ("", "") + data = json.loads(stdout.decode("utf-8", errors="ignore") or "{}") + video_codec = audio_codec = "" + for s in data.get("streams") or []: + if s.get("codec_type") == "video" and not video_codec: + video_codec = (s.get("codec_name") or "").lower() + elif s.get("codec_type") == "audio" and not audio_codec: + audio_codec = (s.get("codec_name") or "").lower() + return (video_codec, audio_codec) + except Exception: + return ("", "") + + +BROWSER_NATIVE_AUDIO = {"aac", "mp3"} + + +async def classify_source(source: Path) -> str: + """Decide the cheapest correct transcode path for a source file: + - 'native': H.264 video + AAC/MP3 audio — browsers already play this file as-is (the + raw-file stream endpoint serves it directly), so no transcode is needed at all. + - 'audio': H.264 video but non-browser audio (AC3/DTS/EAC3/etc) — cheap audio-only + re-encode, video stream-copied untouched. + - 'abr': anything else (HEVC/VP9/etc, or codec probing failed) — needs a real video + re-encode. Also the safe fallback on probe failure: it's the only mode that doesn't + assume the source is already H.264 — 'audio'/'quick' apply the h264_mp4toannexb + bitstream filter unconditionally, which makes ffmpeg hard-fail on non-H.264 input. + """ + if not source.is_file(): + return "abr" + video_codec, audio_codec = await probe_codecs(source) + if not video_codec or video_codec != "h264": + return "abr" + if audio_codec in BROWSER_NATIVE_AUDIO: + return "native" + return "audio" + + # (height, video_bitrate, max_bitrate, buffer, audio_bitrate) ALL_VARIANTS = [ (1080, "5000k", "5500k", "7500k", "192k"), diff --git a/frontend/src/App.js b/frontend/src/App.js index 9151c6c..7cd18f9 100644 --- a/frontend/src/App.js +++ b/frontend/src/App.js @@ -19,6 +19,7 @@ import AdminUpload from "./pages/AdminUpload"; import Settings from "./pages/Settings"; import RadarrImport from "./pages/RadarrImport"; import TranscodeQueue from "./pages/TranscodeQueue"; +import CodecCheckQueue from "./pages/CodecCheckQueue"; import Shows from "./pages/Shows"; import ShowDetail from "./pages/ShowDetail"; import EpisodePlayer from "./pages/EpisodePlayer"; @@ -105,6 +106,7 @@ function App() { } /> } /> } /> + } /> } /> diff --git a/frontend/src/pages/Admin.jsx b/frontend/src/pages/Admin.jsx index 7d20039..38fec14 100644 --- a/frontend/src/pages/Admin.jsx +++ b/frontend/src/pages/Admin.jsx @@ -1,9 +1,67 @@ -import { useEffect, useState } from "react"; +import { useEffect, useState, useRef } from "react"; import api from "../lib/api"; import { toast } from "sonner"; -import { Trash2, Star, Film, Settings as SettingsIcon, Download, RefreshCcw, FolderSearch, Wrench } from "lucide-react"; +import { + Trash2, Star, Film, Settings as SettingsIcon, Download, FolderSearch, Wrench, ScanSearch, + Tv, ClipboardList, ChevronDown, Users as UsersIcon, +} from "lucide-react"; import { useNavigate, Link } from "react-router-dom"; +// Ordered to match the actual content pipeline: a request comes in, gets pulled into +// Radarr/Sonarr (or uploaded/scanned in directly), gets codec-checked, then transcoded if +// needed — after that it's just in the library. Everything else is a standing utility, not a +// step content passes through, so it lives in its own group. +const PIPELINE_LINKS = [ + { to: "/requests", label: "Review Requests", icon: ClipboardList, testId: "admin-requests-link" }, + { to: "/admin/radarr", label: "Radarr Import", icon: Download, testId: "admin-radarr-link" }, + { to: "/admin/sonarr", label: "Sonarr Import", icon: Tv, testId: "admin-sonarr-link" }, + { to: "/admin/upload", label: "Upload Movie", icon: Film, testId: "admin-upload-link" }, + { to: "/admin/scan", label: "Library Scan", icon: FolderSearch, testId: "admin-scan-link" }, + { to: "/admin/codec-check", label: "Codec Check Queue", icon: ScanSearch, testId: "admin-codec-check-link" }, + { to: "/admin/queue", label: "Transcode Queue", icon: SettingsIcon, testId: "admin-queue-link" }, +]; + +const UTILITY_LINKS = [ + { to: "/admin/cleanup", label: "Library Cleanup", icon: Wrench, testId: "admin-cleanup-link" }, + { to: "/admin/users", label: "Users", icon: UsersIcon, testId: "admin-users-link" }, +]; + +function AdminMenu({ label, links }) { + const [open, setOpen] = useState(false); + const ref = useRef(null); + + useEffect(() => { + const onClick = (e) => { if (ref.current && !ref.current.contains(e.target)) setOpen(false); }; + document.addEventListener("mousedown", onClick); + return () => document.removeEventListener("mousedown", onClick); + }, []); + + return ( +
+ + {open && ( +
+ {links.map(({ to, label: linkLabel, icon: Icon, testId }) => ( + setOpen(false)} + className="flex items-center gap-2 px-4 py-3 text-xs uppercase tracking-[0.2em] text-[#8A8A8A] hover:text-white hover:bg-white/5 border-b border-[#222] last:border-b-0" + data-testid={testId} + > + {linkLabel} + + ))} +
+ )} +
+ ); +} + export default function Admin() { const nav = useNavigate(); const [movies, setMovies] = useState([]); @@ -79,27 +137,8 @@ export default function Admin() {

Manage your library, requests, integrations.

- - Upload Movie - - - Radarr Import - - - Library Scan - - - Library Cleanup - - - Queue - - - Users - - - Review Requests - + +
diff --git a/frontend/src/pages/CodecCheckQueue.jsx b/frontend/src/pages/CodecCheckQueue.jsx new file mode 100644 index 0000000..00a415a --- /dev/null +++ b/frontend/src/pages/CodecCheckQueue.jsx @@ -0,0 +1,173 @@ +import { useEffect, useState } from "react"; +import api from "../lib/api"; +import { toast } from "sonner"; +import { RefreshCcw, X, RotateCcw, Pause, Play as PlayIcon, Trash2 } from "lucide-react"; + +const STATUS_COLORS = { + pending: "text-[#fcd34d]", + running: "text-[#fcd34d]", + done: "text-[#86efac]", + failed: "text-[#fca5a5]", + cancelled: "text-[#8A8A8A]", + superseded: "text-[#8A8A8A]", +}; + +const CLASSIFICATION_LABEL = { + native: "Native (no transcode)", + audio: "Audio fix needed", + abr: "Full re-encode needed", +}; + +export default function CodecCheckQueue() { + const [jobs, setJobs] = useState([]); + const [stats, setStats] = useState({ pending: 0, running: 0, done: 0, failed: 0, cancelled: 0, paused: false }); + const [runningJob, setRunningJob] = useState(null); + const [loading, setLoading] = useState(false); + + const load = async () => { + setLoading(true); + try { + const [j, s] = await Promise.all([api.get("/codec-check/queue"), api.get("/codec-check/queue/stats")]); + setJobs(j.data); + setStats(s.data); + } finally { setLoading(false); } + }; + const loadRunning = async () => { + try { const { data } = await api.get("/codec-check/queue/running"); setRunningJob(data); } + catch { /* ignore transient poll failures */ } + }; + useEffect(() => { load(); loadRunning(); }, []); + + useEffect(() => { + const pending = jobs.some((j) => j.status === "pending"); + if (!runningJob && !pending) return; + const t = setInterval(() => { load(); loadRunning(); }, runningJob ? 2000 : 4000); + return () => clearInterval(t); + }, [jobs, runningJob]); + + const listedJobs = jobs.filter((j) => j.status !== "running"); + + const cancel = async (id) => { + try { await api.delete(`/codec-check/queue/${id}`); toast.success("Job cancelled"); load(); } + catch (err) { toast.error(err.response?.data?.detail || "Could not cancel"); } + }; + + const retry = async (id) => { + try { await api.post(`/codec-check/queue/${id}/retry`); toast.success("Re-queued"); load(); } + catch (err) { toast.error(err.response?.data?.detail || "Could not retry"); } + }; + + const retryAllFailed = async () => { + try { + const { data } = await api.post("/codec-check/queue/retry-failed"); + toast.success(`Re-queued ${data.retried} job(s)`); + load(); + } catch (err) { toast.error(err.response?.data?.detail || "Could not retry"); } + }; + + const togglePause = async () => { + try { + const { data } = await api.post("/codec-check/queue/pause", { paused: !stats.paused }); + toast.success(data.paused ? "Queue paused" : "Queue resumed"); + load(); + } catch { toast.error("Could not toggle pause"); } + }; + + const clearFinished = async () => { + if (!window.confirm("Remove all done/failed/cancelled jobs from the history?")) return; + try { const { data } = await api.post("/codec-check/queue/clear"); toast.success(`Cleared ${data.deleted}`); load(); } + catch { toast.error("Could not clear"); } + }; + + return ( +
+
+ Background · Step 1 of 2 +

Codec Check Queue

+

+ Every newly imported movie/episode lands here first. A quick ffprobe decides whether it actually needs + transcoding at all: already-browser-native files (H.264 + AAC/MP3) skip straight to the library with + no transcode job ever created; everything else is handed off to the{" "} + Transcode Queue (step 2). +

+ +
+ {["pending", "running", "done", "failed", "cancelled"].map((k) => ( +
+
{k}
+
{stats[k] || 0}
+
+ ))} +
+ +
+ + + + +
+ + {runningJob && ( +
+
+ + Checking Codec +
+
+ {runningJob.movie_title || runningJob.movie_id.slice(0, 8)} +
+
{runningJob.triggered_by}
+
+ )} + +
+
+ Title + Result + Trigger + Status + Created + Actions +
+ {listedJobs.length === 0 ? ( +
No jobs yet.
+ ) : listedJobs.map((j) => ( +
+
{j.movie_title || j.movie_id.slice(0, 8)}
+ + {j.classification ? CLASSIFICATION_LABEL[j.classification] || j.classification : "—"} + + {j.triggered_by} + + {j.status} + {j.error && j.status === "failed" && {j.error.slice(0, 40)}} + + {new Date(j.created_at).toLocaleDateString()} +
+ {j.status === "pending" && ( + + )} + {(j.status === "failed" || j.status === "cancelled") && ( + + )} +
+
+ ))} +
+
+
+ ); +} diff --git a/frontend/src/pages/Requests.jsx b/frontend/src/pages/Requests.jsx index 8e65b07..f65ed25 100644 --- a/frontend/src/pages/Requests.jsx +++ b/frontend/src/pages/Requests.jsx @@ -40,6 +40,80 @@ function LiveStatus({ requestId }) { ); } +function CandidatePicker({ request, contentType, onClose, onPicked }) { + const [candidates, setCandidates] = useState(null); + const [picking, setPicking] = useState(false); + + useEffect(() => { + let cancelled = false; + (async () => { + try { + const { data } = await api.get(`/requests/${request.id}/candidates`, { params: { content_type: contentType } }); + if (!cancelled) setCandidates(data); + } catch (err) { + if (!cancelled) { + toast.error(err.response?.data?.detail || "Could not search Radarr/Sonarr"); + onClose(); + } + } + })(); + return () => { cancelled = true; }; + }, [request.id, contentType]); + + const pick = async (c) => { + setPicking(true); + try { + await onPicked(c); + } finally { + setPicking(false); + } + }; + + return ( +
+
e.stopPropagation()} + data-testid="candidate-picker" + > +
+

+ Pick the right match for "{request.title}"{request.year ? ` (${request.year})` : ""} +

+ +
+ {candidates === null ? ( +

Searching…

+ ) : candidates.length === 0 ? ( +

No results found.

+ ) : ( +
+ {candidates.map((c) => ( + + ))} +
+ )} +
+
+ ); +} + export default function Requests() { const { user } = useAuth(); const [mine, setMine] = useState([]); @@ -48,6 +122,7 @@ export default function Requests() { const [year, setYear] = useState(""); const [notes, setNotes] = useState(""); const [approving, setApproving] = useState({}); + const [picker, setPicker] = useState(null); // { request, contentType } const load = async () => { const { data } = await api.get("/requests/mine"); @@ -59,11 +134,14 @@ export default function Requests() { }; useEffect(() => { load(); }, [user?.is_admin]); - const approve = async (id, contentType) => { + const approveWithCandidate = async (request, contentType, candidate) => { + const id = request.id; setApproving((p) => ({ ...p, [id]: true })); try { - const { data } = await api.post(`/requests/${id}/approve`, { content_type: contentType }); + const idField = contentType === "movie" ? { tmdb_id: candidate.tmdb_id } : { tvdb_id: candidate.tvdb_id }; + const { data } = await api.post(`/requests/${id}/approve`, { content_type: contentType, ...idField }); toast.success(`Searching for "${data.matched_title || data.title}"`); + setPicker(null); load(); } catch (err) { toast.error(err.response?.data?.detail || "Could not approve request"); @@ -189,10 +267,10 @@ export default function Requests() {
{r.status === "pending" && ( <> - - @@ -211,6 +289,14 @@ export default function Requests() {
)}
+ {picker && ( + setPicker(null)} + onPicked={(c) => approveWithCandidate(picker.request, picker.contentType, c)} + /> + )} ); } diff --git a/frontend/src/pages/TranscodeQueue.jsx b/frontend/src/pages/TranscodeQueue.jsx index 9636fb7..d3cabfe 100644 --- a/frontend/src/pages/TranscodeQueue.jsx +++ b/frontend/src/pages/TranscodeQueue.jsx @@ -92,10 +92,12 @@ export default function TranscodeQueue() { return (
- Background + Background · Step 2 of 2

Transcode Queue

- One job runs at a time, moderate CPU priority, capped threads so it can't starve the rest of the stack. Auto-transcode is currently:{" "} + Only files the Codec Check Queue (step 1) flagged + as needing work land here — already browser-native files skip this queue entirely. One job runs at a time, moderate CPU priority, + capped threads so it can't starve the rest of the stack. Auto-transcode is currently:{" "} {stats.auto_transcode === "off" ? "off" : `auto · ${stats.auto_transcode}`} {" — "}change in Settings. Failed and cancelled jobs never delete the movie — retry re-queues the same file.