"""Per-source adapters — a thin Strategy over yt-dlp, keyed by the media source (youtube / facebook / reddit / generic). yt-dlp stays the download engine; an adapter only contributes source-specific POLICY around it. The base class IS the generic behaviour (yt-dlp does everything, the ref is already a URL), and a subclass overrides ONLY what its source genuinely needs. This replaces knowledge that used to be scattered across worker.py / backfill.py / service.py with no shared source identity. Adapters are stateless singletons; get one with `resolve_adapter(source_kind, source_ref, info)`. Resolution works both before extraction (from source_kind / URL host) and after (from the yt-dlp `extractor_key` in the info dict), so a non-YouTube "url" job still lands on the right adapter. """ from urllib.parse import urlparse from app.config import settings class SourceAdapter: """Generic source: yt-dlp handles everything and the ref is already a fetchable URL.""" key = "generic" # --- extraction (worker._extract_with_fallback / backfill) --------------------------------- def client_sets(self) -> list: """Player-client sets to attempt, in order. A generic source needs no client override, so a single default attempt (`[None]`) — the per-client fallback is a YouTube-only concept.""" return [None] def ydl_opts_extra(self, clients) -> dict: """yt-dlp options this source contributes (merged into `build_ydl_opts`). `clients` is the current entry from `client_sets()`. Generic: nothing.""" return {} def retriable(self, error_text: str) -> bool: """Retry the extraction with the NEXT client set on this error? Generic: no (single attempt).""" return False # --- metadata ----------------------------------------------------------------------------- def uploader_url(self, info: dict) -> str | None: """The channel/uploader page URL. Generic: whatever yt-dlp exposes directly (YouTube gives `channel_url`), else None (reddit / a raw media URL have no derivable channel page).""" return info.get("channel_url") or info.get("uploader_url") def wants_metadata_backfill(self) -> bool: """Whether the one-off metadata backfill should re-fetch this source. Generic: no.""" return False # --- source ref ↔ URL --------------------------------------------------------------------- def source_url(self, source_ref: str) -> str: """A fetchable URL for `source_ref`. Generic: the ref is already a URL.""" return source_ref def nfo_uniqueid_type(self) -> str: """The `` written into the Kodi/Plex `.nfo` sidecar.""" return self.key class YouTubeAdapter(SourceAdapter): key = "youtube" # Errors worth retrying with a different player-client set (YouTube per-client flakiness). _RETRIABLE = ( "not a bot", "DRM protected", "Requested format is not available", "Only images are available", "Sign in to confirm", ) def client_sets(self) -> list: return [ c for c in (settings.player_client_list, settings.player_client_fallback_list) if c ] or [None] def ydl_opts_extra(self, clients) -> dict: # == --extractor-args "youtube:player_client=…;youtubepot-bgutilhttp:base_url=…" (list values). ea: dict = {} if clients: ea["youtube"] = {"player_client": clients} if settings.download_pot_base_url: ea["youtubepot-bgutilhttp"] = {"base_url": [settings.download_pot_base_url]} return {"extractor_args": ea} if ea else {} def retriable(self, error_text: str) -> bool: return any(m in error_text for m in self._RETRIABLE) def wants_metadata_backfill(self) -> bool: return True def source_url(self, source_ref: str) -> str: return f"https://www.youtube.com/watch?v={source_ref}" class FacebookAdapter(SourceAdapter): key = "facebook" def uploader_url(self, info: dict) -> str | None: # Facebook gives neither channel_url nor uploader_url, but exposes a numeric `uploader_id` # (the page id) that resolves at facebook.com/. base = super().uploader_url(info) if base: return base uid = info.get("uploader_id") return f"https://www.facebook.com/{uid}" if uid else None def wants_metadata_backfill(self) -> bool: return True class RedditAdapter(SourceAdapter): key = "reddit" _YOUTUBE = YouTubeAdapter() _FACEBOOK = FacebookAdapter() _REDDIT = RedditAdapter() _GENERIC = SourceAdapter() def _host_is(host: str, *domains: str) -> bool: """True if `host` is exactly one of `domains` or a subdomain of it — a netloc-scoped match, NOT a loose substring, so a look-alike host ('myyoutube.com', 'youtube.com.evil') never matches.""" return any(host == d or host.endswith("." + d) for d in domains) def resolve_adapter( source_kind: str | None, source_ref: str | None = None, info: dict | None = None ) -> SourceAdapter: """Pick the adapter for a job. Matches on `source_kind`, the yt-dlp `extractor_key` from `info` (post-extraction, authoritative), and the netloc of any URL we hold — the `source_ref` (pre-extraction) and yt-dlp's `webpage_url` (post-extraction). Host matching is netloc-scoped, and the webpage_url is used by HOST only (never a substring over its path, which could carry a sibling source's name). Falls back to the generic adapter — behaviour unchanged for any un-dedicated source.""" extractor = (info.get("extractor_key") or info.get("extractor") or "").lower() if info else "" hosts = [] if source_ref: hosts.append(urlparse(source_ref).netloc.lower()) if info and info.get("webpage_url"): hosts.append(urlparse(info["webpage_url"]).netloc.lower()) def host_is(*domains: str) -> bool: return any(_host_is(h, *domains) for h in hosts) if ( source_kind == "youtube" or "youtube" in extractor or host_is("youtube.com", "youtu.be", "youtube-nocookie.com") ): return _YOUTUBE if "facebook" in extractor or host_is("facebook.com", "fb.watch"): return _FACEBOOK if "reddit" in extractor or host_is("reddit.com"): return _REDDIT return _GENERIC