myron/kino-app
1
0
Fork 0
mirror of https://github.com/myronblair/kino-app synced 2026-07-31 21:12:43 -05:00
Code Issues Packages Projects Releases Wiki Activity

Codec-aware transcode routing, two-stage queue, request candidate picker, admin reorg

Browse Source 
- Requests approval now shows a Radarr/Sonarr candidate picker instead of grabbing the first search result (fixes wrong-title/wrong-year matches)
- approve_request reuses an existing Radarr/Sonarr entry instead of erroring when content is already in the library
- New codec-check queue (step 1) probes source codecs and skips transcoding entirely for already browser-native files; only non-native content reaches the transcode queue (step 2)
- Fixes ffmpeg hard-failing on non-H.264 sources via the h264_mp4toannexb bitstream filter (root cause of the fix-audio-all failures)
- Added Codec Check Worker / Transcode Worker to the services health panel with restart
- Reorganized the admin page into Pipeline/Utilities dropdowns; added the missing Sonarr Import link
- Bumped VERSION 1.0.0 -> 1.1.0
This commit is contained in:
root
2026-07-30 20:41:11 -05:00
parent 671074764e
commit b28a49e8fa
13 changed files with 784 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/server.py
+288 -18
View File
@@ -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")