"""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 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` first, then the yt-dlp `extractor_key` from `info` (post-extraction), then the URL host in `source_ref` (pre-extraction). Falls back to the generic adapter — so behaviour is unchanged for any source without a dedicated one.""" # Identify from the yt-dlp extractor key alone — NOT the webpage_url, whose path could contain a # sibling source's name (e.g. a reddit post slug '.../youtube_is_dead') and cause a misresolve. tag = (info.get("extractor_key") or info.get("extractor") or "").lower() if info else "" host = urlparse(source_ref).netloc.lower() if source_ref else "" if ( source_kind == "youtube" or "youtube" in tag or "youtu.be" in host or "youtube.com" in host or "youtube-nocookie.com" in host ): return _YOUTUBE if "facebook" in tag or "facebook.com" in host or "fb.watch" in host: return _FACEBOOK if "reddit" in tag or "reddit.com" in host: return _REDDIT return _GENERIC