"""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 `.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