yorickr/plezy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
c45113fefe6d1bb52e381e36a5d65a252800d381
plezy/scripts/maestro_real_jellyfin.py
T

688 lines
26 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
"""Prepare and bootstrap a disposable real Jellyfin server for Maestro."""
from __future__ import annotations
import argparse
import hashlib
import json
import os
from pathlib import Path
import shutil
import sys
import time
from typing import Any
import urllib.error
import urllib.parse
import urllib.request
import xml.etree.ElementTree as ET
from maestro_fixtures import MEDIA_FIXTURE_SPECS, _AUDIO, _VIDEO
USERNAME = "maestro"
PASSWORD = "maestro"
GUEST_USERNAME = "guest"
GUEST_PASSWORD = "guest"
SERVER_NAME = "Maestro Jellyfin"
BASE_TITLE = "Maestro Movie"
BASE_OVERVIEW = "A deterministic movie used to verify Plezy's end-to-end flows."
GUEST_TITLE = "Guest Galaxy"
SHOW_TITLE = "Maestro Show"
EPISODE_TITLES = ("Maestro Episode 1", "Maestro Episode 2")
MUSIC_ARTIST = "Maestro Artist"
MUSIC_ALBUM = "Regression Album"
MUSIC_TRACK = "Resilient Track"
ALPHABET_TITLES = (
"Alpha Archive",
"Bravo Beacon",
"Charlie Circuit",
"Delta Drive",
"Echo Engine",
"Foxtrot Frame",
"Gamma Garden",
"Hotel Horizon",
"India Index",
"Juliet Junction",
"Kilo Key",
"Lima Loop",
"Mike Matrix",
"November Node",
"Oscar Orbit",
"Papa Pipeline",
"Quebec Queue",
"Romeo Relay",
"Sierra Signal",
"Tango Track",
"Uniform Update",
"Victor View",
"Whiskey Widget",
"Xray XML",
"Yankee Yield",
"Zulu Zone",
)
_MANAGED_MARKER = ".plezy-jellyfin-e2e-media"
DEFAULT_CODEC_BASE_URL = "https://demo-files.plezy.app/media-samples/"
_DOWNLOAD_CHUNK_SIZE = 1024 * 1024
def _write_nfo(path: Path, *, item_id: str, title: str, overview: str, genre: str) -> None:
movie = ET.Element("movie")
values = {
"title": title,
"originaltitle": title,
"sorttitle": title,
"year": "2026",
"premiered": "2026-01-01",
"dateadded": "2026-01-01 00:00:00",
"plot": overview,
"outline": overview,
"studio": "Plezy E2E",
"genre": genre,
"tag": "E2E",
"mpaa": "E2E",
"rating": "8.0",
"lockdata": "true",
}
for key, value in values.items():
ET.SubElement(movie, key).text = value
unique_id = ET.SubElement(movie, "uniqueid", {"type": "plezy", "default": "true"})
unique_id.text = item_id
ET.indent(movie, space=" ")
ET.ElementTree(movie).write(path, encoding="utf-8", xml_declaration=True)
def _write_show_nfo(path: Path) -> None:
show = ET.Element("tvshow")
for key, value in {
"title": SHOW_TITLE,
"sorttitle": SHOW_TITLE,
"year": "2026",
"premiered": "2026-01-01",
"plot": "A deterministic show used to verify episode playback and queue behavior.",
"studio": "Plezy E2E",
"genre": "Test",
"lockdata": "true",
}.items():
ET.SubElement(show, key).text = value
unique_id = ET.SubElement(show, "uniqueid", {"type": "plezy", "default": "true"})
unique_id.text = "maestro-show"
ET.indent(show, space=" ")
ET.ElementTree(show).write(path, encoding="utf-8", xml_declaration=True)
def _write_episode_nfo(path: Path, number: int) -> None:
episode = ET.Element("episodedetails")
title = EPISODE_TITLES[number - 1]
for key, value in {
"title": title,
"showtitle": SHOW_TITLE,
"season": "1",
"episode": str(number),
"aired": f"2026-01-0{number}",
"plot": f"Deterministic episode {number} for player queue coverage.",
"lockdata": "true",
}.items():
ET.SubElement(episode, key).text = value
unique_id = ET.SubElement(episode, "uniqueid", {"type": "plezy", "default": "true"})
unique_id.text = f"maestro-episode-{number}"
ET.indent(episode, space=" ")
ET.ElementTree(episode).write(path, encoding="utf-8", xml_declaration=True)
def _write_music_nfo(path: Path, root_name: str, values: dict[str, str]) -> None:
root = ET.Element(root_name)
for key, value in values.items():
ET.SubElement(root, key).text = value
ET.indent(root, space=" ")
ET.ElementTree(root).write(path, encoding="utf-8", xml_declaration=True)
def _reset_managed_directory(path: Path) -> None:
marker = path / _MANAGED_MARKER
if not path.exists():
path.mkdir(parents=True)
marker.write_text("Managed by scripts/maestro_real_jellyfin.py\n", encoding="utf-8")
return
if not marker.is_file():
if any(path.iterdir()):
raise ValueError(f"Refusing to clear unmanaged media staging directory: {path}")
marker.write_text("Managed by scripts/maestro_real_jellyfin.py\n", encoding="utf-8")
return
for child in path.iterdir():
if child == marker:
continue
if child.is_dir() and not child.is_symlink():
shutil.rmtree(child)
else:
child.unlink()
def _hard_link(source: Path, destination: Path) -> None:
try:
os.link(source, destination)
except OSError as error:
raise ValueError(
f"Could not hard-link {source} into the staging directory; keep both paths on the same filesystem"
) from error
def _codec_digest(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as source:
while chunk := source.read(_DOWNLOAD_CHUNK_SIZE):
digest.update(chunk)
return digest.hexdigest()
def _valid_codec_file(path: Path, expected_size: int, expected_sha256: str) -> bool:
return path.is_file() and path.stat().st_size == expected_size and _codec_digest(path) == expected_sha256
def download_codec_media(output_dir: Path, base_url: str = DEFAULT_CODEC_BASE_URL) -> list[str]:
output = output_dir.expanduser().resolve()
output.mkdir(parents=True, exist_ok=True)
normalized_base_url = base_url.rstrip("/") + "/"
filenames: list[str] = []
for spec in MEDIA_FIXTURE_SPECS:
destination = output / spec.filename
filenames.append(spec.filename)
if _valid_codec_file(destination, spec.size_bytes, spec.sha256):
continue
partial = destination.with_suffix(f"{destination.suffix}.part")
partial.unlink(missing_ok=True)
request = urllib.request.Request(
urllib.parse.urljoin(normalized_base_url, urllib.parse.quote(spec.filename)),
headers={"User-Agent": "plezy-jellyfin-demo-builder"},
)
try:
with urllib.request.urlopen(request, timeout=60) as response:
raw_length = response.headers.get("Content-Length")
if raw_length is not None and int(raw_length) != spec.size_bytes:
raise ValueError(
f"{spec.filename} download is {raw_length} bytes, expected {spec.size_bytes}"
)
total = 0
digest = hashlib.sha256()
with partial.open("wb") as target:
while chunk := response.read(_DOWNLOAD_CHUNK_SIZE):
total += len(chunk)
if total > spec.size_bytes:
raise ValueError(f"{spec.filename} download exceeds {spec.size_bytes} bytes")
digest.update(chunk)
target.write(chunk)
if total != spec.size_bytes:
raise ValueError(f"{spec.filename} download is {total} bytes, expected {spec.size_bytes}")
if digest.hexdigest() != spec.sha256:
raise ValueError(f"{spec.filename} download failed SHA-256 verification")
partial.replace(destination)
except (OSError, ValueError, urllib.error.URLError):
partial.unlink(missing_ok=True)
raise
return filenames
def prepare_media(output_dir: Path, codec_source_dir: Path | None, include_codecs: bool) -> list[str]:
output = output_dir.expanduser().resolve()
_reset_managed_directory(output)
base_dir = output / "movies" / "maestro-movie"
base_dir.mkdir(parents=True)
(base_dir / f"{BASE_TITLE}.mp4").write_bytes(_VIDEO)
_write_nfo(
base_dir / f"{BASE_TITLE}.nfo",
item_id="maestro-movie",
title=BASE_TITLE,
overview=BASE_OVERVIEW,
genre="Test",
)
titles = [BASE_TITLE]
for alphabet_title in ALPHABET_TITLES:
for copy in range(1, 5):
title = alphabet_title if copy == 1 else f"{alphabet_title} {copy}"
item_id = f"alpha-{title.lower().replace(' ', '-')}"
item_dir = output / "movies" / item_id
item_dir.mkdir()
media_path = item_dir / f"{title}.mp4"
media_path.write_bytes(_VIDEO)
_write_nfo(
media_path.with_suffix(".nfo"),
item_id=item_id,
title=title,
overview="A deterministic title for TV alphabet focus coverage.",
genre="E2E Alphabet",
)
titles.append(title)
guest_dir = output / "guest-movies" / "guest-galaxy"
guest_dir.mkdir(parents=True)
guest_media = guest_dir / f"{GUEST_TITLE}.mp4"
guest_media.write_bytes(_VIDEO)
_write_nfo(
guest_media.with_suffix(".nfo"),
item_id="guest-galaxy",
title=GUEST_TITLE,
overview="Content visible only to the Maestro Guest profile.",
genre="E2E Guest",
)
show_dir = output / "shows" / SHOW_TITLE
season_dir = show_dir / "Season 01"
season_dir.mkdir(parents=True)
_write_show_nfo(show_dir / "tvshow.nfo")
for number, title in enumerate(EPISODE_TITLES, start=1):
episode_path = season_dir / f"{SHOW_TITLE} S01E{number:02d} - {title}.mp4"
episode_path.write_bytes(_VIDEO)
_write_episode_nfo(episode_path.with_suffix(".nfo"), number)
album_dir = output / "music" / MUSIC_ARTIST / MUSIC_ALBUM
album_dir.mkdir(parents=True)
(album_dir / f"{MUSIC_TRACK}.wav").write_bytes(_AUDIO)
_write_music_nfo(
album_dir.parent / "artist.nfo",
"artist",
{"name": MUSIC_ARTIST, "sortname": MUSIC_ARTIST, "overview": "Deterministic E2E music artist."},
)
_write_music_nfo(
album_dir / "album.nfo",
"album",
{
"title": MUSIC_ALBUM,
"artist": MUSIC_ARTIST,
"albumartist": MUSIC_ARTIST,
"year": "2026",
"review": "Deterministic E2E music album.",
},
)
if not include_codecs:
return titles
if codec_source_dir is None:
raise ValueError("--codec-source-dir is required with --include-codecs")
source_dir = codec_source_dir.expanduser().resolve()
missing = [spec.filename for spec in MEDIA_FIXTURE_SPECS if not (source_dir / spec.filename).is_file()]
if missing:
raise ValueError(f"Codec fixture directory is missing: {', '.join(missing)}")
for spec in MEDIA_FIXTURE_SPECS:
item_dir = output / "movies" / spec.id
item_dir.mkdir()
media_path = item_dir / f"{spec.title}.mkv"
_hard_link(source_dir / spec.filename, media_path)
_write_nfo(
media_path.with_suffix(".nfo"),
item_id=spec.id,
title=spec.title,
overview=spec.overview,
genre="E2E Codec",
)
titles.append(spec.title)
return titles
class JellyfinApi:
def __init__(self, base_url: str) -> None:
self.base_url = base_url.rstrip("/")
def request(
self,
method: str,
path: str,
*,
payload: Any | None = None,
token: str | None = None,
headers: dict[str, str] | None = None,
timeout: float = 30,
) -> tuple[int, bytes]:
request_headers = {"Accept": "application/json", **(headers or {})}
data: bytes | None = None
if payload is not None:
data = json.dumps(payload).encode("utf-8")
request_headers["Content-Type"] = "application/json"
elif method == "POST":
data = b""
if token is not None:
request_headers["X-Emby-Token"] = token
request = urllib.request.Request(self.base_url + path, data=data, method=method, headers=request_headers)
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
return response.status, response.read()
except urllib.error.HTTPError as error:
return error.code, error.read()
def json(self, method: str, path: str, **kwargs: Any) -> Any:
status, body = self.request(method, path, **kwargs)
if status < 200 or status >= 300:
detail = body.decode("utf-8", errors="replace")[:500]
raise RuntimeError(f"Jellyfin {method} {path} returned HTTP {status}: {detail}")
return json.loads(body) if body else None
def _wait_until(deadline: float, description: str, operation: Any) -> Any:
last_error: Exception | None = None
while time.monotonic() < deadline:
try:
result = operation()
if result is not None:
return result
except (OSError, RuntimeError, urllib.error.URLError) as error:
last_error = error
time.sleep(1)
suffix = f": {last_error}" if last_error is not None else ""
raise TimeoutError(f"Timed out waiting for {description}{suffix}")
def bootstrap_server(
base_url: str,
expected_titles: set[str],
timeout_seconds: int,
expected_music_titles: set[str] | None = None,
) -> dict[str, Any]:
api = JellyfinApi(base_url)
deadline = time.monotonic() + timeout_seconds
def public_info() -> dict[str, Any] | None:
status, body = api.request("GET", "/System/Info/Public", timeout=3)
if status != 200:
return None
return json.loads(body)
info = _wait_until(deadline, "Jellyfin startup", public_info)
version = info.get("Version") or info.get("version")
if version != "10.11.11":
raise RuntimeError(f"Expected Jellyfin 10.11.11, got {version}")
startup_complete = info.get("StartupWizardCompleted", info.get("startupWizardCompleted", False))
if not startup_complete:
def configure_startup() -> bool | None:
status, body = api.request(
"POST",
"/Startup/Configuration",
payload={
"UICulture": "en-US",
"MetadataCountryCode": "US",
"PreferredMetadataLanguage": "en",
},
)
if status == 204:
return True
if status == 503:
return None
detail = body.decode("utf-8", errors="replace")[:500]
raise RuntimeError(f"Jellyfin startup configuration returned HTTP {status}: {detail}")
_wait_until(deadline, "Jellyfin startup API", configure_startup)
def startup_user() -> dict[str, Any] | None:
status, body = api.request("GET", "/Startup/User", timeout=3)
if status != 200:
return None
return json.loads(body)
_wait_until(deadline, "Jellyfin startup user", startup_user)
api.json("POST", "/Startup/User", payload={"Name": USERNAME, "Password": PASSWORD})
api.json("POST", "/Startup/Complete")
authorization = (
'MediaBrowser Client="Plezy E2E Bootstrap", Device="Host", '
'DeviceId="plezy-e2e-bootstrap", Version="1.0"'
)
def authenticate(username: str, password: str) -> dict[str, Any] | None:
status, body = api.request(
"POST",
"/Users/AuthenticateByName",
payload={"Username": username, "Pw": password},
headers={"Authorization": authorization},
timeout=5,
)
if status != 200:
return None
return json.loads(body)
authentication = _wait_until(deadline, "Jellyfin authentication", lambda: authenticate(USERNAME, PASSWORD))
token = authentication["AccessToken"]
user_id = authentication["User"]["Id"]
configuration = api.json("GET", "/System/Configuration", token=token)
if configuration.get("ServerName") != SERVER_NAME:
configuration["ServerName"] = SERVER_NAME
api.json("POST", "/System/Configuration", payload=configuration, token=token)
required_folders = (
("Maestro Movies", "movies", "/media/movies"),
("Guest Movies", "movies", "/media/guest-movies"),
("Maestro Shows", "tvshows", "/media/shows"),
("Maestro Music", "music", "/media/music"),
)
virtual_folders = api.json("GET", "/Library/VirtualFolders", token=token)
existing_folder_names = {folder.get("Name") for folder in virtual_folders}
for name, collection_type, path in required_folders:
if name in existing_folder_names:
continue
query = urllib.parse.urlencode(
{
"name": name,
"collectionType": collection_type,
"paths": path,
"refreshLibrary": "false",
}
)
api.json("POST", f"/Library/VirtualFolders?{query}", token=token)
users = api.json("GET", "/Users", token=token)
main_user = next(user for user in users if user["Id"] == user_id)
guest_user = next((user for user in users if user.get("Name") == GUEST_USERNAME), None)
if guest_user is None:
guest_user = api.json(
"POST",
"/Users/New",
payload={"Name": GUEST_USERNAME, "Password": GUEST_PASSWORD},
token=token,
)
if not guest_user.get("HasPassword", guest_user.get("HasConfiguredPassword", False)):
api.json(
"POST",
f"/Users/{guest_user['Id']}/Password",
payload={"CurrentPw": "", "NewPw": GUEST_PASSWORD},
token=token,
)
unrestricted_policy = dict(main_user["Policy"])
unrestricted_policy["EnableAllFolders"] = True
unrestricted_policy["EnabledFolders"] = []
api.json("POST", f"/Users/{user_id}/Policy", payload=unrestricted_policy, token=token)
views = api.json("GET", f"/Users/{user_id}/Views", token=token).get("Items", [])
views_by_name = {view.get("Name"): view.get("Id") for view in views}
missing_views = [name for name, _, _ in required_folders if not views_by_name.get(name)]
if missing_views:
raise RuntimeError(f"Jellyfin did not create library views: {', '.join(missing_views)}")
main_policy = dict(main_user["Policy"])
main_policy["EnableAllFolders"] = False
main_policy["EnabledFolders"] = [
views_by_name["Maestro Movies"],
views_by_name["Maestro Shows"],
views_by_name["Maestro Music"],
]
api.json("POST", f"/Users/{user_id}/Policy", payload=main_policy, token=token)
guest_policy = dict(guest_user["Policy"])
guest_policy["EnableAllFolders"] = False
guest_policy["EnabledFolders"] = [views_by_name["Guest Movies"]]
api.json("POST", f"/Users/{guest_user['Id']}/Policy", payload=guest_policy, token=token)
api.json("POST", "/Library/Refresh", token=token)
def scanned_items() -> list[dict[str, Any]] | None:
query = urllib.parse.urlencode(
{
"Recursive": "true",
"IncludeItemTypes": "Movie",
"Fields": "MediaSources,MediaStreams",
"Limit": "500",
}
)
result = api.json("GET", f"/Users/{user_id}/Items?{query}", token=token)
items = result.get("Items", [])
by_title = {item.get("Name"): item for item in items}
if not expected_titles.issubset(by_title):
return None
if any(not by_title[title].get("MediaSources") for title in expected_titles):
return None
if GUEST_TITLE in by_title:
raise RuntimeError("Main Jellyfin user can see the guest-only library")
return items
items = _wait_until(deadline, "Jellyfin media scan", scanned_items)
required_music = expected_music_titles or {MUSIC_TRACK}
def scanned_music() -> list[dict[str, Any]] | None:
query = urllib.parse.urlencode(
{
"Recursive": "true",
"IncludeItemTypes": "Audio",
"Fields": "MediaSources,MediaStreams,Album,AlbumId,AlbumArtist,AlbumArtists",
"Limit": "100",
}
)
result = api.json("GET", f"/Users/{user_id}/Items?{query}", token=token)
music_items = result.get("Items", [])
by_title = {item.get("Name"): item for item in music_items}
if not required_music.issubset(by_title):
return None
if any(not by_title[title].get("MediaSources") for title in required_music):
return None
return music_items
music_items = _wait_until(deadline, "Jellyfin music scan", scanned_music)
def scanned_artists() -> list[dict[str, Any]] | None:
query = urllib.parse.urlencode({"UserId": user_id, "Limit": "100"})
result = api.json("GET", f"/Artists/AlbumArtists?{query}", token=token)
artists = result.get("Items", [])
if MUSIC_ARTIST not in {artist.get("Name") for artist in artists}:
return None
return artists
artist_items = _wait_until(deadline, "Jellyfin music artist scan", scanned_artists)
def scanned_episodes() -> list[dict[str, Any]] | None:
query = urllib.parse.urlencode(
{
"Recursive": "true",
"IncludeItemTypes": "Episode",
"Fields": "MediaSources,MediaStreams,SeriesId,ParentId,IndexNumber",
"Limit": "100",
}
)
result = api.json("GET", f"/Users/{user_id}/Items?{query}", token=token)
episodes = result.get("Items", [])
by_title = {item.get("Name"): item for item in episodes}
if not set(EPISODE_TITLES).issubset(by_title):
return None
if any(not by_title[title].get("MediaSources") for title in EPISODE_TITLES):
return None
return episodes
episode_items = _wait_until(deadline, "Jellyfin episode scan", scanned_episodes)
guest_authentication = _wait_until(
deadline,
"Jellyfin guest authentication",
lambda: authenticate(GUEST_USERNAME, GUEST_PASSWORD),
)
guest_token = guest_authentication["AccessToken"]
def scanned_guest_items() -> list[dict[str, Any]] | None:
query = urllib.parse.urlencode(
{
"Recursive": "true",
"IncludeItemTypes": "Movie",
"Fields": "MediaSources,MediaStreams",
"Limit": "100",
}
)
result = api.json("GET", f"/Users/{guest_user['Id']}/Items?{query}", token=guest_token)
guest_items = result.get("Items", [])
by_title = {item.get("Name"): item for item in guest_items}
if GUEST_TITLE not in by_title or not by_title[GUEST_TITLE].get("MediaSources"):
return None
if BASE_TITLE in by_title:
raise RuntimeError("Guest Jellyfin user can see the main library")
return guest_items
guest_items = _wait_until(deadline, "Jellyfin guest media scan", scanned_guest_items)
return {
"server": SERVER_NAME,
"version": "10.11.11",
"userId": user_id,
"guestUserId": guest_user["Id"],
"titles": sorted(item["Name"] for item in items if item.get("Name") in expected_titles),
"musicTitles": sorted(item["Name"] for item in music_items if item.get("Name") in required_music),
"artistTitles": sorted(item["Name"] for item in artist_items if item.get("Name") == MUSIC_ARTIST),
"episodeTitles": sorted(item["Name"] for item in episode_items if item.get("Name") in EPISODE_TITLES),
"guestTitles": sorted(item["Name"] for item in guest_items if item.get("Name") == GUEST_TITLE),
}
def _build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=__doc__)
subparsers = parser.add_subparsers(dest="command", required=True)
prepare = subparsers.add_parser("prepare", help="Create the deterministic media staging tree")
prepare.add_argument("--output-dir", type=Path, required=True)
prepare.add_argument("--codec-source-dir", type=Path)
prepare.add_argument("--include-codecs", action="store_true")
download = subparsers.add_parser("download-codecs", help="Download and verify the hosted codec fixtures")
download.add_argument("--output-dir", type=Path, required=True)
download.add_argument(
"--base-url",
default=os.environ.get("PLEZY_DEMO_MEDIA_BASE_URL", DEFAULT_CODEC_BASE_URL),
)
bootstrap = subparsers.add_parser("bootstrap", help="Configure and verify a fresh Jellyfin server")
bootstrap.add_argument("--url", required=True)
bootstrap.add_argument("--expected-title", action="append", default=[])
bootstrap.add_argument("--expected-music-title", action="append", default=[])
bootstrap.add_argument("--timeout", type=int, default=600)
bootstrap.add_argument("--include-codecs", action="store_true")
return parser
def main() -> int:
args = _build_parser().parse_args()
try:
if args.command == "download-codecs":
filenames = download_codec_media(args.output_dir, args.base_url)
print(json.dumps({"codecDir": str(args.output_dir.resolve()), "files": filenames}, sort_keys=True))
elif args.command == "prepare":
titles = prepare_media(args.output_dir, args.codec_source_dir, args.include_codecs)
print(json.dumps({"mediaDir": str(args.output_dir.resolve()), "titles": titles}, sort_keys=True))
else:
expected = {BASE_TITLE, *args.expected_title}
expected.update(
title if copy == 1 else f"{title} {copy}"
for title in ALPHABET_TITLES
for copy in range(1, 5)
)
if args.include_codecs:
expected.update(spec.title for spec in MEDIA_FIXTURE_SPECS)
expected_music = set(args.expected_music_title) or {MUSIC_TRACK}
result = bootstrap_server(args.url, expected, args.timeout, expected_music)
print(json.dumps(result, sort_keys=True))
return 0
except (OSError, RuntimeError, TimeoutError, ValueError) as error:
print(f"Real Jellyfin E2E setup failed: {error}", file=sys.stderr)
return 1
if __name__ == "__main__":
raise SystemExit(main())
Reference in New Issue View Git Blame Copy Permalink