siftlode/backend/app/downloads/formats.py

"""Format profiles → yt-dlp options, and the cache signature.

A profile `spec` (stored on DownloadProfile / snapshotted on DownloadJob) is the single source
of truth for what file gets produced. `format_sig` hashes exactly the fields that change the
OUTPUT BYTES, so two users picking equivalent settings share one cached MediaAsset.
`build_ydl_opts` turns a spec into a yt-dlp options dict (format selector + postprocessors).

spec shape (all keys optional; defaults + browser-playability clamp applied by `normalize`):
    mode           "av" (audio+video) | "v" (video-only) | "a" (audio-only)
    max_height     int | null (null = best)
    container      always coerced to "mp4" (the only universally browser-playable container)
    audio_format   "m4a" | "mp3" (mode "a" extract target; anything else — e.g. opus — → "m4a")
    vcodec         always coerced to null → H.264 selected (see `normalize`); no opt-out
    embed_subs     bool     embed_chapters bool     embed_thumbnail bool     sponsorblock bool

`normalize` clamps container/vcodec/audio_format so EVERY download plays in-browser; the
`container`/`vcodec` keys survive only so an old stored profile deserializes without error.
"""

import contextlib
import contextvars
import hashlib
import json
import logging
import os
import re
import urllib.parse
from collections.abc import Iterator

from app import logs

log = logs.get_logger(__name__)

# yt-dlp names what it is working on in these shapes, and each one is a row of someone's download
# history: the source URL, the `[extractor] <id>:` prefix on nearly every warning and error, the
# file path it writes to (`storage.rel_path` builds it from the channel and the title), a
# playlist's title, and an id quoted mid-sentence. The job id is already on every line
# (`logs.set_job` in the download child), so none of them adds anything an operator needs. What
# stays: the host (it says which extractor is failing) and the extractor tag. Every pattern below
# is a shape measured in yt-dlp's own source (2026.07.04), pinned in test_ytdlp_logging.py.
_URL_RE = re.compile(r"https?://([^/\s]+)\S*")
# `Destination:` is yt-dlp's own word in `[download] Destination: <path>`, not an id — the path
# after it is what `_PATH_RE` takes out.
_ID_PREFIX_RE = re.compile(r"(\[[\w:.-]+\] )(?!Destination:)[^\s:\[\]]+:")
# The generic extractor's "id" is the URL's basename, percent-DECODED: `My%20Holiday.mp4` is
# `My Holiday` — spaces, brackets, colons and all, which no pattern can bound (`Our Trip: Day 1`
# ends at which `: `?). The job knows its own URL, so the exact name is taken out as a literal
# (`hiding`). This pattern is the fallback for a line logged outside that scope: up to the first
# `: `. Only this extractor: every other one names a site id, and letting the prefix run to `: `
# everywhere would take the words out of lines like `[VideoConvertor] Converting …; Destination:`.
_GENERIC_ID_RE = re.compile(r"(\[generic\] )(?!Destination:)[^\[\]\n]+?:(?= |$)", re.M)
# `[youtube:tab] Playlist <title>: Downloading 5 items` — the title is not an id, and has spaces.
_PLAYLIST_TITLE_RE = re.compile(r"(\] Playlist ).+?(?=: Downloading )")
# `[download] Finished downloading playlist: <title>`, `[download] Downloading playlist: <title>`,
# `… Skipping already downloaded playlist: <title>` — the title runs to the end of the line.
_TITLE_TAIL_RE = re.compile(r"(playlist: |\[download\] Downloading \w+: ).*", re.I)
# An id quoted mid-sentence, not as a prefix: `… not available for <id>`, and `--no-playlist`'s
# `Downloading just the video <id>` (`_yes_playlist`) / `Downloading just video <id>` (multifeed).
_BARE_ID_RE = re.compile(
    r"((?:subtitles not available for|Downloading just (?:the )?\w+) )[^\s,;]+"
)
# Two segments or more, and not glued to a word: `www.youtube.com/…` is what the URL pattern
# leaves behind, and it must survive this one.
_PATH_RE = re.compile(r"(?<![\w.…])/(?:[^/\s\"']+/)+[^/\s\"']*")


_HIDDEN: contextvars.ContextVar[tuple[str, ...]] = contextvars.ContextVar(
    "formats_hidden", default=()
)


def generic_id(url: str) -> str:
    """The name yt-dlp's generic extractor gives a URL: its basename, extension off, percent-
    decoded — `InfoExtractor._generic_id` (2026.07.04), pinned against it in the tests."""
    return urllib.parse.unquote(os.path.splitext(url.rstrip("/").split("/")[-1])[0])


@contextlib.contextmanager
def hiding(url: str | None) -> Iterator[None]:
    """For the work on one URL: `log_safe` also takes out the name yt-dlp's generic extractor
    gives it, exactly, where yt-dlp puts it (`[generic] <name>:`). A context variable, so it
    holds for yt-dlp's logger calls too — they run on the same thread as the extraction."""
    name = generic_id(url) if url else ""
    token = _HIDDEN.set((*_HIDDEN.get(), name) if name else _HIDDEN.get())
    try:
        yield
    finally:
        _HIDDEN.reset(token)


def log_safe(text: object) -> str:
    """yt-dlp's text (its own lines, or a DownloadError's message) with the parts that name a
    person's download taken out. Every log line that carries yt-dlp's words goes through here,
    not just the adapter's: yt-dlp raises the same text it has just logged, so a line that skipped
    it would put back one line later exactly what the adapter took out."""
    text = str(text)
    for name in _HIDDEN.get():
        text = text.replace(f"[generic] {name}:", "[generic] …:")
    text = _URL_RE.sub(lambda m: f"{m.group(1)}/…", text)
    text = _ID_PREFIX_RE.sub(r"\1…:", text)
    text = _GENERIC_ID_RE.sub(r"\1…:", text)
    text = _PLAYLIST_TITLE_RE.sub(r"\1…", text)
    text = _TITLE_TAIL_RE.sub(r"\1…", text)
    text = _BARE_ID_RE.sub(r"\1…", text)
    return _PATH_RE.sub("…", text)


class _YdlLog:
    """yt-dlp's logger protocol, pointed at ours.

    Without it yt-dlp explains itself to nobody: `quiet` sends its output to a stdout we do not
    read, so "Requested format is not available", a signature-extraction failure or a POT refusal
    reached us only as whatever text ended up in the raised exception. Measured against yt-dlp
    2026.07.04: with `logger` set it calls `debug` for extraction and progress lines and
    `warning`/`error` for the rest, and `quiet`/`no_warnings` do NOT gate those calls.

    `debug` goes to TRACE: it is several lines per second during a download, so it shows only
    when the whole process runs at TRACE. (Not under a user's own Teljes consent — that lives in
    the API process's request scope, and the download child that runs yt-dlp has none.)

    `error` goes to WARNING, not ERROR. yt-dlp calls it from `trouble()` just before it raises,
    on EVERY failed attempt — including the ones the pipeline then recovers from with its fallback
    client set. Whether the job failed is the pipeline's verdict (`download.job.failed`), and that
    is the one line whose level should say so."""

    def _emit(self, level: int, msg: str) -> None:
        # The line goes as an ARGUMENT, not as the format string. yt-dlp writes progress as
        # `[download]  53.2% of 12.00MiB`, and while stdlib logging only applies `%` formatting to
        # a record that has args, this way the question never arises — and it is the shape ruff's
        # `G` rules ask for. (The sink's flood limiter groups a bare `%s` by the text itself, so
        # distinct yt-dlp warnings do not share one bucket — `logsink._template`.)
        log.log(level, "%s", log_safe(msg), extra=logs.event("download.ytdlp"))

    def debug(self, msg):
        self._emit(logs.TRACE, msg)

    def info(self, msg):
        self._emit(logging.DEBUG, msg)

    def warning(self, msg):
        self._emit(logging.WARNING, msg)

    def error(self, msg):
        self._emit(logging.WARNING, msg)


_YDL_LOG = _YdlLog()

# Fields that affect the produced file — and therefore the cache identity.
_SIG_KEYS = (
    "mode",
    "max_height",
    "container",
    "audio_format",
    "vcodec",
    "embed_subs",
    "embed_chapters",
    "embed_thumbnail",
    "sponsorblock",
)

_DEFAULTS = {
    "mode": "av",
    "max_height": None,
    "container": "mp4",
    "audio_format": "m4a",
    "vcodec": None,
    # Embedding is OFF by default: it rewrites the whole file (doubles I/O — a 10 GB video
    # gets a 10 GB temp copy), and the Plex `.nfo` + poster sidecars already carry the
    # metadata/thumbnail. Users who want self-contained files opt in via a custom profile.
    "embed_subs": False,
    "embed_chapters": False,
    "embed_thumbnail": False,
    "sponsorblock": False,
}

# Bump when the spec→output-bytes mapping changes so an identical spec gets a fresh cache
# identity instead of hitting a stale asset produced by the old rule. v2: mp4 downloads now
# prefer H.264+AAC (browser/iOS-compatible) rather than yt-dlp's default VP9/Opus best.
# v3: mp4 compat now *selects* H.264 (avc1) with a progressive fallback + re-encodes the rare
# AV1/VP9-only result to H.264, so cached AV1 assets must re-derive (fixes iOS broken-player).
# v4: EVERY download is clamped to the mp4/H.264/AAC compat profile (no more mkv/webm/VP9/AV1/Opus
# opt-out), so any asset cached under an old non-compat spec — or a stale non-H.264 mp4 that slipped
# the v3 net — must re-derive through the current browser-safe guarantee.
# v5: the phone-safe rule (`playable`) also judges the pixel format, so a cached 10-bit / 4:4:4 H.264
# asset (which v4 let through as "H.264") re-derives as 8-bit 4:2:0 instead of being reused.
_SIG_VERSION = 5
# Audio-only downloads version on their own, so a change to the audio bytes doesn't re-download every
# cached video. v5: an extracted .mp3/.m4a is stored as itself instead of remuxed into an .mp4 that
# iPad's WebKit refuses to play, so a cached remuxed audio asset must re-derive.
_AUDIO_SIG_VERSION = 5

_SPONSORBLOCK_CATEGORIES = ["sponsor", "selfpromo", "interaction"]


def normalize(spec: dict) -> dict:
    """Fill defaults + coerce types so signature and yt-dlp opts are deterministic."""
    s = dict(_DEFAULTS)
    for k in _DEFAULTS:
        if spec.get(k) is not None:
            s[k] = spec[k]
    if s["mode"] not in ("av", "v", "a"):
        s["mode"] = "av"
    if s["max_height"] is not None:
        # A malformed max_height (non-numeric inline spec / stored profile) falls back to "best"
        # rather than raising an unhandled 500 at enqueue.
        try:
            s["max_height"] = int(s["max_height"])
        except (TypeError, ValueError) as exc:
            # "best" is the opposite of what a height cap asks for: the user set a ceiling to save
            # bandwidth or disk, and the download quietly takes the largest rendition there is.
            log.warning(
                "the height cap on a download could not be read and was dropped",
                extra=logs.event("download.format.max_height_dropped", reason=type(exc).__name__),
            )
            s["max_height"] = None
    for b in ("embed_subs", "embed_chapters", "embed_thumbnail", "sponsorblock"):
        s[b] = bool(s[b])
    # Audio-only files can't embed subtitles/thumbnails as video; keep the flags meaningful.
    if s["mode"] == "a":
        s["embed_subs"] = False
    # Browser-playability guarantee: the Download module exists to download → share → PLAY
    # in-browser, so every output must play in EVERY browser, iOS/Safari/WebKit included. The only
    # universally-safe combo is mp4 + H.264 + AAC (an .mp3/.m4a for audio), so we CLAMP the spec to
    # it HERE — the single point every path (picker, stored profile, raw API) flows through — instead
    # of only hiding options in the UI. Non-mp4 containers (mkv plays nowhere on iOS; AV1 fails there,
    # and webm/VP9 plays only on recent devices) and any explicit vcodec are coerced to the compat
    # profile; Opus audio falls back to AAC/m4a. A source that exposes no H.264 (or a 10-bit one) is
    # still caught by the worker's phone-safe step (`downloads.playable`), so playback is guaranteed
    # end-to-end.
    s["container"] = "mp4"
    s["vcodec"] = None
    if s["audio_format"] not in ("m4a", "mp3"):
        s["audio_format"] = "m4a"
    return s


def _sig_version(s: dict) -> int:
    return _AUDIO_SIG_VERSION if s["mode"] == "a" else _SIG_VERSION


def format_sig(spec: dict, salt: str | None = None, *, version: int | None = None) -> str:
    """Stable short hash of the output-affecting spec — the MediaAsset cache key component.

    `salt` forks a distinct, non-dedup identity for the same spec (used by force-redownload so it
    always re-fetches into its own asset instead of sharing a cached file with other jobs).
    `version` hashes under a signature version other than the current one (see `format_sigs`)."""
    s = normalize(spec)
    data = {k: s[k] for k in _SIG_KEYS}
    data["_v"] = _sig_version(s) if version is None else version
    if salt is not None:
        data["_salt"] = salt
    payload = json.dumps(data, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(payload.encode()).hexdigest()[:24]


def format_sigs(spec: dict, salt: str | None = None) -> tuple[str, str]:
    """`format_sig` under the current signature version and under the one before it. An asset keeps
    the signature it was created with, so a job queued before a deploy that bumped the version and
    run after it downloads into an asset carrying the previous one — which must still be recognised
    as that user's private asset (`credentials.is_authed_asset`), or it downloads without its cookie."""
    previous = _sig_version(normalize(spec)) - 1
    return format_sig(spec, salt), format_sig(spec, salt, version=previous)


def build_ydl_opts(
    spec: dict,
    outtmpl: str,
    progress_hook,
    extra: dict | None = None,
    cookiefile: str | None = None,
    max_bytes: int | None = None,
) -> dict:
    """Assemble yt-dlp options for a single download to `outtmpl` (a full path template).

    `extra` is source-specific options contributed by the SourceAdapter (e.g. YouTube's
    `extractor_args` with the player-client set + bgutil PO-token base_url); it is merged onto the
    generic options below. Everything here is source-agnostic.

    `cookiefile` is the acting user's per-host cookies.txt path, attached ONLY by the worker
    after the host-scope + private-asset gate in `downloads.credentials`. It is set AFTER the `extra`
    merge so a source adapter can never clobber it.

    `max_bytes` (SEC-13) is the caller's remaining storage budget; yt-dlp refuses a format larger
    than it rather than filling the disk and only then being caught by the quota accounting."""
    s = normalize(spec)
    h = s["max_height"]
    opts: dict = {
        "outtmpl": outtmpl,
        "noplaylist": True,
        # SEC-13: `noplaylist` only demotes a video-inside-a-playlist URL to the single video. A
        # URL that is PURELY a playlist or channel has no such video to fall back to, so it still
        # expands into every entry — one enqueue, an unbounded number of downloads. One job means
        # one item; a user who wants the whole list enqueues it item by item, under their quota.
        "playlist_items": "1",
        "quiet": True,
        "no_warnings": True,
        "noprogress": True,
        "writethumbnail": True,  # kept for the poster.jpg sidecar (and embed, if requested)
        "progress_hooks": [progress_hook],
        "postprocessors": [],
        # Force IPv4: containers often have no working IPv6 route, but googlevideo CDN hosts
        # advertise AAAA records, so ffmpeg/the downloader would try IPv6 and fail with
        # "[Errno -5] No address associated with hostname". Binding to the IPv4 any-address
        # makes every connection IPv4-only. (yt-dlp's --force-ipv4 does the same thing.)
        "source_address": "0.0.0.0",
        # Self-heal transient network blips instead of failing the whole job.
        "retries": 10,
        "fragment_retries": 10,
        "extractor_retries": 3,
        "retry_sleep": {"http": 5, "fragment": 5, "extractor": 5},
        "socket_timeout": 30,
    }

    # Source-specific options from the adapter (e.g. YouTube's extractor_args: player_client + POT).
    if extra:
        opts.update(extra)

    # With a logger attached, `quiet` and `no_warnings` stop deciding whether yt-dlp explains itself
    # and only decide where — so both stay on above and the explanation comes here instead of to a
    # stdout nothing reads. AFTER the adapter merge, like `cookiefile` and the size cap: an adapter
    # must not be able to silence the one account of what went wrong that a failed job leaves.
    opts["logger"] = _YDL_LOG

    # SEC-13: hard ceiling on a single download. AFTER the adapter merge, for the same reason
    # `cookiefile` is — a source adapter must not be able to raise a limit the quota layer set.
    #
    # A non-positive budget is left unset because 0 is not an expressible yt-dlp filter, NOT because
    # nothing needs to happen: the worker refuses that job outright before extraction (`_OwnerRefused`
    # → `quota_exceeded`). Enqueue cannot make that call — it ran while the user was still under the
    # cap. Delete the worker's refusal and this branch silently uncaps the one user already over it.
    if max_bytes and max_bytes > 0:
        opts["max_filesize"] = max_bytes

    # The acting user's per-host cookie. AFTER the adapter merge on purpose: an adapter must not
    # be able to override the authenticated session the worker resolved for this specific user.
    if cookiefile:
        opts["cookiefile"] = cookiefile

    if s["mode"] == "a":
        opts["format"] = "bestaudio/best"
        opts["postprocessors"].append(
            {
                "key": "FFmpegExtractAudio",
                "preferredcodec": s["audio_format"] or "m4a",
                "preferredquality": "0",
            }
        )
    else:
        hcap = f"[height<=?{h}]" if h else ""
        # Every video download is the mp4/H.264 compat profile (normalize() clamps container→mp4,
        # vcodec→None), so shared/downloaded files play in EVERY browser, iOS/Safari included. WebKit
        # (which every iOS browser, Brave included, is forced to use) decodes only H.264/H.265 + AAC
        # inside mp4 — a VP9/AV1-in-mp4 file shows a broken player there — so H.264 is selected at the
        # SELECTION level, not merely preferred: a sort key can't conjure an H.264 stream a source
        # doesn't expose as video-only. A source that ships only non-H.264 is caught after download by
        # the worker's phone-safe step (`downloads.playable`).
        if s["mode"] == "v":
            # Video-only: keep it audio-free. Prefer an H.264 video-only stream, then any video-only,
            # and only fall to a progressive (audio-bearing) `best` as a last resort.
            opts["format"] = (
                f"bestvideo[vcodec^=avc1]{hcap}/bestvideo{hcap}/best[vcodec^=avc1]{hcap}/best{hcap}"
            )
        else:  # av
            # avc1 video-only + m4a audio, then the avc1 progressive, then anything (re-encoded).
            opts["format"] = (
                f"bestvideo[vcodec^=avc1]{hcap}+bestaudio[acodec^=mp4a]"
                f"/best[vcodec^=avc1]{hcap}"
                f"/bestvideo{hcap}+bestaudio/best{hcap}"
            )
        opts["merge_output_format"] = "mp4"
        # Codec preference tiebreaker (ranks H.264/AAC among the selected candidates); the selector
        # above is the hard guarantee, this is belt-and-braces.
        sort = ["vcodec:h264", "acodec:aac"]
        if h:
            sort.append(f"res:{h}")
        opts["format_sort"] = sort

    # SponsorBlock first (marks segments), then strip them from the media + chapters.
    if s["sponsorblock"]:
        opts["postprocessors"].append(
            {"key": "SponsorBlock", "categories": _SPONSORBLOCK_CATEGORIES, "when": "after_filter"}
        )
        opts["postprocessors"].append(
            {"key": "ModifyChapters", "remove_sponsor_segments": _SPONSORBLOCK_CATEGORIES}
        )
    if s["embed_subs"] and s["mode"] != "a":
        opts["writesubtitles"] = True
        opts["subtitleslangs"] = ["en.*", "hu.*", "de.*"]
        opts["postprocessors"].append({"key": "FFmpegEmbedSubtitle"})
    if s["embed_chapters"]:
        opts["postprocessors"].append(
            {"key": "FFmpegMetadata", "add_chapters": True, "add_metadata": True}
        )
    # NOTE: we deliberately do NOT add yt-dlp's `FFmpegThumbnailsConvertor` here. Its jpg conversion
    # runs `ffmpeg -bsf:v mjpeg2jpeg` when the written thumbnail is already JPEG bytes (Reddit serves
    # a JPEG under a `.png` name), and that bitstream filter *fails* on some such images — and since a
    # postprocessor error is fatal, it used to kill the WHOLE download even though the media was fine.
    # The `<name>.jpg` poster sidecar is instead produced robustly by our own pipeline (worker
    # `_download` → `editmod.thumbnail_to_jpg`, with `ensure_poster` cutting a video frame as a
    # fallback), so the og:image share preview (Facebook Messenger / MS Teams) keeps working.
    # The opt-in EmbedThumbnail below is unaffected: it does its OWN format handling — it embeds a
    # jpg/png thumbnail directly and converts anything else to PNG (not the crashing mjpeg2jpeg-to-jpg
    # path), so it never needed the convertor and does not reintroduce the crash.
    if s["embed_thumbnail"]:
        opts["postprocessors"].append({"key": "EmbedThumbnail", "already_have_thumbnail": True})

    return opts