"""Media router — index status, build, stop, and query.""" from __future__ import annotations import logging import os import signal import subprocess import sys import threading import time from pathlib import Path from typing import Any from fastapi import APIRouter, Depends, HTTPException, Query, status from media_library_viewer_api.clients.jellyfin import JellyfinClient from media_library_viewer_api.dependencies import get_jellyfin_client, get_user_id from media_library_viewer_api.observability import record_media_index_build from media_library_viewer_api.services.media_index import MediaIndex logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/media", tags=["media"]) _build_lock = threading.Lock() def get_media_index() -> MediaIndex: return MediaIndex() def _set_build_metadata(index: MediaIndex, state: dict[str, Any]) -> None: for key, value in state.items(): index.set_metadata(key, "" if value is None else value) def _staging_db_path(index: MediaIndex) -> Path: return index.db_path.with_name(f"{index.db_path.stem}.building{index.db_path.suffix}") def _pid_is_alive(pid: int | None) -> bool: if not pid: return False try: os.kill(pid, 0) return True except ProcessLookupError: return False except PermissionError: return True def _clean_stale_build_state(index: MediaIndex) -> Any: status = index.status() if status.build_running and not _pid_is_alive(status.build_pid): logger.warning("Detected stale media build state pid=%s", status.build_pid) _set_build_metadata( index, { "build_running": False, "build_stage": "stale", "build_message": "Previous media index build stopped unexpectedly", "build_cancel_requested": False, "build_pid": "", "build_error": "Worker process is no longer running", }, ) status = index.status() return status def _serialize_status(status: Any) -> dict[str, Any]: return { "exists": status.exists, "item_count": status.item_count, "updated_at": status.updated_at, "updated_at_label": status.updated_at_label, "build_duration_seconds": status.build_duration_seconds, "build_running": status.build_running, "build_stage": status.build_stage, "build_message": status.build_message, "build_progress": status.build_progress, "build_items_processed": status.build_items_processed, "build_items_total": status.build_items_total, "build_current_library": status.build_current_library, "build_library_index": status.build_library_index, "build_libraries_total": status.build_libraries_total, "build_library_progress": status.build_library_progress, "build_library_items_processed": status.build_library_items_processed, "build_library_items_total": status.build_library_items_total, "build_elapsed_seconds": status.build_elapsed_seconds, "build_eta_seconds": status.build_eta_seconds, "build_library_elapsed_seconds": status.build_library_elapsed_seconds, "build_library_eta_seconds": status.build_library_eta_seconds, "build_cancel_requested": status.build_cancel_requested, "build_pid": status.build_pid, "build_error": status.build_error, } def _worker_command(final_db_path: Path, staging_db_path: Path, service_id: str = "") -> list[str]: return [ sys.executable, "-m", "media_library_viewer_api.workers.media_index_worker", "--index-path", str(final_db_path), "--staging-path", str(staging_db_path), "--service-id", service_id, ] def _start_worker(index: MediaIndex, service_id: str = "") -> subprocess.Popen[bytes]: staging_path = _staging_db_path(index) staging_path.unlink(missing_ok=True) return subprocess.Popen( _worker_command(index.db_path, staging_path, service_id), start_new_session=True, env=os.environ.copy(), ) @router.get("/status") def get_index_status(index: MediaIndex = Depends(get_media_index)) -> dict[str, Any]: """Return media index status (exists, count, build progress, and errors).""" with _build_lock: media_status = _clean_stale_build_state(index) logger.info("Media status requested running=%s stage=%s", media_status.build_running, media_status.build_stage) return _serialize_status(media_status) @router.post("/build", status_code=status.HTTP_202_ACCEPTED) def post_build_index( jellyfin_service_id: str | None = None, index: MediaIndex = Depends(get_media_index), ) -> dict[str, Any]: """Start a media index build in a subprocess worker. The worker resolves its own Jellyfin connection from the settings store. We do NOT use Depends(get_jellyfin_client) here because the worker runs in a separate process and needs to resolve the client itself. Validating the connection here would fail if Jellyfin is briefly unreachable, even though the build just needs to start the worker process. """ with _build_lock: current_status = _clean_stale_build_state(index) if current_status.build_running and _pid_is_alive(current_status.build_pid): logger.warning("Media build already running pid=%s", current_status.build_pid) raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Media index build already in progress") logger.info( "Starting media index build service_id=%s", jellyfin_service_id or "", ) process = _start_worker(index, jellyfin_service_id or "") _set_build_metadata( index, { "build_running": True, "build_stage": "queued", "build_message": "Media index build queued", "build_progress": None, "build_items_processed": 0, "build_items_total": 0, "build_current_library": "", "build_library_index": 0, "build_libraries_total": 0, "build_library_progress": None, "build_library_items_processed": 0, "build_library_items_total": 0, "build_elapsed_seconds": None, "build_eta_seconds": None, "build_library_elapsed_seconds": None, "build_library_eta_seconds": None, "build_cancel_requested": False, "build_pid": process.pid, "build_error": "", }, ) media_status = index.status() record_media_index_build(status="started") logger.info("Media index build started pid=%s", process.pid) return {"status": "started", **_serialize_status(media_status)} @router.post("/stop", status_code=status.HTTP_202_ACCEPTED) def stop_build(index: MediaIndex = Depends(get_media_index)) -> dict[str, Any]: """Request cooperative cancellation of a running media index build.""" with _build_lock: media_status = _clean_stale_build_state(index) if not media_status.build_running: logger.warning("Stop requested but no media build is running") raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="No media index build is running") logger.info("Cooperative stop requested for media build pid=%s", media_status.build_pid) _set_build_metadata( index, { "build_running": True, "build_stage": "canceling", "build_message": "Stopping media index build...", "build_cancel_requested": True, "build_error": "", }, ) media_status = index.status() logger.info("Media stop requested acknowledged") return {"status": "stop_requested", **_serialize_status(media_status)} @router.post("/force-stop", status_code=status.HTTP_202_ACCEPTED) def force_stop_build(index: MediaIndex = Depends(get_media_index)) -> dict[str, Any]: """Terminate the media index worker process if it is stuck.""" with _build_lock: media_status = _clean_stale_build_state(index) if not media_status.build_running: logger.warning("Force stop requested but no media build is running") raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="No media index build is running") pid = media_status.build_pid if not pid or not _pid_is_alive(pid): logger.warning("Force stop requested but build worker is not alive pid=%s", pid) _set_build_metadata( index, { "build_running": False, "build_stage": "stale", "build_message": "Media index worker is not running", "build_cancel_requested": False, "build_pid": "", "build_error": "Worker process is not running", }, ) media_status = index.status() return {"status": "already_stopped", **_serialize_status(media_status)} logger.info("Force stopping media build pid=%s", pid) try: os.killpg(pid, signal.SIGTERM) except ProcessLookupError: pass except PermissionError as exc: raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc)) from exc deadline = time.time() + 3.0 while time.time() < deadline and _pid_is_alive(pid): time.sleep(0.1) if _pid_is_alive(pid): try: os.killpg(pid, signal.SIGKILL) except ProcessLookupError: pass except PermissionError as exc: raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc)) from exc _set_build_metadata( index, { "build_running": False, "build_stage": "force-stopped", "build_message": "Media index build force stopped", "build_cancel_requested": False, "build_pid": "", "build_error": "", }, ) media_status = index.status() record_media_index_build(status="force_stopped") return {"status": "force_stopped", **_serialize_status(media_status)} @router.get("/query") def query_media( libraries: str = Query("", description="Comma-separated library IDs"), types: str = Query("Movie,Episode", description="Comma-separated media types"), search: str = Query("", description="Search term"), hdr_filter: str = Query("All", description="All, HDR only, SDR/unknown only"), sort_key: str = Query("title", description="Sort field"), sort_order: str = Query("Ascending", description="Ascending or Descending"), limit: int = Query(100, ge=1, le=1000), offset: int = Query(0, ge=0), jellyfin_service_id: str | None = None, client: JellyfinClient = Depends(get_jellyfin_client), user_id: str = Depends(get_user_id), index: MediaIndex = Depends(get_media_index), ) -> dict[str, Any]: """Query the media index with filters, sorting, and pagination.""" # If no library IDs provided, use all libraries. library_ids = [lid.strip() for lid in libraries.split(",") if lid.strip()] if libraries else None if not library_ids: all_libs = client.libraries(user_id) library_ids = [lib["Id"] for lib in all_libs] media_types = [t.strip() for t in types.split(",") if t.strip()] logger.info( "Media query user_id=%s libraries=%s types=%s search=%s hdr=%s sort=%s/%s limit=%s offset=%s", user_id, len(library_ids or []), ",".join(media_types), search or "", hdr_filter, sort_key, sort_order, limit, offset, ) rows, total = index.query( library_ids=library_ids, media_types=media_types, search=search, hdr_filter=hdr_filter, sort_key=sort_key, sort_order=sort_order, limit=limit, offset=offset, service_id=jellyfin_service_id or "", ) logger.info("Media query returned total=%s rows=%s", total, len(rows)) return { "items": rows, "total": total, "limit": limit, "offset": offset, }