kino-app/backend/server.py

import os
import uuid
import logging
import mimetypes
import asyncio
from pathlib import Path
from datetime import datetime, timezone
from typing import List, Optional, Dict

from fastapi import FastAPI, APIRouter, HTTPException, Depends, UploadFile, File, Form, Header, Query, Request, BackgroundTasks
from fastapi.responses import StreamingResponse, FileResponse, Response
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from dotenv import load_dotenv
from starlette.middleware.cors import CORSMiddleware
from motor.motor_asyncio import AsyncIOMotorClient

from models import (
    UserCreate, UserLogin, UserPublic, TokenResponse, PasswordChange,
    AdminUserCreate, AdminUserUpdate,
    Profile, ProfileUpdate, RATING_ORDER,
    Movie, MovieCreate, MovieUpdate,
    Subtitle,
    WatchlistItem, ProgressUpsert,
    RequestCreate, RequestUpdate, RequestApprove, MovieRequest,
    AppSettings, AppSettingsPublic,
    TMDBSearchResult, RadarrMovieDTO,
    TranscodeJob, QueuePauseToggle,
    Show, ShowCreate, ShowUpdate, Episode, EpisodeCreate, EpisodeProgressUpsert,
    SonarrSeriesDTO, TraktDeviceCodeResponse, TraktStatus, OpenSubsResult,
    Track, TrackUpdate,
)
from auth import hash_password, verify_password, create_token, decode_token
from seed import SAMPLE_MOVIES
import tmdb as tmdb_client
import radarr as radarr_client
import sonarr as sonarr_client
import services as services_client
import library_scan
import library_cleanup
import musicbrainz as musicbrainz_client
import trakt as trakt_client
import opensubtitles as opensubs_client
from transcode import transcode_quick, transcode_abr, transcode_audio_fix, srt_to_vtt


ROOT_DIR = Path(__file__).parent
load_dotenv(ROOT_DIR / ".env")

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
logger = logging.getLogger("kino")

mongo_url = os.environ["MONGO_URL"]
client = AsyncIOMotorClient(mongo_url)
db = client[os.environ["DB_NAME"]]

MEDIA_ROOT = Path(os.environ.get("MEDIA_ROOT", str(ROOT_DIR / "media")))
VIDEOS_DIR = MEDIA_ROOT / "videos"
POSTERS_DIR = MEDIA_ROOT / "posters"
SUBS_DIR = MEDIA_ROOT / "subtitles"
HLS_DIR = MEDIA_ROOT / "hls"
for d in (VIDEOS_DIR, POSTERS_DIR, SUBS_DIR, HLS_DIR):
    d.mkdir(parents=True, exist_ok=True)

CHUNK_SIZE = 1024 * 1024

app = FastAPI(title="StreamHoard")
api = APIRouter(prefix="/api")
bearer = HTTPBearer(auto_error=False)

APP_VERSION = (ROOT_DIR / "VERSION").read_text().strip() if (ROOT_DIR / "VERSION").is_file() else "0.0.0"


@api.get("/version")
async def get_version():
    return {"version": APP_VERSION}


# ---------- Auth deps ----------
async def get_current_user(creds: Optional[HTTPAuthorizationCredentials] = Depends(bearer)) -> dict:
    if not creds:
        raise HTTPException(status_code=401, detail="Not authenticated")
    payload = decode_token(creds.credentials)
    user = await db.users.find_one({"id": payload["sub"]}, {"_id": 0, "password_hash": 0})
    if not user:
        raise HTTPException(status_code=401, detail="User not found")
    return user


async def require_admin(user: dict = Depends(get_current_user)) -> dict:
    if not user.get("is_admin"):
        raise HTTPException(status_code=403, detail="Admin only")
    return user


async def get_active_profile(
    x_profile_id: Optional[str] = Header(None),
    user: dict = Depends(get_current_user),
) -> dict:
    """Resolve current profile via X-Profile-Id header, else default profile."""
    if x_profile_id:
        p = await db.profiles.find_one({"id": x_profile_id, "user_id": user["id"]}, {"_id": 0})
        if p:
            return p
    p = await db.profiles.find_one({"user_id": user["id"]}, {"_id": 0})
    if not p:
        # auto-create default
        prof = Profile(user_id=user["id"], name=user.get("name", "Main")).model_dump()
        await db.profiles.insert_one(prof)
        prof.pop("_id", None)
        return prof
    return p


# ---------- Settings helper ----------
async def get_settings() -> dict:
    doc = await db.settings.find_one({"id": "app"}, {"_id": 0})
    if not doc:
        return AppSettings().model_dump() | {"id": "app"}
    return doc


# ---------- Startup ----------
@app.on_event("startup")
async def on_startup():
    await db.users.create_index("email", unique=True)
    await db.movies.create_index("title")
    await db.profiles.create_index("user_id")
    await db.subtitles.create_index("movie_id")

    admin_email = os.environ.get("ADMIN_EMAIL", "admin@streamhoard.local")
    if not await db.users.find_one({"email": admin_email}):
        admin_user = {
            "id": str(uuid.uuid4()),
            "email": admin_email,
            "name": os.environ.get("ADMIN_NAME", "Admin"),
            "password_hash": hash_password(os.environ.get("ADMIN_PASSWORD", "kino-admin-2026")),
            "is_admin": True,
            "created_at": datetime.now(timezone.utc).isoformat(),
        }
        await db.users.insert_one(admin_user)
        logger.info(f"Seeded admin: {admin_email}")

    if await db.movies.count_documents({}) == 0:
        for m in SAMPLE_MOVIES:
            await db.movies.insert_one(Movie(**m).model_dump())
        logger.info(f"Seeded {len(SAMPLE_MOVIES)} movies")

    # ---- Migration: ensure every user has a default profile, backfill watchlist/progress ----
    async for u in db.users.find({}, {"_id": 0, "id": 1, "name": 1}):
        if not await db.profiles.find_one({"user_id": u["id"]}):
            prof = Profile(user_id=u["id"], name=u.get("name") or "Main").model_dump()
            await db.profiles.insert_one(prof)
            await db.watchlist.update_many(
                {"user_id": u["id"], "$or": [{"profile_id": {"$exists": False}}, {"profile_id": None}]},
                {"$set": {"profile_id": prof["id"]}},
            )
            await db.progress.update_many(
                {"user_id": u["id"], "$or": [{"profile_id": {"$exists": False}}, {"profile_id": None}]},
                {"$set": {"profile_id": prof["id"]}},
            )

    # Now build unique indexes (migration above ensures no null collisions)
    try:
        await db.watchlist.drop_index("user_id_1_movie_id_1")
    except Exception:
        pass
    try:
        await db.watchlist.create_index([("profile_id", 1), ("movie_id", 1)], unique=True)
    except Exception as e:
        logger.warning(f"watchlist unique index skipped: {e}")
    try:
        await db.progress.drop_index("user_id_1_movie_id_1")
    except Exception:
        pass
    try:
        await db.progress.create_index([("profile_id", 1), ("movie_id", 1)], unique=True)
    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():
    client.close()


def _user_public(u: dict) -> UserPublic:
    return UserPublic(
        id=u["id"], email=u["email"], name=u["name"],
        is_admin=u.get("is_admin", False), created_at=u["created_at"],
    )


def _strip(d: dict) -> dict:
    d.pop("_id", None)
    return d


def _rating_allowed(profile: dict, rating: str) -> bool:
    """Profile.max_rating is the strictest a profile can see."""
    cap = profile.get("max_rating", "NR")
    if cap == "NR":
        return True
    try:
        return RATING_ORDER.index(rating) <= RATING_ORDER.index(cap)
    except ValueError:
        return True  # unknown rating → allow


# ============ AUTH ============
@api.post("/auth/register", response_model=TokenResponse)
async def register(payload: UserCreate):
    if await db.users.find_one({"email": payload.email.lower()}):
        raise HTTPException(status_code=400, detail="Email already registered")
    user = {
        "id": str(uuid.uuid4()),
        "email": payload.email.lower(),
        "name": payload.name,
        "password_hash": hash_password(payload.password),
        "is_admin": False,
        "created_at": datetime.now(timezone.utc).isoformat(),
    }
    await db.users.insert_one(user)
    # auto-create default profile
    await db.profiles.insert_one(Profile(user_id=user["id"], name=user["name"]).model_dump())
    return TokenResponse(access_token=create_token(user["id"], False), user=_user_public(user))


@api.post("/auth/login", response_model=TokenResponse)
async def login(payload: UserLogin):
    user = await db.users.find_one({"email": payload.email.lower()})
    if not user or not verify_password(payload.password, user["password_hash"]):
        raise HTTPException(status_code=401, detail="Invalid email or password")
    return TokenResponse(access_token=create_token(user["id"], user.get("is_admin", False)), user=_user_public(user))


@api.get("/auth/me", response_model=UserPublic)
async def me(user: dict = Depends(get_current_user)):
    return _user_public(user)


@api.post("/auth/change-password")
async def change_password(payload: PasswordChange, user: dict = Depends(get_current_user)):
    full_user = await db.users.find_one({"id": user["id"]})
    if not full_user or not verify_password(payload.current_password, full_user["password_hash"]):
        raise HTTPException(status_code=401, detail="Current password is incorrect")
    if len(payload.new_password) < 8:
        raise HTTPException(status_code=400, detail="New password must be at least 8 characters")
    await db.users.update_one(
        {"id": user["id"]}, {"$set": {"password_hash": hash_password(payload.new_password)}}
    )
    return {"ok": True}


# ============ USERS (admin) ============
@api.get("/admin/users", response_model=List[UserPublic])
async def admin_list_users(user: dict = Depends(require_admin)):
    docs = await db.users.find({}, {"_id": 0, "password_hash": 0}).sort("created_at", 1).to_list(500)
    return docs


@api.post("/admin/users", response_model=UserPublic)
async def admin_create_user(payload: AdminUserCreate, user: dict = Depends(require_admin)):
    if await db.users.find_one({"email": payload.email.lower()}):
        raise HTTPException(status_code=400, detail="Email already registered")
    new_user = {
        "id": str(uuid.uuid4()),
        "email": payload.email.lower(),
        "name": payload.name,
        "password_hash": hash_password(payload.password),
        "is_admin": payload.is_admin,
        "created_at": datetime.now(timezone.utc).isoformat(),
    }
    await db.users.insert_one(new_user)
    await db.profiles.insert_one(Profile(user_id=new_user["id"], name=new_user["name"]).model_dump())
    return _user_public(new_user)


@api.patch("/admin/users/{user_id}", response_model=UserPublic)
async def admin_update_user(user_id: str, payload: AdminUserUpdate, user: dict = Depends(require_admin)):
    target = await db.users.find_one({"id": user_id})
    if not target:
        raise HTTPException(status_code=404, detail="User not found")
    updates = {}
    if payload.name is not None:
        updates["name"] = payload.name
    if payload.is_admin is not None:
        if user_id == user["id"] and not payload.is_admin:
            raise HTTPException(status_code=400, detail="Cannot remove your own admin access")
        updates["is_admin"] = payload.is_admin
    if payload.password:
        if len(payload.password) < 8:
            raise HTTPException(status_code=400, detail="Password must be at least 8 characters")
        updates["password_hash"] = hash_password(payload.password)
    if updates:
        await db.users.update_one({"id": user_id}, {"$set": updates})
    target = await db.users.find_one({"id": user_id})
    return _user_public(target)


@api.delete("/admin/users/{user_id}")
async def admin_delete_user(user_id: str, user: dict = Depends(require_admin)):
    if user_id == user["id"]:
        raise HTTPException(status_code=400, detail="Cannot delete your own account")
    target = await db.users.find_one({"id": user_id})
    if not target:
        raise HTTPException(status_code=404, detail="User not found")
    await db.users.delete_one({"id": user_id})
    await db.profiles.delete_many({"user_id": user_id})
    return {"ok": True}


# ============ PROFILE (1:1 with account — no separate sub-profile management) ============
@api.get("/profiles", response_model=List[Profile])
async def list_profiles(user: dict = Depends(get_current_user)):
    """Always exactly one profile per account; get_active_profile auto-creates it if missing."""
    docs = await db.profiles.find({"user_id": user["id"]}, {"_id": 0}).sort("created_at", 1).to_list(1)
    return docs


@api.patch("/profiles/{pid}", response_model=Profile)
async def update_profile(pid: str, payload: ProfileUpdate, user: dict = Depends(get_current_user)):
    """Edit your own single profile (avatar color, kids mode, rating cap) — not a multi-profile switcher."""
    updates = {k: v for k, v in payload.model_dump().items() if v is not None}
    if not updates:
        raise HTTPException(status_code=400, detail="No fields to update")
    res = await db.profiles.find_one_and_update(
        {"id": pid, "user_id": user["id"]},
        {"$set": updates},
        projection={"_id": 0}, return_document=True,
    )
    if not res:
        raise HTTPException(status_code=404, detail="Profile not found")
    return res


# ============ MOVIES ============
def _movie_filter_for_profile(profile: dict) -> dict:
    cap = profile.get("max_rating", "NR")
    if cap == "NR":
        return {}
    allowed = RATING_ORDER[:RATING_ORDER.index(cap) + 1]
    return {"rating": {"$in": allowed + ["NR"]}}


@api.get("/movies", response_model=List[Movie])
async def list_movies(
    genre: Optional[str] = None, q: Optional[str] = None, limit: int = 200,
    profile: dict = Depends(get_active_profile),
):
    query: dict = _movie_filter_for_profile(profile)
    if genre:
        query["genres"] = {"$regex": f"^{genre}$", "$options": "i"}
    if q:
        query["$or"] = [
            {"title": {"$regex": q, "$options": "i"}},
            {"description": {"$regex": q, "$options": "i"}},
            {"director": {"$regex": q, "$options": "i"}},
            {"cast": {"$regex": q, "$options": "i"}},
        ]
    docs = await db.movies.find(query, {"_id": 0}).sort("created_at", -1).to_list(limit)
    return docs


@api.get("/movies/featured", response_model=Movie)
async def get_featured(profile: dict = Depends(get_active_profile)):
    base = _movie_filter_for_profile(profile)
    doc = await db.movies.find_one({**base, "featured": True}, {"_id": 0})
    if not doc:
        doc = await db.movies.find_one(base, {"_id": 0})
    if not doc:
        raise HTTPException(status_code=404, detail="No movies")
    return doc


@api.get("/movies/genres", response_model=List[str])
async def list_genres():
    genres = await db.movies.distinct("genres")
    return sorted([g for g in genres if g])


@api.get("/movies/{movie_id}", response_model=Movie)
async def get_movie(movie_id: str):
    doc = await db.movies.find_one({"id": movie_id}, {"_id": 0})
    if not doc:
        raise HTTPException(status_code=404, detail="Movie not found")
    return doc


@api.post("/movies", response_model=Movie)
async def create_movie(payload: MovieCreate, user: dict = Depends(require_admin)):
    movie = Movie(**payload.model_dump())
    doc = movie.model_dump()
    await db.movies.insert_one(doc)
    return _strip(doc)


@api.patch("/movies/{movie_id}", response_model=Movie)
async def update_movie(movie_id: str, payload: MovieUpdate, user: dict = Depends(require_admin)):
    updates = {k: v for k, v in payload.model_dump().items() if v is not None}
    if not updates:
        raise HTTPException(status_code=400, detail="No fields to update")
    res = await db.movies.find_one_and_update(
        {"id": movie_id}, {"$set": updates}, projection={"_id": 0}, return_document=True,
    )
    if not res:
        raise HTTPException(status_code=404, detail="Movie not found")
    return res


@api.delete("/movies/{movie_id}")
async def delete_movie(movie_id: str, user: dict = Depends(require_admin)):
    movie = await db.movies.find_one({"id": movie_id}, {"_id": 0})
    if not movie:
        raise HTTPException(status_code=404, detail="Movie not found")
    if movie.get("storage_type") == "local" and movie.get("storage_path"):
        try: (VIDEOS_DIR / movie["storage_path"]).unlink(missing_ok=True)
        except Exception: pass
    if movie.get("hls_path"):
        import shutil
        shutil.rmtree(HLS_DIR / movie_id, ignore_errors=True)
    await db.movies.delete_one({"id": movie_id})
    await db.watchlist.delete_many({"movie_id": movie_id})
    await db.progress.delete_many({"movie_id": movie_id})
    await db.subtitles.delete_many({"movie_id": movie_id})
    return {"ok": True}


# ============ UPLOAD VIDEO ============
@api.post("/upload/video", response_model=Movie)
async def upload_video(
    title: str = Form(...),
    description: str = Form(""),
    year: int = Form(2024),
    duration_minutes: int = Form(0),
    rating: str = Form("NR"),
    genres: str = Form(""),
    cast: str = Form(""),
    director: str = Form(""),
    poster_url: str = Form(""),
    backdrop_url: str = Form(""),
    featured: bool = Form(False),
    tmdb_id: Optional[int] = Form(None),
    file: UploadFile = File(...),
    user: dict = Depends(require_admin),
):
    ext = (file.filename.rsplit(".", 1)[-1] if "." in (file.filename or "") else "mp4").lower()
    safe_id = str(uuid.uuid4())
    fname = f"{safe_id}.{ext}"
    target = VIDEOS_DIR / fname
    with target.open("wb") as f:
        while True:
            chunk = await file.read(CHUNK_SIZE)
            if not chunk: break
            f.write(chunk)

    movie = Movie(
        title=title, description=description, year=year,
        duration_minutes=duration_minutes, rating=rating,
        genres=[g.strip() for g in genres.split(",") if g.strip()],
        cast=[c.strip() for c in cast.split(",") if c.strip()],
        director=director, poster_url=poster_url, backdrop_url=backdrop_url,
        video_url="", storage_type="local", storage_path=fname,
        featured=featured, tmdb_id=tmdb_id,
    )
    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)


# ============ STREAM (Range-aware) ============
@api.get("/stream/{movie_id}")
async def stream_movie(
    movie_id: str, request: Request,
    auth: Optional[str] = Query(None),
    authorization: Optional[str] = Header(None),
):
    token = None
    if authorization and authorization.lower().startswith("bearer "):
        token = authorization.split(" ", 1)[1].strip()
    elif auth:
        token = auth
    if not token: raise HTTPException(status_code=401, detail="Not authenticated")
    decode_token(token)

    movie = await db.movies.find_one({"id": movie_id}, {"_id": 0})
    if not movie: raise HTTPException(status_code=404, detail="Movie not found")
    if movie.get("storage_type") not in ("local", "radarr", "scan") or not movie.get("storage_path"):
        raise HTTPException(status_code=400, detail="Movie has no local file; use video_url directly")

    if movie.get("storage_type") in ("radarr", "scan"):
        file_path = Path(movie["storage_path"])  # absolute path on NAS-mounted storage
    else:
        file_path = VIDEOS_DIR / movie["storage_path"]
    if not file_path.is_file():
        raise HTTPException(status_code=404, detail="File missing on disk")

    file_size = file_path.stat().st_size
    content_type, _ = mimetypes.guess_type(str(file_path))
    content_type = content_type or "video/mp4"

    range_header = request.headers.get("range") or request.headers.get("Range")
    if range_header and range_header.startswith("bytes="):
        try:
            start_str, end_str = range_header.replace("bytes=", "").split("-")
            start = int(start_str) if start_str else 0
            end = int(end_str) if end_str else file_size - 1
        except ValueError:
            raise HTTPException(status_code=416, detail="Invalid range")
        if start >= file_size:
            raise HTTPException(status_code=416, detail="Range out of bounds")
        end = min(end, file_size - 1)
        length = end - start + 1

        def iter_file():
            with file_path.open("rb") as f:
                f.seek(start)
                remaining = length
                while remaining > 0:
                    chunk = f.read(min(CHUNK_SIZE, remaining))
                    if not chunk: break
                    remaining -= len(chunk)
                    yield chunk

        return StreamingResponse(iter_file(), status_code=206, media_type=content_type, headers={
            "Content-Range": f"bytes {start}-{end}/{file_size}",
            "Accept-Ranges": "bytes",
            "Content-Length": str(length),
            "Content-Type": content_type,
        })
    return FileResponse(str(file_path), media_type=content_type, headers={"Accept-Ranges": "bytes"})


# ============ HLS ============
async def _set_hls_status(movie_id: str, status: str, path: Optional[str] = None, error: Optional[str] = None):
    update = {"hls_status": status}
    if path is not None: update["hls_path"] = path
    await db.movies.update_one({"id": movie_id}, {"$set": update})
    if error:
        logger.warning(f"HLS {movie_id}: {error}")


async def _process_job(job: dict):
    """Run a transcode job. Handles both movies and episodes."""
    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.transcode_queue.update_one(
            {"id": job["id"]},
            {"$set": {"status": "failed", "error": f"{content_type} not found", "finished_at": datetime.now(timezone.utc).isoformat()}},
        )
        return
    if doc.get("storage_type") not in ("local", "radarr", "sonarr", "scan") or not doc.get("storage_path"):
        await db.transcode_queue.update_one(
            {"id": job["id"]},
            {"$set": {"status": "failed", "error": "Not a local file", "finished_at": datetime.now(timezone.utc).isoformat()}},
        )
        return

    source = (
        Path(doc["storage_path"]) if doc["storage_type"] in ("radarr", "sonarr", "scan")
        else VIDEOS_DIR / doc["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):
        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

    try:
        if job["quality"] == "abr":
            await transcode_abr(source, out_dir, cb)
        elif job["quality"] == "audio":
            await transcode_audio_fix(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 coll.update_one({"id": job["movie_id"]}, {"$set": {"hls_status": "failed"}})

    after = await coll.find_one({"id": job["movie_id"]}, {"_id": 0, "hls_status": 1})
    final_status = "done" if after and 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(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."""
    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.transcode_queue.find_one(
        {"movie_id": content_id, "status": {"$in": ["pending", "running"]}},
        {"_id": 0},
    )
    if existing:
        return existing
    job = TranscodeJob(
        movie_id=content_id, movie_title=title, quality=quality,
        triggered_by=triggered_by, content_type=content_type,
    ).model_dump()
    await db.transcode_queue.insert_one(job)
    await coll.update_one({"id": content_id}, {"$set": {"hls_status": "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"|"audio"} — default "quick". Adds to background queue."""
    quality = (payload or {}).get("quality", "quick")
    if quality not in ("quick", "abr", "audio"):
        raise HTTPException(status_code=400, detail="quality must be 'quick', 'abr', or 'audio'")
    movie = await db.movies.find_one({"id": movie_id}, {"_id": 0})
    if not movie: raise HTTPException(status_code=404, detail="Movie not found")
    if movie.get("storage_type") not in ("local", "radarr", "scan") or not movie.get("storage_path"):
        raise HTTPException(status_code=400, detail="Only local/radarr/scanned movies can be transcoded")
    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")
    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 hls_status AND clear stale hls_path so the row no longer shows as transcoded
        coll = db.movies if job.get("content_type", "movie") == "movie" else db.episodes
        await coll.update_one({"id": job["movie_id"]}, {"$set": {"hls_status": None, "hls_path": 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"), content_type=job.get("content_type", "movie"))
    return {"ok": True, "job_id": new_job["id"]}


@api.post("/transcode/queue/retry-failed")
async def retry_all_failed(user: dict = Depends(require_admin)):
    jobs = await db.transcode_queue.find({"status": {"$in": ["failed", "cancelled"]}}, {"_id": 0}).to_list(500)
    retried = 0
    for job in jobs:
        await _enqueue_transcode(job["movie_id"], job["quality"], triggered_by=job.get("triggered_by", "manual"), content_type=job.get("content_type", "movie"))
        retried += 1
    return {"ok": True, "retried": retried}


@api.post("/transcode/queue/fix-audio-all")
async def fix_audio_all(user: dict = Depends(require_admin)):
    """Replaces every still-pending job (not yet started) with a fast audio-only fix job for
    the same content — for libraries where the source audio is AC3/DTS/EAC3 (unplayable in any
    browser), this gets working sound out much sooner than waiting on the full ABR backlog to
    catch up, since only the audio track gets re-encoded rather than the whole video ladder.
    Jobs already running are left alone; queue a full ABR pass later for real multi-bitrate
    quality once there's no urgency."""
    pending_jobs = await db.transcode_queue.find({"status": "pending"}, {"_id": 0}).to_list(5000)
    cancelled = 0
    for j in pending_jobs:
        await db.transcode_queue.update_one({"id": j["id"]}, {"$set": {"status": "cancelled", "finished_at": datetime.now(timezone.utc).isoformat()}})
        cancelled += 1

    queued = 0
    for coll, content_type in ((db.movies, "movie"), (db.episodes, "episode")):
        async for doc in coll.find({"hls_status": {"$ne": "done"}}, {"_id": 0, "id": 1, "storage_type": 1, "storage_path": 1}):
            if doc.get("storage_type") not in ("local", "radarr", "sonarr", "scan") or not doc.get("storage_path"):
                continue
            await _enqueue_transcode(doc["id"], "audio", triggered_by="auto", content_type=content_type)
            queued += 1
    return {"ok": True, "cancelled": cancelled, "queued": queued}


@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}")
async def serve_hls(
    movie_id: str, filename: str,
    auth: Optional[str] = Query(None),
    authorization: Optional[str] = Header(None),
):
    token = None
    if authorization and authorization.lower().startswith("bearer "):
        token = authorization.split(" ", 1)[1].strip()
    elif auth:
        token = auth
    if not token: raise HTTPException(status_code=401, detail="Not authenticated")
    decode_token(token)

    target = (HLS_DIR / movie_id / filename).resolve()
    base = (HLS_DIR / movie_id).resolve()
    if not str(target).startswith(str(base)):
        raise HTTPException(status_code=400, detail="Invalid path")
    if not target.is_file():
        raise HTTPException(status_code=404, detail="HLS file not found")

    # Rewrite playlists to propagate ?auth= to relative URLs (variants + segments)
    if filename.endswith(".m3u8"):
        text = target.read_text(encoding="utf-8", errors="ignore")
        out_lines = []
        for line in text.splitlines():
            stripped = line.strip()
            if stripped and not stripped.startswith("#") and "://" not in stripped:
                sep = "&" if "?" in stripped else "?"
                line = f"{stripped}{sep}auth={token}"
            out_lines.append(line)
        return Response("\n".join(out_lines) + "\n", media_type="application/vnd.apple.mpegurl")

    media_type = "video/mp2t" if filename.endswith(".ts") else "application/octet-stream"
    return FileResponse(str(target), media_type=media_type)


# ============ SUBTITLES ============
@api.get("/movies/{movie_id}/subtitles", response_model=List[Subtitle])
async def list_subs(movie_id: str, user: dict = Depends(get_current_user)):
    docs = await db.subtitles.find({"movie_id": movie_id}, {"_id": 0}).sort("created_at", 1).to_list(20)
    return docs


@api.get("/episodes/{episode_id}/subtitles", response_model=List[Subtitle])
async def list_episode_subs(episode_id: str, user: dict = Depends(get_current_user)):
    docs = await db.subtitles.find({"episode_id": episode_id}, {"_id": 0}).sort("created_at", 1).to_list(20)
    return docs


@api.post("/movies/{movie_id}/subtitles", response_model=Subtitle)
async def upload_sub(
    movie_id: str,
    language: str = Form("en"),
    label: str = Form("English"),
    is_default: bool = Form(False),
    file: UploadFile = File(...),
    user: dict = Depends(require_admin),
):
    if not await db.movies.find_one({"id": movie_id}, {"_id": 1}):
        raise HTTPException(status_code=404, detail="Movie not found")
    raw = await file.read()
    text = raw.decode("utf-8", errors="ignore")
    is_srt = (file.filename or "").lower().endswith(".srt") or "WEBVTT" not in text[:20]
    if is_srt:
        text = srt_to_vtt(text)
    sid = str(uuid.uuid4())
    fname = f"{sid}.vtt"
    (SUBS_DIR / fname).write_text(text, encoding="utf-8")
    sub = Subtitle(id=sid, movie_id=movie_id, language=language, label=label,
                   storage_path=fname, is_default=is_default).model_dump()
    if is_default:
        await db.subtitles.update_many({"movie_id": movie_id}, {"$set": {"is_default": False}})
    await db.subtitles.insert_one(sub)
    return _strip(sub)


@api.delete("/subtitles/{sub_id}")
async def delete_sub(sub_id: str, user: dict = Depends(require_admin)):
    sub = await db.subtitles.find_one({"id": sub_id}, {"_id": 0})
    if not sub: raise HTTPException(status_code=404, detail="Subtitle not found")
    try: (SUBS_DIR / sub["storage_path"]).unlink(missing_ok=True)
    except Exception: pass
    await db.subtitles.delete_one({"id": sub_id})
    return {"ok": True}


@api.get("/subtitles/{sub_id}/file")
async def serve_sub(
    sub_id: str,
    auth: Optional[str] = Query(None),
    authorization: Optional[str] = Header(None),
):
    token = None
    if authorization and authorization.lower().startswith("bearer "):
        token = authorization.split(" ", 1)[1].strip()
    elif auth:
        token = auth
    if not token: raise HTTPException(status_code=401, detail="Not authenticated")
    decode_token(token)
    sub = await db.subtitles.find_one({"id": sub_id}, {"_id": 0})
    if not sub: raise HTTPException(status_code=404, detail="Subtitle not found")
    return FileResponse(str(SUBS_DIR / sub["storage_path"]), media_type="text/vtt")


# ============ WATCHLIST ============
@api.get("/watchlist", response_model=List[Movie])
async def get_watchlist(profile: dict = Depends(get_active_profile)):
    items = await db.watchlist.find({"profile_id": profile["id"]}, {"_id": 0}).sort("added_at", -1).to_list(500)
    movie_ids = [i["movie_id"] for i in items]
    if not movie_ids: return []
    movies = await db.movies.find({"id": {"$in": movie_ids}}, {"_id": 0}).to_list(500)
    by_id = {m["id"]: m for m in movies}
    return [by_id[mid] for mid in movie_ids if mid in by_id]


@api.post("/watchlist/{movie_id}")
async def add_watchlist(movie_id: str, profile: dict = Depends(get_active_profile)):
    if not await db.movies.find_one({"id": movie_id}, {"_id": 1}):
        raise HTTPException(status_code=404, detail="Movie not found")
    item = WatchlistItem(profile_id=profile["id"], user_id=profile["user_id"], movie_id=movie_id).model_dump()
    try: await db.watchlist.insert_one(item)
    except Exception: pass
    return {"ok": True}


@api.delete("/watchlist/{movie_id}")
async def remove_watchlist(movie_id: str, profile: dict = Depends(get_active_profile)):
    await db.watchlist.delete_one({"profile_id": profile["id"], "movie_id": movie_id})
    return {"ok": True}


# ============ PROGRESS ============
@api.post("/progress")
async def upsert_progress(payload: ProgressUpsert, profile: dict = Depends(get_active_profile)):
    now = datetime.now(timezone.utc).isoformat()
    await db.progress.update_one(
        {"profile_id": profile["id"], "movie_id": payload.movie_id},
        {"$set": {
            "profile_id": profile["id"], "user_id": profile["user_id"],
            "movie_id": payload.movie_id,
            "position_seconds": payload.position_seconds,
            "duration_seconds": payload.duration_seconds,
            "updated_at": now,
        }}, upsert=True,
    )
    # Trakt scrobble hook (fire and forget)
    movie = await db.movies.find_one({"id": payload.movie_id}, {"_id": 0, "tmdb_id": 1})
    if movie:
        asyncio.create_task(_trakt_scrobble_movie(profile["user_id"], movie, payload.position_seconds, payload.duration_seconds))
    return {"ok": True}


@api.get("/progress/continue")
async def continue_watching(profile: dict = Depends(get_active_profile)):
    rows = await db.progress.find({"profile_id": profile["id"]}, {"_id": 0}).sort("updated_at", -1).to_list(50)
    rows = [r for r in rows if r.get("duration_seconds", 0) == 0 or
            r["position_seconds"] / max(r["duration_seconds"], 1) < 0.95]
    if not rows: return []
    movie_ids = [r["movie_id"] for r in rows]
    movies = await db.movies.find({"id": {"$in": movie_ids}}, {"_id": 0}).to_list(50)
    by_id = {m["id"]: m for m in movies}
    return [{**by_id[r["movie_id"]], "progress": r} for r in rows if r["movie_id"] in by_id]


@api.get("/progress/{movie_id}")
async def get_progress(movie_id: str, profile: dict = Depends(get_active_profile)):
    p = await db.progress.find_one({"profile_id": profile["id"], "movie_id": movie_id}, {"_id": 0})
    return p or {"position_seconds": 0, "duration_seconds": 0}


# ============ REQUESTS ============
@api.post("/requests", response_model=MovieRequest)
async def submit_request(payload: RequestCreate, user: dict = Depends(get_current_user)):
    req = MovieRequest(user_id=user["id"], user_name=user.get("name", ""),
                       title=payload.title, year=payload.year, notes=payload.notes)
    doc = req.model_dump()
    await db.requests.insert_one(doc)
    return _strip(doc)


@api.get("/requests/mine", response_model=List[MovieRequest])
async def my_requests(user: dict = Depends(get_current_user)):
    return await db.requests.find({"user_id": user["id"]}, {"_id": 0}).sort("created_at", -1).to_list(200)


@api.get("/requests", response_model=List[MovieRequest])
async def all_requests(user: dict = Depends(require_admin)):
    return await db.requests.find({}, {"_id": 0}).sort("created_at", -1).to_list(500)


@api.patch("/requests/{request_id}", response_model=MovieRequest)
async def update_request(request_id: str, payload: RequestUpdate, user: dict = Depends(require_admin)):
    res = await db.requests.find_one_and_update(
        {"id": request_id}, {"$set": {"status": payload.status}},
        projection={"_id": 0}, return_document=True,
    )
    if not res: raise HTTPException(status_code=404, detail="Request not found")
    return res


@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."""
    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)
    # can't both pass a check-then-act race and each add the content to Radarr/Sonarr.
    req = await db.requests.find_one_and_update(
        {"id": request_id, "status": "pending"}, {"$set": {"status": "approving"}},
        projection={"_id": 0}, return_document=False,
    )
    if not req:
        existing = await db.requests.find_one({"id": request_id}, {"_id": 0, "status": 1})
        if not existing: raise HTTPException(status_code=404, detail="Request not found")
        raise HTTPException(status_code=409, detail=f"Request is already {existing['status']}")
    s = await get_settings()

    try:
        if payload.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")
            try:
                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 not match:
                raise HTTPException(status_code=404, detail="No match found in Radarr")
            try:
                added = await asyncio.to_thread(radarr_client.add_movie, s["radarr_url"], s["radarr_api_key"], match["tmdb_id"], match["title"], match.get("year"))
            except Exception as e:
                raise HTTPException(status_code=502, detail=f"Radarr add error: {e}")
            updates = {"status": "approved", "content_type": "movie", "radarr_id": added["radarr_id"],
                       "matched_title": added.get("title"), "matched_year": added.get("year")}
        else:
            if not s.get("sonarr_url") or not s.get("sonarr_api_key"):
                raise HTTPException(status_code=400, detail="Sonarr not configured")
            try:
                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 not match:
                raise HTTPException(status_code=404, detail="No match found in Sonarr")
            try:
                added = await asyncio.to_thread(sonarr_client.add_series, s["sonarr_url"], s["sonarr_api_key"], match["tvdb_id"], match["title"], match.get("year"))
            except Exception as e:
                raise HTTPException(status_code=502, detail=f"Sonarr add error: {e}")
            updates = {"status": "approved", "content_type": "tv", "sonarr_id": added["sonarr_id"],
                       "matched_title": added.get("title"), "matched_year": added.get("year")}
    except HTTPException:
        # Release the claim so a fixable failure (e.g. Radarr temporarily down) can be retried,
        # instead of leaving the request stuck in "approving" forever.
        await db.requests.update_one({"id": request_id}, {"$set": {"status": "pending"}})
        raise

    res = await db.requests.find_one_and_update(
        {"id": request_id}, {"$set": updates}, projection={"_id": 0}, return_document=True,
    )
    return res


@api.get("/requests/{request_id}/status")
async def request_download_status(request_id: str, user: dict = Depends(require_admin)):
    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 req.get("radarr_id") and s.get("radarr_url") and s.get("radarr_api_key"):
            return await asyncio.to_thread(radarr_client.movie_status, s["radarr_url"], s["radarr_api_key"], req["radarr_id"])
        if req.get("sonarr_id") and s.get("sonarr_url") and s.get("sonarr_api_key"):
            return await asyncio.to_thread(sonarr_client.series_status, s["sonarr_url"], s["sonarr_api_key"], req["sonarr_id"])
    except Exception as e:
        return {"state": "unknown", "error": str(e)}
    return {"state": req.get("status", "pending")}


# ============ SETTINGS (admin) ============
@api.get("/settings", response_model=AppSettingsPublic)
async def read_settings(user: dict = Depends(require_admin)):
    s = await get_settings()
    return AppSettingsPublic(
        tmdb_configured=bool(s.get("tmdb_api_key")),
        radarr_configured=bool(s.get("radarr_api_key") and s.get("radarr_url")),
        sonarr_configured=bool(s.get("sonarr_api_key") and s.get("sonarr_url")),
        opensubs_configured=bool(s.get("opensubs_api_key")),
        trakt_configured=bool(s.get("trakt_client_id") and s.get("trakt_client_secret")),
        tmdb_api_key=s.get("tmdb_api_key", ""),
        radarr_url=s.get("radarr_url", ""),
        radarr_api_key=s.get("radarr_api_key", ""),
        sonarr_url=s.get("sonarr_url", ""),
        sonarr_api_key=s.get("sonarr_api_key", ""),
        opensubs_api_key=s.get("opensubs_api_key", ""),
        trakt_client_id=s.get("trakt_client_id", ""),
        trakt_client_secret=s.get("trakt_client_secret", ""),
        auto_transcode=s.get("auto_transcode", "off"),
        queue_paused=bool(s.get("queue_paused", False)),
    )


@api.put("/settings", response_model=AppSettingsPublic)
async def update_settings(payload: AppSettings, user: dict = Depends(require_admin)):
    doc = payload.model_dump()
    doc["id"] = "app"
    await db.settings.update_one({"id": "app"}, {"$set": doc}, upsert=True)
    return AppSettingsPublic(
        tmdb_configured=bool(doc.get("tmdb_api_key")),
        radarr_configured=bool(doc.get("radarr_api_key") and doc.get("radarr_url")),
        sonarr_configured=bool(doc.get("sonarr_api_key") and doc.get("sonarr_url")),
        opensubs_configured=bool(doc.get("opensubs_api_key")),
        trakt_configured=bool(doc.get("trakt_client_id") and doc.get("trakt_client_secret")),
        tmdb_api_key=doc.get("tmdb_api_key", ""),
        radarr_url=doc.get("radarr_url", ""),
        radarr_api_key=doc.get("radarr_api_key", ""),
        sonarr_url=doc.get("sonarr_url", ""),
        sonarr_api_key=doc.get("sonarr_api_key", ""),
        opensubs_api_key=doc.get("opensubs_api_key", ""),
        trakt_client_id=doc.get("trakt_client_id", ""),
        trakt_client_secret=doc.get("trakt_client_secret", ""),
        auto_transcode=doc.get("auto_transcode", "off"),
        queue_paused=bool(doc.get("queue_paused", False)),
    )


# ============ TMDB (admin) ============
@api.get("/tmdb/search", response_model=List[TMDBSearchResult])
async def tmdb_search(q: str, user: dict = Depends(require_admin)):
    s = await get_settings()
    key = s.get("tmdb_api_key", "")
    if not key: raise HTTPException(status_code=400, detail="TMDB API key not configured")
    try:
        results = await asyncio.to_thread(tmdb_client.search_movies, key, q)
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"TMDB error: {e}")
    return results


@api.get("/tmdb/movie/{tmdb_id}")
async def tmdb_movie(tmdb_id: int, user: dict = Depends(require_admin)):
    s = await get_settings()
    key = s.get("tmdb_api_key", "")
    if not key: raise HTTPException(status_code=400, detail="TMDB API key not configured")
    try:
        return await asyncio.to_thread(tmdb_client.get_movie, key, tmdb_id)
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"TMDB error: {e}")


# ============ RADARR (admin) ============
@api.post("/radarr/test")
async def radarr_test(user: dict = Depends(require_admin)):
    s = await get_settings()
    if not s.get("radarr_url") or not s.get("radarr_api_key"):
        raise HTTPException(status_code=400, detail="Radarr not configured")
    ok = await asyncio.to_thread(radarr_client.test_connection, s["radarr_url"], s["radarr_api_key"])
    return {"ok": ok}


@api.get("/radarr/movies", response_model=List[RadarrMovieDTO])
async def radarr_list(user: dict = Depends(require_admin)):
    s = await get_settings()
    if not s.get("radarr_url") or not s.get("radarr_api_key"):
        raise HTTPException(status_code=400, detail="Radarr not configured")
    try:
        return await asyncio.to_thread(radarr_client.list_movies, s["radarr_url"], s["radarr_api_key"])
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"Radarr error: {e}")


@api.post("/radarr/import")
async def radarr_import(payload: dict, user: dict = Depends(require_admin)):
    """Body: {radarr_ids: [int]}"""
    radarr_ids = payload.get("radarr_ids") or []
    if not radarr_ids:
        raise HTTPException(status_code=400, detail="No radarr_ids provided")
    s = await get_settings()
    if not s.get("radarr_url") or not s.get("radarr_api_key"):
        raise HTTPException(status_code=400, detail="Radarr not configured")
    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
        if await db.movies.find_one({"radarr_id": m["radarr_id"]}, {"_id": 1}): continue
        movie = Movie(
            title=m["title"], description=m.get("overview", ""),
            year=m.get("year") or 2024, rating="NR",
            genres=[], cast=[], director="",
            poster_url=m.get("poster_url", ""),
            backdrop_url=m.get("poster_url", ""),
            video_url="", storage_type="radarr",
            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")
        # Auto-fetch English subtitle (fire and forget)
        asyncio.create_task(_auto_fetch_subtitle("movie", movie["id"], movie.get("tmdb_id")))
        created += 1
    return {"ok": True, "imported": created}


# ============ SHOWS (TV) ============
@api.get("/shows", response_model=List[Show])
async def list_shows(profile: dict = Depends(get_active_profile)):
    query = _movie_filter_for_profile(profile)
    docs = await db.shows.find(query, {"_id": 0}).sort("created_at", -1).to_list(500)
    return docs


@api.get("/shows/featured", response_model=Show)
async def show_featured(profile: dict = Depends(get_active_profile)):
    base = _movie_filter_for_profile(profile)
    doc = await db.shows.find_one({**base, "featured": True}, {"_id": 0})
    if not doc:
        doc = await db.shows.find_one(base, {"_id": 0})
    if not doc:
        raise HTTPException(status_code=404, detail="No shows")
    return doc


@api.get("/shows/{show_id}", response_model=Show)
async def get_show(show_id: str, user: dict = Depends(get_current_user)):
    doc = await db.shows.find_one({"id": show_id}, {"_id": 0})
    if not doc: raise HTTPException(status_code=404, detail="Show not found")
    return doc


@api.post("/shows", response_model=Show)
async def create_show(payload: ShowCreate, user: dict = Depends(require_admin)):
    doc = Show(**payload.model_dump()).model_dump()
    await db.shows.insert_one(doc)
    return _strip(doc)


@api.patch("/shows/{show_id}", response_model=Show)
async def update_show(show_id: str, payload: ShowUpdate, user: dict = Depends(require_admin)):
    updates = {k: v for k, v in payload.model_dump().items() if v is not None}
    if not updates: raise HTTPException(status_code=400, detail="No fields")
    res = await db.shows.find_one_and_update({"id": show_id}, {"$set": updates}, projection={"_id": 0}, return_document=True)
    if not res: raise HTTPException(status_code=404, detail="Not found")
    return res


@api.delete("/shows/{show_id}")
async def delete_show(show_id: str, user: dict = Depends(require_admin)):
    await db.shows.delete_one({"id": show_id})
    await db.episodes.delete_many({"show_id": show_id})
    await db.episode_progress.delete_many({"show_id": show_id})
    return {"ok": True}


@api.get("/shows/{show_id}/episodes", response_model=List[Episode])
async def list_episodes(show_id: str, include_hidden: bool = False, user: dict = Depends(get_current_user)):
    q: dict = {"show_id": show_id}
    if not include_hidden or not user.get("is_admin"):
        q["hidden"] = {"$ne": True}
    docs = await db.episodes.find(q, {"_id": 0}).sort([("season_number", 1), ("episode_number", 1)]).to_list(2000)
    return docs


@api.get("/episodes/{episode_id}", response_model=Episode)
async def get_episode(episode_id: str, user: dict = Depends(get_current_user)):
    doc = await db.episodes.find_one({"id": episode_id}, {"_id": 0})
    if not doc: raise HTTPException(status_code=404, detail="Episode not found")
    return doc


@api.post("/episodes", response_model=Episode)
async def create_episode(payload: EpisodeCreate, user: dict = Depends(require_admin)):
    if not await db.shows.find_one({"id": payload.show_id}, {"_id": 1}):
        raise HTTPException(status_code=404, detail="Show not found")
    doc = Episode(**payload.model_dump()).model_dump()
    await db.episodes.insert_one(doc)
    # Refresh show counts
    seasons = await db.episodes.distinct("season_number", {"show_id": payload.show_id})
    ecount = await db.episodes.count_documents({"show_id": payload.show_id})
    await db.shows.update_one({"id": payload.show_id}, {"$set": {"season_count": len(seasons), "episode_count": ecount}})
    return _strip(doc)


# ============ Episode bulk actions (admin) ============
@api.post("/episodes/bulk-transcode")
async def bulk_transcode_eps(payload: dict, user: dict = Depends(require_admin)):
    ids = payload.get("episode_ids") or []
    quality = payload.get("quality", "abr")
    if quality not in ("quick", "abr", "audio"): raise HTTPException(status_code=400, detail="Invalid quality")
    queued = 0
    for eid in ids:
        ep = await db.episodes.find_one({"id": eid}, {"_id": 0})
        if not ep or ep.get("hls_status") in ("running", "pending"): continue
        if ep.get("storage_type") not in ("local", "sonarr", "scan") or not ep.get("storage_path"): continue
        await _enqueue_transcode(eid, quality, triggered_by="manual", content_type="episode")
        queued += 1
    return {"ok": True, "queued": queued}


@api.post("/episodes/bulk-hide")
async def bulk_hide_eps(payload: dict, user: dict = Depends(require_admin)):
    ids = payload.get("episode_ids") or []
    hidden = bool(payload.get("hidden", True))
    res = await db.episodes.update_many({"id": {"$in": ids}}, {"$set": {"hidden": hidden}})
    return {"ok": True, "modified": res.modified_count}


@api.post("/episodes/bulk-delete")
async def bulk_delete_eps(payload: dict, user: dict = Depends(require_admin)):
    ids = payload.get("episode_ids") or []
    import shutil as _sh
    for eid in ids:
        _sh.rmtree(HLS_DIR / eid, ignore_errors=True)
    res = await db.episodes.delete_many({"id": {"$in": ids}})
    await db.episode_progress.delete_many({"episode_id": {"$in": ids}})
    await db.subtitles.delete_many({"episode_id": {"$in": ids}})
    return {"ok": True, "deleted": res.deleted_count}


@api.post("/episodes/{episode_id}/transcode")
async def transcode_episode(episode_id: str, payload: Optional[dict] = None, user: dict = Depends(require_admin)):
    quality = (payload or {}).get("quality", "abr")
    if quality not in ("quick", "abr", "audio"): raise HTTPException(status_code=400, detail="Invalid quality")
    ep = await db.episodes.find_one({"id": episode_id}, {"_id": 0})
    if not ep: raise HTTPException(status_code=404, detail="Episode not found")
    if ep.get("storage_type") not in ("local", "sonarr", "scan") or not ep.get("storage_path"):
        raise HTTPException(status_code=400, detail="Only local/sonarr/scanned episodes can be transcoded")
    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")
    return {"ok": True, "job_id": job["id"]}


@api.get("/stream/episode/{episode_id}")
async def stream_episode(
    episode_id: str, request: Request,
    auth: Optional[str] = Query(None),
    authorization: Optional[str] = Header(None),
):
    token = authorization.split(" ", 1)[1].strip() if authorization and authorization.lower().startswith("bearer ") else auth
    if not token: raise HTTPException(status_code=401, detail="Not authenticated")
    decode_token(token)
    ep = await db.episodes.find_one({"id": episode_id}, {"_id": 0})
    if not ep: raise HTTPException(status_code=404, detail="Episode not found")
    if ep.get("storage_type") not in ("local", "sonarr", "scan") or not ep.get("storage_path"):
        raise HTTPException(status_code=400, detail="Episode has no local file")
    file_path = Path(ep["storage_path"]) if ep["storage_type"] in ("sonarr", "scan") else VIDEOS_DIR / ep["storage_path"]
    if not file_path.is_file(): raise HTTPException(status_code=404, detail="File missing")
    file_size = file_path.stat().st_size
    content_type = mimetypes.guess_type(str(file_path))[0] or "video/mp4"
    range_header = request.headers.get("range") or request.headers.get("Range")
    if range_header and range_header.startswith("bytes="):
        try:
            start_str, end_str = range_header.replace("bytes=", "").split("-")
            start = int(start_str) if start_str else 0
            end = int(end_str) if end_str else file_size - 1
        except ValueError:
            raise HTTPException(status_code=416, detail="Invalid range")
        end = min(end, file_size - 1); length = end - start + 1
        def iter_file():
            with file_path.open("rb") as f:
                f.seek(start); remaining = length
                while remaining > 0:
                    chunk = f.read(min(CHUNK_SIZE, remaining))
                    if not chunk: break
                    remaining -= len(chunk); yield chunk
        return StreamingResponse(iter_file(), status_code=206, media_type=content_type, headers={
            "Content-Range": f"bytes {start}-{end}/{file_size}",
            "Accept-Ranges": "bytes", "Content-Length": str(length), "Content-Type": content_type,
        })
    return FileResponse(str(file_path), media_type=content_type, headers={"Accept-Ranges": "bytes"})


# Episode progress
@api.post("/progress/episode")
async def upsert_episode_progress(payload: EpisodeProgressUpsert, profile: dict = Depends(get_active_profile)):
    ep = await db.episodes.find_one({"id": payload.episode_id}, {"_id": 0})
    if not ep: raise HTTPException(status_code=404, detail="Episode not found")
    now = datetime.now(timezone.utc).isoformat()
    await db.episode_progress.update_one(
        {"profile_id": profile["id"], "episode_id": payload.episode_id},
        {"$set": {
            "profile_id": profile["id"], "user_id": profile["user_id"],
            "episode_id": payload.episode_id, "show_id": ep["show_id"],
            "position_seconds": payload.position_seconds,
            "duration_seconds": payload.duration_seconds,
            "updated_at": now,
        }}, upsert=True,
    )
    # Trakt scrobble hook (fire and forget)
    asyncio.create_task(_trakt_scrobble_episode(profile["user_id"], ep, payload.position_seconds, payload.duration_seconds))
    return {"ok": True}


@api.get("/progress/episode/{episode_id}")
async def get_episode_progress(episode_id: str, profile: dict = Depends(get_active_profile)):
    p = await db.episode_progress.find_one({"profile_id": profile["id"], "episode_id": episode_id}, {"_id": 0})
    return p or {"position_seconds": 0, "duration_seconds": 0}


@api.get("/progress/episodes/continue")
async def continue_watching_episodes(profile: dict = Depends(get_active_profile)):
    rows = await db.episode_progress.find({"profile_id": profile["id"]}, {"_id": 0}).sort("updated_at", -1).to_list(30)
    rows = [r for r in rows if r.get("duration_seconds", 0) == 0 or r["position_seconds"] / max(r["duration_seconds"], 1) < 0.95]
    if not rows: return []
    ep_ids = [r["episode_id"] for r in rows]
    eps = await db.episodes.find({"id": {"$in": ep_ids}}, {"_id": 0}).to_list(30)
    ep_by_id = {e["id"]: e for e in eps}
    show_ids = list({e["show_id"] for e in eps})
    shows = await db.shows.find({"id": {"$in": show_ids}}, {"_id": 0}).to_list(30)
    show_by_id = {s["id"]: s for s in shows}
    out = []
    for r in rows:
        e = ep_by_id.get(r["episode_id"]); s = show_by_id.get(e["show_id"]) if e else None
        if e and s: out.append({"episode": e, "show": s, "progress": r})
    return out


# ============ SONARR ============
@api.post("/sonarr/test")
async def sonarr_test(user: dict = Depends(require_admin)):
    s = await get_settings()
    if not s.get("sonarr_url") or not s.get("sonarr_api_key"):
        raise HTTPException(status_code=400, detail="Sonarr not configured")
    ok = await asyncio.to_thread(sonarr_client.test_connection, s["sonarr_url"], s["sonarr_api_key"])
    return {"ok": ok}


@api.get("/sonarr/series", response_model=List[SonarrSeriesDTO])
async def sonarr_series(user: dict = Depends(require_admin)):
    s = await get_settings()
    if not s.get("sonarr_url") or not s.get("sonarr_api_key"):
        raise HTTPException(status_code=400, detail="Sonarr not configured")
    try:
        return await asyncio.to_thread(sonarr_client.list_series, s["sonarr_url"], s["sonarr_api_key"])
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"Sonarr error: {e}")


@api.post("/sonarr/import")
async def sonarr_import(payload: dict, user: dict = Depends(require_admin)):
    """Body: {sonarr_ids: [int]} — imports full series + all episodes with files."""
    sonarr_ids = payload.get("sonarr_ids") or []
    if not sonarr_ids: raise HTTPException(status_code=400, detail="No sonarr_ids")
    s = await get_settings()
    if not s.get("sonarr_url") or not s.get("sonarr_api_key"):
        raise HTTPException(status_code=400, detail="Sonarr not configured")
    all_series = await asyncio.to_thread(sonarr_client.list_series, s["sonarr_url"], s["sonarr_api_key"])
    by_id = {x["sonarr_id"]: x for x in all_series}
    created_shows = 0; created_eps = 0
    auto = s.get("auto_transcode", "off")
    for sid in sonarr_ids:
        m = by_id.get(int(sid))
        if not m: continue
        existing = await db.shows.find_one({"sonarr_id": m["sonarr_id"]}, {"_id": 0})
        if existing:
            show_id = existing["id"]
        else:
            show_doc = Show(
                title=m["title"], description=m.get("overview", ""),
                year=m.get("year") or 2024, rating="NR",
                poster_url=m.get("poster_url", ""), backdrop_url=m.get("poster_url", ""),
                sonarr_id=m["sonarr_id"], tvdb_id=m.get("tvdb_id"), tmdb_id=m.get("tmdb_id"),
            ).model_dump()
            await db.shows.insert_one(show_doc)
            show_id = show_doc["id"]; created_shows += 1

        # Fetch episodes
        try:
            eps = await asyncio.to_thread(sonarr_client.list_episodes, s["sonarr_url"], s["sonarr_api_key"], m["sonarr_id"])
        except Exception:
            continue
        for ep in eps:
            if not ep.get("has_file") or not ep.get("file_path"): continue
            if await db.episodes.find_one({"sonarr_episode_id": ep["sonarr_episode_id"]}, {"_id": 1}): continue
            edoc = Episode(
                show_id=show_id, season_number=ep["season_number"], episode_number=ep["episode_number"],
                title=ep.get("title", ""), description=ep.get("description", ""),
                air_date=ep.get("air_date", ""), duration_minutes=ep.get("duration_minutes", 0),
                storage_type="sonarr", storage_path=ep["file_path"],
                sonarr_episode_id=ep["sonarr_episode_id"],
            ).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")
            # 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"):
                asyncio.create_task(_auto_fetch_subtitle(
                    "episode", edoc["id"], show_doc["tmdb_id"],
                    season=ep["season_number"], episode=ep["episode_number"],
                ))
        # Update counts
        seasons = await db.episodes.distinct("season_number", {"show_id": show_id})
        ecount = await db.episodes.count_documents({"show_id": show_id})
        await db.shows.update_one({"id": show_id}, {"$set": {"season_count": len(seasons), "episode_count": ecount}})
    return {"ok": True, "shows_imported": created_shows, "episodes_imported": created_eps}


# ============ LIBRARY SCAN (filesystem — separate from Radarr/Sonarr) ============
@api.get("/scan/movies")
async def scan_movies(user: dict = Depends(require_admin)):
    found = await asyncio.to_thread(library_scan.scan_movies)
    imported_paths = set(await db.movies.distinct("storage_path", {"storage_type": {"$in": ["radarr", "scan", "local"]}}))
    for item in found:
        item["already_imported"] = item["storage_path"] in imported_paths
    return found


@api.get("/scan/tv")
async def scan_tv(user: dict = Depends(require_admin)):
    found = await asyncio.to_thread(library_scan.scan_tv)
    imported_paths = set(await db.episodes.distinct("storage_path", {"storage_type": {"$in": ["sonarr", "scan", "local"]}}))
    for item in found:
        item["already_imported"] = item["storage_path"] in imported_paths
    return found


@api.get("/scan/music")
async def scan_music(user: dict = Depends(require_admin)):
    found = await asyncio.to_thread(library_scan.scan_music)
    imported_paths = set(await db.tracks.distinct("storage_path"))
    for item in found:
        item["already_imported"] = item["storage_path"] in imported_paths
    return found


@api.post("/scan/import")
async def scan_import(payload: dict, user: dict = Depends(require_admin)):
    """Body: {kind: "movies"|"tv"|"music", items: [<scan result dicts as returned by GET /scan/*>]}"""
    kind = payload.get("kind")
    items = payload.get("items") or []
    if kind not in ("movies", "tv", "music"):
        raise HTTPException(status_code=400, detail="kind must be movies, tv, or music")
    if not items:
        raise HTTPException(status_code=400, detail="No items provided")

    s = await get_settings()
    auto = s.get("auto_transcode", "off")
    created = 0

    if kind == "movies":
        for it in items:
            if await db.movies.find_one({"storage_path": it["storage_path"]}, {"_id": 1}):
                continue
            movie = Movie(
                title=it.get("title") or "Untitled", year=it.get("year") or 2024,
                storage_type="scan", storage_path=it["storage_path"],
            ).model_dump()
            await db.movies.insert_one(movie)
            if auto in ("quick", "abr"):
                await _enqueue_transcode(movie["id"], auto, triggered_by="auto")
            created += 1

    elif kind == "tv":
        shows_by_title: Dict[str, str] = {}
        for it in items:
            if await db.episodes.find_one({"storage_path": it["storage_path"]}, {"_id": 1}):
                continue
            show_title = it.get("show_title") or "Untitled"
            show_id = shows_by_title.get(show_title)
            if not show_id:
                existing = await db.shows.find_one({"title": show_title}, {"_id": 0})
                if existing:
                    show_id = existing["id"]
                else:
                    show_doc = Show(title=show_title, year=2024, rating="NR").model_dump()
                    await db.shows.insert_one(show_doc)
                    show_id = show_doc["id"]
                shows_by_title[show_title] = show_id
            edoc = Episode(
                show_id=show_id, season_number=it.get("season_number") or 1,
                episode_number=it.get("episode_number") or 0,
                title=it.get("title") or "", storage_type="scan", storage_path=it["storage_path"],
            ).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")
            created += 1
        for show_id in set(shows_by_title.values()):
            seasons = await db.episodes.distinct("season_number", {"show_id": show_id})
            ecount = await db.episodes.count_documents({"show_id": show_id})
            await db.shows.update_one({"id": show_id}, {"$set": {"season_count": len(seasons), "episode_count": ecount}})

    elif kind == "music":
        for it in items:
            if await db.tracks.find_one({"storage_path": it["storage_path"]}, {"_id": 1}):
                continue
            tags = await asyncio.to_thread(library_scan.tags_for_file, it["storage_path"])
            track = Track(
                title=tags.get("title") or it.get("title") or "Untitled",
                artist=tags.get("artist") or it.get("artist") or "Unknown Artist",
                album=tags.get("album") or it.get("album") or "Unknown Album",
                track_number=tags.get("track_number") or it.get("track_number"),
                duration_seconds=round(tags.get("duration") or 0, 1), storage_path=it["storage_path"],
            ).model_dump()
            await db.tracks.insert_one(track)
            created += 1

    return {"ok": True, "imported": created}


# ============ LIBRARY CLEANUP (admin) ============
_cleanup_scan_status = {"running": False, "findings": None, "error": "", "scan_id": 0}


async def _run_cleanup_scan(scan_id: int):
    try:
        findings = await library_cleanup.scan(db, VIDEOS_DIR, HLS_DIR)
        if _cleanup_scan_status["scan_id"] == scan_id:
            _cleanup_scan_status["findings"] = findings
    except Exception as e:
        if _cleanup_scan_status["scan_id"] == scan_id:
            _cleanup_scan_status["error"] = str(e)
    finally:
        if _cleanup_scan_status["scan_id"] == scan_id:
            _cleanup_scan_status["running"] = False


@api.post("/library/cleanup/scan/start")
async def library_cleanup_scan_start(user: dict = Depends(require_admin)):
    if _cleanup_scan_status["running"]:
        raise HTTPException(status_code=409, detail="Scan already running")
    _cleanup_scan_status["scan_id"] += 1
    _cleanup_scan_status.update(running=True, findings=None, error="")
    asyncio.create_task(_run_cleanup_scan(_cleanup_scan_status["scan_id"]))
    return {"ok": True}


@api.get("/library/cleanup/scan/status")
async def library_cleanup_scan_status(user: dict = Depends(require_admin)):
    return _cleanup_scan_status


@api.post("/library/cleanup/scan/stop")
async def library_cleanup_scan_stop(user: dict = Depends(require_admin)):
    # The already-dispatched file checks finish naturally in the background (there's no clean
    # way to interrupt a ThreadPoolExecutor mid-flight), but bumping scan_id means whenever they
    # do finish, _run_cleanup_scan sees a stale id and discards the result instead of surfacing it.
    _cleanup_scan_status["scan_id"] += 1
    _cleanup_scan_status["running"] = False
    return {"ok": True}


@api.post("/library/cleanup/fix")
async def library_cleanup_fix(payload: dict, user: dict = Depends(require_admin)):
    """Body: {category: str, items: [<finding dicts as returned by GET /library/cleanup/scan>]}"""
    category = payload.get("category")
    items = payload.get("items") or []
    if not items:
        raise HTTPException(status_code=400, detail="No items provided")
    fixed = 0
    for item in items:
        try:
            await library_cleanup.fix_one(db, VIDEOS_DIR, HLS_DIR, category, item)
            fixed += 1
        except Exception as e:
            logger.warning(f"Library cleanup fix failed for {category}/{item.get('id')}: {e}")
    return {"ok": True, "fixed": fixed}


# ============ MUSIC ============
@api.get("/tracks", response_model=List[Track])
async def list_tracks(user: dict = Depends(get_current_user)):
    docs = await db.tracks.find({}, {"_id": 0}).sort([("artist", 1), ("album", 1), ("track_number", 1)]).to_list(10000)
    return docs


@api.patch("/tracks/{track_id}", response_model=Track)
async def update_track(track_id: str, payload: TrackUpdate, user: dict = Depends(require_admin)):
    updates = {k: v for k, v in payload.model_dump().items() if v is not None}
    if not updates:
        raise HTTPException(status_code=400, detail="No fields to update")
    res = await db.tracks.find_one_and_update(
        {"id": track_id}, {"$set": updates}, projection={"_id": 0}, return_document=True,
    )
    if not res:
        raise HTTPException(status_code=404, detail="Track not found")
    return res


@api.delete("/tracks/{track_id}")
async def delete_track(track_id: str, user: dict = Depends(require_admin)):
    res = await db.tracks.delete_one({"id": track_id})
    if res.deleted_count == 0:
        raise HTTPException(status_code=404, detail="Track not found")
    return {"ok": True}


_music_enrich_status = {"running": False, "total": 0, "done": 0, "updated": 0, "error": ""}


async def _run_music_enrich():
    _music_enrich_status.update(running=True, total=0, done=0, updated=0, error="")
    try:
        pairs = await db.tracks.aggregate([
            {"$match": {"album_art_url": {"$in": [None, ""]}}},
            {"$group": {"_id": {"artist": "$artist", "album": "$album"}}},
        ]).to_list(5000)
        _music_enrich_status["total"] = len(pairs)
        for p in pairs:
            artist = p["_id"]["artist"]; album = p["_id"]["album"]
            try:
                release = await asyncio.to_thread(musicbrainz_client.lookup_release, artist, album)
                if release and release.get("mbid"):
                    art_url = await asyncio.to_thread(musicbrainz_client.cover_art_url, release["mbid"])
                    if art_url:
                        res = await db.tracks.update_many(
                            {"artist": artist, "album": album, "album_art_url": {"$in": [None, ""]}},
                            {"$set": {"album_art_url": art_url}},
                        )
                        _music_enrich_status["updated"] += res.modified_count
            except Exception as e:
                logger.warning(f"Music enrich failed for {artist} - {album}: {e}")
            _music_enrich_status["done"] += 1
            await asyncio.sleep(1.1)  # MusicBrainz rate limit: 1 req/sec, unauthenticated
    except Exception as e:
        _music_enrich_status["error"] = str(e)
    finally:
        _music_enrich_status["running"] = False


@api.post("/music/enrich")
async def start_music_enrich(user: dict = Depends(require_admin)):
    if _music_enrich_status["running"]:
        raise HTTPException(status_code=409, detail="Enrichment already running")
    asyncio.create_task(_run_music_enrich())
    return {"ok": True}


@api.get("/music/enrich/status")
async def music_enrich_status(user: dict = Depends(require_admin)):
    return _music_enrich_status


@api.get("/stream/track/{track_id}")
async def stream_track(
    track_id: str, request: Request,
    auth: Optional[str] = Query(None),
    authorization: Optional[str] = Header(None),
):
    token = authorization.split(" ", 1)[1].strip() if authorization and authorization.lower().startswith("bearer ") else auth
    if not token: raise HTTPException(status_code=401, detail="Not authenticated")
    decode_token(token)
    track = await db.tracks.find_one({"id": track_id}, {"_id": 0})
    if not track: raise HTTPException(status_code=404, detail="Track not found")
    file_path = Path(track["storage_path"])
    if not file_path.is_file(): raise HTTPException(status_code=404, detail="File missing on disk")

    file_size = file_path.stat().st_size
    _audio_mime_fallback = {".m4a": "audio/mp4", ".flac": "audio/flac", ".ogg": "audio/ogg", ".wav": "audio/wav", ".aac": "audio/aac"}
    content_type = mimetypes.guess_type(str(file_path))[0] or _audio_mime_fallback.get(file_path.suffix.lower(), "audio/mpeg")
    range_header = request.headers.get("range") or request.headers.get("Range")
    if range_header and range_header.startswith("bytes="):
        try:
            start_str, end_str = range_header.replace("bytes=", "").split("-")
            start = int(start_str) if start_str else 0
            end = int(end_str) if end_str else file_size - 1
        except ValueError:
            raise HTTPException(status_code=416, detail="Invalid range")
        if start >= file_size:
            raise HTTPException(status_code=416, detail="Range out of bounds")
        end = min(end, file_size - 1)
        length = end - start + 1

        def iter_file():
            with file_path.open("rb") as f:
                f.seek(start)
                remaining = length
                while remaining > 0:
                    chunk = f.read(min(CHUNK_SIZE, remaining))
                    if not chunk: break
                    remaining -= len(chunk)
                    yield chunk

        return StreamingResponse(iter_file(), status_code=206, media_type=content_type, headers={
            "Content-Range": f"bytes {start}-{end}/{file_size}",
            "Accept-Ranges": "bytes",
            "Content-Length": str(length),
            "Content-Type": content_type,
        })
    return FileResponse(str(file_path), media_type=content_type, headers={"Accept-Ranges": "bytes"})


# ============ TRAKT ============
async def _trakt_get_token(user_id: str) -> Optional[dict]:
    return await db.trakt_tokens.find_one({"user_id": user_id}, {"_id": 0})


async def _trakt_scrobble_movie(user_id: str, movie: dict, position: float, duration: float):
    if not movie.get("tmdb_id") or duration <= 0: return
    tok = await _trakt_get_token(user_id)
    if not tok: return
    s = await get_settings()
    if not s.get("trakt_client_id"): return
    progress = (position / duration) * 100.0
    action = "stop" if progress >= 80 else ("pause" if progress > 0 else "start")
    try:
        await asyncio.to_thread(trakt_client.scrobble_movie, s["trakt_client_id"], tok["access_token"], movie["tmdb_id"], progress, action)
    except Exception as e:
        logger.warning(f"Trakt scrobble movie failed: {e}")


async def _trakt_scrobble_episode(user_id: str, ep: dict, position: float, duration: float):
    if duration <= 0: return
    tok = await _trakt_get_token(user_id)
    if not tok: return
    s = await get_settings()
    if not s.get("trakt_client_id"): return
    show = await db.shows.find_one({"id": ep["show_id"]}, {"_id": 0})
    if not show or (not show.get("tvdb_id") and not show.get("tmdb_id")): return
    progress = (position / duration) * 100.0
    action = "stop" if progress >= 80 else ("pause" if progress > 0 else "start")
    try:
        await asyncio.to_thread(trakt_client.scrobble_episode, s["trakt_client_id"], tok["access_token"],
                                show.get("tvdb_id"), show.get("tmdb_id"),
                                ep["season_number"], ep["episode_number"], progress, action)
    except Exception as e:
        logger.warning(f"Trakt scrobble episode failed: {e}")


@api.post("/trakt/device-code", response_model=TraktDeviceCodeResponse)
async def trakt_device_code(user: dict = Depends(get_current_user)):
    s = await get_settings()
    if not s.get("trakt_client_id"): raise HTTPException(status_code=400, detail="Trakt not configured (admin sets client_id/secret)")
    try:
        return await asyncio.to_thread(trakt_client.device_code, s["trakt_client_id"])
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"Trakt error: {e}")


@api.post("/trakt/poll")
async def trakt_poll(payload: dict, user: dict = Depends(get_current_user)):
    device_code_val = payload.get("device_code")
    if not device_code_val: raise HTTPException(status_code=400, detail="device_code required")
    s = await get_settings()
    if not s.get("trakt_client_id") or not s.get("trakt_client_secret"):
        raise HTTPException(status_code=400, detail="Trakt not configured")
    try:
        tok = await asyncio.to_thread(trakt_client.poll_token, s["trakt_client_id"], s["trakt_client_secret"], device_code_val)
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"Trakt error: {e}")
    if not tok:
        return {"ok": False, "pending": True}
    try:
        us = await asyncio.to_thread(trakt_client.user_settings, s["trakt_client_id"], tok["access_token"])
        username = ((us or {}).get("user") or {}).get("username", "")
    except Exception:
        username = ""
    await db.trakt_tokens.update_one(
        {"user_id": user["id"]},
        {"$set": {"user_id": user["id"], "access_token": tok["access_token"],
                  "refresh_token": tok.get("refresh_token", ""),
                  "expires_in": tok.get("expires_in", 0), "username": username,
                  "created_at": datetime.now(timezone.utc).isoformat()}},
        upsert=True,
    )
    return {"ok": True, "username": username}


@api.get("/trakt/status", response_model=TraktStatus)
async def trakt_status(user: dict = Depends(get_current_user)):
    tok = await _trakt_get_token(user["id"])
    if not tok: return TraktStatus(connected=False)
    return TraktStatus(connected=True, username=tok.get("username", ""))


@api.delete("/trakt/disconnect")
async def trakt_disconnect(user: dict = Depends(get_current_user)):
    await db.trakt_tokens.delete_one({"user_id": user["id"]})
    return {"ok": True}


# ============ OPENSUBTITLES ============
@api.get("/opensubs/search", response_model=List[OpenSubsResult])
async def opensubs_search(tmdb_id: Optional[int] = None, q: Optional[str] = None,
                          languages: str = "en", season: Optional[int] = None, episode: Optional[int] = None,
                          user: dict = Depends(require_admin)):
    s = await get_settings()
    if not s.get("opensubs_api_key"): raise HTTPException(status_code=400, detail="OpenSubs not configured")
    try:
        return await asyncio.to_thread(opensubs_client.search, s["opensubs_api_key"], tmdb_id, q, languages, episode, season)
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"OpenSubs error: {e}")


@api.post("/opensubs/download")
async def opensubs_download(payload: dict, user: dict = Depends(require_admin)):
    """Body: {file_id, movie_id?, episode_id?, language, label}"""
    file_id = payload.get("file_id")
    if not file_id: raise HTTPException(status_code=400, detail="file_id required")
    movie_id = payload.get("movie_id"); episode_id = payload.get("episode_id")
    if not movie_id and not episode_id: raise HTTPException(status_code=400, detail="movie_id or episode_id required")
    lang = payload.get("language", "en"); label = payload.get("label", "English")
    s = await get_settings()
    if not s.get("opensubs_api_key"): raise HTTPException(status_code=400, detail="OpenSubs not configured")
    try:
        link = await asyncio.to_thread(opensubs_client.get_download_link, s["opensubs_api_key"], file_id)
        srt = await asyncio.to_thread(opensubs_client.download_content, link)
        vtt = srt_to_vtt(srt) if "WEBVTT" not in srt[:20] else srt
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"OpenSubs download error: {e}")
    sid = str(uuid.uuid4()); fname = f"{sid}.vtt"
    (SUBS_DIR / fname).write_text(vtt, encoding="utf-8")
    sub_doc = {
        "id": sid, "movie_id": movie_id or "", "episode_id": episode_id or "",
        "language": lang, "label": label, "storage_path": fname, "is_default": False,
        "created_at": datetime.now(timezone.utc).isoformat(),
    }
    await db.subtitles.insert_one(sub_doc)
    _strip(sub_doc)
    return {"ok": True, "subtitle": sub_doc}


# ============ BULK TMDB ENRICHMENT ============
_movie_enrich_status = {"running": False, "total": 0, "done": 0, "enriched": 0, "skipped": 0, "failed": 0, "error": ""}


async def _run_movie_enrich(key: str):
    _movie_enrich_status.update(running=True, total=0, done=0, enriched=0, skipped=0, failed=0, error="")
    try:
        movies = await db.movies.find({}, {"_id": 0}).to_list(20000)
        _movie_enrich_status["total"] = len(movies)
        for m in movies:
            needs = not m.get("cast") or not m.get("director") or not m.get("description") or not m.get("poster_url")
            if not needs:
                _movie_enrich_status["skipped"] += 1
                _movie_enrich_status["done"] += 1
                continue
            try:
                if m.get("tmdb_id"):
                    data = await asyncio.to_thread(tmdb_client.get_movie, key, m["tmdb_id"])
                else:
                    results = await asyncio.to_thread(tmdb_client.search_movies, key, m["title"])
                    if not results:
                        _movie_enrich_status["failed"] += 1
                        _movie_enrich_status["done"] += 1
                        continue
                    data = await asyncio.to_thread(tmdb_client.get_movie, key, results[0]["tmdb_id"])
                updates = {}
                for field in ("description", "poster_url", "backdrop_url", "director", "duration_minutes", "rating"):
                    if data.get(field) and not m.get(field): updates[field] = data[field]
                if data.get("cast") and not m.get("cast"): updates["cast"] = data["cast"]
                if data.get("genres") and not m.get("genres"): updates["genres"] = data["genres"]
                if data.get("tmdb_id"): updates["tmdb_id"] = data["tmdb_id"]
                if updates:
                    await db.movies.update_one({"id": m["id"]}, {"$set": updates})
                    _movie_enrich_status["enriched"] += 1
            except Exception as e:
                logger.warning(f"Enrich failed for {m.get('title')}: {e}")
                _movie_enrich_status["failed"] += 1
            _movie_enrich_status["done"] += 1
    except Exception as e:
        _movie_enrich_status["error"] = str(e)
    finally:
        _movie_enrich_status["running"] = False


@api.post("/tmdb/enrich-all")
async def tmdb_enrich_all(user: dict = Depends(require_admin)):
    """Kicks off a background sweep of movies with missing metadata, filled from TMDB (by title match)."""
    if _movie_enrich_status["running"]:
        raise HTTPException(status_code=409, detail="Enrichment already running")
    s = await get_settings()
    key = s.get("tmdb_api_key", "")
    if not key: raise HTTPException(status_code=400, detail="TMDB not configured")
    asyncio.create_task(_run_movie_enrich(key))
    return {"ok": True}


@api.get("/tmdb/enrich-all/status")
async def tmdb_enrich_all_status(user: dict = Depends(require_admin)):
    return _movie_enrich_status


async def _auto_fetch_subtitle(content_type: str, content_id: str, tmdb_id: Optional[int],
                                season: Optional[int] = None, episode: Optional[int] = None,
                                language: str = "en", label: str = "English"):
    """Best-effort background: pull first OpenSubs result and attach to movie/episode."""
    if not tmdb_id: return
    s = await get_settings()
    key = s.get("opensubs_api_key", "")
    if not key: return
    try:
        results = await asyncio.to_thread(opensubs_client.search, key, tmdb_id, None, language, episode, season)
        if not results: return
        r = results[0]
        link = await asyncio.to_thread(opensubs_client.get_download_link, key, r["file_id"])
        srt = await asyncio.to_thread(opensubs_client.download_content, link)
        vtt = srt_to_vtt(srt) if "WEBVTT" not in srt[:20] else srt
    except Exception as e:
        logger.warning(f"Auto-subtitle failed for {content_type} {content_id}: {e}")
        return
    sid = str(uuid.uuid4()); fname = f"{sid}.vtt"
    (SUBS_DIR / fname).write_text(vtt, encoding="utf-8")
    doc = {
        "id": sid,
        "movie_id": content_id if content_type == "movie" else "",
        "episode_id": content_id if content_type == "episode" else "",
        "language": language, "label": label, "storage_path": fname,
        "is_default": True,
        "created_at": datetime.now(timezone.utc).isoformat(),
    }
    await db.subtitles.insert_one(doc)
    logger.info(f"Auto-attached subtitle for {content_type} {content_id[:8]}")


# ============ SERVICES (admin) ============
@api.get("/services")
async def services_status(user: dict = Depends(require_admin)):
    return await asyncio.to_thread(services_client.list_service_status)


@api.post("/services/{key}/restart")
async def services_restart(key: str, user: dict = Depends(require_admin)):
    ok = await asyncio.to_thread(services_client.restart_service, key)
    if not ok:
        raise HTTPException(status_code=404, detail="Unknown service")
    return {"ok": True}


# ============ HEALTH ============
@api.get("/")
async def root():
    return {"app": "StreamHoard", "status": "ok"}


app.include_router(api)
app.add_middleware(
    CORSMiddleware,
    allow_credentials=True,
    allow_origins=os.environ.get("CORS_ORIGINS", "*").split(","),
    allow_methods=["*"],
    allow_headers=["*"],
    expose_headers=["Content-Range", "Accept-Ranges", "Content-Length"],
)