From 093315b38217dd532f04c56a1d857a97b368cfd9 Mon Sep 17 00:00:00 2001 From: SpiderFish Date: Wed, 26 Aug 2026 09:53:42 +1000 Subject: [PATCH] Initial portable pretranscode service --- .env.example | 33 ++ .gitignore | 3 + Dockerfile | 12 + LICENSE | 21 ++ README.md | 97 +++++ compose.yaml | 51 +++ pretranscode.py | 719 +++++++++++++++++++++++++++++++++++++ scripts/provision-cache.sh | 42 +++ tests/__init__.py | 0 tests/integration.sh | 30 ++ tests/test_core.py | 69 ++++ 11 files changed, 1077 insertions(+) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 Dockerfile create mode 100644 LICENSE create mode 100644 README.md create mode 100644 compose.yaml create mode 100755 pretranscode.py create mode 100755 scripts/provision-cache.sh create mode 100644 tests/__init__.py create mode 100755 tests/integration.sh create mode 100644 tests/test_core.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..ec5971c --- /dev/null +++ b/.env.example @@ -0,0 +1,33 @@ +# Host paths +MEDIA_LIBRARY_PATH=/srv/media/library +PRETRANSCODE_ROOT=/srv/pretranscode +PRETRANSCODE_VIEW_PATH=/srv/pretranscode/view +PRETRANSCODE_CACHE_PATH=/srv/pretranscode/cache +PRETRANSCODE_STATE_PATH=/srv/pretranscode/state +PRETRANSCODE_BUILD_CONTEXT=/srv/pretranscode/repository + +# Cache allocation (decimal bytes) +CACHE_MAX_BYTES=1000000000000 +PRETRANSCODE_CACHE_IMAGE=/srv/pretranscode/cache.ext4 + +# Container identity and network +PRETRANSCODE_UID=1000 +PRETRANSCODE_GID=1000 +PRETRANSCODE_IMAGE=pretranscode:latest +MEDIA_NETWORK=media-internal + +# Jellyfin API and paths visible inside Jellyfin +JELLYFIN_URL=http://jellyfin:8096 +JELLYFIN_API_KEY=replace-me +JELLYFIN_LIBRARY_ROOT=/data/library +JELLYFIN_ORIGINAL_ROOT=/data/original +JELLYFIN_CACHE_ROOT=/data/cache + +# Worker behavior +SERIES_COOLDOWN_SECONDS=3600 +POLL_SECONDS=5 +RESCAN_SECONDS=300 +H264_CRF=20 +H264_PRESET=medium +H264_MAXRATE=12M +H264_BUFSIZE=24M diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..cff5543 --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +.env +__pycache__/ +*.pyc diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..72fd4a0 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,12 @@ +FROM ubuntu:24.04 + +ENV DEBIAN_FRONTEND=noninteractive +RUN apt-get update \ + && apt-get install -y --no-install-recommends ca-certificates ffmpeg procps python3 util-linux \ + && rm -rf /var/lib/apt/lists/* + +COPY pretranscode.py /app/pretranscode +RUN chmod 0755 /app/pretranscode + +ENTRYPOINT ["nice","-n","19","ionice","-c","3","python3","/app/pretranscode"] +CMD ["run"] diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..04e319d --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 SpiderFish + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..aa28210 --- /dev/null +++ b/README.md @@ -0,0 +1,97 @@ +# Pretranscode + +A small Jellyfin compatibility cache. It converts only known-incompatible video +codecs to uniform 8-bit H.264 while preserving immutable originals. + +## Design + +- Originals are mounted read-only at `/source`. +- Jellyfin reads a mirror tree at `/data/library` containing symlinks. +- Each symlink points either to `/data/original` or a validated file under + `/data/cache`. +- Completed cache files replace symlinks atomically. Partial files are never + exposed. +- Cache eviction restores the original symlink before deleting cached content. +- Jellyfin integration uses only its supported HTTP API. + +There is one worker process. x264 uses available CPU cores; `nice 19`, idle I/O +priority, and minimal Docker CPU shares make normal services take precedence. + +## Requirements + +- Linux, Docker Compose, and ext4 tools +- An existing Jellyfin server and API key +- A shared Docker network between Jellyfin and Pretranscode +- Jellyfin must have read-only access to the originals and cache + +## Installation + +1. Copy `.env.example` to `.env` and set all host paths and the Jellyfin API key. +2. Provision the claimed cache allocation: + + ```bash + sudo scripts/provision-cache.sh .env + ``` + +3. Mount the same paths in Jellyfin while keeping its existing logical library + path: + + ```yaml + volumes: + - ${PRETRANSCODE_VIEW_PATH}:/data/library:ro + - ${MEDIA_LIBRARY_PATH}:/data/original:ro + - ${PRETRANSCODE_CACHE_PATH}:/data/cache:ro + ``` + +4. Start Pretranscode: + + ```bash + docker compose --env-file .env up -d --build + ``` + +For Coolify, import `compose.yaml` as a Custom Docker Compose resource and enter +the `.env.example` values through Coolify's environment-variable UI. + +## Codec policy + +Cached video is H.264 High Profile, Level 4.1, 8-bit `yuv420p`, CRF 20, capped +at 12 Mbps. Existing audio, subtitles, chapters, and Matroska attachments are +preserved when the container supports them. + +The fixed source blacklist is: + +- HEVC/H.265, including Dolby Vision +- H.264 Hi10/10-bit +- MPEG-2 +- VC-1 + +Client error reports never modify this list. + +## Playback locking + +- An active movie locks that movie. +- An active episode locks its complete series. +- Encoding is cancelled and its partial deleted if the item becomes locked. +- A series remains locked for `SERIES_COOLDOWN_SECONDS` after playback. +- Recently stopped content receives queue priority. + +## Failure behavior + +Graceful stops delete partial output. After a crash, startup removes stale +partials and orphan files. Atomic symlinks always point to either an original or +a completed cache file. If the cache mount marker is absent, the worker refuses +to start. + +## Migration + +Copy or recreate these resources on the destination: + +- Repository and `.env` +- State directory +- Symlink-view directory +- Cache image/mount, if retaining completed transcodes + +The cache is optional and can be rebuilt. The source fingerprint in every +manifest prevents stale cache files from replacing changed originals. No +Coolify UUID, Docker volume UUID, container name, or Jellyfin database path is +stored by the application. diff --git a/compose.yaml b/compose.yaml new file mode 100644 index 0000000..7a92baa --- /dev/null +++ b/compose.yaml @@ -0,0 +1,51 @@ +services: + pretranscode: + build: + context: ${PRETRANSCODE_BUILD_CONTEXT:-.} + image: ${PRETRANSCODE_IMAGE:-pretranscode:latest} + restart: unless-stopped + init: true + user: "${PRETRANSCODE_UID:-1000}:${PRETRANSCODE_GID:-1000}" + environment: + SOURCE_ROOT: /source + VIEW_ROOT: /view + CACHE_ROOT: /cache + STATE_ROOT: /state + JELLYFIN_URL: ${JELLYFIN_URL:?} + JELLYFIN_API_KEY: ${JELLYFIN_API_KEY:?} + JELLYFIN_LIBRARY_ROOT: ${JELLYFIN_LIBRARY_ROOT:-/data/library} + JELLYFIN_ORIGINAL_ROOT: ${JELLYFIN_ORIGINAL_ROOT:-/data/original} + JELLYFIN_CACHE_ROOT: ${JELLYFIN_CACHE_ROOT:-/data/cache} + CACHE_MAX_BYTES: ${CACHE_MAX_BYTES:-1000000000000} + SERIES_COOLDOWN_SECONDS: ${SERIES_COOLDOWN_SECONDS:-3600} + POLL_SECONDS: ${POLL_SECONDS:-5} + RESCAN_SECONDS: ${RESCAN_SECONDS:-300} + H264_CRF: ${H264_CRF:-20} + H264_PRESET: ${H264_PRESET:-medium} + H264_MAXRATE: ${H264_MAXRATE:-12M} + H264_BUFSIZE: ${H264_BUFSIZE:-24M} + volumes: + - ${MEDIA_LIBRARY_PATH:?}:/source:ro + - ${PRETRANSCODE_VIEW_PATH:?}:/view + - ${PRETRANSCODE_CACHE_PATH:?}:/cache + - ${PRETRANSCODE_STATE_PATH:?}:/state + networks: + - media + cpu_shares: 2 + read_only: true + tmpfs: + - /tmp + security_opt: + - no-new-privileges:true + stop_grace_period: 30s + healthcheck: + test: ["CMD-SHELL", "pgrep -f '/app/pretranscode run' >/dev/null"] + interval: 30s + timeout: 5s + retries: 3 + start_period: 30s + +networks: + media: + external: true + name: ${MEDIA_NETWORK:-media-internal} diff --git a/pretranscode.py b/pretranscode.py new file mode 100755 index 0000000..9702bc9 --- /dev/null +++ b/pretranscode.py @@ -0,0 +1,719 @@ +#!/usr/bin/env python3 +"""Immutable Jellyfin H.264 cache backed by an atomic symlink view.""" + +import hashlib +import json +import os +import signal +import subprocess +import sys +import time +import urllib.error +import urllib.parse +import urllib.request +from pathlib import Path, PurePosixPath + +VIDEO_EXTENSIONS = {".mkv", ".mp4", ".m4v", ".avi", ".mov", ".ts", ".m2ts", ".webm"} +BLACKLISTED_CODECS = {"hevc", "h265", "mpeg2video", "vc1", "vc-1"} + + +def env_int(name, default): + return int(os.environ.get(name, default)) + + +SOURCE_ROOT = Path(os.environ.get("SOURCE_ROOT", "/source")) +VIEW_ROOT = Path(os.environ.get("VIEW_ROOT", "/view")) +CACHE_ROOT = Path(os.environ.get("CACHE_ROOT", "/cache")) +STATE_ROOT = Path(os.environ.get("STATE_ROOT", "/state")) +STATE_FILE = STATE_ROOT / "state.json" +JELLYFIN_URL = os.environ.get("JELLYFIN_URL", "http://jellyfin:8096").rstrip("/") +JELLYFIN_LIBRARY_ROOT = os.environ.get("JELLYFIN_LIBRARY_ROOT", "/data/library").rstrip("/") +JELLYFIN_ORIGINAL_ROOT = os.environ.get("JELLYFIN_ORIGINAL_ROOT", "/data/original").rstrip("/") +JELLYFIN_CACHE_ROOT = os.environ.get("JELLYFIN_CACHE_ROOT", "/data/cache").rstrip("/") +JELLYFIN_API_KEY = os.environ.get("JELLYFIN_API_KEY", "") +SERIES_COOLDOWN_SECONDS = env_int("SERIES_COOLDOWN_SECONDS", 3600) +POLL_SECONDS = env_int("POLL_SECONDS", 5) +RESCAN_SECONDS = env_int("RESCAN_SECONDS", 300) +H264_CRF = os.environ.get("H264_CRF", "20") +H264_PRESET = os.environ.get("H264_PRESET", "medium") +H264_MAXRATE = os.environ.get("H264_MAXRATE", "12M") +H264_BUFSIZE = os.environ.get("H264_BUFSIZE", "24M") +FFMPEG = os.environ.get("FFMPEG", "/usr/bin/ffmpeg") +FFPROBE = os.environ.get("FFPROBE", "/usr/bin/ffprobe") + + +def log(message): + print(time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), message, flush=True) + + +def atomic_json(path, value): + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_name(path.name + ".tmp") + with temporary.open("w") as handle: + json.dump(value, handle, sort_keys=True, separators=(",", ":")) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary, path) + + +def load_json(path, default): + try: + with path.open() as handle: + return json.load(handle) + except (FileNotFoundError, json.JSONDecodeError, OSError): + return default + + +def fingerprint(path): + stat = path.stat() + return f"{stat.st_size}:{stat.st_mtime_ns}" + + +def relative_source(path): + value = str(path) + for root in (JELLYFIN_LIBRARY_ROOT, str(SOURCE_ROOT)): + prefix = root.rstrip("/") + "/" + if value.startswith(prefix): + return value[len(prefix):] + return None + + +def item_key(relative): + parts = PurePosixPath(relative).parts + if len(parts) >= 2 and parts[0].lower() == "tv": + return "series:" + parts[1] + return "item:" + relative + + +def cache_basename(relative): + digest = hashlib.sha256(relative.encode()).hexdigest() + suffix = ".mp4" if Path(relative).suffix.lower() in {".mp4", ".m4v"} else ".mkv" + return digest + suffix + + +def source_target(relative): + return JELLYFIN_ORIGINAL_ROOT + "/" + PurePosixPath(relative).as_posix() + + +def cache_target(relative): + return JELLYFIN_CACHE_ROOT + "/files/" + cache_basename(relative) + + +class Manager: + def __init__(self): + STATE_ROOT.mkdir(parents=True, exist_ok=True) + if not (CACHE_ROOT / ".pretranscode-cache").is_file(): + raise RuntimeError(f"cache filesystem is not mounted at {CACHE_ROOT}") + files_root = CACHE_ROOT / "files" + files_root.mkdir(parents=True, exist_ok=True) + for partial in files_root.glob("*.partial"): + partial.unlink(missing_ok=True) + for output in list(files_root.glob("*.mkv")) + list(files_root.glob("*.mp4")): + if not output.with_suffix(output.suffix + ".json").is_file(): + output.unlink(missing_ok=True) + for manifest in files_root.glob("*.json"): + if not manifest.with_suffix("").is_file(): + manifest.unlink(missing_ok=True) + VIEW_ROOT.mkdir(parents=True, exist_ok=True) + self.state = load_json(STATE_FILE, {}) + self.state.setdefault("inventory", {}) + self.state.setdefault("priorities", {}) + self.state.setdefault("cooldowns", {}) + self.state.setdefault("retry_after", {}) + self.active_items = set() + self.active_keys = set() + self.previous_active_items = set() + self.previous_active_keys = set() + self.sessions_healthy = False + self.running = True + self.process = None + self.last_scan = 0 + self.last_activity_save = 0 + signal.signal(signal.SIGTERM, self.stop) + signal.signal(signal.SIGINT, self.stop) + + def stop(self, *_): + self.running = False + self.cancel_process("service stopping") + + def save(self): + atomic_json(STATE_FILE, self.state) + + def api(self, path, method="GET"): + request = urllib.request.Request( + JELLYFIN_URL + path, + method=method, + headers={"X-Emby-Token": JELLYFIN_API_KEY, "Content-Length": "0"}, + ) + with urllib.request.urlopen(request, timeout=10) as response: + data = response.read() + return json.loads(data) if data else None + + def read_sessions(self): + try: + sessions = self.api("/Sessions") + except Exception as error: + if self.sessions_healthy: + log(f"Jellyfin sessions unavailable; stopping cache work: {error}") + self.sessions_healthy = False + self.active_items = set() + self.active_keys = set() + return False + + now = time.time() + current_items = set() + current_keys = set() + for session in sessions or []: + item = session.get("NowPlayingItem") or {} + path = item.get("Path") + relative = relative_source(path) if path else None + if not relative and item.get("Id"): + try: + detail = self.api("/Items/" + item["Id"] + "?Fields=Path") or {} + relative = relative_source(detail.get("Path", "")) + except Exception: + relative = None + if not relative: + continue + current_items.add(relative) + current_keys.add(item_key(relative)) + self.touch_cache(relative) + + ended_items = self.previous_active_items - current_items + ended_keys = self.previous_active_keys - current_keys + changed = current_items != self.previous_active_items or current_keys != self.previous_active_keys + for relative in current_items | ended_items: + self.state["priorities"][relative] = now + for key in current_keys | ended_keys: + if key.startswith("series:"): + self.state["cooldowns"][key] = now + SERIES_COOLDOWN_SECONDS + if changed or (current_items and now - self.last_activity_save >= 60): + self.save() + self.last_activity_save = now + + self.previous_active_items = current_items + self.previous_active_keys = current_keys + self.active_items = current_items + self.active_keys = current_keys + self.sessions_healthy = True + return True + + def locked_keys(self): + now = time.time() + expired = [key for key, until in self.state["cooldowns"].items() if float(until) <= now] + if expired: + for key in expired: + self.state["cooldowns"].pop(key, None) + self.save() + return self.active_keys | { + key for key, until in self.state["cooldowns"].items() if float(until) > now + } + + def is_locked(self, relative): + return item_key(relative) in self.locked_keys() + + def cache_paths(self, relative): + output = CACHE_ROOT / "files" / cache_basename(relative) + manifest = output.with_suffix(output.suffix + ".json") + return output, manifest + + def cache_valid(self, relative): + source = SOURCE_ROOT / relative + output, manifest_path = self.cache_paths(relative) + if not source.is_file() or not output.is_file(): + return False + manifest = load_json(manifest_path, {}) + try: + return manifest.get("source_fingerprint") == fingerprint(source) + except OSError: + return False + + def touch_cache(self, relative): + if not self.cache_valid(relative): + return + output, manifest = self.cache_paths(relative) + now = time.time() + for path in (output, manifest): + try: + os.utime(path, (now, now), follow_symlinks=False) + except OSError: + pass + + def ensure_link(self, relative, target, locked=False): + link = VIEW_ROOT / relative + link.parent.mkdir(parents=True, exist_ok=True) + try: + if link.is_symlink() and os.readlink(link) == target: + return False + except OSError: + pass + if locked and (link.exists() or link.is_symlink()): + return False + temporary = link.with_name(link.name + ".pretranscode-tmp") + try: + temporary.unlink(missing_ok=True) + os.symlink(target, temporary) + os.replace(temporary, link) + return True + finally: + temporary.unlink(missing_ok=True) + + def swap_to_original(self, relative): + return self.ensure_link(relative, source_target(relative), locked=self.is_locked(relative)) + + def swap_to_cache(self, relative): + if self.is_locked(relative) or not self.cache_valid(relative): + return False + output, manifest_path = self.cache_paths(relative) + manifest = load_json(manifest_path, {}) + changed = self.ensure_link(relative, cache_target(relative), locked=False) + if changed or manifest.get("refresh_version") != 2: + self.refresh_item(relative) + manifest["refresh_version"] = 2 + atomic_json(manifest_path, manifest) + return changed + + def refresh_item(self, relative): + item_id = (self.state["inventory"].get(relative) or {}).get("item_id") + if not item_id: + return + query = urllib.parse.urlencode({ + "metadataRefreshMode": "FullRefresh", + "imageRefreshMode": "None", + "replaceAllMetadata": "false", + "replaceAllImages": "false", + }) + try: + self.api(f"/Items/{item_id}/Refresh?{query}", method="POST") + except Exception as error: + log(f"Unable to refresh Jellyfin item {relative}: {error}") + + def probe(self, path): + command = [ + FFPROBE, "-v", "error", "-select_streams", "v:0", + "-show_entries", "stream=codec_name,bits_per_raw_sample,pix_fmt,profile", + "-show_entries", "format=duration", "-of", "json", str(path), + ] + result = subprocess.run(command, capture_output=True, text=True, timeout=120) + if result.returncode: + raise RuntimeError(result.stderr.strip() or "ffprobe failed") + data = json.loads(result.stdout) + stream = (data.get("streams") or [{}])[0] + raw_depth = stream.get("bits_per_raw_sample") + try: + depth = int(raw_depth or 0) + except ValueError: + depth = 0 + pixel = (stream.get("pix_fmt") or "").lower() + profile = (stream.get("profile") or "").lower() + if not depth: + if "12" in pixel or "12" in profile: + depth = 12 + elif "10" in pixel or "10" in profile: + depth = 10 + else: + depth = 8 + return { + "codec": (stream.get("codec_name") or "").lower(), + "bit_depth": depth, + "profile": profile, + "duration": float((data.get("format") or {}).get("duration") or 0), + } + + def jellyfin_items(self): + values = {} + start = 0 + limit = 1000 + try: + while True: + query = urllib.parse.urlencode({ + "Recursive": "true", + "IncludeItemTypes": "Movie,Episode", + "Fields": "Path,MediaStreams", + "StartIndex": start, + "Limit": limit, + }) + page = self.api("/Items?" + query) or {} + items = page.get("Items") or [] + for item in items: + relative = relative_source(item.get("Path", "")) + if not relative: + continue + video = next( + (stream for stream in item.get("MediaStreams", []) + if (stream.get("Type") or "").lower() == "video"), + {}, + ) + values[relative] = { + "item_id": item.get("Id"), + "codec": (video.get("Codec") or "").lower(), + "bit_depth": int(video.get("BitDepth") or 8), + "profile": (video.get("Profile") or "").lower(), + "duration": float(item.get("RunTimeTicks") or 0) / 10_000_000, + } + start += len(items) + if not items or start >= int(page.get("TotalRecordCount") or start): + break + except Exception as error: + log(f"Unable to read Jellyfin inventory: {error}") + return values + + def scan_inventory(self): + catalog = self.jellyfin_items() + inventory = self.state["inventory"] + seen = set() + added = False + for root, directories, files in os.walk(SOURCE_ROOT): + directories.sort() + files.sort() + for filename in files: + if Path(filename).suffix.lower() not in VIDEO_EXTENSIONS: + continue + source = Path(root) / filename + relative = source.relative_to(SOURCE_ROOT).as_posix() + seen.add(relative) + try: + current_fingerprint = fingerprint(source) + except OSError: + continue + old = inventory.get(relative) or {} + if old.get("fingerprint") == current_fingerprint: + if relative in catalog and catalog[relative].get("item_id"): + old["item_id"] = catalog[relative]["item_id"] + continue + details = catalog.get(relative) + view = VIEW_ROOT / relative + view_uses_cache = view.is_symlink() and os.readlink(view) == cache_target(relative) + if details and not old and not view_uses_cache: + details = dict(details) + else: + try: + details = self.probe(source) + except Exception as error: + log(f"Probe failed for {relative}: {error}") + self.state["retry_after"][relative] = time.time() + 21600 + continue + if relative in catalog: + details["item_id"] = catalog[relative].get("item_id") + details["fingerprint"] = current_fingerprint + details["size"] = source.stat().st_size + details["mtime"] = source.stat().st_mtime + inventory[relative] = details + self.state["retry_after"].pop(relative, None) + added = True + + removed = set(inventory) - seen + for relative in removed: + if self.is_locked(relative): + continue + self.remove_cache(relative) + inventory.pop(relative, None) + self.state["priorities"].pop(relative, None) + self.state["retry_after"].pop(relative, None) + added = True + if added: + self.save() + return added + + def sync_view(self): + changed = False + source_files = set() + for root, directories, files in os.walk(SOURCE_ROOT): + relative_root = Path(root).relative_to(SOURCE_ROOT) + (VIEW_ROOT / relative_root).mkdir(parents=True, exist_ok=True) + directories.sort() + files.sort() + for filename in files: + relative = (relative_root / filename).as_posix() + source_files.add(relative) + is_video = Path(filename).suffix.lower() in VIDEO_EXTENSIONS + valid_cache = is_video and self.cache_valid(relative) + if valid_cache and not self.is_locked(relative): + changed |= self.swap_to_cache(relative) + else: + changed |= self.ensure_link( + relative, source_target(relative), locked=is_video and self.is_locked(relative) + ) + + for root, directories, files in os.walk(VIEW_ROOT, topdown=False): + for filename in files: + path = Path(root) / filename + relative = path.relative_to(VIEW_ROOT).as_posix() + if relative.endswith(".pretranscode-tmp"): + path.unlink(missing_ok=True) + elif relative not in source_files and path.is_symlink() and not self.is_locked(relative): + path.unlink(missing_ok=True) + changed = True + for directory in directories: + path = Path(root) / directory + try: + path.rmdir() + except OSError: + pass + return changed + + def incompatible(self, info): + codec = (info.get("codec") or "").lower() + depth = int(info.get("bit_depth") or 8) + profile = (info.get("profile") or "").lower() + return codec in BLACKLISTED_CODECS or (codec == "h264" and (depth > 8 or "10" in profile)) + + def candidate_list(self): + now = time.time() + values = [] + for relative, info in self.state["inventory"].items(): + if not self.incompatible(info) or self.cache_valid(relative) or self.is_locked(relative): + continue + if float(self.state["retry_after"].get(relative, 0)) > now: + continue + priority = self.state["priorities"].get(relative) + values.append((0 if priority is not None else 1, -float(priority or info.get("mtime", 0)), relative)) + values.sort() + return values + + def cache_entries(self): + entries = [] + for manifest_path in (CACHE_ROOT / "files").glob("*.json"): + manifest = load_json(manifest_path, {}) + relative = manifest.get("relative") + if not relative: + continue + output = manifest_path.with_suffix("") + if not output.is_file(): + manifest_path.unlink(missing_ok=True) + continue + entries.append((output.stat().st_mtime, relative, output, manifest_path)) + return sorted(entries) + + def remove_cache(self, relative): + if self.is_locked(relative): + return False + output, manifest = self.cache_paths(relative) + link = VIEW_ROOT / relative + if link.is_symlink(): + try: + if os.readlink(link) == cache_target(relative): + self.ensure_link(relative, source_target(relative), locked=False) + self.refresh_item(relative) + except OSError: + pass + output.unlink(missing_ok=True) + manifest.unlink(missing_ok=True) + return True + + def ensure_space(self, required, may_evict): + free = os.statvfs(CACHE_ROOT).f_bavail * os.statvfs(CACHE_ROOT).f_frsize + if free >= required: + return True + if not may_evict: + return False + for _, relative, _, _ in self.cache_entries(): + if self.is_locked(relative): + continue + log(f"Evicting {relative}") + self.remove_cache(relative) + free = os.statvfs(CACHE_ROOT).f_bavail * os.statvfs(CACHE_ROOT).f_frsize + if free >= required: + return True + return False + + def cancel_process(self, reason): + if not self.process or self.process.poll() is not None: + return + log(f"Stopping encode: {reason}") + try: + os.killpg(self.process.pid, signal.SIGTERM) + self.process.wait(timeout=10) + except (ProcessLookupError, subprocess.TimeoutExpired): + try: + os.killpg(self.process.pid, signal.SIGKILL) + except ProcessLookupError: + pass + self.process = None + + def ffmpeg_command(self, source, partial): + common = [ + FFMPEG, "-nostdin", "-hide_banner", "-loglevel", "warning", "-i", str(source), + "-map", "0:v:0", "-map", "0:a?", "-map", "0:s?", "-map_metadata", "0", "-map_chapters", "0", + "-c:v", "libx264", "-preset", H264_PRESET, "-crf", H264_CRF, + "-maxrate", H264_MAXRATE, "-bufsize", H264_BUFSIZE, + "-pix_fmt", "yuv420p", "-profile:v", "high", "-level:v", "4.1", + "-c:a", "copy", "-c:s", "copy", + ] + if partial.name.endswith(".mp4.partial"): + common += ["-movflags", "+faststart", "-f", "mp4"] + else: + common += ["-map", "0:t?", "-c:t", "copy", "-f", "matroska"] + return common + [str(partial)] + + def validate_output(self, source, output, expected_duration): + details = self.probe(output) + if details.get("codec") != "h264" or int(details.get("bit_depth") or 0) > 8: + raise RuntimeError("output is not 8-bit H.264") + source_duration = float(expected_duration or 0) + output_duration = float(details.get("duration") or 0) + if source_duration and abs(source_duration - output_duration) > max(3, source_duration * 0.01): + raise RuntimeError(f"duration mismatch: source={source_duration}, output={output_duration}") + if output.stat().st_size < 1_000_000: + raise RuntimeError("output is unexpectedly small") + + def encode(self, relative, priority): + source = SOURCE_ROOT / relative + if not source.is_file() or self.is_locked(relative): + return "skipped" + output, manifest_path = self.cache_paths(relative) + partial = output.with_name(output.name + ".partial") + partial.unlink(missing_ok=True) + info = self.state["inventory"].get(relative) or {} + estimate = min(max(int(source.stat().st_size * 2.2), 1_000_000_000), 100_000_000_000) + if not self.ensure_space(estimate, may_evict=priority): + return "full" + + log(f"Encoding {relative}") + command = self.ffmpeg_command(source, partial) + log_path = STATE_ROOT / "ffmpeg.log" + with log_path.open("ab", buffering=0) as ffmpeg_log: + self.process = subprocess.Popen( + command, stdout=ffmpeg_log, stderr=ffmpeg_log, start_new_session=True + ) + last_mirror = 0 + while self.running and self.process.poll() is None: + time.sleep(POLL_SECONDS) + if not self.read_sessions() or self.is_locked(relative): + self.cancel_process("item or series is being watched") + partial.unlink(missing_ok=True) + return "cancelled" + if time.time() - last_mirror >= 60: + if self.sync_view(): + try: + self.api("/Library/Refresh", method="POST") + except Exception as error: + log(f"Unable to request library refresh: {error}") + last_mirror = time.time() + returncode = self.process.wait() if self.process else -1 + self.process = None + + if not self.running: + partial.unlink(missing_ok=True) + return "cancelled" + if returncode != 0: + partial.unlink(missing_ok=True) + raise RuntimeError(f"ffmpeg exited with status {returncode}") + if self.is_locked(relative): + partial.unlink(missing_ok=True) + return "cancelled" + + self.validate_output(source, partial, info.get("duration")) + os.replace(partial, output) + atomic_json(manifest_path, { + "relative": relative, + "source_fingerprint": fingerprint(source), + "created_at": time.time(), + }) + if self.is_locked(relative): + output.unlink(missing_ok=True) + manifest_path.unlink(missing_ok=True) + return "cancelled" + self.swap_to_cache(relative) + self.state["priorities"].pop(relative, None) + self.state["retry_after"].pop(relative, None) + self.save() + log(f"Completed {relative}") + return "completed" + + def run(self): + if not JELLYFIN_API_KEY: + raise RuntimeError("JELLYFIN_API_KEY is required") + self.read_sessions() + self.scan_inventory() + self.sync_view() + self.save() + while self.running: + if not self.read_sessions(): + time.sleep(POLL_SECONDS) + continue + now = time.time() + if now - self.last_scan >= RESCAN_SECONDS: + changed = self.scan_inventory() + changed |= self.sync_view() + if changed: + try: + self.api("/Library/Refresh", method="POST") + except Exception as error: + log(f"Unable to request library refresh: {error}") + self.last_scan = now + + candidates = self.candidate_list() + if not candidates: + time.sleep(POLL_SECONDS) + continue + selected = None + for rank, _, relative in candidates: + source = SOURCE_ROOT / relative + estimate = min(max(int(source.stat().st_size * 2.2), 1_000_000_000), 100_000_000_000) + free = os.statvfs(CACHE_ROOT).f_bavail * os.statvfs(CACHE_ROOT).f_frsize + if rank == 0 or free >= estimate: + selected = (rank == 0, relative) + break + if not selected: + time.sleep(60) + continue + priority, relative = selected + try: + result = self.encode(relative, priority) + if result == "full": + time.sleep(60) + except Exception as error: + log(f"Encode failed for {relative}: {error}") + self.state["retry_after"][relative] = time.time() + 21600 + self.save() + output, manifest = self.cache_paths(relative) + output.with_name(output.name + ".partial").unlink(missing_ok=True) + time.sleep(POLL_SECONDS) + + def sync_once(self): + self.scan_inventory() + self.sync_view() + self.save() + + def active(self): + if not self.read_sessions(): + return True + return bool(self.active_items) + + def trim(self, target_bytes): + if self.active(): + raise RuntimeError("cache resize refused while media is playing") + target = max(0, int(target_bytes) - 5_000_000_000) + entries = self.cache_entries() + used = sum(output.stat().st_size + manifest.stat().st_size for _, _, output, manifest in entries) + for _, relative, output, manifest in entries: + if used <= target: + break + size = output.stat().st_size + manifest.stat().st_size + if self.remove_cache(relative): + log(f"Trimmed {relative}") + used -= size + if used > target: + raise RuntimeError("unable to trim cache below target because content is locked") + self.sync_view() + self.save() + + +def main(): + manager = Manager() + command = sys.argv[1] if len(sys.argv) > 1 else "run" + if command == "run": + manager.run() + elif command == "sync": + manager.sync_once() + elif command == "active": + raise SystemExit(0 if not manager.active() else 1) + elif command == "trim" and len(sys.argv) == 3: + manager.trim(int(sys.argv[2])) + else: + raise SystemExit(f"unknown command: {command}") + + +if __name__ == "__main__": + main() diff --git a/scripts/provision-cache.sh b/scripts/provision-cache.sh new file mode 100755 index 0000000..e02501f --- /dev/null +++ b/scripts/provision-cache.sh @@ -0,0 +1,42 @@ +#!/bin/bash +set -euo pipefail + +[ "${EUID}" -eq 0 ] || { echo "Run as root." >&2; exit 1; } +ENV_FILE=${1:-.env} +[ -f "$ENV_FILE" ] || { echo "Missing $ENV_FILE" >&2; exit 1; } +set -a +. "$ENV_FILE" +set +a + +: "${PRETRANSCODE_ROOT:?}" +: "${PRETRANSCODE_VIEW_PATH:?}" +: "${PRETRANSCODE_CACHE_PATH:?}" +: "${PRETRANSCODE_STATE_PATH:?}" +: "${PRETRANSCODE_CACHE_IMAGE:?}" +: "${CACHE_MAX_BYTES:?}" +: "${PRETRANSCODE_UID:?}" +: "${PRETRANSCODE_GID:?}" + +mkdir -p "$PRETRANSCODE_ROOT" "$PRETRANSCODE_VIEW_PATH" \ + "$PRETRANSCODE_CACHE_PATH" "$PRETRANSCODE_STATE_PATH" + +if [ ! -f "$PRETRANSCODE_CACHE_IMAGE" ]; then + fallocate -l "$CACHE_MAX_BYTES" "$PRETRANSCODE_CACHE_IMAGE" + mkfs.ext4 -q -F -m 0 -L pretranscode "$PRETRANSCODE_CACHE_IMAGE" + # mkfs may discard unused extents; claim the configured allocation again. + fallocate -l "$CACHE_MAX_BYTES" "$PRETRANSCODE_CACHE_IMAGE" +elif [ "$(stat -c %s "$PRETRANSCODE_CACHE_IMAGE")" -ne "$CACHE_MAX_BYTES" ]; then + echo "Cache image exists with a different size; resize it explicitly before provisioning." >&2 + exit 1 +fi + +FSTAB_LINE="$PRETRANSCODE_CACHE_IMAGE $PRETRANSCODE_CACHE_PATH ext4 loop,nofail,noatime 0 2" +grep -qF "$FSTAB_LINE" /etc/fstab || echo "$FSTAB_LINE" >> /etc/fstab +mountpoint -q "$PRETRANSCODE_CACHE_PATH" || mount "$PRETRANSCODE_CACHE_PATH" +touch "$PRETRANSCODE_CACHE_PATH/.pretranscode-cache" +chown -R "$PRETRANSCODE_UID:$PRETRANSCODE_GID" \ + "$PRETRANSCODE_VIEW_PATH" "$PRETRANSCODE_STATE_PATH" "$PRETRANSCODE_CACHE_PATH" +chmod 0755 "$PRETRANSCODE_VIEW_PATH" "$PRETRANSCODE_STATE_PATH" "$PRETRANSCODE_CACHE_PATH" + +echo "Cache provisioned at $PRETRANSCODE_CACHE_PATH" +df -h "$PRETRANSCODE_CACHE_PATH" diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/integration.sh b/tests/integration.sh new file mode 100755 index 0000000..0eee6eb --- /dev/null +++ b/tests/integration.sh @@ -0,0 +1,30 @@ +#!/bin/bash +set -euo pipefail +: "${JELLYFIN_API_KEY:?}" +IMAGE=${IMAGE:-pretranscode:test} +NETWORK=${MEDIA_NETWORK:-media-internal} +JELLYFIN_URL=${JELLYFIN_URL:-http://jellyfin:8096} +ROOT=$(mktemp -d) +trap 'rm -rf "$ROOT"' EXIT +mkdir -p "$ROOT"/{source/movies/Test,view,cache,state} +touch "$ROOT/cache/.pretranscode-cache" + +docker build -q -t "$IMAGE" . >/dev/null +docker run --rm --entrypoint ffmpeg -v "$ROOT/source:/source" "$IMAGE" \ + -hide_banner -loglevel error -f lavfi -i testsrc2=size=1280x720:rate=24 \ + -t 6 -c:v libx265 -pix_fmt yuv420p10le -an /source/movies/Test/Test.mkv + +docker run --rm --network "$NETWORK" --entrypoint python3 \ + -e SOURCE_ROOT=/source -e VIEW_ROOT=/view -e CACHE_ROOT=/cache -e STATE_ROOT=/state \ + -e JELLYFIN_URL="$JELLYFIN_URL" -e JELLYFIN_API_KEY \ + -e JELLYFIN_LIBRARY_ROOT=/data/library -e JELLYFIN_ORIGINAL_ROOT=/data/original \ + -e JELLYFIN_CACHE_ROOT=/data/cache \ + -v "$ROOT/source:/source:ro" -v "$ROOT/view:/view" -v "$ROOT/cache:/cache" -v "$ROOT/state:/state" \ + "$IMAGE" -c 'import importlib.machinery; m=importlib.machinery.SourceFileLoader("p","/app/pretranscode").load_module(); x=m.Manager(); assert x.read_sessions(); x.scan_inventory(); x.sync_view(); assert x.encode("movies/Test/Test.mkv",False)=="completed"' + +OUTPUT=$(find "$ROOT/cache/files" -name '*.mkv' ! -name '*.partial' | head -1) +docker run --rm --entrypoint ffprobe -v "$ROOT/cache:/cache:ro" "$IMAGE" \ + -v error -select_streams v:0 -show_entries stream=codec_name,pix_fmt \ + -of default=nw=1 "/cache/files/$(basename "$OUTPUT")" | grep -qx 'codec_name=h264' +test "$(readlink "$ROOT/view/movies/Test/Test.mkv")" = "/data/cache/files/$(basename "$OUTPUT")" +echo "integration test passed" diff --git a/tests/test_core.py b/tests/test_core.py new file mode 100644 index 0000000..696bb6d --- /dev/null +++ b/tests/test_core.py @@ -0,0 +1,69 @@ +import importlib.util +import json +import os +import tempfile +import unittest +from pathlib import Path + + +class CoreTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.temp = tempfile.TemporaryDirectory() + root = Path(cls.temp.name) + for name in ("source", "view", "cache", "state"): + (root / name).mkdir() + (root / "cache" / ".pretranscode-cache").touch() + os.environ.update({ + "SOURCE_ROOT": str(root / "source"), + "VIEW_ROOT": str(root / "view"), + "CACHE_ROOT": str(root / "cache"), + "STATE_ROOT": str(root / "state"), + "JELLYFIN_LIBRARY_ROOT": "/library", + "JELLYFIN_ORIGINAL_ROOT": "/original", + "JELLYFIN_CACHE_ROOT": "/cached", + }) + spec = importlib.util.spec_from_file_location("pretranscode", Path(__file__).parents[1] / "pretranscode.py") + cls.module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(cls.module) + cls.root = root + + @classmethod + def tearDownClass(cls): + cls.temp.cleanup() + + def test_paths_are_configurable(self): + m = self.module + self.assertEqual(m.relative_source("/library/tv/Show/episode.mkv"), "tv/Show/episode.mkv") + self.assertEqual(m.source_target("tv/Show/episode.mkv"), "/original/tv/Show/episode.mkv") + self.assertTrue(m.cache_target("tv/Show/episode.mkv").startswith("/cached/files/")) + + def test_series_lock_key(self): + m = self.module + self.assertEqual(m.item_key("tv/Show/Season 1/Episode.mkv"), "series:Show") + self.assertEqual(m.item_key("movies/Movie/file.mkv"), "item:movies/Movie/file.mkv") + + def test_fixed_codec_blacklist(self): + manager = self.module.Manager() + self.assertTrue(manager.incompatible({"codec": "hevc", "bit_depth": 10})) + self.assertTrue(manager.incompatible({"codec": "h264", "bit_depth": 10})) + self.assertFalse(manager.incompatible({"codec": "h264", "bit_depth": 8})) + self.assertFalse(manager.incompatible({"codec": "av1", "bit_depth": 10})) + + def test_cache_requires_matching_source_fingerprint(self): + m = self.module + manager = m.Manager() + relative = "movies/Test/Test.mkv" + source = self.root / "source" / relative + source.parent.mkdir(parents=True, exist_ok=True) + source.write_bytes(b"original") + output, manifest = manager.cache_paths(relative) + output.write_bytes(b"cached") + manifest.write_text(json.dumps({"relative": relative, "source_fingerprint": m.fingerprint(source)})) + self.assertTrue(manager.cache_valid(relative)) + source.write_bytes(b"changed") + self.assertFalse(manager.cache_valid(relative)) + + +if __name__ == "__main__": + unittest.main()