siftlode/backend/app/plex/queries.py

"""Plex's reads for the HTTP layer: the mirrored rows a page shows, and a user's watch state on them.

A user's own rows (`PlexState`) are read through `userscope`, so "someone else's row" is the same
answer as a missing one. A rating key with a NUL in it names no row (`lookups.by_rating_key`).
"""

from typing import Literal

from sqlalchemy import func, or_, select
from sqlalchemy.orm import Session

from app.collation import collated
from app.models import User
from app.plex import rollup
from app.plex.lookups import by_rating_key
from app.plex.models import (
    PlexCollection,
    PlexItem,
    PlexLibrary,
    PlexPerson,
    PlexSeason,
    PlexShow,
    PlexState,
)
from app.plex.query import content_libs, runtime_by_key, title_col, top_art_keys
from app.userscope import owned_by
from app.utils import like_escape

# --- one row ---------------------------------------------------------------------------------


def item(db: Session, rating_key: str) -> PlexItem | None:
    return by_rating_key(db, PlexItem, rating_key)


def movie(db: Session, rating_key: str) -> PlexItem | None:
    """The movie with this key — an episode's key is not one."""
    it = item(db, rating_key)
    return it if it is not None and it.kind == "movie" else None


def show(db: Session, rating_key: str) -> PlexShow | None:
    return by_rating_key(db, PlexShow, rating_key)


def collection(db: Session, rating_key: str) -> PlexCollection | None:
    return by_rating_key(db, PlexCollection, rating_key)


def show_by_id(db: Session, show_id: int) -> PlexShow | None:
    return db.get(PlexShow, show_id)


def season_by_id(db: Session, season_id: int) -> PlexSeason | None:
    return db.get(PlexSeason, season_id)


def library(db: Session, library_id: int) -> PlexLibrary | None:
    return db.get(PlexLibrary, library_id)


def library_by_key(db: Session, plex_key: str) -> PlexLibrary | None:
    return (
        db.execute(select(PlexLibrary).where(PlexLibrary.plex_key == str(plex_key)))
        .scalars()
        .first()
    )


def library_key(db: Session, library_id: int | None) -> str | None:
    """The Plex section key for a library id — the info/show page passes it to the admin collection
    editor (in the unified cross-library view there's no single library prop otherwise)."""
    if library_id is None:
        return None
    lib = db.get(PlexLibrary, library_id)
    return lib.plex_key if lib else None


def person(db: Session, name: str) -> PlexPerson | None:
    return db.execute(select(PlexPerson).where(PlexPerson.name == name)).scalars().first()


# --- a user's watch state --------------------------------------------------------------------


def state(db: Session, user: User | int, it: PlexItem) -> PlexState | None:
    return (
        db.execute(
            select(PlexState).where(owned_by(PlexState.user_id, user), PlexState.item_id == it.id)
        )
        .scalars()
        .first()
    )


def states(db: Session, user: User | int, item_ids: list[int]) -> dict[int, PlexState]:
    """The user's state on each of these items that has one, by item id."""
    if not item_ids:
        return {}
    return {
        s.item_id: s
        for s in db.execute(
            select(PlexState).where(
                owned_by(PlexState.user_id, user), PlexState.item_id.in_(item_ids)
            )
        ).scalars()
    }


# --- a show ----------------------------------------------------------------------------------


def seasons(db: Session, sh: PlexShow) -> list[PlexSeason]:
    return list(
        db.execute(
            select(PlexSeason)
            .where(PlexSeason.show_id == sh.id)
            .order_by(PlexSeason.season_number.asc().nullsfirst())
        ).scalars()
    )


def episodes(db: Session, sh: PlexShow) -> list[PlexItem]:
    """A show's episodes in season/episode order."""
    return list(
        db.execute(
            select(PlexItem)
            .where(PlexItem.show_id == sh.id, PlexItem.kind == "episode")
            .order_by(
                PlexItem.season_number.asc().nullsfirst(),
                PlexItem.episode_number.asc().nullsfirst(),
            )
        ).scalars()
    )


def episode_keys(db: Session, show_id: int) -> list[str]:
    """The rating keys of a show's episodes, in the order `episodes` gives."""
    return list(
        db.execute(
            select(PlexItem.rating_key)
            .where(PlexItem.show_id == show_id, PlexItem.kind == "episode")
            .order_by(
                PlexItem.season_number.asc().nullsfirst(),
                PlexItem.episode_number.asc().nullsfirst(),
            )
        ).scalars()
    )


def shows_by_keys(db: Session, rating_keys: list[str], exclude: int) -> dict[str, PlexShow]:
    """The mirrored shows among these keys, but `exclude`, by key."""
    return {
        s.rating_key: s
        for s in db.execute(
            select(PlexShow).where(PlexShow.rating_key.in_(rating_keys), PlexShow.id != exclude)
        ).scalars()
    }


# --- libraries -------------------------------------------------------------------------------


def browsable_libraries(db: Session, lang: str) -> list[tuple[PlexLibrary, int]]:
    """The mirrored libraries a user can browse, each with its title count (movies, or shows). An
    HU overlay library is left out: it is never browsed, it only feeds its EN twin's Hungarian
    metadata."""
    libs = db.execute(
        select(PlexLibrary)
        .where(PlexLibrary.overlay_of.is_(None))
        .order_by(PlexLibrary.kind.desc(), collated(PlexLibrary.title, lang))
    ).scalars()
    out = []
    for lib in libs:
        if lib.kind == "movie":
            count = db.scalar(
                select(func.count(PlexItem.id)).where(
                    PlexItem.library_id == lib.id, PlexItem.kind == "movie"
                )
            )
        else:
            count = db.scalar(select(func.count(PlexShow.id)).where(PlexShow.library_id == lib.id))
        out.append((lib, count or 0))
    return out


# --- people ----------------------------------------------------------------------------------


def person_photos(db: Session, names: list[str]) -> dict[str, str | None]:
    return {
        p.name: p.photo_thumb
        for p in db.execute(select(PlexPerson).where(PlexPerson.name.in_(names))).scalars()
    }


def cast_count(db: Session, name: str, movie_lib_ids: list[int], show_lib_ids: list[int]) -> int:
    """How many movies (in `movie_lib_ids`) and shows (in `show_lib_ids`) name this person in their
    cast. An empty id list leaves its side out."""
    count = 0
    if movie_lib_ids:
        count += (
            db.scalar(
                select(func.count(PlexItem.id)).where(
                    PlexItem.kind == "movie",
                    PlexItem.library_id.in_(movie_lib_ids),
                    PlexItem.cast_names.contains([name]),
                )
            )
            or 0
        )
    if show_lib_ids:
        count += (
            db.scalar(
                select(func.count(PlexShow.id)).where(
                    PlexShow.library_id.in_(show_lib_ids),
                    PlexShow.cast_names.contains([name]),
                )
            )
            or 0
        )
    return count


def filmography(
    db: Session,
    kind: Literal["movie", "show"],
    name: str,
    lib_ids: list[int],
    lang: str,
    limit: int,
) -> list:
    """The movies or shows in these libraries whose cast or directors name this person, newest
    first."""
    model: type[PlexItem] | type[PlexShow] = PlexItem if kind == "movie" else PlexShow
    query = select(model).where(
        model.library_id.in_(lib_ids),
        or_(model.cast_names.contains([name]), model.directors.contains([name])),
    )
    if model is PlexItem:
        query = query.where(PlexItem.kind == "movie")
    return list(
        db.execute(
            query.order_by(
                model.year.desc().nullslast(), collated(title_col(model, lang), lang)
            ).limit(limit)
        ).scalars()
    )


# --- collections -----------------------------------------------------------------------------


def collections_page(
    db: Session, library: str | None, q: str | None, keys: list[str] | None, lang: str
) -> list[PlexCollection]:
    """Collections, most-populated first: one library's (by its plex key), else every enabled
    library's; name-filtered by `q`; narrowed to `keys` when given (an empty list asks for none)."""
    if library is not None:
        lib = library_by_key(db, library)
        if lib is None:
            return []
        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 []
        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:
        # An explicit but empty `keys=` asks for nothing, which is not the same as no `keys` (every
        # collection): answered here without a query. The empty IN () would also match nothing.
        if not keys:
            return []
        query = query.where(PlexCollection.rating_key.in_(keys))
    return list(
        db.execute(
            query.order_by(
                PlexCollection.child_count.desc().nullslast(), collated(PlexCollection.title, lang)
            )
        ).scalars()
    )


def library_collections(db: Session, library_id: int, lang: str) -> list[PlexCollection]:
    return list(
        db.execute(
            select(PlexCollection)
            .where(PlexCollection.library_id == library_id)
            .order_by(collated(PlexCollection.title, lang))
        ).scalars()
    )


def collections_by_keys(db: Session, keys: list[str]) -> list[PlexCollection]:
    return list(
        db.execute(select(PlexCollection).where(PlexCollection.rating_key.in_(keys))).scalars()
    )


def collection_named(db: Session, library_id: int, title: str) -> PlexCollection | None:
    """A collection of this library with this name, case aside."""
    return (
        db.execute(
            select(PlexCollection).where(
                PlexCollection.library_id == library_id,
                func.lower(PlexCollection.title) == title.lower(),
            )
        )
        .scalars()
        .first()
    )


def collection_members(
    db: Session, rating_key: str, lang: str
) -> tuple[list[PlexItem], list[PlexShow]]:
    """A collection's movies and shows, each by year then title."""
    movies = db.execute(
        select(PlexItem)
        .where(PlexItem.kind == "movie", PlexItem.collection_keys.contains([rating_key]))
        .order_by(PlexItem.year.asc().nullslast(), collated(title_col(PlexItem, lang), lang))
    ).scalars()
    shows = db.execute(
        select(PlexShow)
        .where(PlexShow.collection_keys.contains([rating_key]))
        .order_by(PlexShow.year.asc().nullslast(), collated(title_col(PlexShow, lang), lang))
    ).scalars()
    return list(movies), list(shows)


def library_movies_in(
    db: Session, library_id: int, rating_key: str, lang: str, limit: int
) -> list[PlexItem]:
    """The movies of one library in this collection, by year then title."""
    return list(
        db.execute(
            select(PlexItem)
            .where(
                PlexItem.library_id == library_id,
                PlexItem.kind == "movie",
                PlexItem.collection_keys.contains([rating_key]),
            )
            .order_by(PlexItem.year.asc().nullslast(), collated(title_col(PlexItem, lang), lang))
            .limit(limit)
        ).scalars()
    )


def show_rollups(db: Session, user_id: int, show_ids: list[int]) -> list:
    """The user's episode rollup (total, watched, in progress, last resumed) of each of these shows
    that has episodes."""
    if not show_ids:
        return []
    return list(
        db.execute(
            rollup.agg_subquery(db, user_id, include_resumed=True).where(
                PlexItem.show_id.in_(show_ids)
            )
        ).all()
    )


def collection_tiles(
    db: Session, keys: list[str]
) -> tuple[dict[str, list[str]], dict[str, list[str]], dict[str, int]]:
    """What each collection's tile shows: the top two movie members' keys, the top two show members'
    keys (a collection lives in one library, so exactly one side has any), and its movies' total
    runtime (a show has no single duration)."""
    movie_ids, show_ids = content_libs(db)
    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)
    return art_m, art_s, runtime