feat(plex): Collections Phase 2 — admin collection editing (write-back to Plex)

Admins can curate Plex collections from a movie's info page ("Collections"
button → PlexCollectionEditor dialog): create a collection (seeded with the
movie), add/remove the movie to/from editable collections, delete a collection,
and "take over" an existing plain Plex collection (mark it editable). All writes
go to Plex (POST/PUT/DELETE via new PlexClient methods) and are reflected on
every client; a targeted single-collection re-sync (sync.resync_collection /
delete_collection_local) updates the local mirror without the full ~4-min sync.

Gating (per the design decisions): editing is ADMIN-ONLY (collections are shared
library-wide); only plain manual collections are editable — smart + external
auto-lists (IMDb/TMDb/…) are always read-only (can_edit = editable && !smart &&
source=="collection"). New admin endpoints under /api/plex/collections
(create/items add+remove/rename/delete/editable). Verified end-to-end incl. the
live Plex write API (create/add/rename/remove/delete all 200, self-cleaned) and
the editor UI (create + delete with confirm) on localdev.
This commit is contained in:
npeter83 2026-07-06 17:33:16 +02:00
parent 4b053a55da
commit 8291f06525
12 changed files with 602 additions and 13 deletions

View file

@ -106,6 +106,59 @@ class PlexClient:
mc = self._get(f"/library/collections/{rating_key}/children")
return mc.get("Metadata", []) or []
# --- Collection editing (P2 — write-back to Plex; all shared/global, admin-gated in the routes) ---
@property
def machine_id(self) -> str:
"""Server machineIdentifier — needed in the `uri` of collection add/create calls. Cached."""
if not getattr(self, "_machine_id", None):
self._machine_id = self._get("/identity").get("machineIdentifier") or ""
return self._machine_id
def _write(self, method: str, path: str) -> dict:
try:
r = self._http.request(method, f"{self.base}{path}")
r.raise_for_status()
except httpx.HTTPError as e:
raise PlexError(str(e)) from e
try:
return (r.json() or {}).get("MediaContainer", {}) or {} if r.content else {}
except ValueError:
return {}
def _metadata_uri(self, *rating_keys: str) -> str:
keys = ",".join(str(k) for k in rating_keys)
return f"server://{self.machine_id}/com.plexapp.plugins.library/library/metadata/{keys}"
def create_collection(self, section_key: str, title: str, item_rating_key: str) -> dict:
"""Create a (dumb) collection seeded with one item; returns its Metadata (incl. ratingKey)."""
uri = quote(self._metadata_uri(item_rating_key), safe="")
mc = self._write(
"POST",
f"/library/collections?type=1&smart=0&sectionId={section_key}"
f"&title={quote(title, safe='')}&uri={uri}",
)
items = mc.get("Metadata", []) or []
if not items:
raise PlexError("Plex did not return the created collection")
return items[0]
def add_collection_item(self, collection_rating_key: str, item_rating_key: str) -> None:
uri = quote(self._metadata_uri(item_rating_key), safe="")
self._write("PUT", f"/library/collections/{collection_rating_key}/items?uri={uri}")
def remove_collection_item(self, collection_rating_key: str, item_rating_key: str) -> None:
self._write("DELETE", f"/library/collections/{collection_rating_key}/children/{item_rating_key}")
def rename_collection(self, section_key: str, collection_rating_key: str, title: str) -> None:
self._write(
"PUT",
f"/library/sections/{section_key}/all?type=18&id={collection_rating_key}"
f"&title.value={quote(title, safe='')}&title.locked=1",
)
def delete_collection(self, collection_rating_key: str) -> None:
self._write("DELETE", f"/library/collections/{collection_rating_key}")
def metadata(self, rating_key: str, markers: bool = False) -> dict | None:
"""Full metadata for one item; include intro/credit markers when requested."""
params = {"includeMarkers": 1} if markers else None