commit 093315b38217dd532f04c56a1d857a97b368cfd9 Author: SpiderFish Date: Wed Aug 26 09:53:42 2026 +1000 Initial portable pretranscode service 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()