siftlode/backend/app/plex/query.py

"""Plex library query-shaping: the full-text field-scope rechecks and the sidebar metadata
filters that /library and /facets share. Extracted from routes/plex/browse.py so the query
builder has one home (and so the HU-overlay column selection lands here, not in the router).
Pure query fragments — every function returns SQLAlchemy expressions or a params dict."""

from dataclasses import dataclass
from datetime import datetime, timedelta, timezone

from sqlalchemy import and_, func, or_

_TS_CONFIG = "public.unaccent_simple"

# Field-scoped search: the scope-pill in the header search box narrows a full-text query to ONE
# facet of the row. `all` uses the GIN-indexed weighted `search_vector` untouched (byte-identical to
# the pre-scope behaviour); a specific field keeps that GIN prefilter AND re-checks a per-column
# tsvector, so the scoped hits are always a strict subset of Everywhere. The per-field expressions
# mirror the `search_vector` Computed definition (same `_TS_CONFIG`, same `left(summary, 1000)`
# truncation) so scoping never surfaces a match the vector itself wouldn't. `episode` is handled
# separately in the query body (it targets episode rows, not the movie/show grid).
_FIELD_COLUMNS = ("title", "people", "summary", "episode")


def _field_recheck(model, field: str, tsq):
    """A per-field `to_tsvector(...) @@ tsq` clause for a non-`all`, non-`episode` scope, or None."""
    if field == "title":
        expr = func.to_tsvector(_TS_CONFIG, func.coalesce(model.title, ""))
    elif field == "people":
        expr = func.to_tsvector(_TS_CONFIG, func.coalesce(model.people_text, ""))
    elif field == "summary":
        expr = func.to_tsvector(
            _TS_CONFIG, func.left(func.coalesce(model.summary, ""), 1000)
        )
    else:
        return None
    return expr.op("@@")(tsq)


def _csv(v: str | None) -> list[str]:
    return [s.strip() for s in v.split(",") if s.strip()] if v else []


_ADDED_WINDOWS = {"24h": 1, "1w": 7, "1m": 30, "6m": 182, "1y": 365}


def _added_cutoff(within: str | None) -> datetime | None:
    days = _ADDED_WINDOWS.get(within or "")
    return datetime.now(timezone.utc) - timedelta(days=days) if days else None


def _meta_filter_conds(model, p: dict) -> list:
    """The shared metadata-filter conditions (identical column names on plex_items + plex_shows) for
    `p` (the request's filter params). Duration is movie-only, handled by the caller. Returned as a
    list so faceted search can apply all-but-one group. DRY for browse, unified, and /facets."""
    conds: list = []
    gsel = _csv(p.get("genres"))
    if gsel:
        gc = [model.genres.contains([g]) for g in gsel]
        conds.append(and_(*gc) if p.get("genre_mode") == "all" else or_(*gc))
    crs = _csv(p.get("content_ratings"))
    if crs:
        conds.append(model.content_rating.in_(crs))
    if p.get("year_min") is not None:
        conds.append(model.year >= p["year_min"])
    if p.get("year_max") is not None:
        conds.append(model.year <= p["year_max"])
    if p.get("rating_min") is not None:
        conds.append(model.rating >= p["rating_min"])
    cutoff = _added_cutoff(p.get("added_within"))
    if cutoff is not None:
        conds.append(model.added_at >= cutoff)
    for d in _csv(p.get("directors")):  # AND
        conds.append(model.directors.contains([d]))
    alist = _csv(p.get("actors"))
    if alist:  # AND (all) or OR (any) — default OR, mirroring genre_mode
        ac = [model.cast_names.contains([a]) for a in alist]
        conds.append(and_(*ac) if p.get("actor_mode") == "all" else or_(*ac))
    stds = _csv(p.get("studios"))
    if stds:  # OR
        conds.append(model.studio.in_(stds))
    if p.get("collection"):
        conds.append(model.collection_keys.contains([p["collection"]]))
    return conds


def _apply_meta_filters(q, model, p: dict):
    conds = _meta_filter_conds(model, p)
    return q.where(*conds) if conds else q


@dataclass
class LibraryFilters:
    """The shared sidebar filter query-params for /library + /facets, declared once and injected via
    ``Depends()``. `as_dict()` yields the `p` dict the query builders read; `duration_*` is included
    for /facets' wsq but ignored by `_meta_filter_conds` (both callers apply duration separately)."""

    genres: str | None = None
    genre_mode: str = "any"
    content_ratings: str | None = None
    year_min: int | None = None
    year_max: int | None = None
    rating_min: float | None = None
    duration_min: int | None = None
    duration_max: int | None = None
    added_within: str | None = None
    directors: str | None = None
    actors: str | None = None
    actor_mode: str = (
        "any"  # how multiple actors combine: any (OR, default) | all (AND)
    )
    studios: str | None = None
    collection: str | None = None

    def as_dict(self) -> dict:
        return {
            "genres": self.genres,
            "genre_mode": self.genre_mode,
            "content_ratings": self.content_ratings,
            "year_min": self.year_min,
            "year_max": self.year_max,
            "rating_min": self.rating_min,
            "duration_min": self.duration_min,
            "duration_max": self.duration_max,
            "added_within": self.added_within,
            "directors": self.directors,
            "actors": self.actors,
            "actor_mode": self.actor_mode,
            "studios": self.studios,
            "collection": self.collection,
        }


def _like_escape(s: str) -> str:
    return s.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")