MrDevBot/pretranscode
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial portable pretranscode service

Browse Source
This commit is contained in:
SpiderFish
2026-08-26 09:53:42 +10:00
commit 1421300f32
11 changed files with 1077 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+33
View File
@@ -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
.gitignore
+3
View File
@@ -0,0 +1,3 @@
.env
__pycache__/
*.pyc
Dockerfile
+12
View File
@@ -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"]
LICENSE
+21
View File
@@ -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.
README.md
+97
View File
@@ -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.
compose.yaml
+51
View File
@@ -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}
pretranscode.py
Executable
+719
View File
@@ -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()
scripts/provision-cache.sh
Executable
+42
View File
@@ -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"
tests/__init__.py
View File
tests/integration.sh
Executable
+30
View File
@@ -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"
tests/test_core.py
+69
View File
@@ -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()