peter
7c0647f87a
Deep-review round 2, on round 1's own fix commit. It found MORE than round 1 did (28 raw, against 13), and most of it was mine -- a new piece of UI attracts findings a backend-only diff never does. The pattern in almost all of them is the same: a principle applied to the site that prompted it and not to its siblings. * `display_text` reached three of the FIVE writers of a user-supplied name. `POST /api/downloads` and `POST /api/downloads/edit` still handed Postgres an unbounded string, so a 300-character name pasted into the Add-download dialog or the editor was a bare 500 (main.py re-raises every DataError but the NUL case). Four of the seven reviewers found it independently. The bound now lives in `models.py`, the three columns are DECLARED with it, and the guard for the next writer is parsed rather than remembered: every `display_name` / `display_uploader` assignment under `app/routes/` must go through the helper. That guard immediately found two more -- the Google sign-in callback writes the profile name into the same 255-wide column -- so it is six, not five. * The connection release reached one of the THREE routes that read their own body. The screenshot route (10 MiB, from a phone) and the client log sink both still held a pooled connection idle-in-transaction for the length of the upload, which is the invariant db.py sizes the pool on. All three now call `db.release_read_connection`, whose precondition -- nothing pending on the shared session -- is CHECKED rather than described, because a whole-session rollback discarding a caller's staged write is a mistake this repo has already made once. `test_body_limit.py` holds the rule for the next such route. * The AST drift registry keyed its sites by (module, limit expression), so a second route copied from the first collapsed onto the same entry and went uncovered -- the exact hole the registry exists to close. Keyed per call site now, and it carries the release requirement too. * The dialog's own refusal was announced once and never again: two over-cap picks leave an identical node with identical text, and an untouched live region says nothing. It is keyed per pick now, the drop zone reads as invalid and points at the sentence, and the reason is reachable from the control rather than only from a one-shot announcement. * Two tests rebuilt the blind spot they were written to remove: the dialog's cap test sat nowhere near the boundary (a `>` loosened to `>=` survived it, which is the mutation round 1 fixed on the server side), and the header-precedence test only covered an exact-case duplicate while a differently-cased one would have been COMBINED onto the wire rather than replaced. Both fixed, and the merge now dedupes case-insensitively because header names are. Also: the upload dialog's unstubbed usage query was firing a real request from jsdom in a neighbouring suite and leaking a delayed "connection lost" toast across test boundaries; `GET /api/downloads/usage` had no test at all though the whole client-side check hangs off it; and the comment claiming the edge 413 is readable now says what was measured -- readable to 200 MiB over loopback, a bare `onerror` through a proxy -- so the pre-send check is named as the mechanism and that branch as the backstop it is. Every test added or changed here was mutation-checked. One mutation SURVIVED and is recorded rather than papered over: the `tooBig` term inside `submit()` cannot be reached, because `ui/Button` swallows the activation itself, so it is marked as belt and braces instead of being described as the guard.
870 lines
41 KiB
Python
870 lines
41 KiB
Python
"""Child-side download/edit pipeline — the yt-dlp download + ffmpeg edit that runs in the spawned subprocess."""
|
|
|
|
import re
|
|
import shutil
|
|
import time
|
|
from datetime import date, datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
import yt_dlp
|
|
|
|
from app import logs, sysconfig
|
|
from app.config import settings
|
|
from app.db import SessionLocal
|
|
from app.downloads import edit as editmod
|
|
from app.downloads import errors as dlerrors
|
|
from app.downloads import adapters, credentials, formats, mediakind, probe, quota, service, storage
|
|
from app.downloads import urlguard
|
|
from app.downloads.jobdb import (
|
|
_edit_staging_dir,
|
|
_job_status,
|
|
_reset_asset_pending,
|
|
_set_job,
|
|
_staging_dir,
|
|
_touch_job,
|
|
)
|
|
from app.downloads.stated import StatedFailure
|
|
from app.downloads.status import JobStatus
|
|
from app.models import DISPLAY_TEXT_MAX, DownloadJob, MediaAsset
|
|
from app.notifications import create_notification
|
|
from app.titles import normalize_title
|
|
|
|
log = logs.get_logger(__name__)
|
|
|
|
|
|
class _Aborted(Exception):
|
|
"""Raised inside the progress hook when the job was paused/canceled mid-download."""
|
|
|
|
|
|
class _OwnerRefused(Exception):
|
|
"""A refusal that belongs to THIS job's owner, not to the source.
|
|
|
|
MediaAssets are shared: two users asking for the same video and format get one row (see
|
|
`service.get_or_create_asset`), and the generic failure handler writes the message onto it —
|
|
correctly, because an upstream failure really is everyone's. A per-user refusal is not. Written
|
|
to the shared row it would fail a second user's queued job for the same video with the first
|
|
user's private reason (`worker._process_job` reads `asset.error` verbatim), telling someone with
|
|
an empty library to free up space. So this is caught separately: the job fails, the asset goes
|
|
back to `pending` for whoever else is holding it."""
|
|
|
|
|
|
# Both progress sources call back FAR faster than the database should be touched: yt-dlp fires its
|
|
# hook per chunk, and ffmpeg's `-progress pipe:1` emits a block of ~12 lines every ~0.5s while
|
|
# run_ffmpeg calls its callbacks once per LINE. Every callback here is a round-trip (`_job_status`
|
|
# opens a session and SELECTs; `_set_job` opens one and UPDATE+COMMITs), so unthrottled they cost
|
|
# ~10^5 queries on a long job. ONE interval, ONE implementation, used by both paths — three
|
|
# hand-rolled copies is how the ffmpeg side ended up unthrottled while the download side was fine.
|
|
_DB_WRITE_INTERVAL_S = 1.0
|
|
_CANCEL_POLL_INTERVAL_S = 1.0
|
|
|
|
_THUMB_SUFFIXES = (".jpg", ".jpeg", ".png", ".webp")
|
|
# yt-dlp's working files in the staging dir. None of them is ever the finished media, but several
|
|
# CAN be the biggest file there (a video-only per-format stream, an abandoned fragment run), so the
|
|
# staging fallback must exclude them by name — picking one would store e.g. a silent video-only
|
|
# stream as a ready asset.
|
|
_YTDLP_TEMP_RE = re.compile(
|
|
r"\.part(-Frag\d+)?$" # in-flight download / fragment
|
|
r"|\.ytdl$" # resume state
|
|
r"|\.temp\.[^.]+$" # merge/postprocess temp
|
|
r"|\.f\d+\.[^.]+$", # per-format stream kept before the merge
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
|
|
def _notify_download_done(db, user_id: int | None, title: str | None, job_id, *, kind: str) -> None:
|
|
"""Post a best-effort 'ready' notice to the owner's inbox — downloads/edits used to finish
|
|
silently while scheduler jobs and playlist-push already notified (R11 feedback symmetry). A
|
|
notification failure must never mask the completed job, so it is isolated + rolled back (mirrors
|
|
scheduler._notify_done). `kind` ("download"/"edit") lets the inbox pick accurate copy; the
|
|
English title here is only a fallback for an unknown type. `user_id`/`title` are captured by the
|
|
caller BEFORE its commit, so reading them here needs no post-commit refresh of an expired row."""
|
|
if user_id is None:
|
|
return
|
|
try:
|
|
create_notification(
|
|
db,
|
|
user_id=user_id,
|
|
type="download_done",
|
|
title=f"Ready to watch: {title}" if title else "Ready to watch",
|
|
data={"title": title, "kind": kind},
|
|
)
|
|
except Exception:
|
|
db.rollback()
|
|
log.exception("failed to write download-done notification for job %s", job_id)
|
|
|
|
|
|
def _parse_upload_date(raw) -> date | None:
|
|
if not raw:
|
|
return None
|
|
try:
|
|
return datetime.strptime(str(raw), "%Y%m%d").date()
|
|
# silent-ok: a date we cannot read becomes a date the card does not show, which is what the
|
|
# card shows for every source that never supplies one. Absent is the honest rendering.
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _fill_asset_meta_early(asset_id: int, info: dict) -> None:
|
|
"""Populate an asset's display metadata from yt-dlp's info_dict the first time it's seen
|
|
(only if not already filled at enqueue from our catalog) — so an ad-hoc URL's row shows a
|
|
real title + thumbnail while it downloads, not just the URL. Zero extra network cost."""
|
|
with SessionLocal() as db:
|
|
asset = db.get(MediaAsset, asset_id)
|
|
if asset is None or asset.title:
|
|
return
|
|
# `or info.get("id")` avoids a blank card while downloading when the cleaned title is empty
|
|
# (e.g. a title that was only a social engagement prefix); the completion path later sets
|
|
# the final title with its own source_ref fallback.
|
|
adapter = adapters.resolve_adapter(None, None, info)
|
|
asset.title = normalize_title(adapter.title(info)) or info.get("id")
|
|
asset.uploader = info.get("uploader") or info.get("channel")
|
|
asset.uploader_url = adapter.uploader_url(info)
|
|
asset.thumbnail_url = info.get("thumbnail")
|
|
asset.duration_s = int(info["duration"]) if info.get("duration") else None
|
|
asset.upload_date = _parse_upload_date(info.get("upload_date"))
|
|
asset.source_webpage_url = info.get("webpage_url")
|
|
db.commit()
|
|
|
|
|
|
def _make_progress_hook(job_id: int, asset_id: int):
|
|
state = {"meta_done": False}
|
|
cancelled = _cancel_poll(job_id) # same policy as the ffmpeg path
|
|
allow_write = _rate_limiter(_DB_WRITE_INTERVAL_S)
|
|
|
|
def hook(d: dict) -> None:
|
|
# Cooperative pause/cancel: if the job is no longer 'running', abort the download.
|
|
if cancelled():
|
|
raise _Aborted
|
|
if not state["meta_done"] and d.get("info_dict"):
|
|
state["meta_done"] = True
|
|
try:
|
|
_fill_asset_meta_early(asset_id, d["info_dict"])
|
|
except Exception: # noqa: BLE001 — metadata is best-effort, never fail a download
|
|
# Best-effort is about the DOWNLOAD, not about knowing. What this swallows is
|
|
# anything from a shape change in yt-dlp's info dict to the database being down,
|
|
# and all the user sees is a card with an id on it until the job finishes.
|
|
log.info(
|
|
"early metadata could not be filled in for this download",
|
|
extra=logs.event("download.meta.early_fill_failed", asset_id=asset_id),
|
|
exc_info=True,
|
|
)
|
|
if d.get("status") != "downloading":
|
|
return
|
|
if not allow_write():
|
|
return
|
|
# yt-dlp downloads the video and audio streams SEPARATELY (each 0→100%), then merges.
|
|
# Label the phase so the user understands the bar resetting between streams instead of
|
|
# seeing a mysterious 95%→5% jump.
|
|
info = d.get("info_dict") or {}
|
|
vcodec = info.get("vcodec") or "none"
|
|
acodec = info.get("acodec") or "none"
|
|
if vcodec != "none" and acodec == "none":
|
|
phase = "video"
|
|
elif acodec != "none" and vcodec == "none":
|
|
phase = "audio"
|
|
else:
|
|
phase = "media"
|
|
total = d.get("total_bytes") or d.get("total_bytes_estimate") or 0
|
|
done = d.get("downloaded_bytes") or 0
|
|
pct = int(done * 100 / total) if total else 0
|
|
_set_job(
|
|
job_id,
|
|
progress=min(pct, 99),
|
|
speed_bps=int(d["speed"]) if d.get("speed") else None,
|
|
eta_s=int(d["eta"]) if d.get("eta") else None,
|
|
phase=phase,
|
|
)
|
|
|
|
return hook
|
|
|
|
|
|
def _rate_limiter(interval_s: float, *, now=time.monotonic):
|
|
"""`should_run()` → True at most once per `interval_s` (the first call always passes).
|
|
`now` is injectable so tests drive a clock instead of patching the global one."""
|
|
# `None`, not 0.0: "never ran" has to be a VALUE, not a falsy timestamp. A truthiness test here
|
|
# reads a legitimate `now() == 0.0` as "never ran" and lets a second call straight through —
|
|
# unreachable with real `time.monotonic()`, but `now` exists to be injected, and a test clock
|
|
# starting at 0 is the obvious choice. That would let a throttle test encode the wrong
|
|
# behaviour and still pass.
|
|
state: dict[str, float | None] = {"at": None}
|
|
|
|
def should_run() -> bool:
|
|
t = now()
|
|
last = state["at"]
|
|
if last is not None and t - last < interval_s:
|
|
return False
|
|
state["at"] = t
|
|
return True
|
|
|
|
return should_run
|
|
|
|
|
|
def _cancel_poll(job_id: int, interval_s: float = _CANCEL_POLL_INTERVAL_S, *, now=time.monotonic):
|
|
"""Cooperative "has this job been paused/cancelled?" poll, rate-limited and sticky: once it
|
|
says yes it never queries again. Shared by the yt-dlp hook and run_ffmpeg so cancel latency is
|
|
one policy, not one per call site."""
|
|
allow = _rate_limiter(interval_s, now=now)
|
|
state = {"cancelled": False}
|
|
|
|
def cancelled() -> bool:
|
|
if not state["cancelled"] and allow():
|
|
state["cancelled"] = _job_status(job_id) not in ("running", None)
|
|
return state["cancelled"]
|
|
|
|
return cancelled
|
|
|
|
|
|
def _ffmpeg_progress_writer(job_id: int, phase: str, *, now=time.monotonic):
|
|
"""`on_progress` for run_ffmpeg: persist a progress row write at most once per interval and only
|
|
when the whole-percent value changed (the bar shows integers), plus a separate once-per-interval
|
|
heartbeat while the percent is stuck (so a slow re-encode isn't mistaken for a stall)."""
|
|
# Progress writes have their OWN 1/interval budget so a heartbeat can never delay a real write (the
|
|
# round-1 concern). The heartbeat, in turn, is gated on `last_touch` — the last DB write OR beat —
|
|
# so a progress write (which already refreshes updated_at) suppresses a redundant heartbeat that
|
|
# same interval, keeping active downloads at ~1 write/interval, not two.
|
|
allow_write = _rate_limiter(_DB_WRITE_INTERVAL_S, now=now)
|
|
state = {"pct": -1}
|
|
last_touch: dict = {"t": None} # monotonic time of the last DB write/heartbeat
|
|
|
|
def write(pct: float) -> None:
|
|
whole = int(pct)
|
|
if whole != state["pct"]:
|
|
if allow_write():
|
|
state["pct"] = whole
|
|
_set_job(job_id, progress=whole, phase=phase)
|
|
last_touch["t"] = now()
|
|
return
|
|
# Same integer percent, but ffmpeg IS still working — a slow re-encode (AV1/VP9 -> H.264 or a
|
|
# crop on the CPU-only host) can sit on one percent for minutes. Heartbeat updated_at so the
|
|
# stall watchdog doesn't kill a live job (B-sub-4) — but only if no write/beat already touched
|
|
# the row this interval, so a progressing download doesn't pay a redundant write.
|
|
t = now()
|
|
if last_touch["t"] is None or t - last_touch["t"] >= _DB_WRITE_INTERVAL_S:
|
|
_touch_job(job_id)
|
|
last_touch["t"] = t
|
|
|
|
return write
|
|
|
|
|
|
def _pp_phase(pp: str) -> str:
|
|
"""Map a yt-dlp postprocessor class name to a user-facing phase label. ffmpeg post-steps
|
|
aren't byte-progress operations (no %), so the UI shows the step name + an indeterminate
|
|
pulse — this makes the long merge/finalize legible instead of a silent 'Processing'."""
|
|
if "Merg" in pp:
|
|
return "merging"
|
|
if "ExtractAudio" in pp:
|
|
return "audio_extract"
|
|
if "Thumbnail" in pp:
|
|
return "thumbnail"
|
|
if "SponsorBlock" in pp or "ModifyChapters" in pp:
|
|
return "sponsorblock"
|
|
if "Metadata" in pp:
|
|
return "metadata"
|
|
return "processing"
|
|
|
|
|
|
def _make_pp_hook(job_id: int):
|
|
"""Postprocessor hook: surface the current ffmpeg step so the bar shows a concrete label
|
|
(Merging / Embedding thumbnail / …) instead of freezing at 100% after the streams finish."""
|
|
|
|
def hook(d: dict) -> None:
|
|
if d.get("status") in ("started", "processing"):
|
|
_set_job(
|
|
job_id, phase=_pp_phase(d.get("postprocessor") or ""), speed_bps=None, eta_s=None
|
|
)
|
|
|
|
return hook
|
|
|
|
|
|
def _find_thumbnail(staging: Path, media: Path) -> Path | None:
|
|
for p in staging.iterdir():
|
|
if p != media and p.suffix.lower() in _THUMB_SUFFIXES:
|
|
return p
|
|
return None
|
|
|
|
|
|
def _produced_file(info: dict, staging: Path) -> Path:
|
|
"""The media file yt-dlp actually wrote.
|
|
|
|
`requested_downloads[0].filepath` is the authoritative answer on YouTube, but an arbitrary-URL
|
|
extractor may not fill it in at all — and `Path("")` is `Path(".")`, which *passes* `.exists()`,
|
|
so a missing value used to sail past the fallback and blow up two steps later (`with_name` on an
|
|
empty name: the BBC-news failure). Anything that isn't a real file counts as absent; then fall
|
|
back to the largest file left in staging that is neither a thumbnail nor one of yt-dlp's own
|
|
working files (see `_YTDLP_TEMP_RE`)."""
|
|
req = (info.get("requested_downloads") or [{}])[0]
|
|
fp = (
|
|
req.get("filepath") or req.get("_filename") or info.get("filepath") or info.get("_filename")
|
|
)
|
|
if fp:
|
|
produced = Path(str(fp))
|
|
if produced.is_file():
|
|
return produced
|
|
cands = [
|
|
p
|
|
for p in staging.iterdir()
|
|
if p.is_file()
|
|
and p.suffix.lower() not in _THUMB_SUFFIXES
|
|
and not _YTDLP_TEMP_RE.search(p.name)
|
|
]
|
|
if not cands:
|
|
raise StatedFailure("yt-dlp produced no output file")
|
|
return max(cands, key=lambda p: p.stat().st_size)
|
|
|
|
|
|
def _media_info(info: dict) -> dict:
|
|
"""Unwrap a playlist/multi_video extraction to the entry that carries the download.
|
|
|
|
A single non-YouTube page (a news article's video, a multi-rendition embed) can extract as a
|
|
wrapper whose own dict has no `requested_downloads` — the media lives on an entry. Pick the
|
|
FIRST entry that actually downloaded something: on a page with a skipped teaser in front of the
|
|
real video, entries[0] carries no file, and taking it blindly would label the produced file
|
|
with the wrong id/title/duration."""
|
|
entries = [e for e in (info.get("entries") or []) if e]
|
|
if not entries or info.get("requested_downloads"):
|
|
return info
|
|
chosen = next((e for e in entries if e.get("requested_downloads")), entries[0])
|
|
return {**info, **chosen}
|
|
|
|
|
|
def _extract_with_fallback(
|
|
job_id, asset_id, url, spec, staging, adapter, cookiefile=None, max_bytes=None
|
|
):
|
|
"""Download via yt-dlp using the source `adapter`'s client sets + options. For YouTube the adapter
|
|
supplies the primary player-client set then a POT-backed fallback (retried on a bot/DRM/format
|
|
error); a Facebook download WITH a cookie is retried as Chrome behind "Cannot parse data"; other
|
|
sources make a single attempt. Returns the info dict of the successful attempt.
|
|
|
|
Before the first attempt the adapter may resolve `url` to the link that really holds the video (a
|
|
Facebook group post → its video's watch link); when the page shows no single video there, or is not
|
|
the post at all (a refused cookie's login page), the adapter raises its own verdict
|
|
(adapters.POST_*_MESSAGE) instead of letting yt-dlp guess, and a failed page read fails the
|
|
download as itself.
|
|
|
|
`cookiefile` (S6) is the acting user's per-host cookies.txt, already gated by
|
|
`credentials.resolve_cookiefile`; passed to every attempt so the fallback stays authenticated.
|
|
|
|
`max_bytes` (SEC-13) is the owner's remaining storage budget, applied to every attempt."""
|
|
client_sets = adapter.client_sets(authed=cookiefile is not None)
|
|
outtmpl = str(staging / "%(id)s.%(ext)s")
|
|
# Resolving may read a page first — network time the stall watchdog must not see as a stall.
|
|
_touch_job(job_id)
|
|
target = adapter.resolve_url(
|
|
url,
|
|
formats.build_ydl_opts(
|
|
spec,
|
|
outtmpl,
|
|
_make_progress_hook(job_id, asset_id),
|
|
cookiefile=cookiefile,
|
|
max_bytes=max_bytes,
|
|
),
|
|
)
|
|
last_exc: Exception | None = None
|
|
for i, clients in enumerate(client_sets):
|
|
opts = formats.build_ydl_opts(
|
|
spec,
|
|
outtmpl,
|
|
_make_progress_hook(job_id, asset_id),
|
|
extra=adapter.ydl_opts_extra(clients),
|
|
cookiefile=cookiefile,
|
|
max_bytes=max_bytes,
|
|
)
|
|
opts["postprocessor_hooks"] = [_make_pp_hook(job_id)]
|
|
_touch_job(job_id) # stall-watchdog heartbeat: extraction writes no progress, so a slow /
|
|
# multi-client-set fallback would otherwise cross the stall timeout with a stale updated_at.
|
|
# (Per-ATTEMPT only; a single extract_info that itself runs past the stall timeout is still
|
|
# killed — acceptable, as socket_timeout + extractor_retries bound one attempt well under it.)
|
|
try:
|
|
with yt_dlp.YoutubeDL(opts) as ydl:
|
|
return ydl.extract_info(target, download=True)
|
|
except yt_dlp.utils.DownloadError as exc:
|
|
last_exc = exc
|
|
if i < len(client_sets) - 1 and adapter.retriable(str(exc)):
|
|
log.info(
|
|
"job %s: clients %s failed (%s); retrying with fallback",
|
|
job_id,
|
|
clients,
|
|
formats.log_safe(exc)[:70],
|
|
)
|
|
for p in list(staging.iterdir()): # drop partials before the retry
|
|
try:
|
|
p.unlink()
|
|
except OSError as exc:
|
|
# A partial left behind is not inert: the next client set resumes into it,
|
|
# so the retry this loop exists to make clean starts dirty instead.
|
|
log.info(
|
|
"a partial file could not be removed before the retry",
|
|
extra=logs.event(
|
|
"download.retry.partial_unlink_failed",
|
|
job_id=job_id,
|
|
reason=type(exc).__name__,
|
|
),
|
|
)
|
|
_set_job(job_id, progress=0, phase="downloading")
|
|
continue
|
|
raise
|
|
raise last_exc # unreachable (loop either returns or raises)
|
|
|
|
|
|
def _download_settings() -> tuple[str, int]:
|
|
"""Layout + retention days from the DB config (admin-overridable on the Configuration page),
|
|
falling back to the env defaults. Read here, not from `settings`, so an admin edit takes effect."""
|
|
with SessionLocal() as db:
|
|
return (
|
|
sysconfig.get_str(db, "download_layout"),
|
|
sysconfig.get_int(db, "download_retention_days"),
|
|
)
|
|
|
|
|
|
def _ensure_browser_playable(job_id: int, produced: Path, spec: dict, info: dict) -> Path:
|
|
"""Make an mp4-compat-profile download play in every browser: H.264 + AAC/MP3 in mp4. If the
|
|
download (whatever the source shipped) isn't already that, re-encode the offending stream(s) —
|
|
or just remux when the codecs are fine but the container isn't. **Scope:** only the mp4 compat
|
|
profile is normalized; an explicit `vcodec` or a non-mp4 custom profile is the user opting out
|
|
of the guarantee and is left as-is (so a `webm` profile can still ship VP9/Opus). Returns the
|
|
file to store (converted, or the original) and corrects `info`'s codec fields to match disk.
|
|
|
|
An audio download (mode "a") is never touched: yt-dlp already extracted it to .mp3 / .m4a, which
|
|
every browser plays as audio/mpeg / audio/mp4. Remuxing it into .mp4 served an mp3 as video/mp4,
|
|
which iPad's WebKit refuses to play (prod job 268)."""
|
|
if spec.get("mode") == "a" or spec.get("container") != "mp4" or spec.get("vcodec"):
|
|
return produced
|
|
vcodec, acodec = editmod.probe_codecs(produced)
|
|
is_mp4 = produced.suffix.lower() == ".mp4"
|
|
needs, reenc_v, reenc_a = editmod.browser_transcode_flags(vcodec, acodec, is_mp4)
|
|
if not needs:
|
|
return produced
|
|
|
|
dest = produced.with_name(produced.stem + ".browser.mp4")
|
|
action = "re-encode" if (reenc_v or reenc_a) else "remux"
|
|
log.info(
|
|
"job %s: %s (v=%s a=%s mp4=%s) -> browser-safe mp4", job_id, action, vcodec, acodec, is_mp4
|
|
)
|
|
_set_job(job_id, phase="processing", progress=0, speed_bps=None, eta_s=None)
|
|
# Prefer the info duration; probe the file when the extractor didn't provide one, so a long
|
|
# re-encode still shows real progress instead of a frozen 0%.
|
|
total_s = int(info["duration"]) if info.get("duration") else editmod.probe_duration(produced)
|
|
editmod.run_ffmpeg(
|
|
editmod.build_browser_safe_cmd(
|
|
produced, dest, reenc_v, reenc_a, threads=editmod.encode_thread_cap()
|
|
),
|
|
total_s=total_s,
|
|
on_progress=_ffmpeg_progress_writer(job_id, "processing"),
|
|
should_cancel=_cancel_poll(job_id),
|
|
)
|
|
produced.unlink(missing_ok=True)
|
|
# Keep the stored asset metadata truthful about what's now on disk.
|
|
if reenc_v:
|
|
info["vcodec"] = "h264"
|
|
if reenc_a:
|
|
info["acodec"] = "aac"
|
|
return dest
|
|
|
|
|
|
def _download(
|
|
job_id: int, asset_id: int, source_kind: str, source_ref: str, spec: dict, user_id: int
|
|
) -> None:
|
|
staging = _staging_dir(asset_id)
|
|
cookiefile: Path | None = None
|
|
url: str | None = None # read by the failure branch, which can run before it is resolved
|
|
try:
|
|
if staging.exists():
|
|
shutil.rmtree(staging, ignore_errors=True)
|
|
staging.mkdir(parents=True, exist_ok=True)
|
|
|
|
_set_job(job_id, phase="downloading", progress=0)
|
|
adapter = adapters.resolve_adapter(source_kind, source_ref)
|
|
url = service.source_url(source_kind, source_ref)
|
|
# SEC-12, second half: the enqueue guard ran when the job was created, which may have been
|
|
# hours ago and against a different DNS answer. This is where the socket actually gets
|
|
# opened, so the check is repeated here and the job fails rather than fetching.
|
|
if source_kind == "url":
|
|
try:
|
|
urlguard.check_url(url)
|
|
except urlguard.BlockedUrl as e:
|
|
raise StatedFailure(f"Refusing to download from this address ({e.reason}).") from e
|
|
# S6: attach the acting user's per-host cookie ONLY when it's host-scoped to this URL AND
|
|
# this asset is that user's private, salted identity (the double gate lives in credentials).
|
|
remaining_bytes: int | None = None
|
|
with SessionLocal() as cdb:
|
|
target = cdb.get(MediaAsset, asset_id)
|
|
if target is not None:
|
|
cookiefile = credentials.resolve_cookiefile(
|
|
cdb, user_id, url, target, spec, staging
|
|
)
|
|
# SEC-13: what this user has left. `check_enqueue` only compared the footprint as it
|
|
# stood when the job was QUEUED — a download's size is unknown until it finishes, so a
|
|
# single job could previously blow through the cap by any amount. Handing the remainder
|
|
# to yt-dlp as `max_filesize` makes the format selector refuse an oversized file up
|
|
# front instead.
|
|
remaining_bytes = quota.remaining_bytes(cdb, user_id)
|
|
# …and zero is a REFUSAL, not "no limit". Earlier jobs of this user may have filled the cap
|
|
# while this one sat in the queue, and 0 is not an expressible yt-dlp `max_filesize`, so
|
|
# passing it down would leave the ceiling unset — the one user already at their cap would be
|
|
# the only one downloading unbounded. Fail the job here instead. Not retryable until they
|
|
# free space, which is what the `quota_exceeded` category tells them.
|
|
if remaining_bytes == 0:
|
|
raise _OwnerRefused(dlerrors.QUOTA_FULL_MESSAGE)
|
|
info = _media_info(
|
|
_extract_with_fallback(
|
|
job_id,
|
|
asset_id,
|
|
url,
|
|
spec,
|
|
staging,
|
|
adapter,
|
|
cookiefile=str(cookiefile) if cookiefile else None,
|
|
max_bytes=remaining_bytes,
|
|
)
|
|
)
|
|
# Re-resolve now that yt-dlp's info reveals the true source (a bare/redirected url only
|
|
# identifies post-extraction) so the .nfo uniqueid type + uploader_url agree on one source.
|
|
adapter = adapters.resolve_adapter(source_kind, source_ref, info)
|
|
|
|
produced = _produced_file(info, staging)
|
|
produced = _ensure_browser_playable(job_id, produced, spec, info)
|
|
ext = produced.suffix.lstrip(".").lower()
|
|
# Codecs/dims/duration come from a PROBE of the produced file — the ground truth — NOT yt-dlp's
|
|
# `info` dict, whose fields are absent for some sources (Facebook among them): deriving
|
|
# media_kind from those nulls misclassified a real H.264 video as a download-only 'file', and a
|
|
# missing duration hid the trim action. Probed before the sidecars so the .nfo runtime has it
|
|
# too. Same probe the upload ingest uses.
|
|
vcodec, acodec, width, height, duration_s = probe.resolve_media_facts(produced, info)
|
|
meta = storage.MediaMeta(
|
|
video_id=info.get("id") or source_ref,
|
|
title=normalize_title(adapter.title(info)) or source_ref,
|
|
uploader=info.get("uploader") or info.get("channel") or "Unknown Channel",
|
|
upload_date=_parse_upload_date(info.get("upload_date")),
|
|
uniqueid_type=adapter.nfo_uniqueid_type(),
|
|
duration_s=duration_s,
|
|
description=info.get("description"),
|
|
thumbnail_url=info.get("thumbnail"),
|
|
)
|
|
|
|
layout, retention_days = _download_settings()
|
|
rel = storage.rel_path(meta, ext, layout, asset_id=asset_id)
|
|
thumb = _find_thumbnail(staging, produced)
|
|
# Convert the raw thumbnail (webp/png/jpeg) to a real JPEG ourselves — robustly — for the
|
|
# poster + og:image sidecar, now that yt-dlp's fatal thumbnail convertor is gone. On failure
|
|
# this is None and write_sidecars writes no poster, so ensure_poster cuts a frame instead.
|
|
poster_jpg = editmod.thumbnail_to_jpg(thumb, staging / "poster.jpg")
|
|
storage.place_file(produced, settings.download_root, rel)
|
|
nfo_ok = storage.write_sidecars(settings.download_root, rel, meta, poster_jpg)
|
|
|
|
final = storage.abs_path(settings.download_root, rel)
|
|
size = final.stat().st_size if final.exists() else None
|
|
expires = datetime.now(timezone.utc) + timedelta(days=retention_days)
|
|
|
|
with SessionLocal() as db:
|
|
asset = db.get(MediaAsset, asset_id)
|
|
if asset:
|
|
asset.status = "ready"
|
|
asset.rel_path = rel
|
|
asset.size_bytes = size
|
|
asset.duration_s = meta.duration_s
|
|
asset.vcodec, asset.acodec, asset.width, asset.height = (
|
|
vcodec,
|
|
acodec,
|
|
width,
|
|
height,
|
|
)
|
|
asset.container = ext
|
|
asset.media_kind = mediakind.derive(asset.vcodec, asset.acodec)
|
|
asset.title = meta.title
|
|
asset.uploader = meta.uploader
|
|
asset.uploader_url = adapter.uploader_url(info)
|
|
asset.upload_date = meta.upload_date
|
|
asset.thumbnail_url = meta.thumbnail_url
|
|
# Record a self-hosted poster for every download: it's the reliable link-preview
|
|
# (og:image) source — a remote thumbnail, esp. Facebook's signed CDN URL, expires
|
|
# and can't be relied on cross-origin. write_sidecars already wrote <base>.jpg from
|
|
# the thumbnail; ensure_poster returns that, or cuts a frame if there was none.
|
|
asset.poster_path = editmod.ensure_poster(asset, settings.download_root)
|
|
asset.source_webpage_url = info.get("webpage_url")
|
|
asset.nfo_written = nfo_ok
|
|
asset.error = None
|
|
asset.last_access_at = datetime.now(timezone.utc)
|
|
asset.expires_at = expires
|
|
job = db.get(DownloadJob, job_id)
|
|
# Capture the owner before commit expires the row (avoids a refresh SELECT just to notify).
|
|
owner_id = job.user_id if job else None
|
|
fields = {
|
|
"status": JobStatus.DONE,
|
|
"progress": 100,
|
|
"phase": None,
|
|
"speed_bps": None,
|
|
"eta_s": None,
|
|
}
|
|
if job and not job.display_name and meta.title:
|
|
# display_name is VARCHAR(255) — clip a long title (see _fill_asset_meta_early).
|
|
fields["display_name"] = meta.title[:DISPLAY_TEXT_MAX]
|
|
# Atomic terminal write: finalizes ONLY while the row is still running, so a child that
|
|
# finishes just after a route cancel/pause can't resurrect it (see finalize_if_running).
|
|
finalized = service.finalize_if_running(db, job_id, **fields)
|
|
db.commit()
|
|
if finalized:
|
|
_notify_download_done(db, owner_id, meta.title, job_id, kind="download")
|
|
# The asset id, not `rel`: the path is built from the channel and the title.
|
|
log.info(
|
|
"job %s done: asset %s",
|
|
job_id,
|
|
asset_id,
|
|
extra=logs.event("download.job.done", job_id=job_id, asset_id=asset_id),
|
|
)
|
|
|
|
except (_Aborted, editmod.EditAborted):
|
|
# Paused/canceled mid-download or mid-transcode: free the asset for a later retry; keep the
|
|
# job's status (the route set it to paused/canceled). ref_count is adjusted by the route.
|
|
_reset_asset_pending(asset_id)
|
|
log.info(
|
|
"job %s aborted (pause/cancel)",
|
|
job_id,
|
|
extra=logs.event("download.job.aborted", job_id=job_id),
|
|
)
|
|
except _OwnerRefused as exc:
|
|
# Fail this job only. The asset is never marked `error`; it goes back to `pending` (below),
|
|
# so another user's job for the same video is claimed and downloaded normally instead of
|
|
# inheriting a reason that is about someone else's account.
|
|
msg = str(exc)[:500]
|
|
with SessionLocal() as db:
|
|
service.finalize_if_running(
|
|
db,
|
|
job_id,
|
|
status=JobStatus.ERROR,
|
|
error=msg,
|
|
error_code=dlerrors.classify(msg),
|
|
error_detail=None,
|
|
phase=None,
|
|
)
|
|
db.commit()
|
|
# Released AFTER the job is terminal: the moment the asset is `pending` another job can claim
|
|
# it, and this one must already be out of the way. (The staging tree is still torn down in
|
|
# `finally` below — the same narrow window the pause/cancel path above has always had.)
|
|
_reset_asset_pending(asset_id)
|
|
log.info(
|
|
"job %s refused for its owner: %s",
|
|
job_id,
|
|
formats.log_safe(msg),
|
|
extra=logs.event("download.job.refused", job_id=job_id),
|
|
)
|
|
except Exception as exc: # noqa: BLE001 — surface any failure to the user, keep worker alive
|
|
msg = str(exc)[:500]
|
|
# The catalog needs the link and whether the user's cookie went with it: the same yt-dlp
|
|
# text is Facebook's login wall without one and a refused cookie with one.
|
|
diagnosis = dlerrors.diagnose(msg, url=url, cookie_attached=cookiefile is not None)
|
|
# SEC-12: redact a transport-level readout for user-supplied URLs (see user_error_text).
|
|
msg = dlerrors.user_error_text(source_kind, msg, diagnosis.code)
|
|
with SessionLocal() as db:
|
|
asset = db.get(MediaAsset, asset_id)
|
|
if asset:
|
|
asset.status = "error"
|
|
asset.error = msg
|
|
# Atomic fail: only while still running, so a route-set cancel/pause is never
|
|
# overwritten with `error` (error also holds the asset → would re-open the
|
|
# double-release class).
|
|
service.finalize_if_running(
|
|
db,
|
|
job_id,
|
|
status=JobStatus.ERROR,
|
|
error=msg,
|
|
error_code=diagnosis.code,
|
|
error_detail=diagnosis.detail,
|
|
phase=None,
|
|
)
|
|
db.commit()
|
|
# After the write: a job the write could not fail is not a failed job. When it raises,
|
|
# the runner reports this failure with it (`runner._run`) and the worker fails the job.
|
|
# silent-ok: not silent — `_log_download_failed` is the report.
|
|
_log_download_failed(job_id, msg, diagnosis.code, exc)
|
|
finally:
|
|
# Unlink the decrypted cookie file explicitly (the rmtree below removes the whole staging
|
|
# tree too, but never leave a plaintext session cookie on disk a moment longer than needed).
|
|
if cookiefile is not None:
|
|
cookiefile.unlink(missing_ok=True)
|
|
shutil.rmtree(staging, ignore_errors=True)
|
|
|
|
|
|
def _failure_stack(exc: BaseException) -> dict[str, str]:
|
|
"""The `extra=` fragment that puts a failure's traceback on its line — or nothing.
|
|
|
|
Nothing for yt-dlp's DownloadError and for our own `StatedFailure` (`ffmpeg failed (1): …`):
|
|
their message is the whole story, and their stack is only the `except` that caught them.
|
|
Everything else keeps it — a TypeError on an odd info dict, a plain RuntimeError, an OSError,
|
|
a database error: that is our code or our platform failing, and the stack is the lead.
|
|
|
|
Scrubbed, link by link (`logs.scrubbed_stack`): a stack's messages are data — an OSError names
|
|
its file, and a database error raised inside `except DownloadError` carries yt-dlp's
|
|
unscrubbed text as its context."""
|
|
if isinstance(exc, (yt_dlp.utils.DownloadError, StatedFailure)):
|
|
return {}
|
|
return logs.scrubbed_stack(exc, formats.log_safe)
|
|
|
|
|
|
def _log_download_failed(job_id: int, msg: str, code: str | None, exc: BaseException) -> None:
|
|
"""The child's `download.job.failed` line. `msg` is the user's copy and keeps their URL; the
|
|
log's copy goes through `log_safe`, or it would put back one line later the URL and id the
|
|
yt-dlp adapter took out of the line before it. The stack only when the classifier could not
|
|
place the failure: a recognised one is fully described by its code."""
|
|
unplaced = code in (None, "", "unknown")
|
|
log.warning(
|
|
"job %s failed: %s",
|
|
job_id,
|
|
formats.log_safe(msg),
|
|
extra={
|
|
**logs.event(
|
|
"download.job.failed",
|
|
job_id=job_id,
|
|
error_code=code,
|
|
error_type=type(exc).__name__,
|
|
),
|
|
**(_failure_stack(exc) if unplaced else {}),
|
|
},
|
|
)
|
|
|
|
|
|
def _fail_edit(job_id: int, asset_id: int, msg: str, exc: BaseException | None = None) -> None:
|
|
"""Fail an edit job, and say so ONCE — every way an edit fails comes through here, so this is
|
|
where its line and its event code live. `exc` is the exception when there was one.
|
|
|
|
Its traceback goes with the line unless the message already tells it all (`_failure_stack`).
|
|
Logged after the write, as on the download error path."""
|
|
with SessionLocal() as db:
|
|
asset = db.get(MediaAsset, asset_id)
|
|
if asset:
|
|
asset.status = "error"
|
|
asset.error = msg
|
|
# Atomic fail: only while still running (see the download error path).
|
|
service.finalize_if_running(
|
|
db,
|
|
job_id,
|
|
status=JobStatus.ERROR,
|
|
error=msg,
|
|
error_code=dlerrors.classify(msg),
|
|
phase=None,
|
|
)
|
|
db.commit()
|
|
# `msg` is the user's copy; ffmpeg's stderr tail in it can name the source file, whose path is
|
|
# built from the channel and the title.
|
|
log.warning(
|
|
"edit job %s failed: %s",
|
|
job_id,
|
|
formats.log_safe(msg),
|
|
extra={
|
|
**logs.event(
|
|
"edit.job.failed",
|
|
job_id=job_id,
|
|
asset_id=asset_id,
|
|
**({"error_type": type(exc).__name__} if exc is not None else {}),
|
|
),
|
|
**(_failure_stack(exc) if exc is not None else {}),
|
|
},
|
|
)
|
|
|
|
|
|
def _process_edit(job_id: int, asset_id: int) -> None:
|
|
"""Produce a per-user trim/crop clip from the source asset's downloaded file via ffmpeg."""
|
|
with SessionLocal() as db:
|
|
job = db.get(DownloadJob, job_id)
|
|
src = db.get(MediaAsset, job.source_asset_id) if job and job.source_asset_id else None
|
|
edit_spec = dict(job.edit_spec or {}) if job else {}
|
|
user_id = job.user_id if job else None
|
|
src_ready = src is not None and src.status == "ready" and bool(src.rel_path)
|
|
src_rel = src.rel_path if src else None
|
|
src_w, src_h, src_dur = (
|
|
(src.width, src.height, src.duration_s) if src else (None, None, None)
|
|
)
|
|
|
|
if not src_ready or not src_rel:
|
|
_fail_edit(job_id, asset_id, "The source download is no longer available.")
|
|
return
|
|
src_path = storage.abs_path(settings.download_root, src_rel)
|
|
if not src_path.exists():
|
|
_fail_edit(job_id, asset_id, "The source file is missing.")
|
|
return
|
|
|
|
src_ext = src_path.suffix.lstrip(".").lower() or "mp4"
|
|
crop = editmod.has_crop(edit_spec)
|
|
out_ext = "mp4" if editmod.needs_reencode(edit_spec) else src_ext
|
|
total_s = editmod.clip_duration(edit_spec, src_dur) or src_dur or 0
|
|
staging = _edit_staging_dir(asset_id)
|
|
|
|
try:
|
|
if staging.exists():
|
|
shutil.rmtree(staging, ignore_errors=True)
|
|
staging.mkdir(parents=True, exist_ok=True)
|
|
dest_staging = staging / f"out.{out_ext}"
|
|
|
|
_set_job(job_id, phase="editing", progress=0, speed_bps=None, eta_s=None)
|
|
if edit_spec.get("segments"):
|
|
cmd, prep = editmod.build_concat_plan(
|
|
src_path, dest_staging, edit_spec, out_ext, staging
|
|
)
|
|
for p, content in prep.items():
|
|
p.write_text(content, encoding="utf-8")
|
|
else:
|
|
cmd = editmod.build_edit_cmd(src_path, dest_staging, edit_spec, out_ext)
|
|
editmod.run_ffmpeg(
|
|
cmd,
|
|
total_s,
|
|
on_progress=_ffmpeg_progress_writer(job_id, "editing"),
|
|
should_cancel=_cancel_poll(job_id),
|
|
)
|
|
if not dest_staging.exists():
|
|
raise StatedFailure("ffmpeg produced no output file")
|
|
|
|
rel = storage.edit_rel_path(user_id, asset_id, out_ext)
|
|
storage.place_file(dest_staging, settings.download_root, rel)
|
|
final = storage.abs_path(settings.download_root, rel)
|
|
size = final.stat().st_size if final.exists() else None
|
|
_, retention_days = _download_settings()
|
|
expires = datetime.now(timezone.utc) + timedelta(days=retention_days)
|
|
|
|
with SessionLocal() as db:
|
|
asset = db.get(MediaAsset, asset_id)
|
|
if asset:
|
|
asset.status = "ready"
|
|
asset.rel_path = rel
|
|
asset.size_bytes = size
|
|
asset.container = out_ext
|
|
asset.duration_s = total_s or asset.duration_s
|
|
if crop:
|
|
asset.width = edit_spec["crop"]["w"]
|
|
asset.height = edit_spec["crop"]["h"]
|
|
else:
|
|
asset.width, asset.height = src_w, src_h
|
|
asset.error = None
|
|
asset.last_access_at = datetime.now(timezone.utc)
|
|
asset.expires_at = expires
|
|
job = db.get(DownloadJob, job_id)
|
|
# Capture owner + title before commit expires the rows (avoids a refresh SELECT).
|
|
owner_id = job.user_id if job else None
|
|
dl_title = asset.title if asset else None
|
|
# Atomic terminal write (see the download path): don't resurrect a route-stopped job.
|
|
finalized = service.finalize_if_running(
|
|
db, job_id, status=JobStatus.DONE, progress=100, phase=None
|
|
)
|
|
db.commit()
|
|
if finalized:
|
|
_notify_download_done(db, owner_id, dl_title, job_id, kind="edit")
|
|
# The asset id, not `rel`: the path is built from the channel and the title.
|
|
log.info(
|
|
"edit job %s done: asset %s",
|
|
job_id,
|
|
asset_id,
|
|
extra=logs.event("edit.job.done", job_id=job_id, asset_id=asset_id),
|
|
)
|
|
|
|
except editmod.EditAborted:
|
|
# Paused/canceled mid-encode: free the asset for a later retry (ref_count is the route's job).
|
|
_reset_asset_pending(asset_id)
|
|
log.info(
|
|
"edit job %s aborted (pause/cancel)",
|
|
job_id,
|
|
extra=logs.event("edit.job.aborted", job_id=job_id),
|
|
)
|
|
except Exception as exc: # noqa: BLE001 — surface the failure, keep the worker alive
|
|
# silent-ok: not silent — `_fail_edit` is the report, and it takes `exc` for its stack.
|
|
_fail_edit(job_id, asset_id, str(exc)[:500], exc)
|
|
finally:
|
|
shutil.rmtree(staging, ignore_errors=True)
|