-
Submit YouTube Playlist
+
Submit YouTube Media
- Paste a playlist link below. It’ll be queued for download and assigned to the next free slot.
+ Paste a YouTube video or playlist link and choose a download type.
{% with messages = get_flashed_messages() %}
@@ -212,12 +259,40 @@ HTML_FORM = """
>
-
Submit Playlist
+
+ Download as:
+
+
+
+
+
Submit
Playlists are kept for roughly 48 hours after they’re downloaded.
+ Utility exports are kept until an operator removes them.
+
+ {% if export_jobs %}
+
+ Recent utility exports
+ {% for job in export_jobs %}
+
+
{{ download_mode_labels[job.mode] }} — {{ export_status_labels[job.status] }} (job {{ job.job_id[:8] }})
+ {% for path in job.output_paths %}
+
{{ path }}
+ {% endfor %}
+ {% if job.error %}
+
{{ job.error }}
+ {% endif %}
+
+ {% endif %}
@@ -238,6 +313,19 @@ MAX_URL_LENGTH = 2048
MAX_SUBMITTER_LENGTH = 200
MAX_NOTE_LENGTH = 1000
MAX_REQUEST_CONTENT_LENGTH = 32 * 1024
+DOWNLOAD_MODE_LABELS = {
+ "player_slot": "Player Slot",
+ "mp4": "MP4 Export",
+ "m4a": "M4A Audio",
+ "mp3": "MP3 Audio",
+}
+EXPORT_STATUS_LABELS = {
+ "queued": "queued",
+ "running": "downloading",
+ "completed": "completed",
+ "failed": "failed",
+ "interrupted": "interrupted",
+}
def _valid_media_url(value: str) -> bool:
@@ -258,6 +346,20 @@ def _valid_media_url(value: str) -> bool:
def _submission_message(result: dict[str, object]) -> str:
status = result.get("status")
+ if result.get("kind") == "export":
+ mode = str(result.get("mode") or "")
+ label = DOWNLOAD_MODE_LABELS.get(mode, "Utility export")
+ if status == "queue_full":
+ return "The utility export queue is full; please try again later."
+ if status == "shutting_down":
+ return "The Controller is shutting down; please try again shortly."
+ if status == "worker_error":
+ return f"The {label} job was reserved, but its worker could not start."
+ if status != "queued":
+ return f"The Controller could not queue that {label} job; please try again."
+ job_id = str(result.get("job_id") or "")[:8]
+ identity = f" (job {job_id})" if job_id else ""
+ return f"{label} queued{identity}. Refresh this page to see its result."
if status == "queue_full":
seconds = max(0, int(result.get("oldest_expires_in_seconds") or 0))
hours = max(1, (seconds + 3599) // 3600)
@@ -308,7 +410,19 @@ def create_app(
@web_app.get("/")
def index():
- return render_template_string(HTML_FORM)
+ try:
+ export_jobs = service.list_export_jobs(
+ limit=min(100, config.downloads.max_queued_exports + 10)
+ )
+ except Exception as exc:
+ LOG.warning("could not load utility export status: %s", exc)
+ export_jobs = []
+ return render_template_string(
+ HTML_FORM,
+ export_jobs=export_jobs,
+ download_mode_labels=DOWNLOAD_MODE_LABELS,
+ export_status_labels=EXPORT_STATUS_LABELS,
+ )
@web_app.errorhandler(413)
def _submission_too_large(_error):
@@ -320,6 +434,8 @@ def create_app(
url = request.form.get("url", "").strip()
submitter = request.form.get("submitter", "").strip() or None
note = request.form.get("note", "").strip() or None
+ raw_mode = request.form.get("download_mode", "player_slot")
+ download_mode = raw_mode.strip() if isinstance(raw_mode, str) else ""
if not _valid_media_url(url):
flash(
@@ -333,9 +449,20 @@ def create_app(
if note is not None and len(note) > MAX_NOTE_LENGTH:
flash(f"The note must be {MAX_NOTE_LENGTH} characters or fewer.")
return redirect(url_for("index"))
+ if download_mode not in DOWNLOAD_MODE_LABELS:
+ flash("Invalid download type. Choose one of the available options.")
+ return redirect(url_for("index"))
try:
- result = service.enqueue_download(url, submitter, note)
+ if download_mode == "player_slot":
+ result = service.enqueue_download(url, submitter, note)
+ else:
+ result = service.enqueue_export(
+ url,
+ download_mode,
+ submitter,
+ note,
+ )
except ControllerUnavailableError as exc:
LOG.warning("playlist submission was rejected: %s", exc)
flash("The Controller is unavailable; please try again shortly.")
diff --git a/config.yaml b/config.yaml
index db24f49..7691495 100644
--- a/config.yaml
+++ b/config.yaml
@@ -51,6 +51,8 @@ downloads:
# Utility exports live below media_base and never participate in slots.
export_directory: exports
export_max_height: 1080
+ # Dedicated capacity prevents utility exports from starving Player work.
+ export_max_workers: 1
# Bounds queued + running utility work; Player slots retain their own bound.
max_queued_exports: 20
diff --git a/controller/config.py b/controller/config.py
index 6c3257c..f42cb85 100644
--- a/controller/config.py
+++ b/controller/config.py
@@ -65,6 +65,7 @@ class DownloadConfig:
max_height: int = 1080
export_directory: Path = Path("exports")
export_max_height: int = 1080
+ export_max_workers: int = 1
max_queued_exports: int = 20
@@ -186,6 +187,9 @@ def _relative_directory(value: Any, name: str) -> Path:
or "\\" in text
):
raise ConfigError(f"{name} must be a safe relative directory")
+ reserved_roots = {f"slot{slot}" for slot in SLOT_IDS} | {".controller-trash"}
+ if path.parts[0].casefold() in reserved_roots:
+ raise ConfigError(f"{name} must not overlap Controller slot storage")
return path
@@ -302,6 +306,11 @@ def load_config(
downloads_raw.get("export_max_height", 1080),
"downloads.export_max_height",
),
+ export_max_workers=_integer(
+ downloads_raw.get("export_max_workers", 1),
+ "downloads.export_max_workers",
+ maximum=10,
+ ),
max_queued_exports=_integer(
downloads_raw.get("max_queued_exports", 20),
"downloads.max_queued_exports",
diff --git a/controller/downloads.py b/controller/downloads.py
index 52da6cd..79170e6 100644
--- a/controller/downloads.py
+++ b/controller/downloads.py
@@ -2,14 +2,18 @@
from __future__ import annotations
+import json
import logging
import os
+import re
import signal
import shutil
+import stat
import subprocess
import tempfile
import threading
import time
+import unicodedata
import uuid
from collections.abc import Iterator
from contextlib import contextmanager
@@ -20,12 +24,31 @@ from typing import Any, Callable
from .config import ControllerConfig
from .contracts import SLOT_IDS
-from .models import DeleteToken, DownloadReservation, SlotProtectedError, StaleOperationError
+from .models import (
+ DeleteToken,
+ DownloadReservation,
+ ExportReservation,
+ SlotProtectedError,
+ StaleOperationError,
+ UTILITY_EXPORT_MODES,
+)
from .store import ControllerStore
LOG = logging.getLogger(__name__)
MEDIA_EXTENSIONS = frozenset({".mp4", ".mkv", ".webm", ".mov", ".m4v"})
+VIDEO_EXPORT_INPUT_EXTENSIONS = frozenset(
+ {*MEDIA_EXTENSIONS, ".avi", ".flv", ".ts"}
+)
+EXPORT_AUDIO_EXTENSIONS = {"m4a": ".m4a", "mp3": ".mp3"}
+MAX_EXPORT_STEM_BYTES = 180
+EXPORT_MANIFEST_NAME = "manifest.json"
+_INVALID_FILENAME_CHARACTERS = frozenset('<>:"/\\|?*')
+_WINDOWS_RESERVED_NAMES = frozenset(
+ {"CON", "PRN", "AUX", "NUL"}
+ | {f"COM{index}" for index in range(1, 10)}
+ | {f"LPT{index}" for index in range(1, 10)}
+)
@dataclass(frozen=True)
@@ -43,6 +66,16 @@ class DownloadEvent:
reused_slot: bool = False
+@dataclass(frozen=True)
+class MediaInspection:
+ format_names: frozenset[str]
+ video_codec: str | None
+ audio_codec: str | None
+ pixel_format: str | None
+ width: int | None
+ height: int | None
+
+
EventCallback = Callable[[DownloadEvent], None]
Runner = Callable[..., subprocess.CompletedProcess[str]]
@@ -63,6 +96,54 @@ def player_format_selector(max_height: int) -> str:
return f"bv*[height<={max_height}]+ba/b[height<={max_height}]"
+def mp4_export_format_selector(max_height: int) -> str:
+ """Select quality first while making native AVC/AAC the equal-quality preference."""
+
+ return f"bv*[height<={max_height}]+ba/b[height<={max_height}]"
+
+
+def safe_export_filename(
+ stem: str,
+ extension: str,
+ *,
+ used_casefolded: set[str] | None = None,
+) -> str:
+ """Return a bounded Unicode filename that cannot introduce a path component."""
+
+ if extension not in {".mp4", ".m4a", ".mp3"}:
+ raise ValueError("unsupported export filename extension")
+ normalized = unicodedata.normalize("NFC", str(stem))
+ cleaned = "".join(
+ "_"
+ if character in _INVALID_FILENAME_CHARACTERS
+ or ord(character) < 32
+ or ord(character) == 127
+ else character
+ for character in normalized
+ )
+ cleaned = re.sub(r"\s+", " ", cleaned).strip(" .") or "untitled"
+ if cleaned.split(".", 1)[0].upper() in _WINDOWS_RESERVED_NAMES:
+ cleaned = f"_{cleaned}"
+
+ def bounded(value: str, suffix: str) -> str:
+ available = MAX_EXPORT_STEM_BYTES - len(suffix.encode("utf-8"))
+ candidate = value
+ while candidate and len(candidate.encode("utf-8")) > available:
+ candidate = candidate[:-1]
+ return (candidate.rstrip(" .") or "untitled") + suffix
+
+ used = used_casefolded if used_casefolded is not None else set()
+ counter = 1
+ while True:
+ collision_suffix = "" if counter == 1 else f" ({counter})"
+ filename = bounded(cleaned, collision_suffix) + extension
+ folded = filename.casefold()
+ if folded not in used:
+ used.add(folded)
+ return filename
+ counter += 1
+
+
def title_from_filename(path: Path) -> str:
stem = path.stem
if " - " in stem:
@@ -83,6 +164,7 @@ class DownloadManager:
on_event: EventCallback | None = None,
runner: Runner | None = None,
executor: ThreadPoolExecutor | None = None,
+ export_executor: ThreadPoolExecutor | None = None,
):
self.config = config
self.store = store
@@ -93,6 +175,19 @@ class DownloadManager:
thread_name_prefix="yt-download",
)
self._owns_executor = executor is None
+ if export_executor is not None:
+ self._export_executor = export_executor
+ self._owns_export_executor = False
+ elif executor is not None:
+ # Deterministic callers/tests may inject one executor explicitly.
+ self._export_executor = executor
+ self._owns_export_executor = False
+ else:
+ self._export_executor = ThreadPoolExecutor(
+ max_workers=config.downloads.export_max_workers,
+ thread_name_prefix="yt-export",
+ )
+ self._owns_export_executor = True
self._lock = threading.Lock()
self._futures: set[Future[Any]] = set()
self._processes: set[subprocess.Popen[str]] = set()
@@ -167,6 +262,49 @@ class DownloadManager:
future.add_done_callback(self._forget_future)
return reservation
+ def enqueue_export(
+ self,
+ url: str,
+ mode: str,
+ submitter: str | None,
+ note: str | None,
+ ) -> ExportReservation:
+ """Reserve and queue a utility export without touching Player slots."""
+
+ if mode not in UTILITY_EXPORT_MODES:
+ raise ValueError("unsupported utility export mode")
+ with self._lock:
+ if self._stopping:
+ return ExportReservation(status="shutting_down")
+ reservation = self.store.reserve_export_job(
+ url,
+ mode,
+ submitter,
+ note,
+ max_active=self.config.downloads.max_queued_exports,
+ )
+ if reservation.status != "queued":
+ return reservation
+ assert reservation.job_id is not None
+ try:
+ future = self._export_executor.submit(
+ self._export,
+ url,
+ mode,
+ reservation.job_id,
+ )
+ except BaseException as exc:
+ message = f"could not start export worker: {exc}"
+ self.store.fail_export_job(reservation.job_id, message)
+ return ExportReservation(
+ status="worker_error",
+ job_id=reservation.job_id,
+ )
+ with self._lock:
+ self._futures.add(future)
+ future.add_done_callback(self._forget_future)
+ return reservation
+
def _forget_future(self, future: Future[Any]) -> None:
with self._lock:
self._futures.discard(future)
@@ -374,12 +512,636 @@ class DownloadManager:
)
)
+ def _export(
+ self,
+ url: str,
+ mode: str,
+ job_id: str,
+ ) -> None:
+ if not self.store.mark_export_running(job_id):
+ return
+ work_directory: Path | None = None
+ published = False
+ completed = False
+ try:
+ work_directory, source_directory, ready_directory, final_parent = (
+ self._prepare_export_directories(job_id, mode)
+ )
+ with self._yt_dlp_arguments() as base_arguments:
+ arguments = base_arguments + self._export_download_arguments(
+ mode,
+ source_directory,
+ url,
+ )
+ self._run_command(
+ arguments,
+ check=True,
+ text=True,
+ timeout=self.config.downloads.process_timeout_seconds,
+ )
+
+ sources = self._export_source_files(source_directory, mode)
+ if not sources:
+ raise RuntimeError("yt-dlp completed without producing export media")
+ used_names: set[str] = set()
+ ready_outputs: list[Path] = []
+ for source in sources:
+ extension = ".mp4" if mode == "mp4" else EXPORT_AUDIO_EXTENSIONS[mode]
+ output = ready_directory / safe_export_filename(
+ source.stem,
+ extension,
+ used_casefolded=used_names,
+ )
+ if mode == "mp4":
+ self._produce_compatible_mp4(source, output)
+ else:
+ inspection = self._inspect_media(source)
+ self._validate_audio_export(inspection, mode)
+ os.replace(source, output)
+ self._fsync_file(output)
+ ready_outputs.append(output)
+
+ self._fsync_directory(ready_directory)
+ self._write_export_manifest(
+ work_directory,
+ job_id=job_id,
+ mode=mode,
+ filenames=[output.name for output in ready_outputs],
+ )
+ published_directory = final_parent / job_id
+ if os.path.lexists(published_directory):
+ raise FileExistsError(
+ f"refusing to overwrite existing export directory: {published_directory}"
+ )
+ os.rename(ready_directory, published_directory)
+ self._fsync_directory(final_parent)
+ published = True
+ media_base = self.config.storage.media_base.resolve()
+ output_paths = tuple(
+ str((published_directory / output.name).resolve().relative_to(media_base))
+ for output in ready_outputs
+ )
+ if not self.store.complete_export_job(job_id, output_paths):
+ raise StaleOperationError(
+ "published export could not transition its durable job"
+ )
+ completed = True
+ except BaseException as exc:
+ message = self._failure_message(exc)
+ try:
+ self.store.fail_export_job(job_id, message)
+ except BaseException:
+ LOG.exception("could not persist utility export failure for %s", job_id)
+ if published:
+ LOG.error(
+ "utility export %s was published but needs restart reconciliation: %s",
+ job_id,
+ message,
+ )
+ else:
+ LOG.warning("utility export %s failed: %s", job_id, message)
+ finally:
+ if work_directory is not None and (not published or completed):
+ self._remove_export_work_directory(work_directory)
+
+ def _export_download_arguments(
+ self,
+ mode: str,
+ source_directory: Path,
+ url: str,
+ ) -> list[str]:
+ arguments = [
+ "--yes-playlist",
+ "--no-overwrites",
+ "--no-progress",
+ "--windows-filenames",
+ "--trim-filenames",
+ str(MAX_EXPORT_STEM_BYTES),
+ "-o",
+ str(
+ source_directory
+ / "%(playlist_index)03d - %(title)s [%(id)s].%(ext)s"
+ ),
+ ]
+ if mode == "mp4":
+ arguments.extend(
+ (
+ "-f",
+ mp4_export_format_selector(
+ self.config.downloads.export_max_height
+ ),
+ "-S",
+ (
+ f"res:{self.config.downloads.export_max_height},"
+ "vcodec:h264,acodec:aac"
+ ),
+ )
+ )
+ elif mode == "m4a":
+ arguments.extend(
+ (
+ "-f",
+ "ba/b",
+ "-S",
+ "abr,acodec:aac",
+ "-x",
+ "--audio-format",
+ "m4a",
+ "--audio-quality",
+ "0",
+ )
+ )
+ elif mode == "mp3":
+ arguments.extend(
+ (
+ "-f",
+ "ba/b",
+ "-x",
+ "--audio-format",
+ "mp3",
+ "--audio-quality",
+ "0",
+ )
+ )
+ else:
+ raise ValueError("unsupported utility export mode")
+ arguments.append(url)
+ return arguments
+
+ def _prepare_export_directories(
+ self,
+ job_id: str,
+ mode: str,
+ ) -> tuple[Path, Path, Path, Path]:
+ if not re.fullmatch(r"[0-9a-f]{32}", job_id):
+ raise ValueError("invalid export job identity")
+ media_base = self.config.storage.media_base
+ media_base.mkdir(parents=True, exist_ok=True)
+ export_root = media_base
+ for part in self.config.downloads.export_directory.parts:
+ export_root = export_root / part
+ self._ensure_real_directory(export_root)
+ work_root = export_root / ".work"
+ final_parent = export_root / ("video" if mode == "mp4" else "audio")
+ self._ensure_real_directory(work_root)
+ self._ensure_real_directory(final_parent)
+ work_directory = work_root / job_id
+ work_directory.mkdir(mode=0o750, exist_ok=False)
+ source_directory = work_directory / "source"
+ ready_directory = work_directory / "ready"
+ source_directory.mkdir(mode=0o750)
+ ready_directory.mkdir(mode=0o750)
+ self._fsync_directory(work_root)
+ return work_directory, source_directory, ready_directory, final_parent
+
+ @staticmethod
+ def _ensure_real_directory(path: Path) -> None:
+ try:
+ path.mkdir(mode=0o750)
+ except FileExistsError:
+ pass
+ metadata = path.lstat()
+ if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
+ raise RuntimeError(f"refusing unsafe export directory: {path}")
+
+ def _write_export_manifest(
+ self,
+ work_directory: Path,
+ *,
+ job_id: str,
+ mode: str,
+ filenames: list[str],
+ ) -> None:
+ if not filenames:
+ raise ValueError("export manifest must contain output filenames")
+ payload = json.dumps(
+ {
+ "version": 1,
+ "job_id": job_id,
+ "mode": mode,
+ "filenames": filenames,
+ },
+ ensure_ascii=False,
+ sort_keys=True,
+ )
+ temporary = work_directory / f".{EXPORT_MANIFEST_NAME}.tmp"
+ manifest = work_directory / EXPORT_MANIFEST_NAME
+ try:
+ with temporary.open("x", encoding="utf-8", newline="\n") as stream:
+ stream.write(payload + "\n")
+ stream.flush()
+ os.fsync(stream.fileno())
+ os.replace(temporary, manifest)
+ self._fsync_directory(work_directory)
+ finally:
+ temporary.unlink(missing_ok=True)
+
+ @staticmethod
+ def _export_source_files(source_directory: Path, mode: str) -> list[Path]:
+ allowed = (
+ VIDEO_EXPORT_INPUT_EXTENSIONS
+ if mode == "mp4"
+ else frozenset({EXPORT_AUDIO_EXTENSIONS[mode]})
+ )
+ media: list[Path] = []
+ for candidate in source_directory.iterdir():
+ if candidate.is_symlink() or not candidate.is_file():
+ raise RuntimeError(
+ f"yt-dlp produced an unsafe export entry: {candidate.name}"
+ )
+ if candidate.suffix.lower() not in allowed:
+ raise RuntimeError(
+ f"yt-dlp produced an unexpected export file: {candidate.name}"
+ )
+ media.append(candidate)
+ return sorted(media, key=lambda path: path.name.casefold())
+
+ def _inspect_media(self, path: Path) -> MediaInspection:
+ result = self._run_command(
+ [
+ "ffprobe",
+ "-v",
+ "error",
+ "-show_streams",
+ "-show_format",
+ "-of",
+ "json",
+ str(path),
+ ],
+ check=True,
+ capture_output=True,
+ text=True,
+ timeout=min(120.0, self.config.downloads.process_timeout_seconds),
+ )
+ try:
+ payload = json.loads(result.stdout or "")
+ streams = payload["streams"]
+ format_name = payload["format"]["format_name"]
+ except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc:
+ raise RuntimeError("ffprobe returned invalid media metadata") from exc
+ if not isinstance(streams, list) or not isinstance(format_name, str):
+ raise RuntimeError("ffprobe returned invalid media metadata")
+ video = next(
+ (
+ stream
+ for stream in streams
+ if isinstance(stream, dict) and stream.get("codec_type") == "video"
+ ),
+ None,
+ )
+ audio = next(
+ (
+ stream
+ for stream in streams
+ if isinstance(stream, dict) and stream.get("codec_type") == "audio"
+ ),
+ None,
+ )
+
+ def positive_integer(value: Any) -> int | None:
+ return (
+ value
+ if isinstance(value, int)
+ and not isinstance(value, bool)
+ and value > 0
+ else None
+ )
+
+ return MediaInspection(
+ format_names=frozenset(
+ value.strip().lower()
+ for value in format_name.split(",")
+ if value.strip()
+ ),
+ video_codec=(
+ video.get("codec_name") if isinstance(video, dict) else None
+ ),
+ audio_codec=(
+ audio.get("codec_name") if isinstance(audio, dict) else None
+ ),
+ pixel_format=(
+ video.get("pix_fmt") if isinstance(video, dict) else None
+ ),
+ width=positive_integer(video.get("width") if isinstance(video, dict) else None),
+ height=positive_integer(video.get("height") if isinstance(video, dict) else None),
+ )
+
+ def _produce_compatible_mp4(self, source: Path, output: Path) -> None:
+ source_info = self._inspect_media(source)
+ if source_info.video_codec is None or source_info.audio_codec is None:
+ raise RuntimeError("MP4 export requires both video and audio")
+ if source_info.height is None:
+ raise RuntimeError("MP4 export source has no verifiable video height")
+ if source_info.height > self.config.downloads.export_max_height:
+ raise RuntimeError("MP4 export source exceeds configured maximum height")
+ video_compatible = (
+ source_info.video_codec == "h264"
+ and source_info.pixel_format in {"yuv420p", "yuvj420p"}
+ )
+ audio_compatible = source_info.audio_codec == "aac"
+ container_compatible = "mp4" in source_info.format_names
+ if video_compatible and audio_compatible and container_compatible:
+ os.replace(source, output)
+ else:
+ temporary = output.with_name(
+ f".{output.stem}.{uuid.uuid4().hex}.tmp.mp4"
+ )
+ arguments = [
+ "ffmpeg",
+ "-nostdin",
+ "-hide_banner",
+ "-loglevel",
+ "error",
+ "-n",
+ "-i",
+ str(source),
+ "-map",
+ "0:v:0",
+ "-map",
+ "0:a:0",
+ "-map_metadata",
+ "0",
+ ]
+ if video_compatible:
+ arguments.extend(("-c:v", "copy"))
+ else:
+ arguments.extend(
+ (
+ "-c:v",
+ "libx264",
+ "-preset",
+ "medium",
+ "-crf",
+ "18",
+ "-pix_fmt",
+ "yuv420p",
+ )
+ )
+ if audio_compatible:
+ arguments.extend(("-c:a", "copy"))
+ else:
+ arguments.extend(("-c:a", "aac", "-b:a", "192k"))
+ arguments.extend(("-movflags", "+faststart", str(temporary)))
+ try:
+ self._run_command(
+ arguments,
+ check=True,
+ text=True,
+ timeout=self.config.downloads.process_timeout_seconds,
+ )
+ converted = self._inspect_media(temporary)
+ self._validate_compatible_mp4(converted, source_info.height)
+ os.replace(temporary, output)
+ finally:
+ temporary.unlink(missing_ok=True)
+ self._validate_compatible_mp4(
+ self._inspect_media(output),
+ source_info.height,
+ )
+
+ def _validate_compatible_mp4(
+ self,
+ inspection: MediaInspection,
+ source_height: int,
+ ) -> None:
+ if (
+ "mp4" not in inspection.format_names
+ or inspection.video_codec != "h264"
+ or inspection.audio_codec != "aac"
+ or inspection.pixel_format not in {"yuv420p", "yuvj420p"}
+ ):
+ raise RuntimeError("MP4 compatibility verification failed")
+ if (
+ inspection.height != source_height
+ or inspection.height > self.config.downloads.export_max_height
+ ):
+ raise RuntimeError("MP4 export resolution verification failed")
+
+ @staticmethod
+ def _validate_audio_export(
+ inspection: MediaInspection,
+ mode: str,
+ ) -> None:
+ expected_codec = "aac" if mode == "m4a" else "mp3"
+ expected_container = "mp4" if mode == "m4a" else "mp3"
+ if inspection.video_codec is not None:
+ raise RuntimeError("audio export unexpectedly contains video")
+ if inspection.audio_codec != expected_codec:
+ raise RuntimeError(f"{mode.upper()} export codec verification failed")
+ if expected_container not in inspection.format_names:
+ raise RuntimeError(f"{mode.upper()} export container verification failed")
+
+ @staticmethod
+ def _fsync_file(path: Path) -> None:
+ descriptor = os.open(path, os.O_RDONLY)
+ try:
+ os.fsync(descriptor)
+ finally:
+ os.close(descriptor)
+
+ def _remove_export_work_directory(self, path: Path) -> None:
+ try:
+ parent = path.parent
+ if os.path.lexists(path):
+ if path.is_symlink() or path.is_file():
+ path.unlink()
+ else:
+ shutil.rmtree(path)
+ self._fsync_directory(parent)
+ except OSError:
+ LOG.exception("could not remove export work directory %s", path)
+
+ def reconcile_export_artifacts(self) -> None:
+ """Recover verified crash-published exports and discard incomplete staging."""
+
+ export_root = self.config.storage.media_base / self.config.downloads.export_directory
+ if not os.path.lexists(export_root):
+ return
+ try:
+ self._verify_existing_export_tree(export_root)
+ except OSError as exc:
+ LOG.error("could not inspect utility export tree: %s", exc)
+ return
+ except RuntimeError as exc:
+ LOG.error("utility export reconciliation refused unsafe storage: %s", exc)
+ return
+
+ jobs = {
+ job.job_id: job for job in self.store.list_reconcilable_export_jobs()
+ }
+ for job in jobs.values():
+ try:
+ self._reconcile_export_job(export_root, job.job_id, job.mode)
+ except BaseException:
+ LOG.exception("could not reconcile utility export %s", job.job_id)
+
+ work_root = export_root / ".work"
+ if not work_root.is_dir() or work_root.is_symlink():
+ return
+ for candidate in work_root.iterdir():
+ if not re.fullmatch(r"[0-9a-f]{32}", candidate.name):
+ LOG.warning("leaving unknown utility export staging entry %s", candidate)
+ continue
+ job = self.store.get_export_job(candidate.name)
+ if job is None:
+ LOG.warning("leaving untracked utility export staging entry %s", candidate)
+ continue
+ if job.status == "completed":
+ self._remove_export_work_directory(candidate)
+ elif job.job_id not in jobs:
+ LOG.warning(
+ "leaving utility export staging for active job %s",
+ candidate.name,
+ )
+
+ def _verify_existing_export_tree(self, export_root: Path) -> None:
+ current = self.config.storage.media_base
+ for part in self.config.downloads.export_directory.parts:
+ current = current / part
+ metadata = current.lstat()
+ if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
+ raise RuntimeError(f"unsafe export directory: {current}")
+ for child_name in (".work", "video", "audio"):
+ child = export_root / child_name
+ if not os.path.lexists(child):
+ continue
+ metadata = child.lstat()
+ if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
+ raise RuntimeError(f"unsafe export directory: {child}")
+
+ def _reconcile_export_job(
+ self,
+ export_root: Path,
+ job_id: str,
+ mode: str,
+ ) -> None:
+ work_directory = export_root / ".work" / job_id
+ final_parent = export_root / ("video" if mode == "mp4" else "audio")
+ final_directory = final_parent / job_id
+ if os.path.lexists(work_directory):
+ work_metadata = work_directory.lstat()
+ if stat.S_ISLNK(work_metadata.st_mode) or not stat.S_ISDIR(
+ work_metadata.st_mode
+ ):
+ raise RuntimeError("utility export staging directory is unsafe")
+ manifest = self._read_export_manifest(work_directory, job_id, mode)
+
+ if os.path.lexists(final_directory):
+ filenames = None if manifest is None else manifest
+ outputs = self._validate_export_output_directory(
+ final_directory,
+ mode,
+ expected_filenames=filenames,
+ )
+ elif manifest is not None:
+ ready_directory = work_directory / "ready"
+ outputs = self._validate_export_output_directory(
+ ready_directory,
+ mode,
+ expected_filenames=manifest,
+ )
+ self._ensure_real_directory(final_parent)
+ os.rename(ready_directory, final_directory)
+ self._fsync_directory(final_parent)
+ else:
+ if os.path.lexists(work_directory):
+ self._remove_export_work_directory(work_directory)
+ return
+
+ media_base = self.config.storage.media_base.resolve()
+ output_paths = tuple(
+ str((final_directory / output.name).resolve().relative_to(media_base))
+ for output in outputs
+ )
+ if self.store.recover_export_job(job_id, output_paths):
+ LOG.warning("recovered crash-published utility export %s", job_id)
+ if os.path.lexists(work_directory):
+ self._remove_export_work_directory(work_directory)
+
+ @staticmethod
+ def _read_export_manifest(
+ work_directory: Path,
+ job_id: str,
+ mode: str,
+ ) -> list[str] | None:
+ manifest = work_directory / EXPORT_MANIFEST_NAME
+ if not manifest.is_file() or manifest.is_symlink():
+ return None
+ if manifest.stat().st_size > 8 * 1024 * 1024:
+ raise RuntimeError("utility export manifest is too large")
+ try:
+ payload = json.loads(manifest.read_text(encoding="utf-8"))
+ except (OSError, UnicodeError, json.JSONDecodeError) as exc:
+ raise RuntimeError("utility export manifest is invalid") from exc
+ if (
+ not isinstance(payload, dict)
+ or payload.get("version") != 1
+ or payload.get("job_id") != job_id
+ or payload.get("mode") != mode
+ ):
+ raise RuntimeError("utility export manifest identity is invalid")
+ filenames = payload.get("filenames")
+ if not isinstance(filenames, list) or not filenames or len(filenames) > 10000:
+ raise RuntimeError("utility export manifest filenames are invalid")
+ expected_extension = ".mp4" if mode == "mp4" else EXPORT_AUDIO_EXTENSIONS[mode]
+ normalized: list[str] = []
+ for filename in filenames:
+ if (
+ not isinstance(filename, str)
+ or not filename
+ or Path(filename).name != filename
+ or Path(filename).suffix.lower() != expected_extension
+ ):
+ raise RuntimeError("utility export manifest contains an unsafe filename")
+ normalized.append(filename)
+ if len({value.casefold() for value in normalized}) != len(normalized):
+ raise RuntimeError("utility export manifest contains duplicate filenames")
+ return normalized
+
+ def _validate_export_output_directory(
+ self,
+ directory: Path,
+ mode: str,
+ *,
+ expected_filenames: list[str] | None,
+ ) -> list[Path]:
+ metadata = directory.lstat()
+ if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
+ raise RuntimeError("utility export output directory is unsafe")
+ outputs = sorted(directory.iterdir(), key=lambda path: path.name.casefold())
+ if not outputs:
+ raise RuntimeError("utility export output directory is empty")
+ if expected_filenames is not None and {
+ path.name for path in outputs
+ } != set(expected_filenames):
+ raise RuntimeError("utility export output does not match its manifest")
+ expected_extension = ".mp4" if mode == "mp4" else EXPORT_AUDIO_EXTENSIONS[mode]
+ for output in outputs:
+ if (
+ output.is_symlink()
+ or not output.is_file()
+ or output.suffix.lower() != expected_extension
+ ):
+ raise RuntimeError("utility export output contains an unsafe entry")
+ inspection = self._inspect_media(output)
+ if mode == "mp4":
+ if inspection.height is None:
+ raise RuntimeError("MP4 export output has no verifiable height")
+ self._validate_compatible_mp4(inspection, inspection.height)
+ else:
+ self._validate_audio_export(inspection, mode)
+ return outputs
+
@staticmethod
def _failure_message(exc: BaseException) -> str:
+ command = getattr(exc, "cmd", None)
+ program = "media command"
+ if isinstance(command, (list, tuple)) and command:
+ program = Path(str(command[0])).name
if isinstance(exc, subprocess.TimeoutExpired):
- return f"yt-dlp timed out after {exc.timeout} seconds"
+ return f"{program} timed out after {exc.timeout} seconds"
if isinstance(exc, subprocess.CalledProcessError):
- return f"yt-dlp failed with exit code {exc.returncode}"
+ return f"{program} failed with exit code {exc.returncode}"
return f"download failed: {exc}"
def _run_command(
@@ -708,6 +1470,8 @@ class DownloadManager:
self._terminate_process(process, force=True)
if self._owns_executor:
self._executor.shutdown(wait=wait, cancel_futures=True)
+ if self._owns_export_executor:
+ self._export_executor.shutdown(wait=wait, cancel_futures=True)
def active_future_count(self) -> int:
with self._lock:
diff --git a/controller/service.py b/controller/service.py
index 1eff7ed..7ccd8ae 100644
--- a/controller/service.py
+++ b/controller/service.py
@@ -148,6 +148,12 @@ class ControllerService:
self.store,
on_event=self._queue_download_event,
)
+ try:
+ self.downloads.reconcile_export_artifacts()
+ except BaseException:
+ LOG.exception(
+ "utility export reconciliation failed; Controller startup continues"
+ )
# Cleanup is deliberately deferred until fresh Player evidence
# (or an explicit administrative reconciliation) makes it safe.
self._reset_cleanup_evidence()
@@ -252,6 +258,26 @@ class ControllerService:
{"url": url, "submitter": submitter, "note": note},
)
+ def enqueue_export(
+ self,
+ url: str,
+ mode: str,
+ submitter: str | None,
+ note: str | None,
+ ) -> dict[str, Any]:
+ return self._call(
+ "enqueue_export",
+ {
+ "url": url,
+ "mode": mode,
+ "submitter": submitter,
+ "note": note,
+ },
+ )
+
+ def list_export_jobs(self, *, limit: int = 10) -> list[dict[str, Any]]:
+ return self._call("list_export_jobs", {"limit": limit})
+
def submit_control(self, payload: Mapping[str, Any]) -> dict[str, Any]:
return self._call("control", dict(payload))
@@ -403,6 +429,35 @@ class ControllerService:
self._handle_mqtt_event(event)
def _handle_call(self, action: str, payload: Any) -> Any:
+ if action == "enqueue_export":
+ assert self.downloads is not None
+ reservation = self.downloads.enqueue_export(
+ payload["url"],
+ payload["mode"],
+ payload.get("submitter"),
+ payload.get("note"),
+ )
+ return {
+ "kind": "export",
+ "status": reservation.status,
+ "job_id": reservation.job_id,
+ "mode": payload["mode"],
+ }
+ if action == "list_export_jobs":
+ jobs = self.store.list_export_jobs(limit=payload["limit"])
+ return [
+ {
+ "job_id": job.job_id,
+ "mode": job.mode,
+ "url": job.url,
+ "status": job.status,
+ "requested_at": job.requested_at,
+ "finished_at": job.finished_at,
+ "output_paths": list(job.output_paths),
+ "error": job.error,
+ }
+ for job in jobs
+ ]
if action == "enqueue_download":
assert self.downloads is not None
url = payload["url"]
diff --git a/controller/store.py b/controller/store.py
index f3701af..21a7be5 100644
--- a/controller/store.py
+++ b/controller/store.py
@@ -1312,6 +1312,47 @@ class ControllerStore:
).rowcount
)
+ def recover_export_job(
+ self,
+ job_id: str,
+ output_paths: Sequence[str],
+ ) -> bool:
+ """Record a fully verified export found during restart reconciliation."""
+
+ normalized = tuple(output_paths)
+ if not normalized or any(
+ not isinstance(path, str) or not path for path in normalized
+ ):
+ raise ValueError("recovered export must contain output paths")
+ with self._transaction() as connection:
+ now = self._now()
+ updated = connection.execute(
+ """
+ UPDATE export_jobs
+ SET status = 'completed', finished_at = ?,
+ output_paths_json = ?, error = NULL
+ WHERE job_id = ?
+ AND status IN ('queued','running','failed','interrupted')
+ """,
+ (now, json.dumps(normalized), job_id),
+ ).rowcount
+ if updated:
+ connection.execute(
+ """
+ INSERT INTO audit_events
+ (occurred_at, event_type, actor, details_json)
+ VALUES (?, 'utility_export_recovered', 'startup', ?)
+ """,
+ (
+ now,
+ json.dumps(
+ {"job_id": job_id, "output_paths": normalized},
+ sort_keys=True,
+ ),
+ ),
+ )
+ return bool(updated)
+
def get_export_job(self, job_id: str) -> ExportJob | None:
with self._lock:
row = self.connection.execute(
@@ -1326,12 +1367,27 @@ class ControllerStore:
rows = self.connection.execute(
"""
SELECT * FROM export_jobs
- ORDER BY requested_at DESC, job_id DESC LIMIT ?
+ ORDER BY
+ CASE WHEN status IN ('queued','running') THEN 0 ELSE 1 END,
+ requested_at DESC,
+ job_id DESC
+ LIMIT ?
""",
(limit,),
).fetchall()
return [self._row_to_export_job(row) for row in rows]
+ def list_reconcilable_export_jobs(self) -> list[ExportJob]:
+ with self._lock:
+ rows = self.connection.execute(
+ """
+ SELECT * FROM export_jobs
+ WHERE status IN ('failed','interrupted')
+ ORDER BY requested_at, job_id
+ """
+ ).fetchall()
+ return [self._row_to_export_job(row) for row in rows]
+
def protect_slot(
self,
slot: int,
diff --git a/tests/test_app.py b/tests/test_app.py
index 9a5b068..fc9cdde 100644
--- a/tests/test_app.py
+++ b/tests/test_app.py
@@ -14,7 +14,14 @@ from controller.service import ControllerService, ControllerUnavailableError
class FakeControllerService:
- def __init__(self, *, enqueue_result=None, enqueue_error=None, health=None):
+ def __init__(
+ self,
+ *,
+ enqueue_result=None,
+ enqueue_error=None,
+ health=None,
+ export_jobs=None,
+ ):
self.enqueue_result = enqueue_result or {
"status": "queued",
"slot": 3,
@@ -29,7 +36,9 @@ class FakeControllerService:
"router_state": "video",
"degraded_reasons": [],
}
+ self.export_jobs = list(export_jobs or [])
self.enqueue_calls = []
+ self.export_calls = []
def enqueue_download(self, url, submitter, note):
self.enqueue_calls.append((url, submitter, note))
@@ -37,6 +46,20 @@ class FakeControllerService:
raise self.enqueue_error
return dict(self.enqueue_result)
+ def enqueue_export(self, url, mode, submitter, note):
+ self.export_calls.append((url, mode, submitter, note))
+ if self.enqueue_error is not None:
+ raise self.enqueue_error
+ return {
+ "kind": "export",
+ "status": "queued",
+ "job_id": "export-job",
+ "mode": mode,
+ }
+
+ def list_export_jobs(self, *, limit=10):
+ return [dict(job) for job in self.export_jobs[:limit]]
+
def health(self):
return dict(self.health_result)
@@ -71,11 +94,94 @@ class FlaskFactoryTests(unittest.TestCase):
response = self._app(FakeControllerService()).test_client().get("/")
self.assertEqual(response.status_code, 200)
- self.assertIn(b"Submit YouTube Playlist", response.data)
+ self.assertIn(b"Submit YouTube Media", response.data)
self.assertIn(b'action="/submit"', response.data)
self.assertIn(b'name="url"', response.data)
self.assertIn(b'name="submitter"', response.data)
self.assertIn(b'name="note"', response.data)
+ self.assertIn(b'name="download_mode"', response.data)
+ self.assertIn(b'value="player_slot" checked', response.data)
+ self.assertIn(b"MP4 Export", response.data)
+ self.assertIn(b"M4A Audio", response.data)
+ self.assertIn(b"MP3 Audio", response.data)
+
+ def test_older_clients_omitting_mode_still_use_player_slot(self):
+ service = FakeControllerService()
+
+ response = self._post(service, submitter=" Alice ")
+
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(
+ service.enqueue_calls,
+ [("https://www.youtube.com/playlist?list=PL1234567890", "Alice", None)],
+ )
+ self.assertEqual(service.export_calls, [])
+
+ def test_each_utility_mode_uses_export_path_without_player_slot(self):
+ for mode, label in (
+ ("mp4", b"MP4 Export queued"),
+ ("m4a", b"M4A Audio queued"),
+ ("mp3", b"MP3 Audio queued"),
+ ):
+ with self.subTest(mode=mode):
+ service = FakeControllerService()
+ response = self._post(
+ service,
+ download_mode=mode,
+ submitter="Operator",
+ note="Utility",
+ )
+ self.assertEqual(response.status_code, 200)
+ self.assertIn(label, response.data)
+ self.assertEqual(service.enqueue_calls, [])
+ self.assertEqual(
+ service.export_calls,
+ [
+ (
+ "https://www.youtube.com/playlist?list=PL1234567890",
+ mode,
+ "Operator",
+ "Utility",
+ )
+ ],
+ )
+
+ def test_invalid_download_mode_is_rejected_server_side(self):
+ service = FakeControllerService()
+
+ response = self._post(service, download_mode="../../slot1")
+
+ self.assertEqual(response.status_code, 200)
+ self.assertIn(b"Invalid download type", response.data)
+ self.assertEqual(service.enqueue_calls, [])
+ self.assertEqual(service.export_calls, [])
+
+ def test_recent_export_status_shows_results_and_failures(self):
+ service = FakeControllerService(
+ export_jobs=[
+ {
+ "job_id": "one",
+ "mode": "mp4",
+ "status": "completed",
+ "output_paths": ["exports/video/one/Service.mp4"],
+ "error": None,
+ },
+ {
+ "job_id": "two",
+ "mode": "mp3",
+ "status": "failed",
+ "output_paths": [],
+ "error": "ffmpeg failed",
+ },
+ ]
+ )
+
+ response = self._app(service).test_client().get("/")
+
+ self.assertEqual(response.status_code, 200)
+ self.assertIn(b"Recent utility exports", response.data)
+ self.assertIn(b"exports/video/one/Service.mp4", response.data)
+ self.assertIn(b"ffmpeg failed", response.data)
def test_submit_rejects_non_youtube_and_malformed_urls(self):
invalid_urls = (
diff --git a/tests/test_config_contracts.py b/tests/test_config_contracts.py
index 50d2100..f9999c7 100644
--- a/tests/test_config_contracts.py
+++ b/tests/test_config_contracts.py
@@ -84,6 +84,7 @@ class ConfigurationTests(unittest.TestCase):
self.assertEqual(config.mqtt.router_echo_grace_seconds, 2.0)
self.assertEqual(config.downloads.export_directory, Path("exports"))
self.assertEqual(config.downloads.export_max_height, 1080)
+ self.assertEqual(config.downloads.export_max_workers, 1)
self.assertEqual(config.downloads.max_queued_exports, 20)
def test_documented_environment_overrides_apply(self):
@@ -190,9 +191,23 @@ class ConfigurationTests(unittest.TestCase):
with self.assertRaisesRegex(ConfigError, "safe relative directory"):
load_config(path, environ={})
+ def test_export_directory_cannot_overlap_player_slot_storage(self):
+ invalid = ("slot1", "SLOT10/exports", ".controller-trash/exports")
+ for value in invalid:
+ with self.subTest(value=value), tempfile.TemporaryDirectory() as directory:
+ path = Path(directory) / "config.yaml"
+ path.write_text(
+ f"downloads:\n export_directory: {value!r}\n",
+ encoding="utf-8",
+ )
+ with self.assertRaisesRegex(ConfigError, "slot storage"):
+ load_config(path, environ={})
+
def test_export_limits_are_bounded_positive_integers(self):
cases = (
("export_max_height", 0),
+ ("export_max_workers", 0),
+ ("export_max_workers", 11),
("max_queued_exports", 0),
("max_queued_exports", 1001),
)
diff --git a/tests/test_exports.py b/tests/test_exports.py
new file mode 100644
index 0000000..54974a0
--- /dev/null
+++ b/tests/test_exports.py
@@ -0,0 +1,625 @@
+import json
+import subprocess
+import tempfile
+import unittest
+from concurrent.futures import Future
+from pathlib import Path
+
+from yt_dlp import YoutubeDL
+
+from controller.config import ControllerConfig, DownloadConfig, StorageConfig
+from controller.downloads import (
+ DownloadManager,
+ mp4_export_format_selector,
+ safe_export_filename,
+)
+from controller.store import ControllerStore
+
+
+class ImmediateExecutor:
+ def submit(self, function, *args, **kwargs):
+ future = Future()
+ try:
+ future.set_result(function(*args, **kwargs))
+ except BaseException as exc:
+ future.set_exception(exc)
+ return future
+
+
+class QueuedExecutor:
+ def __init__(self):
+ self.calls = []
+
+ def submit(self, function, *args, **kwargs):
+ future = Future()
+ self.calls.append((future, function, args, kwargs))
+ return future
+
+
+class FakeExportRunner:
+ def __init__(
+ self,
+ *,
+ video_codec="h264",
+ audio_codec="aac",
+ pixel_format="yuv420p",
+ container="mp4",
+ height=1080,
+ playlist_items=1,
+ fail_download=False,
+ ):
+ self.video_codec = video_codec
+ self.audio_codec = audio_codec
+ self.pixel_format = pixel_format
+ self.container = container
+ self.height = height
+ self.playlist_items = playlist_items
+ self.fail_download = fail_download
+ self.calls = []
+
+ def __call__(self, arguments, **kwargs):
+ arguments = list(arguments)
+ self.calls.append((arguments, dict(kwargs)))
+ program = Path(arguments[0]).name
+ if program == "yt-dlp-test":
+ if self.fail_download:
+ raise subprocess.CalledProcessError(9, arguments)
+ template = Path(arguments[arguments.index("-o") + 1])
+ template.parent.mkdir(parents=True, exist_ok=True)
+ if "--audio-format" in arguments:
+ extension = "." + arguments[arguments.index("--audio-format") + 1]
+ else:
+ extension = {
+ "mp4": ".mp4",
+ "matroska": ".mkv",
+ "webm": ".webm",
+ }[self.container]
+ for index in range(1, self.playlist_items + 1):
+ (template.parent / f"{index:03d} - Tést Item {index} [id{index}]{extension}").write_bytes(
+ f"source-{index}".encode()
+ )
+ return subprocess.CompletedProcess(arguments, 0, stdout="", stderr="")
+ if program == "ffmpeg":
+ Path(arguments[-1]).write_bytes(b"converted")
+ return subprocess.CompletedProcess(arguments, 0, stdout="", stderr="")
+ if program == "ffprobe":
+ path = Path(arguments[-1])
+ converted = path.is_file() and path.read_bytes() == b"converted"
+ if path.suffix == ".m4a":
+ payload = {
+ "streams": [{"codec_type": "audio", "codec_name": "aac"}],
+ "format": {"format_name": "mov,mp4,m4a,3gp,3g2,mj2"},
+ }
+ elif path.suffix == ".mp3":
+ payload = {
+ "streams": [{"codec_type": "audio", "codec_name": "mp3"}],
+ "format": {"format_name": "mp3"},
+ }
+ else:
+ payload = {
+ "streams": [
+ {
+ "codec_type": "video",
+ "codec_name": "h264" if converted else self.video_codec,
+ "pix_fmt": "yuv420p" if converted else self.pixel_format,
+ "width": 1920 if self.height == 1080 else 1280,
+ "height": self.height,
+ },
+ {
+ "codec_type": "audio",
+ "codec_name": "aac" if converted else self.audio_codec,
+ },
+ ],
+ "format": {
+ "format_name": "mov,mp4,m4a,3gp,3g2,mj2"
+ if converted or self.container == "mp4"
+ else self.container
+ },
+ }
+ return subprocess.CompletedProcess(
+ arguments,
+ 0,
+ stdout=json.dumps(payload),
+ stderr="",
+ )
+ raise AssertionError(f"unexpected command: {arguments!r}")
+
+ @property
+ def yt_dlp_calls(self):
+ return [arguments for arguments, _ in self.calls if Path(arguments[0]).name == "yt-dlp-test"]
+
+ @property
+ def ffmpeg_calls(self):
+ return [arguments for arguments, _ in self.calls if Path(arguments[0]).name == "ffmpeg"]
+
+
+class UtilityExportTests(unittest.TestCase):
+ def setUp(self):
+ self.tempdir = tempfile.TemporaryDirectory()
+ base = Path(self.tempdir.name)
+ self.media = base / "media"
+ self.data = base / "data"
+ self.config = ControllerConfig(
+ storage=StorageConfig(media_base=self.media, data_base=self.data),
+ downloads=DownloadConfig(
+ yt_dlp_binary="yt-dlp-test",
+ cookie_path=None,
+ max_workers=1,
+ process_timeout_seconds=30,
+ max_height=1080,
+ export_directory=Path("exports"),
+ export_max_height=1080,
+ max_queued_exports=5,
+ ),
+ )
+ self.store = ControllerStore(self.config.storage).open()
+
+ def tearDown(self):
+ self.store.close()
+ self.tempdir.cleanup()
+
+ def manager(self, runner, events=None):
+ return DownloadManager(
+ self.config,
+ self.store,
+ runner=runner,
+ executor=ImmediateExecutor(),
+ export_executor=ImmediateExecutor(),
+ on_event=(events if events is not None else []).append,
+ )
+
+ def assert_player_slots_empty(self):
+ self.assertTrue(
+ all(slot.media_state == "empty" for slot in self.store.list_slots())
+ )
+
+ def test_compatible_mp4_exports_without_slot_or_transcode(self):
+ events = []
+ runner = FakeExportRunner()
+ result = self.manager(runner, events).enqueue_export(
+ "https://www.youtube.com/watch?v=one", "mp4", "Alice", "Slides"
+ )
+
+ self.assertEqual(result.status, "queued")
+ job = self.store.get_export_job(result.job_id)
+ self.assertEqual(job.status, "completed")
+ self.assertEqual(len(job.output_paths), 1)
+ self.assertTrue(job.output_paths[0].startswith(f"exports/video/{result.job_id}/"))
+ self.assertTrue(job.output_paths[0].endswith(".mp4"))
+ self.assertTrue((self.media / job.output_paths[0]).is_file())
+ self.assertEqual(events, [])
+ self.assertEqual(runner.ffmpeg_calls, [])
+ self.assert_player_slots_empty()
+
+ arguments = runner.yt_dlp_calls[0]
+ self.assertIn("--yes-playlist", arguments)
+ self.assertIn("--windows-filenames", arguments)
+ self.assertIn("--no-overwrites", arguments)
+ self.assertEqual(
+ arguments[arguments.index("-f") + 1],
+ "bv*[height<=1080]+ba/b[height<=1080]",
+ )
+ self.assertEqual(
+ arguments[arguments.index("-S") + 1],
+ "res:1080,vcodec:h264,acodec:aac",
+ )
+
+ def test_incompatible_mp4_transcodes_only_incompatible_codecs(self):
+ runner = FakeExportRunner(
+ video_codec="vp9",
+ audio_codec="opus",
+ pixel_format="yuv420p",
+ container="webm",
+ )
+ result = self.manager(runner).enqueue_export(
+ "https://www.youtube.com/watch?v=vp9", "mp4", None, None
+ )
+
+ self.assertEqual(self.store.get_export_job(result.job_id).status, "completed")
+ self.assertEqual(len(runner.ffmpeg_calls), 1)
+ arguments = runner.ffmpeg_calls[0]
+ self.assertIn("libx264", arguments)
+ self.assertIn("-crf", arguments)
+ self.assertEqual(arguments[arguments.index("-crf") + 1], "18")
+ self.assertIn("yuv420p", arguments)
+ self.assertIn("aac", arguments)
+ self.assertIn("192k", arguments)
+ self.assertNotIn("-vf", arguments)
+ self.assertNotIn("scale", " ".join(arguments))
+ self.assert_player_slots_empty()
+
+ def test_compatible_h264_aac_in_other_container_is_remuxed_only(self):
+ runner = FakeExportRunner(container="matroska")
+ result = self.manager(runner).enqueue_export(
+ "https://www.youtube.com/watch?v=mkv", "mp4", None, None
+ )
+
+ self.assertEqual(self.store.get_export_job(result.job_id).status, "completed")
+ arguments = runner.ffmpeg_calls[0]
+ self.assertEqual(arguments[arguments.index("-c:v") + 1], "copy")
+ self.assertEqual(arguments[arguments.index("-c:a") + 1], "copy")
+ self.assertNotIn("libx264", arguments)
+
+ def test_lower_resolution_mp4_is_preserved_without_scaling(self):
+ runner = FakeExportRunner(height=720)
+ result = self.manager(runner).enqueue_export(
+ "https://www.youtube.com/watch?v=720", "mp4", None, None
+ )
+
+ self.assertEqual(self.store.get_export_job(result.job_id).status, "completed")
+ self.assertEqual(runner.ffmpeg_calls, [])
+
+ def test_over_height_mp4_is_rejected_even_if_downloader_misbehaves(self):
+ runner = FakeExportRunner(height=2160)
+ result = self.manager(runner).enqueue_export(
+ "https://www.youtube.com/watch?v=2160", "mp4", None, None
+ )
+
+ job = self.store.get_export_job(result.job_id)
+ self.assertEqual(job.status, "failed")
+ self.assertIn("maximum height", job.error)
+ self.assertFalse((self.media / "exports" / "video" / result.job_id).exists())
+ self.assert_player_slots_empty()
+
+ def test_audio_exports_use_intentional_audio_only_strategies(self):
+ for mode, expected_selector in (
+ ("m4a", "ba/b"),
+ ("mp3", "ba/b"),
+ ):
+ with self.subTest(mode=mode):
+ events = []
+ runner = FakeExportRunner()
+ result = self.manager(runner, events).enqueue_export(
+ f"https://www.youtube.com/watch?v={mode}", mode, None, None
+ )
+ job = self.store.get_export_job(result.job_id)
+ self.assertEqual(job.status, "completed")
+ self.assertEqual(len(job.output_paths), 1)
+ self.assertTrue(job.output_paths[0].endswith(f".{mode}"))
+ arguments = runner.yt_dlp_calls[0]
+ self.assertEqual(arguments[arguments.index("-f") + 1], expected_selector)
+ self.assertIn("-x", arguments)
+ self.assertEqual(
+ arguments[arguments.index("--audio-format") + 1], mode
+ )
+ self.assertEqual(
+ arguments[arguments.index("--audio-quality") + 1], "0"
+ )
+ if mode == "m4a":
+ self.assertEqual(
+ arguments[arguments.index("-S") + 1],
+ "abr,acodec:aac",
+ )
+ self.assertEqual(runner.ffmpeg_calls, [])
+ self.assertEqual(events, [])
+ self.assert_player_slots_empty()
+
+ def test_playlist_items_publish_as_individual_collision_safe_files(self):
+ runner = FakeExportRunner(playlist_items=3)
+ result = self.manager(runner).enqueue_export(
+ "https://www.youtube.com/playlist?list=three", "m4a", None, None
+ )
+
+ job = self.store.get_export_job(result.job_id)
+ self.assertEqual(job.status, "completed")
+ self.assertEqual(len(job.output_paths), 3)
+ self.assertEqual(len(set(job.output_paths)), 3)
+ self.assertTrue(all((self.media / path).is_file() for path in job.output_paths))
+
+ def test_export_failure_and_expiry_never_mutate_player_or_export_files(self):
+ slot_one = self.media / "slot1"
+ slot_one.mkdir(parents=True)
+ protected_file = slot_one / "active.mp4"
+ protected_file.write_bytes(b"active")
+ failed = self.manager(FakeExportRunner(fail_download=True)).enqueue_export(
+ "https://www.youtube.com/watch?v=fail", "mp3", None, None
+ )
+ self.assertEqual(self.store.get_export_job(failed.job_id).status, "failed")
+ self.assertEqual(protected_file.read_bytes(), b"active")
+ self.assert_player_slots_empty()
+
+ completed = self.manager(FakeExportRunner()).enqueue_export(
+ "https://www.youtube.com/watch?v=keep", "m4a", None, None
+ )
+ export_path = self.media / self.store.get_export_job(completed.job_id).output_paths[0]
+ self.manager(FakeExportRunner()).expire_due_slots()
+ self.assertTrue(export_path.is_file())
+ self.assertEqual(protected_file.read_bytes(), b"active")
+
+ def test_symlinked_export_root_cannot_redirect_into_player_slot(self):
+ slot_one = self.media / "slot1"
+ slot_one.mkdir(parents=True)
+ protected_file = slot_one / "active.mp4"
+ protected_file.write_bytes(b"active")
+ (self.media / "exports").symlink_to(slot_one, target_is_directory=True)
+
+ result = self.manager(FakeExportRunner()).enqueue_export(
+ "https://www.youtube.com/watch?v=escape", "m4a", None, None
+ )
+
+ self.assertEqual(self.store.get_export_job(result.job_id).status, "failed")
+ self.assertIn("unsafe export directory", self.store.get_export_job(result.job_id).error)
+ self.assertEqual(protected_file.read_bytes(), b"active")
+
+ def test_export_work_uses_dedicated_executor_capacity(self):
+ player_executor = QueuedExecutor()
+ export_executor = QueuedExecutor()
+ manager = DownloadManager(
+ self.config,
+ self.store,
+ runner=FakeExportRunner(),
+ executor=player_executor,
+ export_executor=export_executor,
+ )
+
+ manager.enqueue_export(
+ "https://www.youtube.com/watch?v=export", "mp4", None, None
+ )
+ manager.enqueue(
+ "https://www.youtube.com/watch?v=player", None, None
+ )
+
+ self.assertEqual(len(export_executor.calls), 1)
+ self.assertEqual(len(player_executor.calls), 1)
+
+ def test_restart_reconciles_published_export_and_cleans_manifest(self):
+ runner = FakeExportRunner()
+ manager = self.manager(runner)
+ job_id = "a" * 32
+ self.store.reserve_export_job(
+ "https://www.youtube.com/watch?v=recover",
+ "mp4",
+ None,
+ None,
+ max_active=5,
+ job_id=job_id,
+ )
+ self.store.mark_export_running(job_id)
+ self.store.fail_export_job(job_id, "database completion failed")
+ work = self.media / "exports" / ".work" / job_id
+ published = self.media / "exports" / "video" / job_id
+ work.mkdir(parents=True)
+ published.mkdir(parents=True)
+ filename = "Recovered.mp4"
+ (published / filename).write_bytes(b"compatible")
+ manager._write_export_manifest(
+ work,
+ job_id=job_id,
+ mode="mp4",
+ filenames=[filename],
+ )
+
+ manager.reconcile_export_artifacts()
+
+ job = self.store.get_export_job(job_id)
+ self.assertEqual(job.status, "completed")
+ self.assertEqual(
+ job.output_paths,
+ (f"exports/video/{job_id}/{filename}",),
+ )
+ self.assertFalse(work.exists())
+ self.assertTrue((published / filename).is_file())
+
+ def test_restart_atomically_publishes_manifested_ready_export(self):
+ runner = FakeExportRunner()
+ manager = self.manager(runner)
+ job_id = "b" * 32
+ self.store.reserve_export_job(
+ "https://www.youtube.com/watch?v=staged",
+ "m4a",
+ None,
+ None,
+ max_active=5,
+ job_id=job_id,
+ )
+ self.store.mark_export_running(job_id)
+ self.store.fail_export_job(job_id, "interrupted after conversion")
+ work = self.media / "exports" / ".work" / job_id
+ ready = work / "ready"
+ ready.mkdir(parents=True)
+ filename = "Recovered.m4a"
+ (ready / filename).write_bytes(b"aac")
+ manager._write_export_manifest(
+ work,
+ job_id=job_id,
+ mode="m4a",
+ filenames=[filename],
+ )
+
+ manager.reconcile_export_artifacts()
+
+ job = self.store.get_export_job(job_id)
+ output = self.media / job.output_paths[0]
+ self.assertEqual(job.status, "completed")
+ self.assertTrue(output.is_file())
+ self.assertFalse(work.exists())
+
+ def test_restart_discards_only_known_incomplete_export_staging(self):
+ manager = self.manager(FakeExportRunner())
+ job_id = "c" * 32
+ self.store.reserve_export_job(
+ "https://www.youtube.com/watch?v=incomplete",
+ "mp3",
+ None,
+ None,
+ max_active=5,
+ job_id=job_id,
+ )
+ self.store.fail_export_job(job_id, "interrupted download")
+ work = self.media / "exports" / ".work" / job_id
+ work.mkdir(parents=True)
+ (work / "partial.part").write_bytes(b"partial")
+ unknown = self.media / "exports" / ".work" / ("d" * 32)
+ unknown.mkdir()
+ (unknown / "operator-file").write_bytes(b"keep")
+
+ manager.reconcile_export_artifacts()
+
+ self.assertFalse(work.exists())
+ self.assertTrue((unknown / "operator-file").is_file())
+
+ def test_safe_filename_blocks_traversal_preserves_unicode_and_deduplicates(self):
+ used = set()
+ first = safe_export_filename(
+ "../../slot1/CON:Tést\x00?", ".mp4", used_casefolded=used
+ )
+ second = safe_export_filename(
+ "../../slot1/CON:Tést\x00?", ".mp4", used_casefolded=used
+ )
+ long_name = safe_export_filename("🎬" * 500, ".mp3")
+
+ self.assertNotIn("/", first)
+ self.assertNotIn("\\", first)
+ self.assertNotIn("..", Path(first).parts)
+ self.assertIn("Tést", first)
+ self.assertNotEqual(first.casefold(), second.casefold())
+ self.assertTrue(second.endswith(" (2).mp4"))
+ self.assertLessEqual(len(Path(long_name).stem.encode("utf-8")), 180)
+
+ def test_mp4_sorting_keeps_resolution_ahead_of_codec_preference(self):
+ formats = [
+ {
+ "format_id": "av1-1080",
+ "url": "https://example.test/av1",
+ "ext": "webm",
+ "height": 1080,
+ "width": 1920,
+ "vcodec": "av01.0.08M.08",
+ "acodec": "none",
+ "tbr": 4000,
+ "filesize": 1,
+ },
+ {
+ "format_id": "h264-720",
+ "url": "https://example.test/h264",
+ "ext": "mp4",
+ "height": 720,
+ "width": 1280,
+ "vcodec": "avc1.64001f",
+ "acodec": "none",
+ "tbr": 3000,
+ "filesize": 1,
+ },
+ {
+ "format_id": "aac",
+ "url": "https://example.test/aac",
+ "ext": "m4a",
+ "vcodec": "none",
+ "acodec": "mp4a.40.2",
+ "tbr": 128,
+ "filesize": 1,
+ },
+ ]
+ with YoutubeDL(
+ {
+ "quiet": True,
+ "format_sort": ["res:1080", "vcodec:h264", "acodec:aac"],
+ }
+ ) as ydl:
+ info = {"formats": formats}
+ ydl.sort_formats(info)
+ selector = ydl.build_format_selector(mp4_export_format_selector(1080))
+ selected = list(ydl._select_formats(info["formats"], selector))
+
+ self.assertEqual(selected[0]["height"], 1080)
+ self.assertEqual(selected[0]["requested_formats"][0]["format_id"], "av1-1080")
+
+ def test_mp4_sorting_prefers_native_h264_aac_at_equal_quality(self):
+ formats = [
+ {
+ "format_id": "vp9-1080",
+ "url": "https://example.test/vp9",
+ "ext": "webm",
+ "height": 1080,
+ "width": 1920,
+ "vcodec": "vp9",
+ "acodec": "none",
+ "tbr": 4200,
+ "filesize": 1,
+ },
+ {
+ "format_id": "h264-1080",
+ "url": "https://example.test/h264",
+ "ext": "mp4",
+ "height": 1080,
+ "width": 1920,
+ "vcodec": "avc1.640028",
+ "acodec": "none",
+ "tbr": 4000,
+ "filesize": 1,
+ },
+ {
+ "format_id": "opus",
+ "url": "https://example.test/opus",
+ "ext": "webm",
+ "vcodec": "none",
+ "acodec": "opus",
+ "tbr": 160,
+ "filesize": 1,
+ },
+ {
+ "format_id": "aac",
+ "url": "https://example.test/aac",
+ "ext": "m4a",
+ "vcodec": "none",
+ "acodec": "mp4a.40.2",
+ "tbr": 128,
+ "filesize": 1,
+ },
+ ]
+ with YoutubeDL(
+ {
+ "quiet": True,
+ "format_sort": ["res:1080", "vcodec:h264", "acodec:aac"],
+ }
+ ) as ydl:
+ info = {"formats": formats}
+ ydl.sort_formats(info)
+ selector = ydl.build_format_selector(mp4_export_format_selector(1080))
+ selected = list(ydl._select_formats(info["formats"], selector))
+
+ self.assertEqual(
+ [item["format_id"] for item in selected[0]["requested_formats"]],
+ ["h264-1080", "aac"],
+ )
+
+ def test_m4a_sorting_keeps_audio_quality_ahead_of_native_aac_preference(self):
+ formats = [
+ {
+ "format_id": "aac-low",
+ "url": "https://example.test/aac-low",
+ "ext": "m4a",
+ "vcodec": "none",
+ "acodec": "mp4a.40.2",
+ "abr": 48,
+ "tbr": 48,
+ "filesize": 1,
+ },
+ {
+ "format_id": "opus-high",
+ "url": "https://example.test/opus-high",
+ "ext": "webm",
+ "vcodec": "none",
+ "acodec": "opus",
+ "abr": 160,
+ "tbr": 160,
+ "filesize": 1,
+ },
+ ]
+ with YoutubeDL(
+ {"quiet": True, "format_sort": ["abr", "acodec:aac"]}
+ ) as ydl:
+ info = {"formats": formats}
+ ydl.sort_formats(info)
+ selected = list(
+ ydl._select_formats(
+ info["formats"],
+ ydl.build_format_selector("ba/b"),
+ )
+ )
+
+ self.assertEqual(selected[0]["format_id"], "opus-high")
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_service.py b/tests/test_service.py
index c6ccd72..1f42862 100644
--- a/tests/test_service.py
+++ b/tests/test_service.py
@@ -89,11 +89,26 @@ class FakeDownloadManager:
self.stopped = False
self.delete_requests = []
self.enqueue_calls = []
+ self.export_calls = []
+ self.export_reconciled = False
def enqueue(self, url, submitter, note):
self.enqueue_calls.append((url, submitter, note))
return self.store.reserve_download(url, submitter, note)
+ def enqueue_export(self, url, mode, submitter, note):
+ self.export_calls.append((url, mode, submitter, note))
+ return self.store.reserve_export_job(
+ url,
+ mode,
+ submitter,
+ note,
+ max_active=20,
+ )
+
+ def reconcile_export_artifacts(self):
+ self.export_reconciled = True
+
def recover_pending_deletions(self):
self.recovered = True
@@ -171,6 +186,7 @@ class ControllerServiceTests(unittest.TestCase):
)
self.service.start()
self.assertFalse(self.downloads.recovered)
+ self.assertTrue(self.downloads.export_reconciled)
return self.service
def ready_slot(self, slot=1):
@@ -497,6 +513,62 @@ class ControllerServiceTests(unittest.TestCase):
self.assertIn("mqtt_disconnected", health["degraded_reasons"])
self.assertIn("awaiting_player_evidence", health["degraded_reasons"])
+ def test_utility_export_is_slot_free_and_not_gated_by_player_evidence(self):
+ self.start_service()
+ self.assertIn(
+ "awaiting_player_evidence",
+ self.service.health()["degraded_reasons"],
+ )
+
+ result = self.service.enqueue_export(
+ "https://www.youtube.com/playlist?list=utility",
+ "mp4",
+ "Alice",
+ "ProPresenter",
+ )
+
+ self.assertEqual(result["kind"], "export")
+ self.assertEqual(result["status"], "queued")
+ self.assertEqual(result["mode"], "mp4")
+ self.assertIsNotNone(result["job_id"])
+ self.assertEqual(
+ self.downloads.export_calls,
+ [
+ (
+ "https://www.youtube.com/playlist?list=utility",
+ "mp4",
+ "Alice",
+ "ProPresenter",
+ )
+ ],
+ )
+ self.assertTrue(
+ all(slot.media_state == "empty" for slot in self.store.list_slots())
+ )
+ self.assertEqual(self.mqtt.json_messages, [])
+ self.assertEqual(self.mqtt.legacy_messages, [])
+ jobs = self.service.list_export_jobs(limit=10)
+ self.assertEqual(len(jobs), 1)
+ self.assertEqual(jobs[0]["job_id"], result["job_id"])
+ self.assertEqual(jobs[0]["output_paths"], [])
+
+ def test_invalid_utility_mode_is_rejected_without_player_side_effects(self):
+ self.start_service()
+
+ with self.assertRaisesRegex(ValueError, "unsupported utility export mode"):
+ self.service.enqueue_export(
+ "https://www.youtube.com/watch?v=one",
+ "slot1",
+ None,
+ None,
+ )
+
+ self.assertTrue(
+ all(slot.media_state == "empty" for slot in self.store.list_slots())
+ )
+ self.assertEqual(self.mqtt.json_messages, [])
+ self.assertEqual(self.mqtt.legacy_messages, [])
+
def test_cleanup_waits_for_post_start_player_evidence_but_new_ingest_does_not(self):
self.start_service()
self.service._next_expiry = self.monotonic()
diff --git a/tests/test_store.py b/tests/test_store.py
index bddb384..18e03b4 100644
--- a/tests/test_store.py
+++ b/tests/test_store.py
@@ -150,6 +150,39 @@ class FreshStoreTests(StoreTestCase):
)
self.assertTrue(all(slot.media_state == "empty" for slot in store.list_slots()))
+ def test_export_listing_keeps_live_jobs_visible_before_recent_history(self):
+ with self.open_store() as store:
+ store.reserve_export_job(
+ "https://www.youtube.com/watch?v=active",
+ "mp4",
+ None,
+ None,
+ max_active=5,
+ job_id="active-old",
+ )
+ self.clock.value += 1
+ for index in range(3):
+ job_id = f"completed-{index}"
+ store.reserve_export_job(
+ f"https://www.youtube.com/watch?v={index}",
+ "mp3",
+ None,
+ None,
+ max_active=5,
+ job_id=job_id,
+ )
+ store.mark_export_running(job_id)
+ store.complete_export_job(
+ job_id,
+ (f"exports/audio/{job_id}/item.mp3",),
+ )
+ self.clock.value += 1
+
+ listed = store.list_export_jobs(limit=2)
+ self.assertEqual(listed[0].job_id, "active-old")
+ self.assertEqual(listed[0].status, "queued")
+ self.assertEqual(listed[1].job_id, "completed-2")
+
def test_restart_marks_live_exports_interrupted_without_touching_slots(self):
store = self.open_store()
store.reserve_export_job(