siftlode/backend/app/downloads/adapters.py

"""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 yt_dlp.networking.impersonate import ImpersonateTarget

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, authed: bool = False) -> list:
        """Attempts to make, in order — each entry is handed back to `ydl_opts_extra`, and the next one
        is tried when `retriable` says so. A generic source makes a single default attempt (`[None]`).
        `authed` is whether the acting user's cookie file goes with the download."""
        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 `<uniqueid type=...>` 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, authed: bool = False) -> 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"
    # Measured (E11.5, yt-dlp 2026.07.04): with the user's cookie, some reels answer a plain request
    # with "Cannot parse data" and parse only when the request looks like Chrome. Impersonation is a
    # SECOND attempt behind that one error, not the default: the links that already download with the
    # cookie keep their proven path, and only the ones stuck at the wall go through curl_cffi. Without
    # a cookie the same error is the login wall, which no disguise gets past.
    _IMPERSONATE = "impersonate"
    _WALL = "cannot parse data"

    def client_sets(self, authed: bool = False) -> list:
        return [None, self._IMPERSONATE] if authed else [None]

    def ydl_opts_extra(self, clients) -> dict:
        if clients == self._IMPERSONATE:
            return {"impersonate": ImpersonateTarget.from_str("chrome")}
        return {}

    def retriable(self, error_text: str) -> bool:
        return self._WALL in error_text.lower()

    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/<id>.
        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