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,
    "embed_subs": False,
    "embed_chapters": True,
    "embed_thumbnail": True,
    "sponsorblock": False,
}

_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:
        s["max_height"] = int(s["max_height"])
    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)
    payload = json.dumps({k: s[k] for k in _SIG_KEYS}, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(payload.encode()).hexdigest()[:24]


def target_ext(spec: dict) -> str:
    """Best-effort final extension for the produced file (also used to name the staging output)."""
    s = normalize(spec)
    if s["mode"] == "a":
        return s["audio_format"] or "m4a"
    return s["container"] or "mp4"


def build_ydl_opts(spec: dict, outtmpl: str, progress_hook) -> dict:
    """Assemble yt-dlp options for a single download to `outtmpl` (a full path template)."""
    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,
    }

    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 = []
        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