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 (E7); the
`container`/`vcodec` keys survive only so an old stored profile deserializes without error.
"""

import hashlib
import json
import logging
import re

from app import logs

log = logs.get_logger(__name__)

# yt-dlp puts the full source URL in most of its lines, and a source URL is a row of someone's
# download history. The job id is already on every line (`logs.set_job` in the download child), so
# the URL adds nothing an operator needs and plenty they should not keep.
_URL_RE = re.compile(r"https?://([^/\s]+)\S*")


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, and TRACE is the
    level a user turns on for their own session. Everything else keeps its level."""

    @staticmethod
    def _clean(msg: str) -> str:
        return _URL_RE.sub(lambda m: f"{m.group(1)}/…", str(msg))

    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.
        log.log(level, "%s", self._clean(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.ERROR, 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.
_SIG_VERSION = 4
# 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 (E7): 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/MP3, 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 doesn't play at all; webm/VP9/AV1 fail iOS) and any
    # explicit vcodec are coerced to the compat profile; Opus audio (Safari-flaky as a bare file)
    # falls back to AAC/m4a. A source that exposes no H.264 is still caught by the worker's
    # browser-safe re-encode (`_ensure_browser_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 (S6), 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 (S6). 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 browser-safe re-encode (`_ensure_browser_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