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 applied by `normalize`):
    mode           "av" (audio+video) | "v" (video-only) | "a" (audio-only)
    max_height     int | null (null = best)
    container      "mp4" | "mkv" | "webm" | null (merge/remux target for av/v)
    audio_format   "m4a" | "mp3" | "opus" | null (extract target for mode "a")
    vcodec         "h264" | "vp9" | "av1" | null (preference, best-effort)
    embed_subs     bool     embed_chapters bool     embed_thumbnail bool     sponsorblock bool
"""
import hashlib
import json

# 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.
_SIG_VERSION = 2

_SPONSORBLOCK_CATEGORIES = ["sponsor", "selfpromo", "interaction"]
_VCODEC_SORT = {"h264": "vcodec:h264", "vp9": "vcodec:vp9", "av1": "vcodec:av01"}


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):
            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
    return s


def format_sig(spec: dict) -> str:
    """Stable short hash of the output-affecting spec — the MediaAsset cache key component."""
    s = normalize(spec)
    data = {k: s[k] for k in _SIG_KEYS}
    data["_v"] = _SIG_VERSION
    payload = json.dumps(data, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(payload.encode()).hexdigest()[:24]


def build_ydl_opts(
    spec: dict,
    outtmpl: str,
    progress_hook,
    pot_base_url: str | None = None,
    player_clients: list[str] | None = None,
) -> dict:
    """Assemble yt-dlp options for a single download to `outtmpl` (a full path template).

    `pot_base_url` points the bgutil PO-token provider plugin at the sidecar server;
    `player_clients` picks which YouTube clients to use (POT-capable ones like web use the token)."""
    s = normalize(spec)
    h = s["max_height"]
    opts: dict = {
        "outtmpl": outtmpl,
        "noplaylist": True,
        "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,
    }

    # Extractor args: which YouTube clients to use + where the bgutil PO-token sidecar lives
    # (== --extractor-args "youtube:player_client=…;youtubepot-bgutilhttp:base_url=…"). Values
    # are lists in the Python API.
    ea: dict = {}
    if player_clients:
        ea["youtube"] = {"player_client": player_clients}
    if pot_base_url:
        ea["youtubepot-bgutilhttp"] = {"base_url": [pot_base_url]}
    if ea:
        opts["extractor_args"] = ea

    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 ""
        if s["mode"] == "v":
            opts["format"] = f"bestvideo{hcap}/best{hcap}"
        else:  # av
            opts["format"] = f"bestvideo{hcap}+bestaudio/best{hcap}"
        if s["container"]:
            opts["merge_output_format"] = s["container"]
        sort = []
        # Compatibility-first for mp4 (the universal container): prefer H.264 video + AAC
        # audio so shared/downloaded files play in EVERY browser, including iOS/Safari.
        # WebKit (which every iOS browser, Brave included, is forced to use) decodes only
        # H.264/H.265 + AAC inside mp4 — a VP9/Opus-in-mp4 file shows a broken player there,
        # even though Chromium plays it fine. These sort keys rank codec ABOVE resolution, so
        # a 720p H.264 stream is chosen over a 1080p-VP9-only one; VP9/AV1 remain a graceful
        # fallback only when no H.264 stream exists. A custom profile that explicitly sets
        # `vcodec` opts out (its codec choice wins over the compatibility default).
        if s["container"] == "mp4" and s["vcodec"] is None:
            sort += ["vcodec:h264", "acodec:aac"]
        if h:
            sort.append(f"res:{h}")
        if s["vcodec"] in _VCODEC_SORT:
            sort.append(_VCODEC_SORT[s["vcodec"]])
        if sort:
            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}
        )
    # Normalize the written thumbnail to jpg so we always have a `<name>.jpg` poster sidecar.
    # `already_have_thumbnail` keeps that file on disk after embedding (default deletes it).
    opts["postprocessors"].append({"key": "FFmpegThumbnailsConvertor", "format": "jpg"})
    if s["embed_thumbnail"]:
        opts["postprocessors"].append(
            {"key": "EmbedThumbnail", "already_have_thumbnail": True}
        )

    return opts