Script the Jellyfin library retirement
Matches libraries by path, never by name, and reads the name to delete back from the API rather than assuming it: the endpoint takes a name, matches loosely on some versions, and "YouTube" is a prefix of "YouTube (stream)". Refuses to retire the old library unless ytstream's has episodes, so a broken replacement cannot leave the server with no YouTube library. Dry run by default. Also records that /opt/youtube-automate is verified fully pushed to its bare repo -- clean tree, specs.md tracked -- so it is safe to delete, while /var/lib/youtube-automate holds subs.db, which is state and not in the repo. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Claude
Executable
+163
@@ -0,0 +1,163 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Retire the old *YouTube* Jellyfin library and rename ytstream's to take its name.
|
||||
|
||||
sudo -u susan /opt/ytstream/deploy/retire-jellyfin-library.py # show the plan
|
||||
sudo -u susan /opt/ytstream/deploy/retire-jellyfin-library.py --yes # do it
|
||||
|
||||
plan.md §12 step 3. This is the one decommission step that throws something away:
|
||||
removing a library discards Jellyfin's own state for those items — watch history,
|
||||
resume positions, favourites. The *files* are untouched, and re-adding the library
|
||||
at the same path returns the same ItemId and reuses the old items (measured in the
|
||||
PoC), so the loss is bounded, but it is a loss.
|
||||
|
||||
Libraries are matched by PATH, never by name, and the name to delete is read back
|
||||
from the API rather than assumed. Deleting by a guessed name is how you remove the
|
||||
wrong library: the endpoint takes a name, matches loosely on some versions, and
|
||||
"YouTube" is a prefix of "YouTube (stream)".
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sqlite3
|
||||
import sys
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
|
||||
BASE = "http://127.0.0.1:8096"
|
||||
DB = "/var/lib/ytstream/ytstream.db"
|
||||
|
||||
OLD_PATH = "/disks/Plex/YouTube" # youtube-automate's downloads
|
||||
NEW_PATH = "/disks/Plex/_ytstream" # ytstream's .strm tree
|
||||
FINAL_NAME = "YouTube"
|
||||
|
||||
|
||||
def api_key() -> str:
|
||||
try:
|
||||
with sqlite3.connect(f"file:{DB}?mode=ro", uri=True) as conn:
|
||||
row = conn.execute(
|
||||
"SELECT value FROM setting WHERE key='jellyfin_api_key'").fetchone()
|
||||
except sqlite3.Error as exc:
|
||||
sys.exit(f"cannot read {DB}: {exc}\nRun me as susan or root.")
|
||||
if not row or not row[0]:
|
||||
sys.exit("no jellyfin_api_key in the settings table")
|
||||
return row[0]
|
||||
|
||||
|
||||
def request(key: str, method: str, path: str, params: dict | None = None):
|
||||
url = BASE + path
|
||||
if params:
|
||||
url += "?" + urllib.parse.urlencode(params)
|
||||
req = urllib.request.Request(
|
||||
url, method=method, headers={"X-Emby-Token": key, "Accept": "application/json"})
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=60) as response:
|
||||
body = response.read()
|
||||
except urllib.error.HTTPError as exc:
|
||||
sys.exit(f"{method} {path} -> HTTP {exc.code}: {exc.read()[:200]!r}")
|
||||
except OSError as exc:
|
||||
sys.exit(f"{method} {path} -> {exc}\nIs Jellyfin running?")
|
||||
return json.loads(body) if body else None
|
||||
|
||||
|
||||
def folders(key: str) -> list[dict]:
|
||||
return request(key, "GET", "/Library/VirtualFolders") or []
|
||||
|
||||
|
||||
def by_path(all_folders: list[dict], target: str) -> dict | None:
|
||||
target = target.rstrip("/")
|
||||
for folder in all_folders:
|
||||
for location in folder.get("Locations") or []:
|
||||
if str(location).rstrip("/") == target:
|
||||
return folder
|
||||
return None
|
||||
|
||||
|
||||
def episode_count(key: str, library: dict) -> int:
|
||||
result = request(key, "GET", "/Items", {
|
||||
"parentId": library["ItemId"],
|
||||
"includeItemTypes": "Episode",
|
||||
"recursive": "true",
|
||||
"limit": "0",
|
||||
})
|
||||
return (result or {}).get("TotalRecordCount", 0)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
ap = argparse.ArgumentParser(description=__doc__,
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||
ap.add_argument("--yes", action="store_true", help="actually make the changes")
|
||||
args = ap.parse_args()
|
||||
|
||||
key = api_key()
|
||||
current = folders(key)
|
||||
old = by_path(current, OLD_PATH)
|
||||
new = by_path(current, NEW_PATH)
|
||||
|
||||
if new is None:
|
||||
sys.exit(f"no Jellyfin library covers {NEW_PATH}. Refusing to touch anything "
|
||||
f"-- retiring the old library without a working new one would leave "
|
||||
f"no YouTube library at all.")
|
||||
|
||||
# A healthy replacement is the whole precondition. An empty new library means
|
||||
# the scan never ran or the tree is unreadable, and removing the old one then
|
||||
# would leave nothing to watch.
|
||||
episodes = episode_count(key, new)
|
||||
print(f"ytstream library {new['Name']!r} at {NEW_PATH}: {episodes} episodes")
|
||||
if episodes == 0:
|
||||
sys.exit("the ytstream library has 0 episodes -- fix that first "
|
||||
"(`ytstream materialise --all`, then a Jellyfin scan)")
|
||||
|
||||
if old is None:
|
||||
print(f"old library none found at {OLD_PATH} (already retired)")
|
||||
else:
|
||||
print(f"old library {old['Name']!r} at {OLD_PATH}: "
|
||||
f"{episode_count(key, old)} episodes")
|
||||
|
||||
renaming = new["Name"] != FINAL_NAME
|
||||
print()
|
||||
print("Planned changes:")
|
||||
if old is not None:
|
||||
print(f" 1. DELETE library {old['Name']!r} (files in {OLD_PATH} untouched)")
|
||||
if renaming:
|
||||
print(f" {'2' if old is not None else '1'}. RENAME {new['Name']!r} -> {FINAL_NAME!r}")
|
||||
if old is None and not renaming:
|
||||
print(" nothing to do.")
|
||||
return
|
||||
print()
|
||||
print("Jellyfin's watch history and resume positions for the deleted library go")
|
||||
print("with it. The video files do not.")
|
||||
|
||||
if not args.yes:
|
||||
print()
|
||||
print("Dry run. Re-run with --yes to apply.")
|
||||
return
|
||||
|
||||
if old is not None:
|
||||
# Name read back from the API, not guessed. refreshLibrary=false: a scan
|
||||
# here would be pointless work and, on the ytstream side, unwanted traffic.
|
||||
request(key, "DELETE", "/Library/VirtualFolders",
|
||||
{"name": old["Name"], "refreshLibrary": "false"})
|
||||
print(f"==> deleted {old['Name']!r}")
|
||||
|
||||
if renaming:
|
||||
request(key, "POST", "/Library/VirtualFolders/Name",
|
||||
{"id": new["ItemId"], "newName": FINAL_NAME})
|
||||
print(f"==> renamed to {FINAL_NAME!r}")
|
||||
|
||||
after = folders(key)
|
||||
print()
|
||||
print("Now:")
|
||||
for folder in after:
|
||||
locations = folder.get("Locations") or []
|
||||
if any(str(p).rstrip("/") in (OLD_PATH, NEW_PATH) for p in locations):
|
||||
print(f" {folder['Name']!r} {locations}")
|
||||
if by_path(after, OLD_PATH) is None and old is not None:
|
||||
print(f"\n{OLD_PATH} is no longer a library. The files are still there:")
|
||||
print(f" rm -rf {OLD_PATH} # when you are ready")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user