#!/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()