peter
dae0b80f6a
"Add to collection…" opens a panel beside the card: the title's library's manual collections tick it in or out, and a name + Enter creates a new one with the title in it. It used to take the Collections page, a new collection, its name and a title search, then another search per title. A lean endpoint lists the collections with whether each holds the title. Admin only, as on the title's page. Also: a collection seeded with a show is made as a show collection (Plex type 2). Measured: as a movie one it came out empty and refused the show after.
584 lines
22 KiB
Python
584 lines
22 KiB
Python
"""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.mailtext import lang_for_user
|
|
from app.collation import collated, sorted_words
|
|
from app import logs
|
|
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 display_title, movie_card, norm_lang, 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, runtime_by_key, title_col, top_art_keys
|
|
from app.utils import like_escape
|
|
from app.refusal import refused
|
|
from app.routes._common import not_found
|
|
|
|
log = logs.get_logger(__name__)
|
|
|
|
# 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(),
|
|
collated(PlexCollection.title, lang_for_user(user)),
|
|
)
|
|
)
|
|
.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
|
|
]
|
|
}
|
|
|
|
|
|
@router.get("/collections/for/{item_rating_key}")
|
|
def collections_for_item(
|
|
item_rating_key: str,
|
|
user: User = Depends(current_user),
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
"""What a movie or show card's "Add to collection" lists: the editable collections of the title's
|
|
library (a collection lives in one library, and only a manual one can take a title — the same
|
|
`can_edit` the editor offers), each saying whether the title is in it already, by name."""
|
|
movie = by_rating_key(db, PlexItem, item_rating_key)
|
|
title: PlexItem | PlexShow | None = (
|
|
movie if movie is not None and movie.kind == "movie" else None
|
|
) or by_rating_key(db, PlexShow, item_rating_key)
|
|
if title is None:
|
|
raise not_found("item")
|
|
member_of = set(title.collection_keys or [])
|
|
rows = (
|
|
db.execute(
|
|
select(PlexCollection)
|
|
.where(PlexCollection.library_id == title.library_id)
|
|
.order_by(collated(PlexCollection.title, lang_for_user(user)))
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
return {
|
|
"collections": [
|
|
{**_collection_card(c), "contains": c.rating_key in member_of}
|
|
for c in rows
|
|
if _collection_editable(c)
|
|
]
|
|
}
|
|
|
|
|
|
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, alpha: dict[str, int] | None = None
|
|
) -> 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.
|
|
|
|
`alpha` ranks each title in the reader's alphabet (`collation.sorted_words`); without it, titles
|
|
compare as Python strings, which puts every accented initial after "z"."""
|
|
by_title = (lambda m: alpha[m.title]) if alpha is not None else (lambda m: m.title)
|
|
key = MEMBER_SORT_KEYS.get(sort, MEMBER_SORT_KEYS[DEFAULT_MEMBER_SORT])
|
|
if key is MEMBER_SORT_KEYS["title"]:
|
|
key = by_title
|
|
ordered = sorted(members, key=by_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",
|
|
lang: str = "en",
|
|
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."""
|
|
lang = norm_lang(lang)
|
|
col = by_rating_key(db, PlexCollection, rating_key)
|
|
if col is None:
|
|
raise not_found("collection")
|
|
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()
|
|
.all()
|
|
)
|
|
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()
|
|
.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=display_title(m, lang).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=display_title(sh, lang).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
|
|
]
|
|
alpha = {w: i for i, w in enumerate(sorted_words(db, [m.title for m in members], lang))}
|
|
members = sort_members(members, sort, sort_dir, alpha)
|
|
items = [
|
|
movie_card(m.obj, mstate.get(m.obj.id), lang=lang)
|
|
if m.kind == "movie"
|
|
else show_card(m.obj, sstate.get(m.obj.id, "new"), lang=lang)
|
|
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 not_found("collection")
|
|
if not _collection_editable(col):
|
|
raise refused(403, "collection_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."""
|
|
log.warning(
|
|
"a Plex collection write did not settle",
|
|
extra=logs.event(
|
|
"plex.collections.write_failed", wrote=wrote, reason=type(e).__name__, error=str(e)
|
|
),
|
|
)
|
|
return refused(422, "collection_pending_sync") if wrote else refused(422, "plex_write_failed")
|
|
|
|
|
|
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 refused(400, "collection_fields_required")
|
|
seed_show = by_rating_key(db, PlexShow, seed)
|
|
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 seed_show
|
|
lib = db.get(PlexLibrary, seed_item.library_id) if seed_item else None
|
|
if lib is None:
|
|
raise not_found("library")
|
|
_reject_duplicate_name(db, lib.id, title)
|
|
wrote = False
|
|
try:
|
|
with PlexClient(db) as plex:
|
|
# A show collection is made as one (Plex type 2): as a movie one it stayed empty.
|
|
new = plex.create_collection(lib.plex_key, title, seed, 2 if seed_show else 1)
|
|
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) from 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) from 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) from 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 refused(400, "title_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) from 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 _write_or_pending_error(False, e) from None
|
|
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 not_found("collection")
|
|
want = body.editable
|
|
if want and (col.smart or _collection_source(col) != "collection"):
|
|
raise refused(400, "collection_smart")
|
|
col.editable = want
|
|
db.commit()
|
|
return _collection_card(col)
|