120 lines
4.1 KiB
Python
120 lines
4.1 KiB
Python
"""Patreon campaign-ID resolver.
|
|
|
|
When gallery-dl fails with "Failed to extract campaign ID" on a Patreon
|
|
source, call resolve_campaign_id(vanity, cookies_path) to look up the
|
|
campaign ID via Patreon's campaigns API. The caller can then retry with
|
|
the id-URL form (patreon.com/id:<id>) to bypass the broken vanity lookup.
|
|
"""
|
|
import asyncio
|
|
import http.cookiejar
|
|
import logging
|
|
import os
|
|
from typing import Optional
|
|
|
|
import aiohttp
|
|
|
|
logger = 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"
|
|
)
|
|
_REQUEST_TIMEOUT_SECONDS = 10.0
|
|
|
|
|
|
def _load_cookie_jar(cookies_path: Optional[str]) -> Optional[aiohttp.CookieJar]:
|
|
"""Load a Netscape-format cookie file into an aiohttp CookieJar.
|
|
|
|
Returns None if the path is None, the file is missing, or parsing fails.
|
|
Never raises.
|
|
"""
|
|
if not cookies_path or not os.path.isfile(cookies_path):
|
|
return None
|
|
try:
|
|
moz = http.cookiejar.MozillaCookieJar(cookies_path)
|
|
moz.load(ignore_discard=True, ignore_expires=True)
|
|
except (OSError, http.cookiejar.LoadError) as e:
|
|
logger.debug(f"Could not load cookies from {cookies_path}: {e}")
|
|
return None
|
|
|
|
jar = aiohttp.CookieJar()
|
|
for cookie in moz:
|
|
jar.update_cookies(
|
|
{cookie.name: cookie.value},
|
|
response_url=aiohttp.helpers.URL(
|
|
f"https://{cookie.domain.lstrip('.')}/"
|
|
),
|
|
)
|
|
return jar
|
|
|
|
|
|
async def resolve_campaign_id(
|
|
vanity: str,
|
|
cookies_path: Optional[str],
|
|
) -> Optional[str]:
|
|
"""Resolve a Patreon vanity (creator slug) to a campaign ID.
|
|
|
|
Returns the campaign ID as a string, or None on any failure (network
|
|
error, auth error, empty response, unexpected shape). Never raises.
|
|
"""
|
|
jar = _load_cookie_jar(cookies_path)
|
|
headers = {
|
|
"User-Agent": _USER_AGENT,
|
|
"Accept": "application/vnd.api+json",
|
|
}
|
|
params = {
|
|
"filter[vanity]": vanity,
|
|
"fields[campaign]": "name",
|
|
}
|
|
timeout = aiohttp.ClientTimeout(total=_REQUEST_TIMEOUT_SECONDS)
|
|
|
|
try:
|
|
async with aiohttp.ClientSession(
|
|
cookie_jar=jar,
|
|
headers=headers,
|
|
timeout=timeout,
|
|
) as session:
|
|
async with session.get(_CAMPAIGNS_URL, params=params) as resp:
|
|
if resp.status != 200:
|
|
logger.warning(
|
|
f"Patreon campaigns API returned HTTP {resp.status} for vanity={vanity}"
|
|
)
|
|
return None
|
|
try:
|
|
payload = await resp.json(content_type=None)
|
|
except (aiohttp.ContentTypeError, ValueError) as e:
|
|
logger.warning(
|
|
f"Patreon campaigns API returned non-JSON for vanity={vanity}: {e}"
|
|
)
|
|
return None
|
|
except asyncio.TimeoutError:
|
|
logger.warning(f"Patreon campaigns API timed out for vanity={vanity}")
|
|
return None
|
|
except aiohttp.ClientError as e:
|
|
logger.warning(
|
|
f"Patreon campaigns API request failed for vanity={vanity}: "
|
|
f"{type(e).__name__}: {e}"
|
|
)
|
|
return None
|
|
|
|
data = payload.get("data") if isinstance(payload, dict) else None
|
|
if not isinstance(data, list) or not data:
|
|
logger.warning(
|
|
f"Patreon campaigns API returned empty/missing data for vanity={vanity}: "
|
|
f"payload keys={list(payload) if isinstance(payload, dict) else type(payload).__name__}"
|
|
)
|
|
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:
|
|
logger.warning(
|
|
f"Patreon campaigns API response missing 'id' for vanity={vanity}: "
|
|
f"first entry keys={list(first) if isinstance(first, dict) else type(first).__name__}"
|
|
)
|
|
return None
|
|
|
|
logger.info(f"Resolved Patreon vanity={vanity} → campaign_id={campaign_id}")
|
|
return campaign_id
|