siftlode/backend/app/routes/plex/collections.py

"""Plex collections: browse mirrored collections + the admin manual-collection editor."""

from collections.abc import Callable
from datetime import date, datetime
from typing import Any, NamedTuple

from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app.auth import admin_user, current_user
from app.db import get_db
from app.models import (
    PlexCollection,
    PlexItem,
    PlexLibrary,
    PlexShow,
    PlexState,
    User,
)
from app.plex import rollup
from app.plex import sync as plex_sync
from app.plex.cards import movie_card, show_card
from app.plex.client import PlexClient, PlexError, PlexNotConfigured
from app.plex.lookups import by_rating_key
from app.plex.query import content_libs, like_escape, runtime_by_key, top_art_keys

# The most-restrictive → least ordering used to pick a collection's effective content rating from its
# members (Plex shows one cert on the collection; we don't store it, so derive it).
_CERT_RANK = {
    "G": 1,
    "TV-Y": 1,
    "TV-Y7": 1,
    "TV-G": 1,
    "PG": 2,
    "TV-PG": 2,
    "PG-13": 3,
    "TV-14": 3,
    "R": 4,
    "TV-MA": 4,
    "NC-17": 5,
}


def _collection_cert(certs: list[str | None]) -> str | None:
    ranked = [(c, _CERT_RANK.get(c, 0)) for c in certs if c]
    return max(ranked, key=lambda x: x[1])[0] if ranked else None


router = APIRouter(prefix="/api/plex", tags=["plex"])

_STRIP_PROVIDERS = ("imdb", "tmdb", "tvdb", "trakt")


def _collection_source(col: PlexCollection) -> str:
    first = (col.title or "").strip().split(" ", 1)[0].lower()
    if first in _STRIP_PROVIDERS:
        return first
    if col.smart:
        return "smart"
    return "collection"


def _collection_editable(c: PlexCollection) -> bool:
    """Effective editability: the admin marked it editable AND it's a plain manual collection — never
    smart, never an external auto-list (IMDb/TMDb/…), which are managed outside Plex."""
    return bool(c.editable) and not c.smart and _collection_source(c) == "collection"


def _collection_card(c: PlexCollection) -> dict:
    return {
        "id": c.rating_key,
        "title": c.title,
        "summary": c.summary,
        "thumb": f"/api/plex/image/{c.rating_key}" if c.thumb_key else None,
        "child_count": c.child_count,
        "smart": c.smart,
        "source": _collection_source(c),
        "editable": c.editable,
        "can_edit": _collection_editable(
            c
        ),  # whether the edit controls should be offered (admin UI)
    }


@router.get("/collections")
def collections(
    library: str | None = None,
    q: str | None = None,
    keys: str | None = None,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Mirrored collections as cards (most-populated first), optionally name-filtered by `q`. With a
    `library` plex_key → that one library's collections (used by the collection editor). WITHOUT one →
    the UNION across all enabled libraries (the unified-scope sidebar picker, which has no single
    library selected).

    `keys` (CSV of rating_keys) narrows to just those. It exists because a Plex link carries the
    collection FILTER as bare rating_keys — nothing in the URL says what "17045" is called — so the
    sidebar token and the applied-filter chip had nothing to render but the raw id. Resolving them
    through this endpoint costs one small row each; pulling the whole list to find two titles is
    ~200 KB."""
    if library is not None:
        lib = (
            db.execute(select(PlexLibrary).where(PlexLibrary.plex_key == str(library)))
            .scalars()
            .first()
        )
        if lib is None:
            return {"collections": []}
        query = select(PlexCollection).where(PlexCollection.library_id == lib.id)
    else:
        enabled_ids = list(
            db.execute(
                select(PlexLibrary.id).where(
                    PlexLibrary.enabled.is_(True), PlexLibrary.overlay_of.is_(None)
                )
            ).scalars()
        )
        if not enabled_ids:
            return {"collections": []}
        query = select(PlexCollection).where(PlexCollection.library_id.in_(enabled_ids))
    term = (q or "").strip()
    if term:
        query = query.where(PlexCollection.title.ilike(f"%{like_escape(term)}%"))
    if keys is not None:
        wanted = [k for k in (keys or "").split(",") if k]
        # An explicit but empty `keys=` asks for nothing — answer nothing, rather than falling through
        # to "every collection" (which is what an unguarded empty IN () would have to become).
        if not wanted:
            return {"collections": []}
        query = query.where(PlexCollection.rating_key.in_(wanted))
    rows = (
        db.execute(
            query.order_by(
                PlexCollection.child_count.desc().nullslast(),
                func.lower(PlexCollection.title),
            )
        )
        .scalars()
        .all()
    )
    # A montage of the top titles per tile + the collection's total runtime. Art unions MOVIE and SHOW
    # members (a collection lives in one library, so exactly one side has art) — a show collection would
    # otherwise render blank; the collection's own poster is the next fallback, then the colour hash.
    # Runtime is movie-only (shows have no single duration).
    movie_ids, show_ids = content_libs(db)
    keys = [c.rating_key for c in rows]
    art_m = top_art_keys(db, PlexItem, movie_ids, PlexItem.collection_keys, keys, per=2)
    art_s = top_art_keys(db, PlexShow, show_ids, PlexShow.collection_keys, keys, per=2)
    runtime = runtime_by_key(db, PlexItem, movie_ids, PlexItem.collection_keys, keys)

    def _art(c: PlexCollection) -> list[str]:
        member = art_m.get(c.rating_key) or art_s.get(c.rating_key)
        return member or ([c.rating_key] if c.thumb_key else [])

    return {
        "collections": [
            {
                **_collection_card(c),
                "art": _art(c),
                "total_runtime_s": runtime.get(c.rating_key, 0),
            }
            for c in rows
        ]
    }


class Member(NamedTuple):
    """One collection member flattened to everything the Sort control can order by. The page's two
    legs are different tables (movies and shows), so there is no single query to ORDER BY — the
    ordering is a pure list sort over these, which is also what makes it unit-testable."""

    kind: str  # "movie" | "show"
    obj: Any  # PlexItem | PlexShow, carried through to the card serializer
    title: str  # lower-cased; never NULL (the column is not nullable)
    year: int | None
    rating: float | None
    added: datetime | None
    release: date | None
    duration: int | None  # movies only — a show has no single runtime
    last_watched: datetime | None


# Sort key -> what to read off a Member. Pinned to `PLEX_SORT_KEYS` by a test: the collection page
# shows the SAME Sort control as the browse grid, so a key this map is missing would be an option
# that silently does nothing here.
MEMBER_SORT_KEYS: dict[str, Callable[[Member], Any]] = {
    "added": lambda m: m.added,
    "lastwatched": lambda m: m.last_watched,
    "release": lambda m: m.release,
    "year": lambda m: m.year,
    "rating": lambda m: m.rating,
    "duration": lambda m: m.duration,
    "title": lambda m: m.title,
}

# What the page ordered by before it had a control, and still its default: a film series wants to be
# read in release order, not by whenever Plex happened to ingest it.
DEFAULT_MEMBER_SORT = "year"


def sort_members(members: list[Member], sort: str, sort_dir: str) -> list[Member]:
    """Order a collection's members by one of `PLEX_SORT_KEYS`.

    Two properties worth stating because they are easy to get wrong with a plain `reverse=True`:
    members with no value for the key sink to the END in BOTH directions (matching the browse grid's
    `nullslast()` — a movie with no rating is not "the worst rated"), and ties always resolve
    title-ascending, so flipping the direction reverses the ranking rather than the alphabet inside
    it. An unknown key falls back to the page's default instead of returning an arbitrary order."""
    key = MEMBER_SORT_KEYS.get(sort, MEMBER_SORT_KEYS[DEFAULT_MEMBER_SORT])
    ordered = sorted(members, key=lambda m: m.title)  # stable secondary, applied first
    have = [m for m in ordered if key(m) is not None]
    missing = [m for m in ordered if key(m) is None]
    have.sort(key=key, reverse=sort_dir != "asc")
    return have + missing


@router.get("/collections/{rating_key}")
def collection_detail(
    rating_key: str,
    sort: str = DEFAULT_MEMBER_SORT,
    sort_dir: str = "asc",
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """One collection's hero data + its members (movies + shows) in the requested order — the
    Collection info page. The sort vocabulary is the browse grid's (`PLEX_SORT_KEYS`) so the same Sort
    control means the same thing on both surfaces; the default is release year ascending, which is
    what this page did before it had a control. Read-only here; edits go through the admin mutation
    endpoints below."""
    col = by_rating_key(db, PlexCollection, rating_key)
    if col is None:
        raise HTTPException(status_code=404, detail="Unknown collection")
    movies = (
        db.execute(
            select(PlexItem)
            .where(PlexItem.kind == "movie", PlexItem.collection_keys.contains([rating_key]))
            .order_by(PlexItem.year.asc().nullslast(), func.lower(PlexItem.title))
        )
        .scalars()
        .all()
    )
    shows = (
        db.execute(
            select(PlexShow)
            .where(PlexShow.collection_keys.contains([rating_key]))
            .order_by(PlexShow.year.asc().nullslast(), func.lower(PlexShow.title))
        )
        .scalars()
        .all()
    )
    mstate = rollup.item_state_map(db, user.id, [m.id for m in movies])
    # ONE rollup pass over the show members, for both the aggregate badge status and `resumed_at` —
    # the browse grid's "last watched" for a show (the newest resume/watched across its episodes). The
    # cheaper `rollup.state_map` gives only the status, which would leave `lastwatched` unorderable on
    # a show collection while the control still offered it.
    show_rows = (
        db.execute(
            rollup.agg_subquery(db, user.id, include_resumed=True).where(
                PlexItem.show_id.in_([s.id for s in shows])
            )
        ).all()
        if shows
        else []
    )
    sstate = {r.show_id: rollup.show_status(r.total, r.watched, r.inprog) for r in show_rows}
    sresumed = {r.show_id: r.resumed_at for r in show_rows}

    def _movie_last_watched(st: PlexState | None) -> datetime | None:
        # Same definition as the browse grid's: when it was last resumed, else when it was marked
        # watched (coalesce(progress_updated_at, watched_at)).
        return (st.progress_updated_at or st.watched_at) if st else None

    # Merge the two type-homogeneous lists into ONE member list, then order it.
    members = [
        Member(
            kind="movie",
            obj=m,
            title=(m.title or "").lower(),
            year=m.year,
            rating=m.rating,
            added=m.added_at,
            release=m.originally_available_at,
            duration=m.duration_s,
            last_watched=_movie_last_watched(mstate.get(m.id)),
        )
        for m in movies
    ] + [
        Member(
            kind="show",
            obj=sh,
            title=(sh.title or "").lower(),
            year=sh.year,
            rating=sh.rating,
            added=sh.added_at,
            release=sh.originally_available_at,
            duration=None,
            last_watched=sresumed.get(sh.id),
        )
        for sh in shows
    ]
    members = sort_members(members, sort, sort_dir)
    items = [
        movie_card(m.obj, mstate.get(m.obj.id))
        if m.kind == "movie"
        else show_card(m.obj, sstate.get(m.obj.id, "new"))
        for m in members
    ]
    years = [m.year for m in members if m.year is not None]
    certs = [m.content_rating for m in movies] + [s.content_rating for s in shows]
    return {
        **_collection_card(col),
        "content_rating": _collection_cert(certs),
        "year_min": min(years) if years else None,
        "year_max": max(years) if years else None,
        "total_runtime_s": sum(m.duration_s or 0 for m in movies),
        "items": items,
    }


def _editable_collection_or_403(db: Session, rating_key: str) -> PlexCollection:
    col = by_rating_key(db, PlexCollection, rating_key)
    if col is None:
        raise HTTPException(status_code=404, detail="Unknown collection")
    if not _collection_editable(col):
        raise HTTPException(status_code=403, detail="This collection is read-only")
    return col


def _reject_duplicate_name(
    db: Session, library_id: int, title: str, exclude_rk: str | None = None
) -> None:
    """No two collections in a section may share a name: Plex CONFLATES same-named collections'
    membership (both rating_keys resolve to one merged member list — a real footgun), so forbid it up
    front on create + rename rather than mirror the mess."""
    dup = (
        db.execute(
            select(PlexCollection).where(
                PlexCollection.library_id == library_id,
                func.lower(PlexCollection.title) == title.lower(),
            )
        )
        .scalars()
        .first()
    )
    if dup is not None and dup.rating_key != exclude_rk:
        # Structured, so the bilingual UI says it in the reader's language (it was toasted in
        # Hungarian beside an English error dialog saying the same thing).
        raise HTTPException(status_code=409, detail={"code": "collection", "reason": "name_taken"})


def _write_or_pending_error(wrote: bool, e: Exception) -> HTTPException:
    """Map a PlexError raised inside a mutate-then-resync block to the honest message. If the mutating
    call itself failed (``wrote`` False) the write never landed → "Plex write failed". If the write
    succeeded but the follow-up resync READ (Plex metadata/children) failed, the change IS on the Plex
    server → don't claim a write failure; say the local copy will reconcile on the next sync.

    Both use 422, NOT a 5xx: the frontend treats 502/503/504 as "gateway unreachable" and raises a
    sticky offline banner (see api/core.ts RETRIABLE_GATEWAY). A 422 surfaces the detail as an ordinary
    action-error toast, which is what both of these are — the app answered, it didn't go unreachable."""
    if wrote:
        detail = (
            "Collection updated in Plex, but refreshing the local copy failed; "
            f"it will reconcile on the next sync ({e})"
        )
    else:
        detail = f"Plex write failed: {e}"
    return HTTPException(status_code=422, detail=detail)


class CreateCollectionIn(BaseModel):
    library: str = ""
    title: str = ""
    item_rating_key: str = ""


@router.post("/collections")
def create_collection(
    body: CreateCollectionIn,
    _: User = Depends(admin_user),
    db: Session = Depends(get_db),
) -> dict:
    """Create a manual collection (seeded with one item — Plex needs at least one). Editable by default."""
    library = body.library
    title = body.title.strip()
    seed = body.item_rating_key
    if not title or not seed:
        raise HTTPException(status_code=400, detail="title and item_rating_key are required")
    if library:
        lib = (
            db.execute(select(PlexLibrary).where(PlexLibrary.plex_key == library)).scalars().first()
        )
    else:
        # The "New collection" flow only knows the seed title → derive its library.
        seed_item = by_rating_key(db, PlexItem, seed) or by_rating_key(db, PlexShow, seed)
        lib = db.get(PlexLibrary, seed_item.library_id) if seed_item else None
    if lib is None:
        raise HTTPException(status_code=404, detail="Unknown library")
    _reject_duplicate_name(db, lib.id, title)
    wrote = False
    try:
        with PlexClient(db) as plex:
            new = plex.create_collection(lib.plex_key, title, seed)
            wrote = True
            col = plex_sync.resync_collection(db, plex, lib, str(new.get("ratingKey")))
            col.editable = True
        db.commit()
    except (PlexError, PlexNotConfigured) as e:
        db.rollback()
        raise _write_or_pending_error(wrote, e)
    return _collection_card(col)


@router.post("/collections/{rating_key}/items/{item_rating_key}")
def collection_add_item(
    rating_key: str,
    item_rating_key: str,
    _: User = Depends(admin_user),
    db: Session = Depends(get_db),
) -> dict:
    col = _editable_collection_or_403(db, rating_key)
    lib = db.get(PlexLibrary, col.library_id)
    wrote = False
    try:
        with PlexClient(db) as plex:
            plex.add_collection_item(rating_key, str(item_rating_key))
            wrote = True
            col = plex_sync.resync_collection(db, plex, lib, rating_key)
        db.commit()
    except (PlexError, PlexNotConfigured) as e:
        db.rollback()
        raise _write_or_pending_error(wrote, e)
    return _collection_card(col)


@router.delete("/collections/{rating_key}/items/{item_rating_key}")
def collection_remove_item(
    rating_key: str,
    item_rating_key: str,
    _: User = Depends(admin_user),
    db: Session = Depends(get_db),
) -> dict:
    col = _editable_collection_or_403(db, rating_key)
    lib = db.get(PlexLibrary, col.library_id)
    wrote = False
    try:
        with PlexClient(db) as plex:
            plex.remove_collection_item(rating_key, str(item_rating_key))
            wrote = True
            col = plex_sync.resync_collection(db, plex, lib, rating_key)
        db.commit()
    except (PlexError, PlexNotConfigured) as e:
        db.rollback()
        raise _write_or_pending_error(wrote, e)
    return _collection_card(col)


class CollectionRenameIn(BaseModel):
    title: str = ""


@router.patch("/collections/{rating_key}")
def collection_rename(
    rating_key: str,
    body: CollectionRenameIn,
    _: User = Depends(admin_user),
    db: Session = Depends(get_db),
) -> dict:
    col = _editable_collection_or_403(db, rating_key)
    title = body.title.strip()
    if not title:
        raise HTTPException(status_code=400, detail="title is required")
    _reject_duplicate_name(db, col.library_id, title, exclude_rk=rating_key)
    lib = db.get(PlexLibrary, col.library_id)
    wrote = False
    try:
        with PlexClient(db) as plex:
            plex.rename_collection(lib.plex_key, rating_key, title)
            wrote = True
            col = plex_sync.resync_collection(db, plex, lib, rating_key)
        db.commit()
    except (PlexError, PlexNotConfigured) as e:
        db.rollback()
        raise _write_or_pending_error(wrote, e)
    return _collection_card(col)


@router.delete("/collections/{rating_key}")
def collection_delete(
    rating_key: str, _: User = Depends(admin_user), db: Session = Depends(get_db)
) -> dict:
    col = _editable_collection_or_403(db, rating_key)
    lib = db.get(PlexLibrary, col.library_id)
    try:
        with PlexClient(db) as plex:
            plex.delete_collection(rating_key)
        plex_sync.delete_collection_local(db, lib, rating_key)
        db.commit()
    except (PlexError, PlexNotConfigured) as e:
        db.rollback()
        raise HTTPException(status_code=422, detail=f"Plex write failed: {e}")
    return {"deleted": rating_key}


class CollectionEditableIn(BaseModel):
    editable: bool = False


@router.post("/collections/{rating_key}/editable")
def collection_set_editable(
    rating_key: str,
    body: CollectionEditableIn,
    _: User = Depends(admin_user),
    db: Session = Depends(get_db),
) -> dict:
    """Mark an existing Plex collection as 'mine to edit' (or unmark). Smart / auto-list collections
    (IMDb/TMDb/…) can never be made editable — they're regenerated outside Plex."""
    col = by_rating_key(db, PlexCollection, rating_key)
    if col is None:
        raise HTTPException(status_code=404, detail="Unknown collection")
    want = body.editable
    if want and (col.smart or _collection_source(col) != "collection"):
        raise HTTPException(
            status_code=400,
            detail="Smart / auto-list collections can't be made editable",
        )
    col.editable = want
    db.commit()
    return _collection_card(col)