FabledCurator/backend/app/services/native_ingest_common.py

"""Shared primitives for the native-ingest platform adapters (Patreon,
SubscribeStar, …) — the single home for logic the per-platform client/downloader
modules would otherwise each copy.

DRY pass 2026-06-17 (#899): these used to live in `patreon_*` with the
SubscribeStar modules importing patreon privates (wrong owner + sibling-coupling).
They're platform-agnostic, so they live here and both adapters import them. The
per-platform modules keep only what genuinely differs (feed parsing, the media
shape, Patreon's Mux/yt-dlp video branch + detail-fetch enrichment).

FC runs on a plain-HTTP homelab; nothing here uses a secure-context Web API.
"""

from __future__ import annotations

import contextlib
import http.cookiejar
import json
import logging
import os
import time
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from urllib.parse import urlsplit

import requests

from ..utils.paths import filehash_from_url, safe_ext
from .file_validator import is_validatable, quarantine_file, validate_file

log = logging.getLogger(__name__)

_USER_AGENT = (
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36"
)

# 429 backoff (plan #703): ride out a transient API rate-limit instead of failing
# the whole walk. Honor the server's Retry-After; else exponential, capped.
_MAX_429_RETRIES = 3
_BACKOFF_BASE_SECONDS = 2.0
_BACKOFF_CAP_SECONDS = 30.0

# Media-download tuning (shared by every platform downloader).
_TIMEOUT_SECONDS = 120.0
_CHUNK = 1 << 16
_MAX_MEDIA_RETRIES = 3
_TRANSIENT_TRANSPORT_EXC = (
    requests.ConnectionError,
    requests.Timeout,
    requests.exceptions.ChunkedEncodingError,
)

_TITLE_MAX = 40
# Windows/gallery-dl path-restrict forbidden set + path separators.
_FORBIDDEN = set('<>:"/\\|?*')


# -- shared exception taxonomy --------------------------------------------
# Every native client raises one of these (platform subclasses keep an
# isinstance-distinct platform name AND the semantic Auth/Drift class), so the
# base Ingester._failure_result can map them platform-agnostically.

class NativeIngestError(Exception):
    """Base for a native-ingest client failure. `status_code` carries the HTTP
    status when the failure was an HTTP response (None for transport/parse);
    `retry_after` carries the server's 429 Retry-After hint so the cooldown can
    match it (plan #708 B1)."""

    def __init__(
        self,
        message: str,
        *,
        status_code: int | None = None,
        retry_after: float | None = None,
    ):
        super().__init__(message)
        self.status_code = status_code
        self.retry_after = retry_after


class NativeAuthError(NativeIngestError):
    """Authentication/authorization failure — expired/missing credential or an
    insufficient tier. The fix is rotating the credential, NOT updating the
    ingester. Maps to error_type 'auth_error'."""


class NativeDriftError(NativeIngestError):
    """A response did not match the shape the ingester depends on (JSON:API field
    set, or scraped HTML structure). Fail loud so the import step flags 'the
    platform changed' instead of silently importing nothing. Maps to API_DRIFT."""


# -- HTTP session ----------------------------------------------------------

def make_session(
    cookies_path: str | Path | None,
    *,
    accept: str = "*/*",
    extra_headers: dict | None = None,
) -> requests.Session:
    """Build a requests.Session loaded with the Netscape cookies.txt
    CredentialService materializes. `accept` sets the Accept header (the JSON:API
    vs HTML feed differ); `extra_headers` adds platform headers (e.g.
    X-Requested-With). Missing/unparseable cookies log a warning, never fail."""
    session = requests.Session()
    headers = {"User-Agent": _USER_AGENT, "Accept": accept}
    if extra_headers:
        headers.update(extra_headers)
    session.headers.update(headers)
    if cookies_path and os.path.isfile(str(cookies_path)):
        try:
            jar = http.cookiejar.MozillaCookieJar(str(cookies_path))
            jar.load(ignore_discard=True, ignore_expires=True)
            session.cookies = jar  # type: ignore[assignment]
        except (OSError, http.cookiejar.LoadError) as exc:
            log.warning("Could not load cookies from %s: %s", cookies_path, exc)
    return session


def retry_after_seconds(
    resp: requests.Response,
    attempt: int,
    *,
    base: float = _BACKOFF_BASE_SECONDS,
    cap: float = _BACKOFF_CAP_SECONDS,
) -> float:
    """Backoff for a 429: the numeric Retry-After header if present, else
    exponential base·2^(attempt-1), both capped."""
    header = resp.headers.get("Retry-After")
    if header:
        try:
            return min(float(header), cap)
        except (TypeError, ValueError):
            pass
    return min(base * (2 ** max(0, attempt - 1)), cap)


# -- filename / path helpers -----------------------------------------------

def sanitize_segment(name: str) -> str:
    """Make `name` safe for one filesystem path segment: replace separators, the
    Windows-forbidden set, and control chars with `_`; strip trailing dots/spaces
    (gallery-dl path-restrict). Never empty (falls back to `_`)."""
    out = ["_" if (ch in _FORBIDDEN or ord(ch) < 32) else ch for ch in name]
    cleaned = "".join(out).rstrip(". ")
    return cleaned or "_"


def basename_from_url(url: str) -> str:
    """Derive a sane filename from a URL when the media has no name: path basename
    with a junk-extension guard (safe_ext), bounded stem; falls back to the URL's
    content hash, then "file"."""
    path = urlsplit(url).path
    base = os.path.basename(path)
    if base:
        ext = safe_ext(base)
        stem = base[: -len(Path(base).suffix)] if Path(base).suffix else base
        stem = stem[:120] or "file"
        return f"{stem}{ext}"
    return filehash_from_url(url) or "file"


def post_dir_name(post: dict) -> str:
    """`<YYYY-MM-DD>_<post_id>_<title40>` matching gallery-dl's layout (date prefix
    omitted when published_at is missing/unparseable; title is empty for platforms
    with no title field). Accepts both ISO and trailing-`Z` published_at."""
    post_id = str(post.get("id") or "")
    attrs = post.get("attributes") or {}
    title = attrs.get("title")
    title40 = (title if isinstance(title, str) else "")[:_TITLE_MAX]
    published = attrs.get("published_at")
    date_prefix = None
    if isinstance(published, str) and published:
        s = published.strip()
        if s.endswith("Z"):
            s = s[:-1] + "+00:00"
        try:
            date_prefix = f"{datetime.fromisoformat(s):%Y-%m-%d}"
        except ValueError:
            date_prefix = None
    raw = f"{date_prefix}_{post_id}_{title40}" if date_prefix else f"{post_id}_{title40}"
    return sanitize_segment(raw)


# -- one downloadable file, and its seen-ledger identity -------------------

@dataclass
class MediaItem:
    """One resolved downloadable file belonging to a post, as every platform
    client returns it.

    - `url`: what to fetch. `filename`: the media's own name when the platform
      gives one, else the URL basename (never a network call).
    - `kind`: the content type, using gallery-dl's `files` names ("images",
      "attachments", …) so per-source content_types filter the same way.
    - `post_id`: the owning post, so a flattened list stays traceable.
    - `filehash`: the 32-hex content hash in a content-addressed CDN URL
      (Patreon), else None.
    - `media_id`: the platform's stable per-file id where the URL has no hash:
      SubscribeStar's gallery id, Discord's attachment id (or, for an embed, a
      hash of its URL path; the query string is a signature that changes on
      every fetch). Empty when the platform has none.
    - `extension` / `num`: Discord only. gallery-dl splits the name from its
      extension and numbers files 1-based across the message; both name the
      file on disk.
    """

    url: str
    filename: str
    kind: str
    post_id: str
    filehash: str | None = None
    media_id: str = ""
    extension: str = ""
    num: int = 0


# The seen-ledger tables' key column width.
_LEDGER_KEY_MAX = 128


def ledger_key(media: MediaItem) -> str:
    """A file's identity in the cross-run seen-ledger, stable across walks.

    The content hash when the URL carries one. Otherwise the file is keyed
    within its post: by the platform's per-file id, or by filename when there is
    none (a Patreon Mux video or a hashless inline image). A positional key is
    never used, because an edit that removes a file renumbers the ones after it
    and would make a different file read as seen."""
    if media.filehash:
        return media.filehash
    return f"{media.post_id}:{media.media_id or media.filename}"[:_LEDGER_KEY_MAX]


# -- per-item download outcomes (shared dataclasses) -----------------------

@dataclass
class MediaOutcome:
    """Per-media result of a download_post pass. status ∈ downloaded /
    skipped_seen / skipped_disk / quarantined / error. `path` is the on-disk file
    (downloaded / skipped_disk), the quarantine dest (quarantined), or None;
    `error` is the failure/validation reason (error/quarantined) else None."""

    media: object
    status: str
    path: Path | None
    error: str | None


@dataclass
class PostRecordOutcome:
    """Result of write_post_record — mirrors the MediaOutcome contract so the core
    reports per-post handling. `path` is the _post.json sidecar (None when the post
    had no id); the rest is the captured body's shape for the run log."""

    path: Path | None
    post_type: str | None
    title: str | None
    body_chars: int


# -- membership roster seam (shared dataclass, #387 C2/C7) -----------------

@dataclass
class Membership:
    """One membership the ACCOUNT holds, as the roster needs it (#387 C2).

    Lives HERE rather than in the platform module that first produced it, for
    the same reason `PostRecordOutcome` does: it is the seam's contract, not
    Patreon's. C7 moved it — while it sat in `patreon_client` a second platform
    would have had to import its contract from the first platform's module,
    which inverts the dependency and is how a "portable" seam quietly becomes
    Patreon-shaped.

    Deliberately not a raw upstream row: the sweep should not have to know that
    a tier lives behind a JSON:API `reward` relationship, and
    `platform_membership` should not gain columns because one platform shapes
    things a certain way.

    `status` carries the PLATFORM's own word, verbatim and unmapped
    (`active_patron`, `former_patron`, ...). Deciding what it means is the read
    site's job — `has_paid_access`, below — precisely so an
    unrecognised word records as evidence rather than as a decision.

    `is_free_member` is SEPARATE from status and must stay that way. Patreon
    expresses a free follow as this boolean rather than as a status value, so
    "does the account pay for this" is `status == "active_patron" and not
    is_free_member` — a question the status string alone cannot answer. NOTE:
    the C0 capture contains no ACTIVE free member, so the two fields are
    perfectly correlated in that sample; the separation is what the schema
    says, not something the sample proves.

    A platform that lacks a field supplies the empty answer, never a guess:
    no tiers -> `[]`, no pledge -> `amount_cents=None` (absent stays
    distinguishable from zero — "free" and "we don't know" are different
    answers), no vanity -> None and identity falls back to the URL tail.
    """

    campaign_id: str
    display_name: str | None
    url: str | None
    vanity: str | None
    status: str | None
    is_free_member: bool
    tier_names: list[str]
    amount_cents: int | None
    currency: str | None
    # Everything the roster did not model, kept so a later question can be
    # answered without another authenticated round-trip. Scoped to the
    # membership's own attributes plus the creator's — never the raw page,
    # which is where the card/address resources live.
    details: dict


# -- base downloader (shared fetch/validate plumbing) ----------------------

class BaseNativeDownloader:
    """Shared download plumbing for native-platform downloaders: one post's
    files (`download_post`, with the seen/disk skips and the time-box), the
    streaming GET (transient-retry + Range-resume), and file
    validation/quarantine.

    A platform downloader subclasses this, implements `write_post_record`, and
    overrides only the seams where its layout differs: `_post_folder`,
    `_media_name`, `_media_sidecar`, and for Patreon's Mux/yt-dlp video,
    `_existing_output` and `_fetch_media`. PURE: no DB; the seen-skip is an
    injected predicate.
    """

    def __init__(
        self,
        images_root: Path,
        cookies_path: str | None = None,
        *,
        platform: str,
        validate: bool = True,
        rate_limit: float = 0.0,
        session: requests.Session | None = None,
    ):
        self.images_root = Path(images_root)
        self.cookies_path = str(cookies_path) if cookies_path else None
        self.platform = platform
        self._validate = validate
        self._rate_limit = rate_limit or 0.0
        self.session = session if session is not None else make_session(cookies_path)

    # -- download seams ----------------------------------------------------

    def _fetch_get(self, url: str, dest: Path) -> Path:
        """Stream `url` to a .part then atomic-rename to `dest`."""
        part = dest.with_name(dest.name + ".part")
        try:
            self._fetch_to_file(url, part)
        except Exception:
            with contextlib.suppress(OSError):
                part.unlink()
            raise
        os.replace(part, dest)
        return dest

    def _fetch_to_file(self, url: str, dest: Path) -> None:
        """Stream a URL to `dest`, retrying TRANSIENT failures (transport blips,
        429 honoring Retry-After, 5xx) with backoff + resume-from-disk (Range);
        failing fast on permanent 4xx (404/403). Resume: a retry with bytes on
        disk asks `Range: bytes=<have>-`; 206 → append, 200 → restart clean, 416 →
        already complete. The caller stages into a `.part` so a non-range server
        never corrupts the output."""
        attempt = 0
        while True:
            have = dest.stat().st_size if dest.exists() else 0
            headers = {"Range": f"bytes={have}-"} if have > 0 else None
            try:
                resp = self.session.get(
                    url, stream=True, timeout=_TIMEOUT_SECONDS, headers=headers,
                )
                if (resp.status_code == 429 or resp.status_code >= 500) \
                        and attempt < _MAX_MEDIA_RETRIES:
                    attempt += 1
                    delay = retry_after_seconds(resp, attempt)
                    log.warning(
                        "%s media transient HTTP %d (%s) — backing off %.1fs "
                        "(retry %d/%d)",
                        self.platform, resp.status_code, url, delay, attempt,
                        _MAX_MEDIA_RETRIES,
                    )
                    time.sleep(delay)
                    continue
                if have > 0 and resp.status_code == 416:
                    return
                resp.raise_for_status()
                mode = "ab" if (have > 0 and resp.status_code == 206) else "wb"
                with open(dest, mode) as fh:
                    for chunk in resp.iter_content(chunk_size=_CHUNK):
                        if chunk:
                            fh.write(chunk)
                return
            except _TRANSIENT_TRANSPORT_EXC as exc:
                if attempt >= _MAX_MEDIA_RETRIES:
                    raise
                attempt += 1
                delay = min(2.0 * (2 ** (attempt - 1)), _BACKOFF_CAP_SECONDS)
                log.warning(
                    "%s media transport error (%s) — backing off %.1fs "
                    "(retry %d/%d): %s",
                    self.platform, url, delay, attempt, _MAX_MEDIA_RETRIES, exc,
                )
                time.sleep(delay)

    # -- validation --------------------------------------------------------

    def _validate_path(
        self, path: Path, artist_slug: str, source_url: str | None = None
    ) -> tuple[str | None, Path | None]:
        """Validate a freshly-written file; quarantine if bad (shared
        file_validator move + provenance sidecar). Returns (reason,
        quarantine_dest) when quarantined, else (None, None). Logs the quarantine
        so a corrupt file is visible in the worker logs, not just counted (#899
        L2)."""
        if not self._validate or not is_validatable(path):
            return None, None
        try:
            result = validate_file(path)
        except Exception as exc:
            log.warning("Validator raised on %s: %s", path, exc)
            return None, None
        if result.ok:
            return None, None
        dest = quarantine_file(
            self.images_root, path, artist_slug, self.platform,
            url=source_url, result=result,
        )
        reason = result.reason or "validation failed"
        log.warning(
            "%s quarantined %s (%s) — %s",
            self.platform, dest or path, artist_slug, reason,
        )
        return reason, (dest or path)

    # -- one post's files ---------------------------------------------------

    def download_post(
        self,
        post: dict,
        media_items: list[MediaItem],
        artist_slug: str,
        *,
        is_seen: Callable[[MediaItem], bool] = lambda m: False,
        should_stop: Callable[[], bool] = lambda: False,
        recapture: bool = False,
    ) -> list[MediaOutcome]:
        """Download every file of one post and return one outcome per file.

        - Two-tier skip: the injected seen-ledger predicate first, then the disk.
        - `should_stop()` is polled before each file, so a file-dense post
          cannot run a time-boxed chunk far past its deadline. The files not
          reached are not marked seen and are fetched by the next chunk.
        - One file's failure is an "error" outcome for that file; the rest
          proceed.
        - `recapture` re-reads what is on disk without re-downloading: a seen
          file that is on disk comes back `skipped_disk` with its path (so the
          engine can backfill its source filehash), and a seen file that is gone
          is left alone, since refetching is recovery's job.
        """
        folder = self._post_folder(post, artist_slug)
        outcomes: list[MediaOutcome] = []
        for index, media in enumerate(media_items, start=1):
            if should_stop():
                break
            try:
                outcomes.append(self._download_one(
                    post, media, folder, artist_slug, index, is_seen,
                    recapture=recapture,
                ))
            except Exception as exc:  # resilient: isolate one item's failure
                log.warning(
                    "%s media failed (post %s, item %d): %s",
                    self.platform, post.get("id"), index, exc,
                )
                outcomes.append(
                    MediaOutcome(media=media, status="error", path=None, error=str(exc))
                )
        return outcomes

    def _download_one(
        self,
        post: dict,
        media: MediaItem,
        folder: Path,
        artist_slug: str,
        index: int,
        is_seen: Callable[[MediaItem], bool],
        *,
        recapture: bool = False,
    ) -> MediaOutcome:
        seen = is_seen(media)
        if seen and not recapture:
            return MediaOutcome(media=media, status="skipped_seen", path=None, error=None)
        path = folder / self._media_name(post, media, index)
        existing = self._existing_output(path, media)
        if existing is not None:
            return MediaOutcome(media=media, status="skipped_disk", path=existing, error=None)
        if seen:
            return MediaOutcome(media=media, status="skipped_seen", path=None, error=None)

        folder.mkdir(parents=True, exist_ok=True)
        # Real downloads are paced; the skips above have already returned.
        if self._rate_limit > 0:
            time.sleep(self._rate_limit)
        return self._fetch_media(post, media, path, artist_slug)

    # -- platform seams ----------------------------------------------------
    # Where a platform lays its files out. Each matches the layout gallery-dl
    # wrote, so files from before a platform moved to its native downloader
    # are found on disk instead of fetched again.

    def _post_folder(self, post: dict, artist_slug: str) -> Path:
        return self.images_root / artist_slug / self.platform / post_dir_name(post)

    def _media_name(self, post: dict, media: MediaItem, index: int) -> str:
        return sanitize_segment(f"{index:02d}_{media.filename}")

    def _existing_output(self, path: Path, media: MediaItem) -> Path | None:
        """The file already on disk for this media, or None."""
        return path if path.exists() else None

    def _fetch_media(
        self, post: dict, media: MediaItem, path: Path, artist_slug: str,
    ) -> MediaOutcome:
        """Fetch one file, validate it, and write its sidecar."""
        out = self._fetch_get(media.url, path)
        reason, quarantined = self._validate_path(out, artist_slug, media.url)
        if reason is not None:
            # Distinct from an error, so the run reports a real
            # files_quarantined count and the paths.
            return MediaOutcome(media=media, status="quarantined", path=quarantined, error=reason)
        self._write_media_sidecar(post, out, media)
        return MediaOutcome(media=media, status="downloaded", path=out, error=None)

    def _media_sidecar(self, post: dict, media: MediaItem) -> dict:
        """The per-file sidecar's content: the file's identity only. The post's
        body, title and links live in its post record alone, so they are never
        copied beside every file."""
        return {"category": self.platform, "id": str(post.get("id") or ""),
                "source_url": media.url}

    def _write_media_sidecar(self, post: dict, media_path: Path, media: MediaItem) -> Path:
        """Write the per-file sidecar beside the file, named like it minus the
        extension, which is what `find_sidecar` pairs first."""
        sidecar_path = media_path.with_suffix(".json")
        sidecar_path.write_text(json.dumps(self._media_sidecar(post, media), indent=2))
        return sidecar_path

# --- membership status vocabulary (#387) ------------------------------------
#
# Lives here, beside `Membership`, rather than in `membership_roster`. It is
# pure platform knowledge with no database behind it, and the platform clients
# need it too. Patreon's must tell a lapsed membership to a deleted creator
# (skippable) from a paid one it cannot attribute (drift), and a client may not
# import `membership_roster`: test_gated_reason.py forbids any fetch path from
# reaching the roster, so the roster can explain a skip but never cause one.
#
# Platform word -> whether the account currently has paid access.
#
# Every entry here must come from a CHARACTERISED response, never from API docs
# or a plausible guess — project rule 130, and inventing a status before seeing
# it in a real payload is exactly the failure it names.
#
# patreon: from a live capture of the operator's own session, 2026-09-10
# (Scribe note #3886). Only two values were OBSERVED in `patron_status` and
# only those two are here.
#
# `declined_patron` is deliberately ABSENT even though it looks obviously
# right. It appears in the request's `filter[membership_type]`, and the capture
# proved that filter is NOT the same vocabulary as the attribute — a row
# selected by the filter as `free_member` came back with
# `patron_status: former_patron`, a word the filter does not contain. Reading
# the filter as an enum is the specific mistake the capture caught; adding
# `declined_patron` on the strength of it would be repeating that mistake one
# step later.
#
# Unknown words are NOT an error: an unrecognised status means the roster
# records evidence it cannot yet interpret, which is a better state than
# dropping the row or asserting a meaning for it.
#
# subscribestar: from a live capture of the account's /subscriptions page,
# 2026-09-13 (Scribe note #3989). SubscribeStar gives NO per-row status word —
# a membership's state is which of two tables it sits in — so the "word" stored
# is the table card's own `data-identifier`, verbatim. Those two identifiers are
# the whole vocabulary; there is nothing further to characterise later.
MEMBERSHIP_STATUS: dict[str, dict[str, bool]] = {
    "patreon": {
        "active_patron": True,
        "former_patron": False,
    },
    "subscribestar": {
        "active_subscriptions": True,
        "cancelled_subscriptions": False,
    },
}


def has_paid_access(
    platform: str, status: str | None, *, is_free_member: bool = False,
) -> bool | None:
    """Does this membership mean the account currently PAYS for access?

    Returns None for a status this code has not been taught, which callers must
    treat as "unknown" rather than as False. The difference matters: False says
    the operator has lost access, and asserting that from an unrecognised word
    would tell them to cancel a source they are still paying for.

    `is_free_member` is a second axis, not a status, and that is Patreon's
    design rather than ours: the capture shows a free follow expressed as a
    boolean alongside `patron_status`, so a "current" membership can still be
    one nobody is paying for. Taking status alone would report a free follower
    as a paying patron, and C4 would then never offer to clean it up.

    (Honest limit: the capture contains no ACTIVE free member, so it cannot
    demonstrate the two axes coming apart. The separation is what the payload's
    shape says; the sample only shows it is possible, not that it happens.)
    """
    if status is None:
        return None
    known = MEMBERSHIP_STATUS.get(platform, {}).get(status)
    if known is None:
        return None
    if not known:
        return False
    return not is_free_member