"""The comfort shelf: pin a thing you can always fall back to, and have it be one click from playing. Five kinds are pinnable — a video, a YouTube channel, a Plex show, a Plex movie, a playlist — because comfort media is almost always "put on X", a series or a channel, and only sometimes one particular video. `models.ComfortPin` carries that union as five nullable FKs under a `num_nonnulls(...) = 1` check; this module is where the union is spoken about in one vocabulary, so no call site invents a second one. Every shelf card carries a resolved `next`: the thing that actually plays when you click the tile — a show's in-progress-or-next episode, a channel's newest unwatched video, a playlist's first unwatched item. It costs a query per non-leaf pin, which is affordable precisely because this shelf is short and hand-curated, and it buys the property the whole surface exists for: one click, no intermediate page, no second decision. Domain module, so nothing here raises HTTP. `resolve_ref` returns None for "no such thing, or not yours" and the route turns that into its 404 — core never imports routes. """ from sqlalchemy import select from sqlalchemy.orm import Session from app.discover import cards, queries from app.models import Channel, ComfortPin, PlexItem, PlexShow, Playlist, User, Video from app.plex import rollup from app.plex.cards import leaf_card, show_card from app.plex.lookups import by_rating_key # kind -> the ComfortPin column that holds it. The client's vocabulary and the schema's, in one place. PIN_COLUMNS: dict[str, str] = { "video": "video_id", "channel": "channel_id", "plex_item": "plex_item_id", "plex_show": "plex_show_id", "playlist": "playlist_id", } PIN_KINDS = tuple(PIN_COLUMNS) def resolve_ref(db: Session, user: User, kind: str, ref: str) -> tuple[str, object] | None: """Translate the client's (kind, ref) into the (column, value) the row stores, or None. The client speaks the ids the rest of the API speaks: a video id, a channel id, a Plex `rating_key`, a playlist id. Plex's internal integer keys never leave the backend, so the translation belongs here rather than in five call sites.""" if kind == "video": return ("video_id", ref) if db.get(Video, ref) is not None else None if kind == "channel": return ("channel_id", ref) if db.get(Channel, ref) is not None else None if kind == "plex_item": it = by_rating_key(db, PlexItem, ref) return ("plex_item_id", it.id) if it is not None else None if kind == "plex_show": sh = by_rating_key(db, PlexShow, ref) return ("plex_show_id", sh.id) if sh is not None else None if kind == "playlist": try: pid = int(ref) except ValueError: return None pl = db.get(Playlist, pid) # A playlist is per-user, unlike the other four (which are shared catalog rows whose PRIVACY # lives in the state tables). Someone else's list must read as "no such thing", not as a 403 — # the same reason `owned_or_404` 404s rather than 403s. return ("playlist_id", pl.id) if pl is not None and pl.user_id == user.id else None return None def list_pins(db: Session, user: User) -> list[ComfortPin]: return list( db.execute( select(ComfortPin) .where(ComfortPin.user_id == user.id) .order_by(ComfortPin.position.asc(), ComfortPin.id.asc()) ) .scalars() .all() ) def next_position(db: Session, user: User) -> int: """New pins land at the end of the shelf.""" return len(list_pins(db, user)) def _shelf_card(db: Session, user: User, pin: ComfortPin) -> dict | None: """One pin as a unified card with its resolved `next`. None when the pinned row has gone (the FK cascade removes the pin itself, so this is only the window between a delete and a refetch).""" card: dict | None = None nxt: dict | None = None if pin.video_id is not None: row = queries.yt_video_row(db, user, pin.video_id) if row is None: return None card = cards.from_video(row) nxt = dict(card) # a video plays itself elif pin.channel_id is not None: ch = db.get(Channel, pin.channel_id) if ch is None: return None card = cards.from_channel(ch, queries.channel_unwatched_count(db, user, ch.id)) row = queries.channel_next_video(db, user, ch.id) nxt = cards.from_video(row) if row else None elif pin.plex_item_id is not None: it = db.get(PlexItem, pin.plex_item_id) if it is None: return None st = rollup.item_state_map(db, user.id, [it.id]).get(it.id) card = cards.from_plex(leaf_card(it, st)) nxt = dict(card) # a movie or a single episode plays itself elif pin.plex_show_id is not None: sh = db.get(PlexShow, pin.plex_show_id) if sh is None: return None status = rollup.state_map(db, user.id, [sh.id]).get(sh.id, "new") card = cards.from_plex(show_card(sh, status)) ep = queries.next_episode(db, user.id, sh.id) nxt = cards.from_plex(ep) if ep else None elif pin.playlist_id is not None: pl = db.get(Playlist, pin.playlist_id) if pl is None: return None count, row = queries.playlist_summary(db, user, pl.id) card = cards.from_playlist(pl, count, row.get("thumbnail_url") if row else None) nxt = cards.from_video(row) if row else None if card is None: return None # A COPY, never the card itself: a self-playing pin (a video, a movie) would otherwise make # `card["next"] is card`, which serializes as a circular reference and 500s the whole shelf. The # copy is taken before `pin_id`/`next` are attached, so `next` stays a plain playable card. card["pin_id"] = pin.id card["next"] = nxt return card def shelf(db: Session, user: User) -> list[dict]: """The whole shelf, in the user's own order.""" out = [] for pin in list_pins(db, user): card = _shelf_card(db, user, pin) if card is not None: out.append(card) return out def playables(shelf_cards: list[dict]) -> list[dict]: """The `next` of every shelf entry that has one — what the hero may draw from. A pinned channel with nothing unwatched left, or an empty playlist, simply is not an answer right now.""" return [c["next"] for c in shelf_cards if c.get("next")]