Compare commits

..
5 Commits
14 changed files with 2587 additions and 43 deletions
+104 -11
View File
@@ -23,7 +23,8 @@ Browser / HA / Companion
Flask -> bounded event queue -> single Controller coordinator Flask -> bounded event queue -> single Controller coordinator
| | | | | |
| | +-> SQLite state/audit | | +-> SQLite state/audit
| +------------> bounded yt-dlp workers | +------------> Player-slot workers
| +------------> utility-export workers
+---------------------> MQTT +---------------------> MQTT
| |
+-----------------------------+------------------+ +-----------------------------+------------------+
@@ -411,6 +412,62 @@ state rather than a completed delete.
Expiry never compacts slots. An expired slot becomes empty; every other slot Expiry never compacts slots. An expired slot becomes empty; every other slot
keeps its number and canonical directory. keeps its number and canonical directory.
## Utility exports
The submission page has four explicit modes. `Player Slot` is selected by
default, and an older client that omits `download_mode` still takes that path.
`MP4 Export`, `M4A Audio`, and `MP3 Audio` use a separate durable job queue and
dedicated worker pool. They never reserve or mutate a Player slot, publish
ingest/HA/MQTT state, acquire the Audio Router, create Player metadata, or enter
Player playback, expiry, deletion, or reconciliation gates.
Player Slot remains quality-first and does not force MP4 or H.264. Its strict
yt-dlp selector is `bv*[height<=H]+ba/b[height<=H]`: separate 1080p video and
audio are eligible, a lower-resolution source is not upscaled, and no
unfiltered fallback may exceed `downloads.max_height`.
Utility modes behave as follows:
- MP4 selects the best source at or below `downloads.export_max_height`, with
native H.264/AAC preferred only among otherwise equal-quality choices. It
probes every result with ffprobe. An existing MP4/H.264/AAC/yuv420p result is
moved without remuxing or transcoding; otherwise ffmpeg copies each compatible
stream and converts only an incompatible video stream to H.264 or audio stream
to AAC. The output height must equal the selected source height, so export
never upscales, and the final file is verified as MP4/H.264/AAC.
- M4A selects best available audio, prefers native AAC among equal-quality
choices, extracts audio only, and verifies an AAC stream in an M4A/MP4
container. Native AAC is losslessly repackaged by yt-dlp; another source codec
is encoded only when required.
- MP3 selects best available audio, extracts audio only, encodes with yt-dlp's
highest-quality VBR setting when required, and verifies an MP3-only result.
Playlist exports preserve item order and use bounded, human-readable Unicode
filenames with unsafe path/control characters removed and case-insensitive
collision suffixes. Each job publishes atomically into its own directory:
```text
/mnt/Media/YTPlayer/exports/
video/<job-id>/*.mp4
audio/<job-id>/*.m4a
audio/<job-id>/*.mp3
```
Work is staged under `exports/.work/<job-id>`. A fsynced manifest makes the
publication boundary recoverable: after restart, a dedicated export worker verifies and
finishes a fully prepared or already-published job, while a known partial job
is marked interrupted and its private staging directory is removed. It does
not delay web readiness, MQTT reconnect, or Player coordination and does not
resume a partial yt-dlp process. Unknown directories are left untouched.
The web page lists live jobs before recent completed/failed/interrupted jobs and
shows their output paths or error. It intentionally provides no file-serving,
cancellation, or deletion endpoint.
Utility output is deliberately outside slot retention. Controller never
expires it; operators should monitor capacity and remove an entire completed
`video/<job-id>` or `audio/<job-id>` directory according to local policy. Do not
write untrusted files or symlinks into the media/export tree.
## Restart uncertainty and administrative reconciliation ## Restart uncertainty and administrative reconciliation
On startup, a last-known managed active slot or pending play is converted to a On startup, a last-known managed active slot or pending play is converted to a
@@ -552,15 +609,16 @@ Controller v2 uses one embedded SQLite database at
`/controller-data/controller.sqlite3`. It introduces no external database `/controller-data/controller.sqlite3`. It introduces no external database
service. service.
Schema version 2 contains: Schema version 3 contains:
| Table | Purpose and important constraints | | Table | Purpose and important constraints |
|---|---| |---|---|
| `schema_migrations` | Applied internal schema versions; v1 databases upgrade transactionally to v2 | | `schema_migrations` | Applied internal schema versions; v1/v2 databases upgrade transactionally to v3 |
| `data_migrations` | One-time import provenance, source hash, backup path, and warnings | | `data_migrations` | One-time import provenance, source hash, backup path, and warnings |
| `slots` | Exactly one row for each stable slot 1-10, generation, media state, identity, timestamps, expiry, and error | | `slots` | Exactly one row for each stable slot 1-10, generation, media state, identity, timestamps, expiry, and error |
| `slot_items` | Ordered item title/path metadata keyed by slot and position | | `slot_items` | Ordered item title/path metadata keyed by slot and position |
| `download_jobs` | Reservation/worker lifecycle keyed by job ID and slot generation; one live job per slot | | `download_jobs` | Reservation/worker lifecycle keyed by job ID and slot generation; one live job per slot |
| `export_jobs` | Slot-free utility job mode/status, request metadata, output paths, and error; no slot foreign key |
| `slot_guards` | Active, pending, restart-uncertain, and legacy-uncertain playback protection; one live guard per slot | | `slot_guards` | Active, pending, restart-uncertain, and legacy-uncertain playback protection; one live guard per slot |
| `player_observation` | Singleton last status/telemetry reduction, generation, active identity, pause, and reconciliation state/token | | `player_observation` | Singleton last status/telemetry reduction, generation, active identity, pause, and reconciliation state/token |
| `audio_router_observation` | Singleton last mode, freshness, timestamp, and warning | | `audio_router_observation` | Singleton last mode, freshness, timestamp, and warning |
@@ -615,8 +673,10 @@ discard the original or hash-named backup.
Once `controller.sqlite3` exists, startup never imports YAML again and later Once `controller.sqlite3` exists, startup never imports YAML again and later
YAML edits are ignored. That database-presence rule plus the `data_migrations` YAML edits are ignored. That database-presence rule plus the `data_migrations`
record makes the migration idempotent. An existing schema-v1 Controller SQLite record makes the migration idempotent. Existing schema-v1 or schema-v2
database is upgraded transactionally rather than reimporting YAML. Controller SQLite databases are upgraded transactionally to schema v3 rather
than reimporting YAML. The v2-to-v3 upgrade adds only the slot-free
`export_jobs` table, its recent-jobs index, and migration record.
Legacy state mapping is explicit: Legacy state mapping is explicit:
@@ -686,8 +746,9 @@ The checked-in `config.yaml` is the canonical example. Important defaults are:
queue, and 2-second QoS 1 PUBACK timeout. queue, and 2-second QoS 1 PUBACK timeout.
- Storage: 48-hour retention, 60-second expiry scan, and 15-minute uncertainty - Storage: 48-hour retention, 60-second expiry scan, and 15-minute uncertainty
review deadline. review deadline.
- Downloads: two workers, 2-hour subprocess timeout, maximum height 1080, and - Downloads: two Player-slot workers, one dedicated utility-export worker,
optional cookies. 2-hour subprocess timeout, strict Player/export maximum height 1080, at most
20 queued/running exports, output below `exports`, and optional cookies.
- Router: required mode `video`, 3-second confirmation timeout, unknown-state - Router: required mode `video`, 3-second confirmation timeout, unknown-state
policy `attempt`, continue after confirmation failure, and no fallback restore policy `attempt`, continue after confirmation failure, and no fallback restore
mode. mode.
@@ -718,8 +779,10 @@ permissions, and rotate it as a credential.
The HTTP surface is intentionally small: The HTTP surface is intentionally small:
- `GET /` renders the playlist submission form. - `GET /` renders the submission form and recent utility-export status.
- `POST /submit` accepts an HTTP(S) URL plus optional submitter and note. - `POST /submit` accepts an HTTP(S) URL, optional submitter and note, and
`download_mode` equal to `player_slot`, `mp4`, `m4a`, or `mp3`. Omitting the
field remains equivalent to `player_slot`.
- `GET /healthz` returns service and reconciliation state as JSON. - `GET /healthz` returns service and reconciliation state as JSON.
Health includes `status`, `started`, `coordinator_alive`, `mqtt_connected`, Health includes `status`, `started`, `coordinator_alive`, `mqtt_connected`,
@@ -752,6 +815,8 @@ The production image and Compose file intentionally differ from Controller v1:
of the old `/cookies_ro.txt` path. of the old `/cookies_ro.txt` path.
- The image includes ffmpeg and a Node 22 runtime required by the pinned yt-dlp - The image includes ffmpeg and a Node 22 runtime required by the pinned yt-dlp
YouTube JavaScript support. YouTube JavaScript support.
- Utility exports use the existing writable media bind mount and ffmpeg/ffprobe
installation; they add no service, port, volume, or external database.
- The build currently uses `node:22.23.1-bookworm-slim` as a JavaScript-runtime - The build currently uses `node:22.23.1-bookworm-slim` as a JavaScript-runtime
stage and `python:3.12-slim-bookworm` as the final base. Runtime Python stage and `python:3.12-slim-bookworm` as the final base. Runtime Python
requirements are exact pins, including Flask 3.1.3, Gunicorn 26.0.0, requirements are exact pins, including Flask 3.1.3, Gunicorn 26.0.0,
@@ -850,11 +915,22 @@ retry only while no final SQLite database exists. Returning to the recorded v1
image with the untouched v1 data/media snapshot remains safe because migration image with the untouched v1 data/media snapshot remains safe because migration
does not modify them. does not modify them.
### Upgrade from an earlier Controller v2 build
Stop Controller and back up the complete data directory before installing this
build. On first start, schema v1 or v2 is upgraded in one transaction to schema
v3; existing slots, guards, observations, leases, jobs, and audit rows are not
rewritten. Confirm `exports/.work`, `exports/video`, and `exports/audio` can be
created by the configured container UID/GID, then submit one controlled export
and inspect the path shown on the web page. The legacy `playlists.yaml` import
does not rerun when the SQLite database already exists.
## Rollback ## Rollback
Code rollback and state rollback are different operations. Controller v2 does Code rollback and state rollback are different operations. Controller v2 does
not write changes back to `playlists.yaml`; v1 does not understand not write changes back to `playlists.yaml`; v1 does not understand
`controller.sqlite3`. After any v2 download, delete, expiry, or slot reuse, the `controller.sqlite3`. Earlier Controller v2 builds also do not understand
schema v3. After any v2 download, delete, expiry, or slot reuse, the
pre-migration YAML and current media tree may no longer agree. pre-migration YAML and current media tree may no longer agree.
1. Stop Controller v2 cleanly. Leave Player Node unchanged. 1. Stop Controller v2 cleanly. Leave Player Node unchanged.
@@ -881,6 +957,14 @@ For a migration-only failure before final SQLite installation, simply keep v2
stopped and restart the recorded v1 deployment against the untouched original stopped and restart the recorded v1 deployment against the untouched original
data and media; no v2 database state needs conversion. data and media; no v2 database state needs conversion.
To roll back only this utility-export extension to an earlier v2 build, stop the
service and restore the complete schema-v2 data-directory backup made before
upgrade; do not point the older binary at the schema-v3 database. Utility files
already published below `exports` are independent of slots and may be archived
or left in place, but the older build will not list or manage them. Rolling
forward again requires restoring a matched newer database backup or allowing a
fresh schema-v2 backup to upgrade once more.
## Development and tests ## Development and tests
Application imports are side-effect free. Constructing the Flask app does not Application imports are side-effect free. Constructing the Flask app does not
@@ -904,7 +988,10 @@ Player generation, pathless busy/error handling and pause-timeout reduction,
durable playback plus connection-scoped cleanup reconciliation, Audio Router durable playback plus connection-scoped cleanup reconciliation, Audio Router
timeouts and ownership loss, service integration, HA compatibility, HTTP timeouts and ownership loss, service integration, HA compatibility, HTTP
health, and Docker/runtime assets. It uses fakes and temporary directories; a health, and Docker/runtime assets. It uses fakes and temporary directories; a
real-broker/container smoke test is still required for a deployment. real-broker/container smoke test is still required for a deployment. Utility
coverage additionally exercises actual pinned yt-dlp selection semantics,
Player/export isolation, queue bounds, codec command paths, safe names,
failure/restart publication recovery, and schema-v2-to-v3 migration.
Useful static and container checks are: Useful static and container checks are:
@@ -965,6 +1052,12 @@ PYTHONDONTWRITEBYTECODE=1 .venv/bin/python -m unittest discover -s tests -v
timeout. The current job fails instead of resuming. A container kill before timeout. The current job fails instead of resuming. A container kill before
that bookkeeping completes is recovered as `interrupted`/`error`; queued that bookkeeping completes is recovered as `interrupted`/`error`; queued
deletion work remains durably `deleting` for the evidence-gated retry path. deletion work remains durably `deleting` for the evidence-gated retry path.
- Utility exports are not subject to slot retention and have no automatic disk
quota or cleanup policy. Monitor the media filesystem and remove completed
job directories administratively. Interrupted partial jobs are not resumed.
- Export-path checks reject configured slot/trash overlap and encountered
symlinks, but the writable media tree is not a hostile multi-user sandbox.
Only trusted host accounts should be able to modify it while Controller runs.
- The container is read-only and non-root, but the data and media mounts must be - The container is read-only and non-root, but the data and media mounts must be
writable. Controller can intentionally delete canonical expired or writable. Controller can intentionally delete canonical expired or
administratively deleted slot directories; maintain independent backups. administratively deleted slot directories; maintain independent backups.
+132 -5
View File
@@ -18,7 +18,7 @@ HTML_FORM = """
<html> <html>
<head> <head>
<meta charset="utf-8"> <meta charset="utf-8">
<title>YouTube Playlist Submit</title> <title>YouTube Media Submit</title>
<!-- Make it behave on mobile --> <!-- Make it behave on mobile -->
<meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1"> <meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1">
@@ -96,6 +96,33 @@ HTML_FORM = """
font-size: 1rem; 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, input::placeholder,
textarea::placeholder { textarea::placeholder {
color: #6b7280; color: #6b7280;
@@ -139,6 +166,26 @@ HTML_FORM = """
color: #9ca3af; 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) { @media (max-width: 480px) {
.page-wrapper { .page-wrapper {
padding: 1rem; padding: 1rem;
@@ -167,9 +214,9 @@ HTML_FORM = """
<body> <body>
<div class="page-wrapper"> <div class="page-wrapper">
<div class="card"> <div class="card">
<h2>Submit YouTube Playlist</h2> <h2>Submit YouTube Media</h2>
<p class="subtitle"> <p class="subtitle">
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.
</p> </p>
{% with messages = get_flashed_messages() %} {% with messages = get_flashed_messages() %}
@@ -212,12 +259,40 @@ HTML_FORM = """
></textarea> ></textarea>
</div> </div>
<button type="submit">Submit Playlist</button> <fieldset class="form-group">
<legend>Download as:</legend>
<div class="download-options">
<label><input type="radio" name="download_mode" value="player_slot" checked> Player Slot</label>
<label><input type="radio" name="download_mode" value="mp4"> MP4 Export</label>
<label><input type="radio" name="download_mode" value="m4a"> M4A Audio</label>
<label><input type="radio" name="download_mode" value="mp3"> MP3 Audio</label>
</div>
</fieldset>
<button type="submit">Submit</button>
</form> </form>
<p class="helper"> <p class="helper">
Playlists are kept for roughly 48 hours after they’re downloaded. Playlists are kept for roughly 48 hours after they’re downloaded.
Utility exports are kept until an operator removes them.
</p> </p>
{% if export_jobs %}
<section class="export-status">
<strong>Recent utility exports</strong>
{% for job in export_jobs %}
<div class="export-job">
<div>{{ download_mode_labels[job.mode] }} — {{ export_status_labels[job.status] }} (job {{ job.job_id[:8] }})</div>
{% for path in job.output_paths %}
<div class="export-path">{{ path }}</div>
{% endfor %}
{% if job.error %}
<div class="export-error">{{ job.error }}</div>
{% endif %}
</div>
{% endfor %}
</section>
{% endif %}
</div> </div>
</div> </div>
</body> </body>
@@ -238,6 +313,19 @@ MAX_URL_LENGTH = 2048
MAX_SUBMITTER_LENGTH = 200 MAX_SUBMITTER_LENGTH = 200
MAX_NOTE_LENGTH = 1000 MAX_NOTE_LENGTH = 1000
MAX_REQUEST_CONTENT_LENGTH = 32 * 1024 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: 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: def _submission_message(result: dict[str, object]) -> str:
status = result.get("status") 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": if status == "queue_full":
seconds = max(0, int(result.get("oldest_expires_in_seconds") or 0)) seconds = max(0, int(result.get("oldest_expires_in_seconds") or 0))
hours = max(1, (seconds + 3599) // 3600) hours = max(1, (seconds + 3599) // 3600)
@@ -308,7 +410,19 @@ def create_app(
@web_app.get("/") @web_app.get("/")
def index(): 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) @web_app.errorhandler(413)
def _submission_too_large(_error): def _submission_too_large(_error):
@@ -320,6 +434,8 @@ def create_app(
url = request.form.get("url", "").strip() url = request.form.get("url", "").strip()
submitter = request.form.get("submitter", "").strip() or None submitter = request.form.get("submitter", "").strip() or None
note = request.form.get("note", "").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): if not _valid_media_url(url):
flash( flash(
@@ -333,9 +449,20 @@ def create_app(
if note is not None and len(note) > MAX_NOTE_LENGTH: if note is not None and len(note) > MAX_NOTE_LENGTH:
flash(f"The note must be {MAX_NOTE_LENGTH} characters or fewer.") flash(f"The note must be {MAX_NOTE_LENGTH} characters or fewer.")
return redirect(url_for("index")) 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: try:
if download_mode == "player_slot":
result = service.enqueue_download(url, submitter, note) result = service.enqueue_download(url, submitter, note)
else:
result = service.enqueue_export(
url,
download_mode,
submitter,
note,
)
except ControllerUnavailableError as exc: except ControllerUnavailableError as exc:
LOG.warning("playlist submission was rejected: %s", exc) LOG.warning("playlist submission was rejected: %s", exc)
flash("The Controller is unavailable; please try again shortly.") flash("The Controller is unavailable; please try again shortly.")
+7
View File
@@ -48,6 +48,13 @@ downloads:
max_workers: 2 max_workers: 2
process_timeout_seconds: 7200 process_timeout_seconds: 7200
max_height: 1080 max_height: 1080
# 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
router: router:
required_mode: video required_mode: video
+42
View File
@@ -63,6 +63,10 @@ class DownloadConfig:
max_workers: int = 2 max_workers: int = 2
process_timeout_seconds: float = 2 * 3600.0 process_timeout_seconds: float = 2 * 3600.0
max_height: int = 1080 max_height: int = 1080
export_directory: Path = Path("exports")
export_max_height: int = 1080
export_max_workers: int = 1
max_queued_exports: int = 20
@dataclass(frozen=True) @dataclass(frozen=True)
@@ -172,6 +176,23 @@ def _optional_path(value: Any, name: str) -> Path | None:
return Path(text) return Path(text)
def _relative_directory(value: Any, name: str) -> Path:
text = _text(value, name)
assert text is not None
path = Path(text)
if (
path.is_absolute()
or not path.parts
or any(part in {".", ".."} for part in path.parts)
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
def _env_value(env: Mapping[str, str], key: str, fallback: Any) -> Any: def _env_value(env: Mapping[str, str], key: str, fallback: Any) -> Any:
return env[key] if key in env else fallback return env[key] if key in env else fallback
@@ -277,6 +298,27 @@ def load_config(
max_workers=_integer(downloads_raw.get("max_workers", 2), "downloads.max_workers", maximum=10), max_workers=_integer(downloads_raw.get("max_workers", 2), "downloads.max_workers", maximum=10),
process_timeout_seconds=_number(downloads_raw.get("process_timeout_seconds", 2 * 3600), "downloads.process_timeout_seconds", minimum=1.0), process_timeout_seconds=_number(downloads_raw.get("process_timeout_seconds", 2 * 3600), "downloads.process_timeout_seconds", minimum=1.0),
max_height=_integer(downloads_raw.get("max_height", 1080), "downloads.max_height"), max_height=_integer(downloads_raw.get("max_height", 1080), "downloads.max_height"),
export_directory=_relative_directory(
downloads_raw.get("export_directory", "exports"),
"downloads.export_directory",
),
export_max_height=_integer(
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",
# The web status endpoint deliberately returns at most 100 rows.
# Keep the queue bound within that view so every active job remains
# visible even when the configured queue is full.
maximum=100,
),
) )
if downloads.cookie_required and downloads.cookie_path is None: if downloads.cookie_required and downloads.cookie_path is None:
raise ConfigError("downloads.cookie_path is required when cookie_required is true") raise ConfigError("downloads.cookie_path is required when cookie_required is true")
+798 -8
View File
@@ -2,14 +2,18 @@
from __future__ import annotations from __future__ import annotations
import json
import logging import logging
import os import os
import re
import signal import signal
import shutil import shutil
import stat
import subprocess import subprocess
import tempfile import tempfile
import threading import threading
import time import time
import unicodedata
import uuid import uuid
from collections.abc import Iterator from collections.abc import Iterator
from contextlib import contextmanager from contextlib import contextmanager
@@ -20,12 +24,31 @@ from typing import Any, Callable
from .config import ControllerConfig from .config import ControllerConfig
from .contracts import SLOT_IDS 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 from .store import ControllerStore
LOG = logging.getLogger(__name__) LOG = logging.getLogger(__name__)
MEDIA_EXTENSIONS = frozenset({".mp4", ".mkv", ".webm", ".mov", ".m4v"}) 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) @dataclass(frozen=True)
@@ -43,6 +66,16 @@ class DownloadEvent:
reused_slot: bool = False 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] EventCallback = Callable[[DownloadEvent], None]
Runner = Callable[..., subprocess.CompletedProcess[str]] Runner = Callable[..., subprocess.CompletedProcess[str]]
@@ -57,6 +90,60 @@ def canonical_playlist_path(media_base: Path, slot: int) -> Path:
return canonical_slot_directory(media_base, slot) / "playlist.m3u" return canonical_slot_directory(media_base, slot) / "playlist.m3u"
def player_format_selector(max_height: int) -> str:
"""Select the best Player media without exceeding its configured height."""
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: def title_from_filename(path: Path) -> str:
stem = path.stem stem = path.stem
if " - " in stem: if " - " in stem:
@@ -77,6 +164,7 @@ class DownloadManager:
on_event: EventCallback | None = None, on_event: EventCallback | None = None,
runner: Runner | None = None, runner: Runner | None = None,
executor: ThreadPoolExecutor | None = None, executor: ThreadPoolExecutor | None = None,
export_executor: ThreadPoolExecutor | None = None,
): ):
self.config = config self.config = config
self.store = store self.store = store
@@ -87,6 +175,19 @@ class DownloadManager:
thread_name_prefix="yt-download", thread_name_prefix="yt-download",
) )
self._owns_executor = executor is None 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._lock = threading.Lock()
self._futures: set[Future[Any]] = set() self._futures: set[Future[Any]] = set()
self._processes: set[subprocess.Popen[str]] = set() self._processes: set[subprocess.Popen[str]] = set()
@@ -161,13 +262,72 @@ class DownloadManager:
future.add_done_callback(self._forget_future) future.add_done_callback(self._forget_future)
return reservation 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 schedule_export_reconciliation(self) -> bool:
"""Queue crash-artifact recovery without delaying core service startup."""
with self._lock:
if self._stopping:
return False
try:
future = self._export_executor.submit(self.reconcile_export_artifacts)
except BaseException:
LOG.exception("could not schedule utility export reconciliation")
return False
with self._lock:
self._futures.add(future)
future.add_done_callback(self._forget_future)
return True
def _forget_future(self, future: Future[Any]) -> None: def _forget_future(self, future: Future[Any]) -> None:
with self._lock: with self._lock:
self._futures.discard(future) self._futures.discard(future)
try: try:
future.result() future.result()
except BaseException: except BaseException:
LOG.exception("download worker escaped its failure boundary") LOG.exception("media worker escaped its failure boundary")
def _emit(self, event: DownloadEvent) -> None: def _emit(self, event: DownloadEvent) -> None:
try: try:
@@ -297,10 +457,7 @@ class DownloadManager:
/ "%(playlist_index)03d - %(title)s [%(id)s].%(ext)s" / "%(playlist_index)03d - %(title)s [%(id)s].%(ext)s"
), ),
"-f", "-f",
( player_format_selector(self.config.downloads.max_height),
f"bv*[height<={self.config.downloads.max_height}]+ba/"
f"b[height<={self.config.downloads.max_height}]/b"
),
url, url,
] ]
self._run_command( self._run_command(
@@ -371,12 +528,643 @@ 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"
# ffprobe reports the same comma-separated demuxer aliases for MOV and
# MP4. A .mov source therefore needs at least a stream-copy remux; it
# must never be renamed to .mp4 solely because the alias list includes
# "mp4".
container_compatible = (
source.suffix.lower() == ".mp4"
and "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 @staticmethod
def _failure_message(exc: BaseException) -> str: 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): 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): 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}" return f"download failed: {exc}"
def _run_command( def _run_command(
@@ -705,6 +1493,8 @@ class DownloadManager:
self._terminate_process(process, force=True) self._terminate_process(process, force=True)
if self._owns_executor: if self._owns_executor:
self._executor.shutdown(wait=wait, cancel_futures=True) 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: def active_future_count(self) -> int:
with self._lock: with self._lock:
+22
View File
@@ -10,6 +10,7 @@ MEDIA_STATES = frozenset(
) )
LIVE_JOB_STATES = frozenset({"reserved", "running"}) LIVE_JOB_STATES = frozenset({"reserved", "running"})
LIVE_GUARD_STATES = frozenset({"active", "review_required"}) LIVE_GUARD_STATES = frozenset({"active", "review_required"})
UTILITY_EXPORT_MODES = frozenset({"mp4", "m4a", "mp3"})
@dataclass(frozen=True) @dataclass(frozen=True)
@@ -54,6 +55,27 @@ class DownloadJob:
error: str | None error: str | None
@dataclass(frozen=True)
class ExportReservation:
status: str
job_id: str | None = None
@dataclass(frozen=True)
class ExportJob:
job_id: str
mode: str
url: str
status: str
submitter: str | None
note: str | None
requested_at: float
started_at: float | None
finished_at: float | None
output_paths: tuple[str, ...]
error: str | None
@dataclass(frozen=True) @dataclass(frozen=True)
class DeleteToken: class DeleteToken:
slot: int slot: int
+54
View File
@@ -165,6 +165,11 @@ class ControllerService:
self._coordinator.start() self._coordinator.start()
self.mqtt.start() self.mqtt.start()
self._started = True self._started = True
# A crash-published playlist may contain many files to probe.
# Queue that bounded work only after core readiness and keep it
# on the dedicated export executor, so MQTT reconnect and
# Player coordination never wait for recovery to finish.
self.downloads.schedule_export_reconciliation()
except BaseException: except BaseException:
self._stop_event.set() self._stop_event.set()
try: try:
@@ -252,6 +257,26 @@ class ControllerService:
{"url": url, "submitter": submitter, "note": note}, {"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]: def submit_control(self, payload: Mapping[str, Any]) -> dict[str, Any]:
return self._call("control", dict(payload)) return self._call("control", dict(payload))
@@ -403,6 +428,35 @@ class ControllerService:
self._handle_mqtt_event(event) self._handle_mqtt_event(event)
def _handle_call(self, action: str, payload: Any) -> Any: 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": if action == "enqueue_download":
assert self.downloads is not None assert self.downloads is not None
url = payload["url"] url = payload["url"]
+255 -3
View File
@@ -25,6 +25,8 @@ from .models import (
DeleteToken, DeleteToken,
DownloadJob, DownloadJob,
DownloadReservation, DownloadReservation,
ExportJob,
ExportReservation,
MigrationError, MigrationError,
MigrationReport, MigrationReport,
PlayerObservation, PlayerObservation,
@@ -33,10 +35,11 @@ from .models import (
SlotProtectedError, SlotProtectedError,
SlotRecord, SlotRecord,
StaleOperationError, StaleOperationError,
UTILITY_EXPORT_MODES,
) )
SCHEMA_VERSION = 2 SCHEMA_VERSION = 3
LEGACY_MIGRATION_NAME = "legacy_playlists_yaml_v1" LEGACY_MIGRATION_NAME = "legacy_playlists_yaml_v1"
MAX_LEGACY_YAML_BYTES = 8 * 1024 * 1024 MAX_LEGACY_YAML_BYTES = 8 * 1024 * 1024
@@ -104,6 +107,24 @@ CREATE UNIQUE INDEX IF NOT EXISTS one_live_download_per_slot
ON download_jobs(slot_number) ON download_jobs(slot_number)
WHERE status IN ('reserved','running'); WHERE status IN ('reserved','running');
CREATE TABLE IF NOT EXISTS export_jobs (
job_id TEXT PRIMARY KEY,
mode TEXT NOT NULL CHECK (mode IN ('mp4','m4a','mp3')),
url TEXT NOT NULL,
status TEXT NOT NULL CHECK (status IN
('queued','running','completed','failed','interrupted')),
submitter TEXT,
note TEXT,
requested_at REAL NOT NULL,
started_at REAL,
finished_at REAL,
output_paths_json TEXT NOT NULL DEFAULT '[]',
error TEXT
);
CREATE INDEX IF NOT EXISTS export_jobs_recent
ON export_jobs(requested_at DESC, job_id DESC);
CREATE TABLE IF NOT EXISTS slot_guards ( CREATE TABLE IF NOT EXISTS slot_guards (
guard_id TEXT PRIMARY KEY, guard_id TEXT PRIMARY KEY,
slot_number INTEGER NOT NULL REFERENCES slots(slot_number), slot_number INTEGER NOT NULL REFERENCES slots(slot_number),
@@ -401,6 +422,13 @@ class ControllerStore:
""", """,
(now,), (now,),
) )
connection.execute(
"""
INSERT OR IGNORE INTO schema_migrations(version, name, applied_at)
VALUES (3, 'utility_export_jobs', ?)
""",
(now,),
)
connection.execute(f"PRAGMA user_version = {SCHEMA_VERSION}") connection.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")
def _upgrade_database(self, connection: sqlite3.Connection) -> None: def _upgrade_database(self, connection: sqlite3.Connection) -> None:
@@ -409,12 +437,13 @@ class ControllerStore:
version = connection.execute("PRAGMA user_version").fetchone()[0] version = connection.execute("PRAGMA user_version").fetchone()[0]
if version == SCHEMA_VERSION: if version == SCHEMA_VERSION:
return return
if version != 1: if version not in {1, 2}:
raise MigrationError( raise MigrationError(
f"unsupported Controller database schema {version}; expected 1 or {SCHEMA_VERSION}" f"unsupported Controller database schema {version}; expected 1, 2, or {SCHEMA_VERSION}"
) )
connection.execute("BEGIN IMMEDIATE") connection.execute("BEGIN IMMEDIATE")
try: try:
if version < 2:
columns = { columns = {
row["name"] row["name"]
for row in connection.execute( for row in connection.execute(
@@ -432,6 +461,37 @@ class ControllerStore:
""", """,
(self._now(),), (self._now(),),
) )
connection.execute(
"""
CREATE TABLE IF NOT EXISTS export_jobs (
job_id TEXT PRIMARY KEY,
mode TEXT NOT NULL CHECK (mode IN ('mp4','m4a','mp3')),
url TEXT NOT NULL,
status TEXT NOT NULL CHECK (status IN
('queued','running','completed','failed','interrupted')),
submitter TEXT,
note TEXT,
requested_at REAL NOT NULL,
started_at REAL,
finished_at REAL,
output_paths_json TEXT NOT NULL DEFAULT '[]',
error TEXT
)
"""
)
connection.execute(
"""
CREATE INDEX IF NOT EXISTS export_jobs_recent
ON export_jobs(requested_at DESC, job_id DESC)
"""
)
connection.execute(
"""
INSERT OR IGNORE INTO schema_migrations(version, name, applied_at)
VALUES (3, 'utility_export_jobs', ?)
""",
(self._now(),),
)
connection.execute(f"PRAGMA user_version = {SCHEMA_VERSION}") connection.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")
except BaseException: except BaseException:
connection.rollback() connection.rollback()
@@ -837,6 +897,15 @@ class ControllerStore:
row["generation"], row["generation"],
), ),
) )
connection.execute(
"""
UPDATE export_jobs
SET status = 'interrupted', finished_at = ?,
error = 'export interrupted by Controller restart'
WHERE status IN ('queued','running')
""",
(now,),
)
def list_slots(self) -> list[SlotRecord]: def list_slots(self) -> list[SlotRecord]:
with self._lock: with self._lock:
@@ -1136,6 +1205,189 @@ class ControllerStore:
error=row["error"], error=row["error"],
) )
@staticmethod
def _row_to_export_job(row: sqlite3.Row) -> ExportJob:
try:
raw_paths = json.loads(row["output_paths_json"])
except (TypeError, ValueError, json.JSONDecodeError):
raw_paths = []
output_paths = tuple(
value for value in raw_paths if isinstance(value, str)
) if isinstance(raw_paths, list) else ()
return ExportJob(
job_id=row["job_id"],
mode=row["mode"],
url=row["url"],
status=row["status"],
submitter=row["submitter"],
note=row["note"],
requested_at=row["requested_at"],
started_at=row["started_at"],
finished_at=row["finished_at"],
output_paths=output_paths,
error=row["error"],
)
def reserve_export_job(
self,
url: str,
mode: str,
submitter: str | None,
note: str | None,
*,
max_active: int,
job_id: str | None = None,
) -> ExportReservation:
if not isinstance(url, str) or not url.strip():
raise ValueError("export URL must be non-empty")
if mode not in UTILITY_EXPORT_MODES:
raise ValueError("unsupported utility export mode")
if not isinstance(max_active, int) or isinstance(max_active, bool) or max_active < 1:
raise ValueError("max_active must be a positive integer")
job_id = job_id or uuid.uuid4().hex
with self._transaction() as connection:
active = connection.execute(
"""
SELECT COUNT(*) FROM export_jobs
WHERE status IN ('queued','running')
"""
).fetchone()[0]
if active >= max_active:
return ExportReservation(status="queue_full")
connection.execute(
"""
INSERT INTO export_jobs
(job_id, mode, url, status, submitter, note, requested_at)
VALUES (?, ?, ?, 'queued', ?, ?, ?)
""",
(job_id, mode, url.strip(), submitter, note, self._now()),
)
return ExportReservation(status="queued", job_id=job_id)
def mark_export_running(self, job_id: str) -> bool:
with self._transaction() as connection:
return bool(
connection.execute(
"""
UPDATE export_jobs SET status = 'running', started_at = ?
WHERE job_id = ? AND status = 'queued'
""",
(self._now(), job_id),
).rowcount
)
def complete_export_job(
self,
job_id: str,
output_paths: Sequence[str],
) -> bool:
normalized = tuple(output_paths)
if not normalized or any(
not isinstance(path, str) or not path for path in normalized
):
raise ValueError("completed export must contain output paths")
with self._transaction() as connection:
return bool(
connection.execute(
"""
UPDATE export_jobs
SET status = 'completed', finished_at = ?,
output_paths_json = ?, error = NULL
WHERE job_id = ? AND status = 'running'
""",
(self._now(), json.dumps(normalized), job_id),
).rowcount
)
def fail_export_job(self, job_id: str, error: str) -> bool:
with self._transaction() as connection:
return bool(
connection.execute(
"""
UPDATE export_jobs
SET status = 'failed', finished_at = ?, error = ?
WHERE job_id = ? AND status IN ('queued','running')
""",
(self._now(), error, job_id),
).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(
"SELECT * FROM export_jobs WHERE job_id = ?", (job_id,)
).fetchone()
return None if row is None else self._row_to_export_job(row)
def list_export_jobs(self, *, limit: int = 20) -> list[ExportJob]:
if not isinstance(limit, int) or isinstance(limit, bool) or not 1 <= limit <= 100:
raise ValueError("export job limit must be between 1 and 100")
with self._lock:
rows = self.connection.execute(
"""
SELECT * FROM export_jobs
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( def protect_slot(
self, self,
slot: int, slot: int,
+108 -2
View File
@@ -14,7 +14,14 @@ from controller.service import ControllerService, ControllerUnavailableError
class FakeControllerService: 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 { self.enqueue_result = enqueue_result or {
"status": "queued", "status": "queued",
"slot": 3, "slot": 3,
@@ -29,7 +36,9 @@ class FakeControllerService:
"router_state": "video", "router_state": "video",
"degraded_reasons": [], "degraded_reasons": [],
} }
self.export_jobs = list(export_jobs or [])
self.enqueue_calls = [] self.enqueue_calls = []
self.export_calls = []
def enqueue_download(self, url, submitter, note): def enqueue_download(self, url, submitter, note):
self.enqueue_calls.append((url, submitter, note)) self.enqueue_calls.append((url, submitter, note))
@@ -37,6 +46,20 @@ class FakeControllerService:
raise self.enqueue_error raise self.enqueue_error
return dict(self.enqueue_result) 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): def health(self):
return dict(self.health_result) return dict(self.health_result)
@@ -71,11 +94,94 @@ class FlaskFactoryTests(unittest.TestCase):
response = self._app(FakeControllerService()).test_client().get("/") response = self._app(FakeControllerService()).test_client().get("/")
self.assertEqual(response.status_code, 200) 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'action="/submit"', response.data)
self.assertIn(b'name="url"', response.data) self.assertIn(b'name="url"', response.data)
self.assertIn(b'name="submitter"', response.data) self.assertIn(b'name="submitter"', response.data)
self.assertIn(b'name="note"', 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): def test_submit_rejects_non_youtube_and_malformed_urls(self):
invalid_urls = ( invalid_urls = (
+46
View File
@@ -82,6 +82,10 @@ class ConfigurationTests(unittest.TestCase):
self.assertEqual(config.storage.uncertain_review_after_seconds, 900.0) self.assertEqual(config.storage.uncertain_review_after_seconds, 900.0)
self.assertEqual(config.mqtt.publish_ack_timeout_seconds, 2.0) self.assertEqual(config.mqtt.publish_ack_timeout_seconds, 2.0)
self.assertEqual(config.mqtt.router_echo_grace_seconds, 2.0) 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): def test_documented_environment_overrides_apply(self):
env = { env = {
@@ -175,6 +179,48 @@ class ConfigurationTests(unittest.TestCase):
with self.assertRaisesRegex(ConfigError, f"{field} must be an absolute path"): with self.assertRaisesRegex(ConfigError, f"{field} must be an absolute path"):
load_config(path, environ={}) load_config(path, environ={})
def test_export_directory_must_stay_relative_to_media_root(self):
invalid = ("/tmp/exports", "../exports", "nested/../exports", r"nested\\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, "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", 101),
)
for field, value in cases:
with self.subTest(field=field), tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "config.yaml"
path.write_text(
f"downloads:\n {field}: {value}\n",
encoding="utf-8",
)
with self.assertRaisesRegex(ConfigError, field):
load_config(path, environ={})
class ImportLifecycleTests(unittest.TestCase): class ImportLifecycleTests(unittest.TestCase):
def test_controller_core_import_does_not_connect_or_create_runtime_paths(self): def test_controller_core_import_does_not_connect_or_create_runtime_paths(self):
+121 -2
View File
@@ -8,8 +8,14 @@ import unittest
from concurrent.futures import Future from concurrent.futures import Future
from pathlib import Path from pathlib import Path
from yt_dlp import YoutubeDL
from controller.config import ControllerConfig, DownloadConfig, StorageConfig from controller.config import ControllerConfig, DownloadConfig, StorageConfig
from controller.downloads import DownloadManager, canonical_playlist_path from controller.downloads import (
DownloadManager,
canonical_playlist_path,
player_format_selector,
)
from controller.models import SlotProtectedError from controller.models import SlotProtectedError
from controller.store import ControllerStore from controller.store import ControllerStore
@@ -169,7 +175,120 @@ class DownloadManagerTests(unittest.TestCase):
self.assertTrue(all(not path.exists() for path in working_paths)) self.assertTrue(all(not path.exists() for path in working_paths))
self.assertEqual(self.cookie.read_text(encoding="utf-8"), "cookie data") self.assertEqual(self.cookie.read_text(encoding="utf-8"), "cookie data")
download_arguments = runner.calls[-1][0] download_arguments = runner.calls[-1][0]
self.assertIn("bv*[height<=720]+ba/b[height<=720]/b", download_arguments) self.assertIn("bv*[height<=720]+ba/b[height<=720]", download_arguments)
@staticmethod
def _select_player_format(max_height, formats):
with YoutubeDL({"quiet": True}) as ydl:
selector = ydl.build_format_selector(
player_format_selector(max_height)
)
selected = list(ydl._select_formats(formats, selector))
return selected
def test_player_selector_chooses_separate_true_1080_video_and_audio(self):
formats = [
{
"format_id": "22",
"url": "https://example.test/720.mp4",
"ext": "mp4",
"height": 720,
"width": 1280,
"vcodec": "avc1.64001f",
"acodec": "mp4a.40.2",
"tbr": 2500,
"filesize": 1,
},
{
"format_id": "399",
"url": "https://example.test/1080.mp4",
"ext": "mp4",
"height": 1080,
"width": 1920,
"vcodec": "av01.0.08M.08",
"acodec": "none",
"tbr": 4500,
"filesize": 1,
},
{
"format_id": "251",
"url": "https://example.test/audio.webm",
"ext": "webm",
"vcodec": "none",
"acodec": "opus",
"tbr": 160,
"filesize": 1,
},
]
selected = self._select_player_format(1080, formats)
self.assertEqual(len(selected), 1)
self.assertEqual(selected[0]["height"], 1080)
self.assertEqual(
[item["format_id"] for item in selected[0]["requested_formats"]],
["399", "251"],
)
self.assertEqual(selected[0]["vcodec"], "av01.0.08M.08")
self.assertEqual(selected[0]["acodec"], "opus")
def test_player_selector_preserves_lower_source_resolution(self):
formats = [
{
"format_id": "18",
"url": "https://example.test/360.mp4",
"ext": "mp4",
"height": 360,
"width": 640,
"vcodec": "avc1.42001e",
"acodec": "mp4a.40.2",
"tbr": 700,
"filesize": 1,
},
{
"format_id": "22",
"url": "https://example.test/720.mp4",
"ext": "mp4",
"height": 720,
"width": 1280,
"vcodec": "avc1.64001f",
"acodec": "mp4a.40.2",
"tbr": 2500,
"filesize": 1,
},
]
selected = self._select_player_format(1080, formats)
self.assertEqual(len(selected), 1)
self.assertEqual(selected[0]["format_id"], "22")
self.assertEqual(selected[0]["height"], 720)
def test_player_selector_never_uses_unfiltered_over_height_fallback(self):
formats = [
{
"format_id": "401",
"url": "https://example.test/2160.mp4",
"ext": "mp4",
"height": 2160,
"width": 3840,
"vcodec": "av01.0.12M.08",
"acodec": "none",
"tbr": 12000,
"filesize": 1,
},
{
"format_id": "251",
"url": "https://example.test/audio.webm",
"ext": "webm",
"vcodec": "none",
"acodec": "opus",
"tbr": 160,
"filesize": 1,
},
]
self.assertEqual(self._select_player_format(1080, formats), [])
def test_worker_failure_is_contained_and_persisted(self): def test_worker_failure_is_contained_and_persisted(self):
events = [] events = []
+655
View File
@@ -0,0 +1,655 @@
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",
"mov": ".mov",
}[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 in {"mp4", "mov"}
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_quicktime_alias_is_remuxed_instead_of_renamed_to_mp4(self):
runner = FakeExportRunner(container="mov")
result = self.manager(runner).enqueue_export(
"https://www.youtube.com/watch?v=mov", "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_export_reconciliation_is_submitted_to_dedicated_executor(self):
export_executor = QueuedExecutor()
manager = DownloadManager(
self.config,
self.store,
runner=FakeExportRunner(),
executor=ImmediateExecutor(),
export_executor=export_executor,
)
self.assertTrue(manager.schedule_export_reconciliation())
self.assertEqual(len(export_executor.calls), 1)
_, function, arguments, keyword_arguments = export_executor.calls[0]
self.assertEqual(function, manager.reconcile_export_artifacts)
self.assertEqual(arguments, ())
self.assertEqual(keyword_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()
+73
View File
@@ -89,11 +89,27 @@ class FakeDownloadManager:
self.stopped = False self.stopped = False
self.delete_requests = [] self.delete_requests = []
self.enqueue_calls = [] self.enqueue_calls = []
self.export_calls = []
self.export_reconciled = False
def enqueue(self, url, submitter, note): def enqueue(self, url, submitter, note):
self.enqueue_calls.append((url, submitter, note)) self.enqueue_calls.append((url, submitter, note))
return self.store.reserve_download(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 schedule_export_reconciliation(self):
self.export_reconciled = True
return True
def recover_pending_deletions(self): def recover_pending_deletions(self):
self.recovered = True self.recovered = True
@@ -171,6 +187,7 @@ class ControllerServiceTests(unittest.TestCase):
) )
self.service.start() self.service.start()
self.assertFalse(self.downloads.recovered) self.assertFalse(self.downloads.recovered)
self.assertTrue(self.downloads.export_reconciled)
return self.service return self.service
def ready_slot(self, slot=1): def ready_slot(self, slot=1):
@@ -497,6 +514,62 @@ class ControllerServiceTests(unittest.TestCase):
self.assertIn("mqtt_disconnected", health["degraded_reasons"]) self.assertIn("mqtt_disconnected", health["degraded_reasons"])
self.assertIn("awaiting_player_evidence", 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): def test_cleanup_waits_for_post_start_player_evidence_but_new_ingest_does_not(self):
self.start_service() self.start_service()
self.service._next_expiry = self.monotonic() self.service._next_expiry = self.monotonic()
+160 -2
View File
@@ -50,7 +50,7 @@ class FreshStoreTests(StoreTestCase):
self.assertTrue(all(slot.media_state == "empty" for slot in slots)) self.assertTrue(all(slot.media_state == "empty" for slot in slots))
self.assertEqual(store.player_observation().phase, "unknown") self.assertEqual(store.player_observation().phase, "unknown")
def test_schema_one_database_upgrades_external_reconciliation_atomically(self): def test_schema_one_database_upgrades_reconciliation_and_exports_atomically(self):
with self.open_store() as store: with self.open_store() as store:
database_path = store.path database_path = store.path
connection = sqlite3.connect(database_path) connection = sqlite3.connect(database_path)
@@ -67,7 +67,7 @@ class FreshStoreTests(StoreTestCase):
with self.open_store() as upgraded: with self.open_store() as upgraded:
self.assertEqual( self.assertEqual(
upgraded.connection.execute("PRAGMA user_version").fetchone()[0], upgraded.connection.execute("PRAGMA user_version").fetchone()[0],
2, 3,
) )
columns = { columns = {
row["name"] row["name"]
@@ -80,6 +80,164 @@ class FreshStoreTests(StoreTestCase):
"SELECT name FROM schema_migrations WHERE version = 2" "SELECT name FROM schema_migrations WHERE version = 2"
).fetchone() ).fetchone()
self.assertEqual(migration["name"], "external_playback_reconciliation") self.assertEqual(migration["name"], "external_playback_reconciliation")
export_migration = upgraded.connection.execute(
"SELECT name FROM schema_migrations WHERE version = 3"
).fetchone()
self.assertEqual(export_migration["name"], "utility_export_jobs")
tables = {
row["name"]
for row in upgraded.connection.execute(
"SELECT name FROM sqlite_master WHERE type = 'table'"
).fetchall()
}
self.assertIn("export_jobs", tables)
def test_schema_two_database_adds_only_utility_export_jobs(self):
with self.open_store() as store:
database_path = store.path
connection = sqlite3.connect(database_path)
try:
connection.execute("DROP TABLE export_jobs")
connection.execute("DELETE FROM schema_migrations WHERE version = 3")
connection.execute("PRAGMA user_version = 2")
connection.commit()
finally:
connection.close()
with self.open_store() as upgraded:
self.assertEqual(
upgraded.connection.execute("PRAGMA user_version").fetchone()[0],
3,
)
self.assertEqual(upgraded.list_export_jobs(), [])
def test_export_jobs_are_slot_free_bounded_and_durable(self):
with self.open_store() as store:
first = store.reserve_export_job(
"https://www.youtube.com/watch?v=one",
"mp4",
"Alice",
"Sunday",
max_active=1,
job_id="export-one",
)
self.assertEqual(first.status, "queued")
self.assertEqual(first.job_id, "export-one")
self.assertEqual(
store.reserve_export_job(
"https://www.youtube.com/watch?v=two",
"mp3",
None,
None,
max_active=1,
).status,
"queue_full",
)
self.assertTrue(store.mark_export_running("export-one"))
self.assertTrue(
store.complete_export_job(
"export-one",
("exports/video/export-one/Service.mp4",),
)
)
job = store.get_export_job("export-one")
self.assertEqual(job.mode, "mp4")
self.assertEqual(job.status, "completed")
self.assertEqual(job.submitter, "Alice")
self.assertEqual(
job.output_paths,
("exports/video/export-one/Service.mp4",),
)
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(
"https://www.youtube.com/playlist?list=one",
"m4a",
None,
None,
max_active=2,
job_id="queued-export",
)
store.reserve_export_job(
"https://www.youtube.com/watch?v=two",
"mp3",
None,
None,
max_active=2,
job_id="running-export",
)
store.mark_export_running("running-export")
store.close()
with self.open_store() as reopened:
jobs = {job.job_id: job for job in reopened.list_export_jobs()}
self.assertEqual(jobs["queued-export"].status, "interrupted")
self.assertEqual(jobs["running-export"].status, "interrupted")
self.assertTrue(
all("Controller restart" in job.error for job in jobs.values())
)
self.assertTrue(
all(slot.media_state == "empty" for slot in reopened.list_slots())
)
def test_export_failures_do_not_corrupt_player_inventory(self):
with self.open_store() as store:
slot = store.reserve_download(
"https://www.youtube.com/watch?v=managed",
None,
None,
job_id="managed",
)
store.fail_download(
"managed", slot.slot, slot.generation, "managed failure"
)
before = store.get_slot(slot.slot)
export = store.reserve_export_job(
"https://www.youtube.com/watch?v=export",
"mp3",
None,
None,
max_active=5,
job_id="failed-export",
)
self.assertEqual(export.status, "queued")
self.assertTrue(store.fail_export_job("failed-export", "ffmpeg failed"))
self.assertEqual(store.get_slot(slot.slot), before)
def test_slot_reservation_is_committed_before_worker_transition(self): def test_slot_reservation_is_committed_before_worker_transition(self):
with self.open_store() as store: with self.open_store() as store: