"""Native Patreon JSON:API client (build step 1 of the native ingester).
Clean-room reimplementation of the Patreon `/api/posts` read path. This is a
plain, synchronous client over `requests` — the orchestrator already wraps
sync importer calls, so nothing here needs to be async (mirrors
patreon_resolver.py, which wraps its sync lookup in run_in_executor at the
call site).
Scope (build step 1): fetch + page + parse only. This module is NOT wired into
download_service yet — that is a later step. The public surface here exists so
the later step can drive it:
- PatreonClient(cookies_path).iter_posts(campaign_id)
→ (post, included_index, page_cursor)
- extract_media(post, included_index) → list[MediaItem]
- parse_cursor_from_url(url) → cursor
Drift detection is loud on purpose: Patreon ships JSON:API and the shapes we
depend on (top-level `data`, media resources carrying `file_name`/`url`) are
the contract. If a response comes back as an HTML login page or a media
resource is missing the fields we resolve against, we raise PatreonDriftError
rather than silently yielding empty media — so the later import step surfaces
"Patreon changed something" instead of "creator has no posts".
FC runs on a plain-HTTP homelab; nothing here uses a secure-context Web API.
"""
from __future__ import annotations
import logging
import re
import time
from collections.abc import Iterator
from dataclasses import dataclass
from html import unescape
from pathlib import Path
from urllib.parse import parse_qs, urlsplit
import requests
from ..utils.paths import filehash_from_url
from ..utils.prosemirror import post_body_html
from .native_ingest_common import (
_MAX_429_RETRIES,
NativeAuthError,
NativeDriftError,
NativeIngestError,
basename_from_url,
make_session,
retry_after_seconds,
)
log = logging.getLogger(__name__)
_POSTS_URL = "https://www.patreon.com/api/posts"
_TIMEOUT_SECONDS = 30.0
# JSON:API request contract (observed from real traffic — see module plan).
_INCLUDE = (
"campaign,access_rules,attachments,attachments_media,audio,images,media,"
"native_video_insights,user,user_defined_tags,ti_checks"
)
_FIELDS_POST = (
# `content` is the legacy flat-HTML body (Patreon now returns it null);
# `content_json_string` is the current ProseMirror-doc body. Request BOTH —
# post_body_html() prefers content (old posts) and falls back to converting
# content_json_string (current posts). #842.
"content,content_json_string,post_file,image,post_type,published_at,title,"
"url,patreon_url,current_user_can_view"
)
_FIELDS_MEDIA = "id,image_urls,download_url,metadata,file_name"
_FIELDS_CAMPAIGN = "name,url"
# Inline post `content` is HTML; images are emitted as 
.
# Pull every src; downstream dedup collapses any that duplicate a gallery item
# by filehash. Tolerant of attribute ordering and single/double quotes.
_CONTENT_IMG_RE = re.compile(r"]*?\bsrc=[\"']([^\"']+)[\"']", re.IGNORECASE)
class PatreonAPIError(NativeIngestError):
"""Base for native Patreon client failures. status_code / retry_after are
inherited from NativeIngestError (HTTP status; 429 Retry-After hint)."""
class PatreonAuthError(PatreonAPIError, NativeAuthError):
"""Authentication / authorization failure — missing or expired session
cookies, an insufficient pledge tier, or an HTML login/challenge page served
where JSON was expected. DISTINCT from drift: the fix is rotating the
credential, not updating the ingester. Maps to error_type 'auth_error'.
"""
class PatreonDriftError(PatreonAPIError, NativeDriftError):
"""A JSON response did not match the JSON:API shape we depend on.
Raised for: a missing top-level `data` list, `data` not a list, or a media
resource lacking the `file_name`/`url` fields we resolve against. Fail loud
so the import step flags API drift (ingester needs update) instead of
silently importing nothing. An HTML-login / non-JSON body is auth, not
drift — that raises PatreonAuthError.
"""
@dataclass
class MediaItem:
"""One resolved downloadable item belonging to a post.
Fields:
url — the CDN/download URL to fetch.
filename — media `file_name` when present; otherwise the URL basename
(NEVER a network call). Bounded/sane extension.
kind — one of: "images", "image_large", "attachments", "postfile",
"content". Mirrors gallery-dl's `files` content-type names so
the later step can honor the same per-source content_types.
filehash — the 32-char hex (MD5) segment from the CDN URL, or None if
the URL carries no such segment. Used for in-post dedup and
the cross-run seen-ledger.
post_id — the owning post's id (so a flattened media list stays
traceable to its post).
"""
url: str
filename: str
kind: str
filehash: str | None
post_id: str
def _filehash(url: str) -> str | None:
# Delegate to the shared extractor (utils.paths) so capture-time persistence
# and render-time inline-image matching use the EXACT same identity.
return filehash_from_url(url)
def parse_cursor_from_url(url: str | None) -> str | None:
"""Extract the `page[cursor]` query param from a links.next URL."""
if not url:
return None
query = urlsplit(url).query
values = parse_qs(query).get("page[cursor]")
if values and values[0]:
return values[0]
return None
class PatreonClient:
"""Synchronous Patreon JSON:API read client.
Construct with a path to a Netscape cookies.txt (the same file
CredentialService.get_cookies_path materializes). Cookies are loaded into a
requests.Session; no secure-context APIs are used.
"""
def __init__(
self,
cookies_path: str | Path | None,
*,
request_sleep: float = 0.0,
max_retries: int = _MAX_429_RETRIES,
):
self.cookies_path = str(cookies_path) if cookies_path else None
self._session = make_session(cookies_path, accept="application/vnd.api+json")
# Politeness: seconds to sleep before each /api/posts page fetch (paces
# the rate-limited API endpoint). 0 = no pacing. plan #703.
self._request_sleep = request_sleep or 0.0
self._max_retries = max_retries
# -- request -----------------------------------------------------------
def _params(self, campaign_id: str, cursor: str | None) -> dict[str, str]:
params = {
"include": _INCLUDE,
"fields[post]": _FIELDS_POST,
"fields[media]": _FIELDS_MEDIA,
"fields[campaign]": _FIELDS_CAMPAIGN,
"filter[campaign_id]": campaign_id,
"filter[contains_exclusive_posts]": "true",
"filter[is_draft]": "false",
"sort": "-published_at",
"json-api-version": "1.0",
}
if cursor:
params["page[cursor]"] = cursor
return params
def _fetch(self, campaign_id: str, cursor: str | None) -> dict:
if self._request_sleep > 0:
time.sleep(self._request_sleep) # pace the API endpoint
attempt = 0
while True:
try:
resp = self._session.get(
_POSTS_URL,
params=self._params(campaign_id, cursor),
timeout=_TIMEOUT_SECONDS,
)
except requests.RequestException as exc:
raise PatreonAPIError(
f"Patreon posts request failed (campaign_id={campaign_id}): {exc}"
) from exc
# Transient rate-limit: back off and retry rather than failing the
# whole walk. Only a PERSISTENT 429 (retries exhausted) falls
# through to the terminal RATE_LIMITED raise below.
if resp.status_code == 429 and attempt < self._max_retries:
attempt += 1
delay = retry_after_seconds(resp, attempt)
log.warning(
"Patreon 429 (campaign_id=%s) — backing off %.1fs (retry %d/%d)",
campaign_id, delay, attempt, self._max_retries,
)
time.sleep(delay)
continue
break
if resp.status_code in (401, 403):
# Auth rejected — expired/missing cookies or an insufficient tier.
# Actionable as "rotate credentials", so it's auth, not drift/http.
raise PatreonAuthError(
f"Patreon posts API returned HTTP {resp.status_code} — auth "
f"rejected (cookies expired or tier insufficient; "
f"campaign_id={campaign_id})",
status_code=resp.status_code,
)
if resp.status_code != 200:
# A persistent 429 (retries exhausted) is terminal RATE_LIMITED — carry
# the server's raw Retry-After seconds so the cooldown matches its hint
# (plan #708 B1). Header is uncapped here; the cooldown clamps it.
retry_after = None
if resp.status_code == 429:
hdr = resp.headers.get("Retry-After")
if hdr:
try:
retry_after = float(hdr)
except (TypeError, ValueError):
retry_after = None
raise PatreonAPIError(
f"Patreon posts API returned HTTP {resp.status_code} "
f"(campaign_id={campaign_id})",
status_code=resp.status_code,
retry_after=retry_after,
)
try:
payload = resp.json()
except ValueError as exc:
# A non-JSON body here is almost always the HTML login/challenge
# page served when cookies are missing/expired — that is an AUTH
# failure (rotate cookies), not API drift (update the ingester) and
# not a transient network error.
raise PatreonAuthError(
"Patreon posts API returned a non-JSON response (likely an "
f"HTML login/challenge page — session expired; "
f"campaign_id={campaign_id}): {exc}"
) from exc
return payload
# -- parsing -----------------------------------------------------------
@staticmethod
def _transform(response: dict) -> dict:
"""Flatten the JSON:API `included` array for relationship resolution.
Returns a dict keyed by `(type, id)` → that resource's `attributes`
(so a post's relationships can be resolved in O(1)). Missing/oddly
shaped `included` entries are skipped rather than fatal — drift
detection for the top-level shape lives in _validate_response.
"""
index: dict[tuple[str, str], dict] = {}
for inc in response.get("included") or []:
if not isinstance(inc, dict):
continue
rtype = inc.get("type")
rid = inc.get("id")
if rtype is None or rid is None:
continue
index[(str(rtype), str(rid))] = inc.get("attributes") or {}
return index
@staticmethod
def _validate_response(response: dict) -> None:
if not isinstance(response, dict):
raise PatreonDriftError("Patreon response was not a JSON object")
if "data" not in response:
raise PatreonDriftError("Patreon response missing top-level 'data' key")
data = response.get("data")
if not isinstance(data, list):
raise PatreonDriftError("Patreon response 'data' was not a list")
def _related_ids(self, post: dict, rel_name: str) -> list[str]:
rels = post.get("relationships") or {}
rel = rels.get(rel_name) or {}
data = rel.get("data")
if isinstance(data, dict): # to-one relationship
data = [data]
if not isinstance(data, list):
return []
ids: list[str] = []
for ref in data:
if isinstance(ref, dict) and ref.get("id") is not None:
ids.append(str(ref["id"]))
return ids
@staticmethod
def _media_url(attrs: dict) -> str | None:
"""Pick the best fetchable URL for a media resource.
Prefer the full-size `download_url`; fall back to the largest
`image_urls` size. gallery-dl prefers download_url too, only dipping
into image_urls when a smaller configured size is requested — FC
always wants the original, so download_url first.
"""
download_url = attrs.get("download_url")
if isinstance(download_url, str) and download_url:
return download_url
image_urls = attrs.get("image_urls")
if isinstance(image_urls, dict):
for key in ("original", "full", "large", "default"):
candidate = image_urls.get(key)
if isinstance(candidate, str) and candidate:
return candidate
# Otherwise take any non-empty string value.
for candidate in image_urls.values():
if isinstance(candidate, str) and candidate:
return candidate
return None
def _media_item(self, attrs: dict, kind: str, post_id: str) -> MediaItem:
url = self._media_url(attrs)
if not url:
raise PatreonDriftError(
f"Patreon media (post {post_id}, kind={kind}) had no resolvable URL "
f"(no download_url / image_urls)"
)
# file_name is OPTIONAL: Patreon legitimately serves some gallery images
# without it (operator-flagged 2026-06-07, BlenderKnight post 73665615),
# and the URL basename is a fine fallback — the same thing gallery-dl
# uses. A genuine schema change shows up as no URL (above) or a media id
# absent from `included` (caller), not a missing name.
file_name = attrs.get("file_name")
filename = file_name if isinstance(file_name, str) and file_name else basename_from_url(url)
return MediaItem(
url=url,
filename=filename,
kind=kind,
filehash=_filehash(url),
post_id=post_id,
)
def extract_media(self, post: dict, included_index: dict) -> list[MediaItem]:
"""Resolve all downloadable media for one post.
Walks the kinds in the same order gallery-dl does — images,
image_large (post cover), attachments, postfile, content (inline
![]()
) — and dedups within the post by filehash (first wins). The
image_large cover commonly duplicates a gallery image; deduping by
filehash collapses them to the gallery item (encountered first).
"""
post_id = str(post.get("id") or "")
attrs = post.get("attributes") or {}
items: list[MediaItem] = []
def _resolve_rel(rel_name: str, kind: str) -> None:
for mid in self._related_ids(post, rel_name):
media_attrs = included_index.get(("media", mid))
if media_attrs is None:
# Referenced but not in `included`: a media id with no
# resource is drift (we asked for include=media).
raise PatreonDriftError(
f"Patreon post {post_id} references media {mid} "
f"({rel_name}) not present in 'included'"
)
items.append(self._media_item(media_attrs, kind, post_id))
# 1. gallery images
_resolve_rel("images", "images")
# 2. image_large — the post-level cover (`image.large_url`). Not a
# media relationship; it lives on the post attributes.
image = attrs.get("image")
if isinstance(image, dict):
large_url = image.get("large_url") or image.get("url")
if isinstance(large_url, str) and large_url:
items.append(
MediaItem(
url=large_url,
filename=basename_from_url(large_url),
kind="image_large",
filehash=_filehash(large_url),
post_id=post_id,
)
)
# 3. attachments
_resolve_rel("attachments_media", "attachments")
# 4. postfile — the post's primary attached file (`post_file`).
post_file = attrs.get("post_file")
if isinstance(post_file, dict):
pf_url = post_file.get("url") or post_file.get("download_url")
if isinstance(pf_url, str) and pf_url:
pf_name = post_file.get("name")
filename = (
pf_name
if isinstance(pf_name, str) and pf_name
else basename_from_url(pf_url)
)
items.append(
MediaItem(
url=pf_url,
filename=filename,
kind="postfile",
filehash=_filehash(pf_url),
post_id=post_id,
)
)
# 5. content — inline ![]()
in the post HTML body.
content = attrs.get("content")
if isinstance(content, str) and content:
for raw_src in _CONTENT_IMG_RE.findall(content):
src = unescape(raw_src)
if not src:
continue
items.append(
MediaItem(
url=src,
filename=basename_from_url(src),
kind="content",
filehash=_filehash(src),
post_id=post_id,
)
)
return _dedup_by_filehash(items)
@staticmethod
def post_meta(post: dict) -> dict:
"""Title + published date for a post — for the preview sample (plan #708
B4). Part of the client contract `ingest_core.Ingester.preview` calls."""
attrs = post.get("attributes") or {}
title = attrs.get("title")
published = attrs.get("published_at")
return {
"title": title if isinstance(title, str) else None,
"date": published if isinstance(published, str) else None,
}
@staticmethod
def post_is_gated(post: dict) -> bool:
"""True when the authenticated account CANNOT view this post's content
(#874). Patreon serves only BLURRED locked-preview thumbnails for
paywalled / insufficient-tier posts, and `current_user_can_view` on the
post attributes is the access flag (it IS in `_FIELDS_POST`). The walk
skips a gated post ENTIRELY — no media, no post-record stub — so those
unusable previews never get downloaded.
Gate ONLY on an explicit `current_user_can_view == False`. A missing /
None flag (older posts, a sparse fieldset, or API drift) is treated as
viewable, so we never over-filter accessible posts on an absent field.
Part of the client contract the core consumes via getattr — an optional
seam, so stub clients / not-yet-migrated platforms simply never gate."""
attrs = post.get("attributes") or {}
return attrs.get("current_user_can_view") is False
@staticmethod
def post_record_key(post: dict) -> tuple[str, str] | None:
"""`(ledger_key, post_id)` for a media-less post's seen-ledger entry, or
None when the post has no id. The synthetic `post:` key lets the
generic core gate text-post capture through the SAME seen-ledger as media
— so a text post's body is detail-fetched + recorded ONCE, not re-fetched
every tick. Part of the client contract the core uses via getattr."""
pid = post.get("id")
pid = str(pid) if pid is not None else ""
if not pid:
return None
return (f"post:{pid}", pid)
# -- iteration ---------------------------------------------------------
def iter_posts(
self, campaign_id: str, cursor: str | None = None
) -> Iterator[tuple[dict, dict, str | None]]:
"""Yield (post, included_index, page_cursor) for every post in the feed.
Pages newest→oldest via `links.next`, validating each response for
drift before yielding. The triple gives a caller everything it needs to
resolve and checkpoint without re-fetching:
- post — the raw post resource.
- included_index — the page's flattened `included` (the same object
for every post on a page), to pass straight to
extract_media(post, included_index).
- page_cursor — the cursor that FETCHED this post's page (None for
the first page). The caller checkpoints THIS value,
matching the existing backfill cursor logic where
the saved cursor re-fetches the page being
processed (so a chunk cut mid-page resumes the page,
not the one after it).
"""
current_cursor = cursor
while True:
response = self._fetch(campaign_id, current_cursor)
self._validate_response(response)
page_cursor = current_cursor
included_index = self._transform(response)
for post in response.get("data") or []:
if isinstance(post, dict):
yield post, included_index, page_cursor
next_url = (response.get("links") or {}).get("next")
next_cursor = parse_cursor_from_url(next_url)
if not next_cursor:
return
current_cursor = next_cursor
# -- detail (full body enrichment) -------------------------------------
def fetch_post_detail_content(self, post_id: str) -> str | None:
"""Best-effort fetch of a post's body (as HTML) from the per-post DETAIL
endpoint (`/api/posts/{id}`).
Patreon deprecated the flat `content` HTML field — it returns null on the
feed AND the detail endpoint, for every post type. The real body now lives
in `content_json_string` (a ProseMirror doc), and is only returned under
the DEFAULT post fieldset: a sparse `fields[post]=content` request OMITS
it (confirmed against the live API 2026-06-15). So we request the default
fieldset (no `fields[post]`) and resolve the body via `post_body_html`
(content → else convert content_json_string). The downloader calls this to
enrich a post whose feed body was empty before writing the sidecar.
Best-effort BY DESIGN: a body we can't fetch must never fail the walk.
Every failure path returns None rather than raising — distinct from the
loud drift/auth raises on the feed path, which gate real downloads.
"""
if not post_id:
return None
if self._request_sleep > 0:
time.sleep(self._request_sleep) # pace the API endpoint (plan #703)
# No `fields[post]` — the default fieldset is the only shape that returns
# content_json_string (the body). A sparse fieldset nulls it out.
try:
resp = self._session.get(f"{_POSTS_URL}/{post_id}", timeout=_TIMEOUT_SECONDS)
except requests.RequestException as exc:
log.warning("Patreon post-detail fetch failed (post %s): %s", post_id, exc)
return None
if resp.status_code != 200:
log.warning(
"Patreon post-detail fetch HTTP %s (post %s)", resp.status_code, post_id
)
return None
try:
payload = resp.json()
except ValueError:
return None
data = payload.get("data") if isinstance(payload, dict) else None
attrs = data.get("attributes") if isinstance(data, dict) else None
body = post_body_html(attrs)
if body and body.strip():
log.info("post-detail: fetched %d chars (post %s)", len(body), post_id)
return body
ptype = attrs.get("post_type") if isinstance(attrs, dict) else None
log.info(
"post-detail: no body (post %s, post_type=%s)", post_id, ptype,
)
return None
# -- verify ------------------------------------------------------------
def verify_auth(self, campaign_id: str) -> tuple[bool | None, str]:
"""Cheap auth probe: fetch the first `/api/posts` page and report whether
the credential authenticated, WITHOUT downloading anything.
Returns `(ok, message)` matching the credential-verify contract:
- True — authenticated (the feed returned a valid JSON:API page).
- False — the credential was rejected (PatreonAuthError: 401/403, or an
HTML login page → cookies expired / tier insufficient).
- None — inconclusive: API drift (our parser is stale, not a cred
problem) or a transient network/HTTP error.
"""
try:
response = self._fetch(campaign_id, None)
self._validate_response(response)
except PatreonAuthError as exc:
return False, f"Patreon rejected the credential — {exc}"
except PatreonDriftError as exc:
return None, f"Couldn't verify — Patreon's API shape changed: {exc}"
except PatreonAPIError as exc:
return None, f"Couldn't verify (network/HTTP issue): {exc}"
return True, "Credentials valid — the Patreon feed authenticated."
def _dedup_by_filehash(items: list[MediaItem]) -> list[MediaItem]:
"""Drop later items sharing a filehash with an earlier one (first wins).
Items with no filehash (None) are never deduped against each other — we
can't prove they're the same file, so keep them all.
"""
seen: set[str] = set()
out: list[MediaItem] = []
for item in items:
if item.filehash is not None:
if item.filehash in seen:
continue
seen.add(item.filehash)
out.append(item)
return out