"""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

Milestone 387 added a SECOND read path on the same session: the membership
roster — what the ACCOUNT subscribes to, as opposed to what one creator has
posted.
  - iter_memberships(user_id) → Iterator[Membership]
  - current_user_id() → str

It is an OPTIONAL seam by construction, probed with
`getattr(client, "iter_memberships", None)` exactly as `post_is_gated` already
is. A client that does not implement it (Discord, HentaiFoundry) makes the
whole feature invisible for that platform — no flag, no config row, no
"unsupported" branch to keep alive.

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,
    Membership,
    NativeAuthError,
    NativeDriftError,
    NativeIngestError,
    basename_from_url,
    has_paid_access,
    make_session,
    retry_after_seconds,
)

log = logging.getLogger(__name__)

_POSTS_URL = "https://www.patreon.com/api/posts"
_MEMBERS_URL = "https://www.patreon.com/api/members"
_CURRENT_USER_URL = "https://www.patreon.com/api/current_user"
_TIMEOUT_SECONDS = 30.0

# --- membership roster contract (#387 C2) ---------------------------------
# Characterized from a real capture of the operator's own session — Scribe note
# #3886. NOT from Patreon's public v2 API, which is the CREATOR api behind
# OAuth scopes and a different surface entirely (project rule 130).
#
# DELIBERATELY MINIMAL, and that is a privacy decision rather than a
# performance one. The web app's own include set pulls `latest_pledge.card`
# and `address`; the card resources come back carrying the ACCOUNT HOLDER'S
# EMAIL in `merchant_name`. Copying the browser's query string wholesale — the
# obvious move — would have FC fetching payment PII it has no use for and can
# only mishandle. We ask for the creator and the tier, and nothing else.
_MEMBERS_INCLUDE = "campaign,reward"
_FIELDS_MEMBER = (
    "patron_status,is_free_member,is_gifted,pledge_amount_cents,currency,"
    "pledge_cadence,next_charge_date,access_expires_at"
)
_FIELDS_MEMBERS_CAMPAIGN = "name,url,vanity,is_active"
_FIELDS_REWARD = "title"
# The browser sends 1000. Whether a server-side ceiling applies below that is
# untested (note #3886, open question 4), so page conservatively: a wrong guess
# costs one extra request, and the paging loop is driven by meta.pagination
# rather than by this number.
_MEMBERS_PAGE_COUNT = 200

# 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 <img ... src="...">.
# 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"<img\b[^>]*?\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 _request(self, url: str, params: dict[str, str], *, what: str, scope: str) -> dict:
        """One paced, retried, error-classified GET returning parsed JSON.

        Extracted from `_fetch` so the membership endpoint (#387 C2) rides the
        SAME request path rather than growing a second copy of the 429 backoff,
        the auth-vs-drift classification and the Retry-After plumbing. Two
        copies of this would drift, and the half that drifted would be the one
        that only runs once a day.

        `what` / `scope` only shape the messages ("posts"/"campaign_id=123"),
        so a failure still says which call failed and against what.
        """
        if self._request_sleep > 0:
            time.sleep(self._request_sleep)  # pace the API endpoint
        attempt = 0
        while True:
            try:
                resp = self._session.get(url, params=params, timeout=_TIMEOUT_SECONDS)
            except requests.RequestException as exc:
                raise PatreonAPIError(
                    f"Patreon {what} request failed ({scope}): {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 (%s) — backing off %.1fs (retry %d/%d)",
                    scope, 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 {what} API returned HTTP {resp.status_code} — auth "
                f"rejected (cookies expired or tier insufficient; {scope})",
                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 {what} API returned HTTP {resp.status_code} ({scope})",
                status_code=resp.status_code,
                retry_after=retry_after,
            )
        try:
            return 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(
                f"Patreon {what} API returned a non-JSON response (likely an "
                f"HTML login/challenge page — session expired; {scope}): {exc}"
            ) from exc

    def _fetch(self, campaign_id: str, cursor: str | None) -> dict:
        return self._request(
            _POSTS_URL, self._params(campaign_id, cursor),
            what="posts", scope=f"campaign_id={campaign_id}",
        )

    # -- 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
        <img>) — 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 <img> 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:<id>` 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

    # -- membership roster (#387 C2) ---------------------------------------

    def current_user_id(self) -> str:
        """The signed-in account's own numeric user id.

        Needed because `/api/members` is filtered by `filter[user_id]` — the
        endpoint answers "who are the members of X", and the account asking
        about ITSELF still has to say so.

        INFERRED, NOT CHARACTERIZED. C0 captured `/api/members`, not this; what
        is relied on here is only the JSON:API envelope (`data.id`), which this
        same API demonstrably uses everywhere else. If that inference is wrong
        it raises drift rather than returning something plausible — which is
        the right failure, because the alternative is a confidently empty
        roster and an empty roster means "cancel everything" to C4.
        """
        payload = self._request(
            _CURRENT_USER_URL, {"json-api-version": "1.0"},
            what="current_user", scope="self",
        )
        data = (payload or {}).get("data")
        if not isinstance(data, dict) or not data.get("id"):
            raise PatreonDriftError(
                "Patreon current_user response had no data.id — cannot scope "
                "the membership roster to this account"
            )
        return str(data["id"])

    def _members_params(self, user_id: str | None, offset: int) -> dict[str, str]:
        params = {
            "include": _MEMBERS_INCLUDE,
            "fields[member]": _FIELDS_MEMBER,
            "fields[campaign]": _FIELDS_MEMBERS_CAMPAIGN,
            "fields[reward]": _FIELDS_REWARD,
            "page[offset]": str(offset),
            "page[count]": str(_MEMBERS_PAGE_COUNT),
            "json-api-version": "1.0",
            "json-api-use-default-includes": "false",
        }
        if user_id:
            params["filter[user_id]"] = user_id
        # NOTE: `filter[membership_type]` is deliberately NOT sent. The browser
        # sends the six values its settings page wants to show, and the capture
        # proves that list is NOT the same vocabulary as the `patron_status`
        # attribute — a row selected as `free_member` came back with
        # `patron_status: former_patron`, a word absent from the filter. Sending
        # no filter asks for everything the endpoint will give, which is what a
        # roster wants: a membership that DISAPPEARS is the signal C4 reads, and
        # a filter tuned for a UI that hides lapses would manufacture exactly
        # that disappearance. (Note #3886, open question 1.)
        return params

    @staticmethod
    def _validate_members_response(response: dict) -> None:
        """Drift checks specific to the roster.

        Stricter than the posts path about pagination on purpose: `iter_posts`
        can treat a missing `links.next` as "that was the last page", but here
        a missing total is indistinguishable from a truncated page — and a
        roster that silently stops half way reads downstream as "you cancelled
        those", which is the worst wrong answer this feature can give.
        """
        PatreonClient._validate_response(response)
        meta = response.get("meta")
        if not isinstance(meta, dict):
            raise PatreonDriftError("Patreon members response missing 'meta'")
        pagination = meta.get("pagination")
        if not isinstance(pagination, dict) or "total" not in pagination:
            raise PatreonDriftError(
                "Patreon members response missing meta.pagination.total — "
                "cannot tell a complete roster from a truncated one"
            )

    def _membership(self, member: dict, index: dict) -> Membership | None:
        """One member row as a Membership, or None for a row the roster can skip.

        The one skippable row is a LAPSED membership whose creator no longer
        exists. The live roster (note #3886, CORRECTION 3) returned 104 rows,
        because FC sends no membership-type filter and so gets lapses going back
        years. One of them, a membership that ended in 2017, carried no
        `campaign` relationship at all: the key is absent, not null, and its
        reward names no campaign either. The creator's page is gone.

        Raising on that row made the whole roster unusable over one membership
        nobody can act on. Skipping it changes no conclusion. A lapsed
        membership already means "not paying", absence means the same, and no
        Source can be matched to a campaign that no longer has an id.

        The refusal stays for every other row. An active or unrecognised
        membership without a creator is something FC cannot vouch for, and
        dropping it would read downstream as a cancellation.
        """
        attrs = member.get("attributes") or {}
        if "patron_status" not in attrs:
            raise PatreonDriftError(
                "Patreon member resource has no patron_status attribute"
            )

        campaign_ids = self._related_ids(member, "campaign")
        if not campaign_ids:
            paid = has_paid_access(
                "patreon", attrs.get("patron_status"),
                is_free_member=bool(attrs.get("is_free_member")),
            )
            if paid is False:
                log.info(
                    "Patreon roster: skipping a lapsed membership with no campaign "
                    "(creator deleted); status=%s access_expires_at=%s",
                    attrs.get("patron_status"), attrs.get("access_expires_at"),
                )
                return None
            raise PatreonDriftError(
                "Patreon member resource has no campaign relationship — a "
                "membership we cannot attribute to a creator is not usable"
            )
        campaign_id = campaign_ids[0]
        campaign = index.get(("campaign", campaign_id)) or {}

        # A member has at most one reward, and `reward.data` is legitimately
        # null — an active patron with no tier. Absence is a fact about the
        # membership, not a parse failure.
        tier_names: list[str] = []
        for reward_id in self._related_ids(member, "reward"):
            title = (index.get(("reward", reward_id)) or {}).get("title")
            if title:
                tier_names.append(str(title))

        return Membership(
            campaign_id=campaign_id,
            display_name=campaign.get("name"),
            url=campaign.get("url"),
            vanity=campaign.get("vanity"),
            status=attrs.get("patron_status"),
            # Default False, not None: the attribute is always present in the
            # capture, and treating a missing one as "free" would understate
            # access rather than overstate it.
            is_free_member=bool(attrs.get("is_free_member")),
            tier_names=tier_names,
            # The MEMBER's amount, never the reward's. `reward.amount_cents` is
            # the creator's list price in the CREATOR's currency (the capture
            # has CAD, DKK and EUR rewards sitting on USD pledges), so reading
            # it would report a number the operator has never been charged.
            amount_cents=attrs.get("pledge_amount_cents"),
            currency=attrs.get("currency"),
            details={"member": attrs, "campaign": campaign},
        )

    def iter_memberships(self, user_id: str | None = None) -> Iterator[Membership]:
        """Yield every membership the account holds.

        Pages on `page[offset]`/`page[count]` against `meta.pagination.total` —
        NOT on `links`. The response's own `links.first` is built without the
        `/api/` prefix the request uses, so following it verbatim would hit the
        web page instead of the API (note #3886).

        `user_id` omitted means the `filter[user_id]` parameter is omitted.
        Whether the endpoint then defaults to self is UNTESTED — pass
        `current_user_id()` unless you are deliberately probing that.
        """
        user_id = user_id or None
        offset = 0
        seen = 0
        while True:
            response = self._request(
                _MEMBERS_URL, self._members_params(user_id, offset),
                what="members", scope="membership roster",
            )
            self._validate_members_response(response)
            index = self._transform(response)
            rows = [m for m in (response.get("data") or []) if isinstance(m, dict)]
            for member in rows:
                membership = self._membership(member, index)
                if membership is not None:
                    yield membership

            seen += len(rows)
            total = int(response["meta"]["pagination"]["total"] or 0)
            # An empty page terminates regardless of what `total` claims. Trusting
            # the total alone would spin forever against a server that reports
            # more rows than it will hand over.
            if not rows or seen >= total:
                return
            offset += len(rows)

    # -- 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