peter
90ee34a1b4
Given a group post's link, yt-dlp either found no formats or took another video from the page's feed data, so a "successful" retry of such a post was a wrong download. With the user's cookie the Facebook adapter now reads the post page, takes the single video node in the script blocks that carry the post id, and downloads that video's watch link (the Chrome retry does the rest). No single video in the post fails with its own message and the group-post copy, reworded to say the post has no video; the post link is never handed to yt-dlp as a fallback. Without a cookie nothing changes.
228 lines
10 KiB
Python
228 lines
10 KiB
Python
"""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.
|
||
"""
|
||
|
||
import re
|
||
from urllib.parse import urlparse
|
||
|
||
import yt_dlp
|
||
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 resolve_url(self, url: str, ydl_opts: dict) -> str | None:
|
||
"""The link to hand yt-dlp for `url`, resolved BEFORE the first attempt, or None when the source
|
||
itself shows there is no single video to download there. `ydl_opts` are the download's own
|
||
options (cookie, timeouts, address family) for any page this needs to read. Generic: `url`."""
|
||
return url
|
||
|
||
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"
|
||
# A post page is ~4–5 MB of inline JSON; anything far past that is not a post page.
|
||
_POST_PAGE_MAX_BYTES = 16 * 1024 * 1024
|
||
|
||
def resolve_url(self, url: str, ydl_opts: dict) -> str | None:
|
||
# Measured (E11.5 S4): given a group post's link, yt-dlp either finds no formats or — worse —
|
||
# downloads ANOTHER video from the page's feed data. The post's own video is the one Video node
|
||
# in the page scripts that carry the post id; its watch link downloads (plain first, then the
|
||
# Chrome retry above). Needs the cookie: signed out, the page is the login wall.
|
||
post_id = facebook_group_post_id(url)
|
||
if post_id is None or not ydl_opts.get("cookiefile"):
|
||
return url
|
||
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
||
raw = ydl.urlopen(url).read(self._POST_PAGE_MAX_BYTES)
|
||
video_id = facebook_post_video_id(raw.decode("utf-8", "replace"), post_id)
|
||
# Never fall back to the post link: that is exactly where the wrong video comes from.
|
||
return f"https://www.facebook.com/watch/?v={video_id}" if video_id else None
|
||
|
||
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
|
||
|
||
|
||
_FB_GROUP_POST_PATH = re.compile(r"^/groups/[^/]+/(?:permalink|posts)/(\d+)/?$")
|
||
_SCRIPT_BLOCK = re.compile(r"<script[^>]*>(.*?)</script>", re.S)
|
||
_VIDEO_NODE = re.compile(r'"__typename":"Video","id":"(\d+)"')
|
||
|
||
|
||
def facebook_group_post_id(url: str) -> str | None:
|
||
"""The post id of a Facebook group post link (`/groups/<group>/permalink|posts/<id>/`), else None."""
|
||
parsed = urlparse(url)
|
||
if not _host_is(parsed.netloc.lower(), "facebook.com"):
|
||
return None
|
||
match = _FB_GROUP_POST_PATH.match(parsed.path)
|
||
return match.group(1) if match else None
|
||
|
||
|
||
def facebook_post_video_id(page: str, post_id: str) -> str | None:
|
||
"""The id of the one video attached to `post_id`, read from its page: the Video nodes of the script
|
||
blocks that carry the post id (the page's other ~10 videos — feed, reels — sit in other blocks).
|
||
None for no video AND for more than one: guessing between two would be the wrong-video bug again."""
|
||
quoted = f'"{post_id}"'
|
||
ids = {
|
||
vid
|
||
for block in _SCRIPT_BLOCK.findall(page)
|
||
if quoted in block
|
||
for vid in _VIDEO_NODE.findall(block)
|
||
}
|
||
return ids.pop() if len(ids) == 1 else None
|
||
|
||
|
||
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
|