From 7e6b916963932d10b352bb6c127f12e0ff75fa67 Mon Sep 17 00:00:00 2001 From: Media Date: Wed, 12 Aug 2026 18:48:03 -0500 Subject: [PATCH] feat: add isolated utility export modes --- app.py | 139 +++++- config.yaml | 2 + controller/config.py | 9 + controller/downloads.py | 770 ++++++++++++++++++++++++++++++++- controller/service.py | 55 +++ controller/store.py | 58 ++- tests/test_app.py | 110 ++++- tests/test_config_contracts.py | 15 + tests/test_exports.py | 625 ++++++++++++++++++++++++++ tests/test_service.py | 72 +++ tests/test_store.py | 33 ++ 11 files changed, 1876 insertions(+), 12 deletions(-) create mode 100644 tests/test_exports.py diff --git a/app.py b/app.py index d98df83..d4d87d8 100644 --- a/app.py +++ b/app.py @@ -18,7 +18,7 @@ HTML_FORM = """ - YouTube Playlist Submit + YouTube Media Submit @@ -96,6 +96,33 @@ HTML_FORM = """ font-size: 1rem; } + input[type="radio"] { + width: auto; + margin: 0; + } + + fieldset { + border: 0; + padding: 0; + } + + legend { + margin-bottom: 0.35rem; + font-size: 0.92rem; + } + + .download-options { + display: grid; + gap: 0.4rem; + } + + .download-options label { + display: flex; + align-items: center; + gap: 0.55rem; + margin: 0; + } + input::placeholder, textarea::placeholder { color: #6b7280; @@ -139,6 +166,26 @@ HTML_FORM = """ color: #9ca3af; } + .export-status { + margin-top: 1rem; + padding-top: 0.9rem; + border-top: 1px solid rgba(148, 163, 184, 0.25); + } + + .export-job { + margin-top: 0.65rem; + padding: 0.65rem 0.75rem; + border-radius: 0.55rem; + background: rgba(30, 41, 59, 0.7); + font-size: 0.82rem; + } + + .export-path, + .export-error { + overflow-wrap: anywhere; + color: #cbd5e1; + } + @media (max-width: 480px) { .page-wrapper { padding: 1rem; @@ -167,9 +214,9 @@ HTML_FORM = """
-

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 = """ >
- +
+ Download as: +
+ + + + +
+
+ +

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 %} +
+ {% endfor %} +
+ {% 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(