peter
14a44bff6e
The collection browse-tile montage queried only movie members, so show-library collections (IMDb Popular, TMDb Trending, … in the TV library) rendered as bare colour tiles. Union the art from PlexShow members as well (a collection lives in one library, so exactly one side has art); fall back to the collection's own poster, then the colour hash. Runtime stays movie-only (shows have no single duration). Verified live: every collection now shows a montage (0 art-less tiles, was ~all the show collections). Gate: backend ruff clean, 5 collections tests (+ a show-collection art assertion).
280 lines
9.8 KiB
Python
280 lines
9.8 KiB
Python
"""Plex collections: browse mirrored collections + the admin manual-collection editor."""
|
|
|
|
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,
|
|
User,
|
|
)
|
|
from app.plex import sync as plex_sync
|
|
from app.plex.client import PlexClient, PlexError, PlexNotConfigured
|
|
from app.plex.lookups import by_rating_key
|
|
from app.plex.query import like_escape, runtime_by_key, top_art_keys
|
|
from app.routes.plex._shared import _content_libs
|
|
|
|
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,
|
|
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)."""
|
|
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)}%"))
|
|
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
|
|
]
|
|
}
|
|
|
|
|
|
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
|
|
|
|
|
|
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"
|
|
)
|
|
lib = db.execute(
|
|
select(PlexLibrary).where(PlexLibrary.plex_key == library)
|
|
).scalars().first()
|
|
if lib is None:
|
|
raise HTTPException(status_code=404, detail="Unknown library")
|
|
try:
|
|
with PlexClient(db) as plex:
|
|
new = plex.create_collection(lib.plex_key, title, seed)
|
|
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 HTTPException(status_code=422, detail=f"Plex write failed: {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)
|
|
try:
|
|
with PlexClient(db) as plex:
|
|
plex.add_collection_item(rating_key, str(item_rating_key))
|
|
col = plex_sync.resync_collection(db, plex, lib, rating_key)
|
|
db.commit()
|
|
except (PlexError, PlexNotConfigured) as e:
|
|
db.rollback()
|
|
raise HTTPException(status_code=422, detail=f"Plex write failed: {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)
|
|
try:
|
|
with PlexClient(db) as plex:
|
|
plex.remove_collection_item(rating_key, str(item_rating_key))
|
|
col = plex_sync.resync_collection(db, plex, lib, rating_key)
|
|
db.commit()
|
|
except (PlexError, PlexNotConfigured) as e:
|
|
db.rollback()
|
|
raise HTTPException(status_code=422, detail=f"Plex write failed: {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")
|
|
lib = db.get(PlexLibrary, col.library_id)
|
|
try:
|
|
with PlexClient(db) as plex:
|
|
plex.rename_collection(lib.plex_key, rating_key, title)
|
|
col = plex_sync.resync_collection(db, plex, lib, rating_key)
|
|
db.commit()
|
|
except (PlexError, PlexNotConfigured) as e:
|
|
db.rollback()
|
|
raise HTTPException(status_code=422, detail=f"Plex write failed: {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)
|