103 lines
3.2 KiB
Python
103 lines
3.2 KiB
Python
"""Patreon vanity → campaign ID resolver.
|
|
|
|
When gallery-dl fails with "Failed to extract campaign ID" on a Patreon
|
|
source URL, call resolve_campaign_id(vanity, cookies_path) to look up
|
|
the campaign ID via Patreon's public campaigns API. Caller then retries
|
|
with `patreon.com/id:<campaign_id>` to bypass the broken vanity path.
|
|
|
|
Uses the stdlib `requests` (already a transitive dep via gallery-dl) so
|
|
we don't add `aiohttp`. The sync call is wrapped in run_in_executor so
|
|
the calling async code stays non-blocking.
|
|
|
|
Never raises. Returns None on any error (network, auth, parse, missing
|
|
match).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import http.cookiejar
|
|
import logging
|
|
import os
|
|
|
|
import requests
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
_CAMPAIGNS_URL = "https://www.patreon.com/api/campaigns"
|
|
_USER_AGENT = (
|
|
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
|
|
"(KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36"
|
|
)
|
|
_TIMEOUT_SECONDS = 10.0
|
|
|
|
|
|
def _load_cookie_jar(cookies_path: str | None) -> http.cookiejar.MozillaCookieJar | None:
|
|
if not cookies_path or not os.path.isfile(cookies_path):
|
|
return None
|
|
try:
|
|
jar = http.cookiejar.MozillaCookieJar(cookies_path)
|
|
jar.load(ignore_discard=True, ignore_expires=True)
|
|
return jar
|
|
except (OSError, http.cookiejar.LoadError) as exc:
|
|
log.debug("Could not load cookies from %s: %s", cookies_path, exc)
|
|
return None
|
|
|
|
|
|
def _sync_lookup(vanity: str, cookies_path: str | None) -> str | None:
|
|
jar = _load_cookie_jar(cookies_path)
|
|
headers = {
|
|
"User-Agent": _USER_AGENT,
|
|
"Accept": "application/vnd.api+json",
|
|
}
|
|
params = {
|
|
"filter[vanity]": vanity,
|
|
"fields[campaign]": "name",
|
|
}
|
|
try:
|
|
resp = requests.get(
|
|
_CAMPAIGNS_URL,
|
|
params=params,
|
|
headers=headers,
|
|
cookies=jar,
|
|
timeout=_TIMEOUT_SECONDS,
|
|
)
|
|
except requests.RequestException as exc:
|
|
log.warning("Patreon campaigns API request failed for vanity=%s: %s", vanity, exc)
|
|
return None
|
|
|
|
if resp.status_code != 200:
|
|
log.warning(
|
|
"Patreon campaigns API returned HTTP %d for vanity=%s",
|
|
resp.status_code, vanity,
|
|
)
|
|
return None
|
|
|
|
try:
|
|
payload = resp.json()
|
|
except ValueError as exc:
|
|
log.warning("Patreon campaigns API returned non-JSON for vanity=%s: %s", vanity, exc)
|
|
return None
|
|
|
|
if not isinstance(payload, dict):
|
|
return None
|
|
data = payload.get("data")
|
|
if not isinstance(data, list) or not data:
|
|
return None
|
|
first = data[0] if isinstance(data[0], dict) else None
|
|
campaign_id = first.get("id") if first else None
|
|
if not isinstance(campaign_id, str) or not campaign_id:
|
|
return None
|
|
log.info("Resolved Patreon vanity=%s → campaign_id=%s", vanity, campaign_id)
|
|
return campaign_id
|
|
|
|
|
|
async def resolve_campaign_id(
|
|
vanity: str,
|
|
cookies_path: str | None,
|
|
) -> str | None:
|
|
"""Async wrapper. Returns the campaign id string or None on any failure.
|
|
Never raises."""
|
|
loop = asyncio.get_running_loop()
|
|
return await loop.run_in_executor(None, _sync_lookup, vanity, cookies_path)
|