720 lines
28 KiB
Python
Executable File
720 lines
28 KiB
Python
Executable File
#!/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()
|