#!/usr/bin/env python3 """Immutable Jellyfin H.264 cache backed by an atomic symlink view.""" import hashlib import json import os import shutil 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", "") CACHE_MAX_BYTES = env_int("CACHE_MAX_BYTES", 1_000_000_000_000) 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", "18") H264_PRESET = os.environ.get("H264_PRESET", "medium") H264_MAXRATE = os.environ.get("H264_MAXRATE", "15M") H264_BUFSIZE = os.environ.get("H264_BUFSIZE", "30M") LEGACY_ENCODE_PROFILE = f"h264-high-8bit-crf{H264_CRF}-{H264_PRESET}-{H264_MAXRATE}-quality-v3" ENCODE_PROFILE = f"h264-high-8bit-crf{H264_CRF}-{H264_PRESET}-{H264_MAXRATE}-quality-v4" 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) and manifest.get("encode_profile") == ENCODE_PROFILE ) except OSError: return False def touch_cache(self, relative): if not self.cache_valid(relative): return _, manifest = self.cache_paths(relative) try: os.utime(manifest, None, 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,color_transfer,color_primaries,color_space:" "stream_side_data=side_data_type,dv_profile,rpu_present_flag,el_present_flag,bl_present_flag:" "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 side_data = stream.get("side_data_list") or [] return { "codec": (stream.get("codec_name") or "").lower(), "bit_depth": depth, "profile": profile, "color_transfer": (stream.get("color_transfer") or "").lower(), "color_primaries": (stream.get("color_primaries") or "").lower(), "color_space": (stream.get("color_space") or "").lower(), "dolby_vision": any( "dovi" in (value.get("side_data_type") or "").lower() for value in side_data ), "duration": float((data.get("format") or {}).get("duration") or 0), } def source_info(self, relative): source = SOURCE_ROOT / relative details = self.probe(source) previous = self.state["inventory"].get(relative) or {} if previous.get("item_id"): details["item_id"] = previous["item_id"] details["fingerprint"] = fingerprint(source) stat = source.stat() details["size"] = stat.st_size details["mtime"] = stat.st_mtime self.state["inventory"][relative] = details return details 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(), "color_transfer": (video.get("ColorTransfer") or "").lower(), "color_primaries": (video.get("ColorPrimaries") or "").lower(), "color_space": (video.get("ColorSpace") 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 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 is_hdr(self, info): return (info.get("color_transfer") or "").lower() in {"smpte2084", "arib-std-b67"} 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((manifest_path.stat().st_mtime, relative, output, manifest_path)) return sorted(entries) def remove_cache(self, relative, force=False): if not force and 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 cache_usage(self): return sum( path.stat().st_size for path in (CACHE_ROOT / "files").iterdir() if path.is_file() ) def cache_available(self): stat = os.statvfs(CACHE_ROOT) filesystem_free = stat.f_bavail * stat.f_frsize configured_free = max(0, CACHE_MAX_BYTES - self.cache_usage()) return min(filesystem_free, configured_free) def purge_incompatible_cache_profiles(self): for _, relative, _, _ in self.cache_entries(): if self.cache_valid(relative) or self.is_locked(relative): continue log(f"Dropping obsolete cache rendition for {relative}") self.remove_cache(relative) def enforce_cache_limit(self): usage = self.cache_usage() for _, relative, output, manifest in self.cache_entries(): if usage <= CACHE_MAX_BYTES: break if self.is_locked(relative): continue size = output.stat().st_size + manifest.stat().st_size log(f"Evicting {relative} after cache resize") if self.remove_cache(relative): usage -= size def ensure_space(self, required, may_evict): if self.cache_available() >= 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) if self.cache_available() >= 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, info): hdr = self.is_hdr(info) if hdr: transfer = (info.get("color_transfer") or "").lower() if transfer == "arib-std-b67": input_colors = "tin=arib-std-b67:pin=bt2020:min=bt2020nc:" else: input_colors = "" video_filter = ( f"zscale={input_colors}t=linear:npl=100,format=gbrpf32le," "zscale=p=bt709,tonemap=tonemap=hable:desat=0," "zscale=t=bt709:m=bt709:r=tv:dither=error_diffusion,format=yuv420p" ) else: video_filter = "zscale=dither=error_diffusion,format=yuv420p" common = [ FFMPEG, "-nostdin", "-hide_banner", "-loglevel", "warning", "-i", str(source), "-filter_complex", f"[0:v:0]{video_filter}[video]", "-map", "[video]", "-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", "-c:a", "copy", "-c:s", "copy", ] if hdr: common += [ "-color_primaries", "bt709", "-color_trc", "bt709", "-colorspace", "bt709", "-color_range", "tv", ] 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") if details.get("dolby_vision"): raise RuntimeError("output incorrectly retains Dolby Vision stream metadata") 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 video_stream_hash(self, path): command = [ FFMPEG, "-nostdin", "-hide_banner", "-loglevel", "error", "-i", str(path), "-map", "0:v:0", "-c:v", "copy", "-bsf:v", "h264_mp4toannexb", "-f", "hash", "-hash", "sha256", "-", ] result = subprocess.run(command, capture_output=True, text=True, timeout=3600) if result.returncode: raise RuntimeError(result.stderr.strip() or "unable to hash video stream") return result.stdout.strip() def sanitize_cache_metadata(self, relative, output): if output.suffix.lower() not in {".mkv", ".mp4"}: raise RuntimeError(f"unsupported cache container for metadata repair: {output.suffix}") stage = output.with_name(output.name + ".sanitize-video.mp4.partial") repaired = output.with_name(output.name + ".sanitize.partial") stage.unlink(missing_ok=True) repaired.unlink(missing_ok=True) original_hash = self.video_stream_hash(output) log(f"Losslessly removing inherited Dolby Vision metadata from {relative}") first = [ FFMPEG, "-nostdin", "-hide_banner", "-loglevel", "warning", "-i", str(output), "-map", "0:v:0", "-c:v", "copy", "-an", "-sn", "-dn", "-map_metadata", "-1", "-map_chapters", "-1", "-movflags", "+faststart", "-f", "mp4", str(stage), ] second = [ FFMPEG, "-nostdin", "-hide_banner", "-loglevel", "warning", "-i", str(stage), "-i", str(output), "-map", "0:v:0", "-map", "1:a?", "-map", "1:s?", "-map_metadata", "1", "-map_chapters", "1", "-c", "copy", ] if output.suffix.lower() == ".mkv": second += ["-map", "1:t?", "-c:t", "copy", "-f", "matroska"] else: second += ["-movflags", "+faststart", "-f", "mp4"] second.append(str(repaired)) log_path = STATE_ROOT / "ffmpeg.log" try: with log_path.open("ab", buffering=0) as ffmpeg_log: for command in (first, second): result = subprocess.run(command, stdout=ffmpeg_log, stderr=ffmpeg_log) if result.returncode: raise RuntimeError(f"metadata repair ffmpeg exited with status {result.returncode}") details = self.probe(repaired) if details.get("dolby_vision"): raise RuntimeError("metadata repair did not remove Dolby Vision stream metadata") if self.video_stream_hash(repaired) != original_hash: raise RuntimeError("metadata repair changed the encoded video stream") os.replace(repaired, output) finally: stage.unlink(missing_ok=True) repaired.unlink(missing_ok=True) def migrate_v4(self): if self.active(): raise RuntimeError("manifest migration refused while media is playing") backup = STATE_FILE.with_name(f"state.json.pre-v4-{int(time.time())}") if STATE_FILE.is_file(): shutil.copy2(STATE_FILE, backup) log(f"Backed up state to {backup}") migrated = 0 sanitized = 0 skipped = 0 for _, relative, output, manifest_path in self.cache_entries(): manifest = load_json(manifest_path, {}) if manifest.get("encode_profile") not in {LEGACY_ENCODE_PROFILE, ENCODE_PROFILE}: log(f"Skipping manifest with unsupported profile: {relative}") skipped += 1 continue source = SOURCE_ROOT / relative if not source.is_file() or manifest.get("source_fingerprint") != fingerprint(source): log(f"Skipping stale manifest: {relative}") skipped += 1 continue source_details = self.source_info(relative) output_details = self.probe(output) if output_details.get("dolby_vision"): self.sanitize_cache_metadata(relative, output) manifest.pop("refresh_version", None) sanitized += 1 self.validate_output(source, output, source_details.get("duration")) manifest["encode_profile"] = ENCODE_PROFILE manifest["migration_version"] = 4 manifest["migrated_at"] = time.time() atomic_json(manifest_path, manifest) migrated += 1 self.save() log(f"Manifest migration complete: migrated={migrated}, sanitized={sanitized}, skipped={skipped}") 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.source_info(relative) 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, info) 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(), "encode_profile": ENCODE_PROFILE, }) 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.purge_incompatible_cache_profiles() self.enforce_cache_limit() self.scan_inventory() self.sync_view() self.save() while self.running: if not self.read_sessions(): time.sleep(POLL_SECONDS) continue self.purge_incompatible_cache_profiles() self.enforce_cache_limit() 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) if rank == 0 or self.cache_available() >= 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 purge(self, force=False): if not force and self.active(): raise RuntimeError("cache purge refused while media is playing") for _, relative, _, _ in self.cache_entries(): log(f"Purging {relative}") self.remove_cache(relative, force=force) self.sync_view() self.save() 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])) elif command == "purge": manager.purge(force="--force" in sys.argv[2:]) elif command == "migrate-v4": manager.migrate_v4() else: raise SystemExit(f"unknown command: {command}") if __name__ == "__main__": main()