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


# -- 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 — `membership_roster.has_paid_access` — 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: the streaming
    GET (transient-retry + Range-resume) and file validation/quarantine. Platform
    downloaders subclass this and implement `download_post` / `write_post_record`
    / the per-media sidecar (and any platform-specific fetch, e.g. Patreon's
    Mux/yt-dlp video branch). 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)

    # -- sidecar (per-media, minimal) --------------------------------------

    def _write_minimal_sidecar(
        self, post: dict, media_path: Path, *, source_url: str | None = None
    ) -> Path:
        """Post-first per-media sidecar (#856): image identity ONLY
        (category/id/source_url). The post body/links live solely in _post.json."""
        data: dict = {"category": self.platform, "id": str(post.get("id") or "")}
        if source_url:
            data["source_url"] = source_url
        sidecar_path = media_path.with_suffix(".json")
        sidecar_path.write_text(json.dumps(data, indent=2))
        return sidecar_path