From f6788190932a4bf150e182f41ddff9ebbc94cc24 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 17 Jun 2026 10:04:52 -0400 Subject: [PATCH 1/7] feat(subscribestar): seen/failed ledger models + migration 0054 (#889) Phase 1, step 1 of moving SubscribeStar off gallery-dl onto the native core ingester (milestone: SubscribeStar native). Mirror of the Patreon ledger: SubscribeStarSeenMedia (skip already-ingested media on routine walks; recovery bypasses) and SubscribeStarFailedMedia (dead-letter so persistently-failing media stops re-burning backfill chunks). Per operator decision, dedicated per-platform tables (not a generalized shared ledger). filehash is String(128): a CDN content hash when the URL carries one, else a synthesized : key. UNIQUE (source_id, filehash) upsert key. Registered in models/__init__; migration 0054 creates both tables (down 0053). Co-Authored-By: Claude Opus 4.8 --- .../versions/0054_subscribestar_ledgers.py | 82 +++++++++++++++++++ backend/app/models/__init__.py | 4 + .../app/models/subscribestar_failed_media.py | 44 ++++++++++ .../app/models/subscribestar_seen_media.py | 40 +++++++++ 4 files changed, 170 insertions(+) create mode 100644 alembic/versions/0054_subscribestar_ledgers.py create mode 100644 backend/app/models/subscribestar_failed_media.py create mode 100644 backend/app/models/subscribestar_seen_media.py diff --git a/alembic/versions/0054_subscribestar_ledgers.py b/alembic/versions/0054_subscribestar_ledgers.py new file mode 100644 index 0000000..59972ae --- /dev/null +++ b/alembic/versions/0054_subscribestar_ledgers.py @@ -0,0 +1,82 @@ +"""subscribestar_seen_media + subscribestar_failed_media: per-source ledgers + +Revision ID: 0054 +Revises: 0053 +Create Date: 2026-06-17 + +SubscribeStar native ingester (phase 1 of the gallery-dl → native-core +migration). Mirrors the Patreon ledger tables (0037/0038): a seen-ledger so +routine walks skip already-ingested media (recovery bypasses it) and a +dead-letter ledger so persistently-failing media stops re-burning backfill +chunks. `filehash` is a CDN content hash when present, else a synthesized +``:`` key — hence String(128). UNIQUE (source_id, filehash) +is the upsert key on each. +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0054" +down_revision: Union[str, None] = "0053" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "subscribestar_seen_media", + sa.Column("id", sa.Integer, primary_key=True), + sa.Column( + "source_id", + sa.Integer, + sa.ForeignKey("source.id", ondelete="CASCADE"), + nullable=False, + index=True, + ), + sa.Column("filehash", sa.String(128), nullable=False), + sa.Column("post_id", sa.String(64), nullable=True), + sa.Column( + "seen_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("NOW()"), + ), + sa.UniqueConstraint( + "source_id", "filehash", name="uq_subscribestar_seen_media_source_id" + ), + ) + op.create_table( + "subscribestar_failed_media", + sa.Column("id", sa.Integer, primary_key=True), + sa.Column( + "source_id", + sa.Integer, + sa.ForeignKey("source.id", ondelete="CASCADE"), + nullable=False, + index=True, + ), + sa.Column("filehash", sa.String(128), nullable=False), + sa.Column("attempts", sa.Integer, nullable=False, server_default="1"), + sa.Column("last_error", sa.Text, nullable=True), + sa.Column( + "first_failed_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("NOW()"), + ), + sa.Column( + "last_failed_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("NOW()"), + ), + sa.UniqueConstraint( + "source_id", "filehash", name="uq_subscribestar_failed_media_source_id" + ), + ) + + +def downgrade() -> None: + op.drop_table("subscribestar_failed_media") + op.drop_table("subscribestar_seen_media") diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 4433dc2..6061f31 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -24,6 +24,8 @@ from .series_chapter import SeriesChapter from .series_page import SeriesPage from .series_suggestion import SeriesSuggestion from .source import Source +from .subscribestar_failed_media import SubscribeStarFailedMedia +from .subscribestar_seen_media import SubscribeStarSeenMedia from .tag import Tag, TagKind, image_tag from .tag_alias import TagAlias from .tag_allowlist import TagAllowlist @@ -41,6 +43,8 @@ __all__ = [ "Credential", "PatreonFailedMedia", "PatreonSeenMedia", + "SubscribeStarFailedMedia", + "SubscribeStarSeenMedia", "Post", "PostAttachment", "SeriesChapter", diff --git a/backend/app/models/subscribestar_failed_media.py b/backend/app/models/subscribestar_failed_media.py new file mode 100644 index 0000000..9201aa7 --- /dev/null +++ b/backend/app/models/subscribestar_failed_media.py @@ -0,0 +1,44 @@ +"""SubscribeStarFailedMedia — per-source dead-letter ledger of SubscribeStar +media that keeps failing to download/validate. + +Mirror of PatreonFailedMedia. Media that fails every walk (404'd CDN URL, +deleted post, persistently-corrupt bytes) would otherwise re-error forever and +re-burn backfill chunks. After ``attempts`` reaches the dead-letter threshold +the ingester skips it on routine tick/backfill walks (recovery still +re-attempts). A later clean download clears the row. + +`filehash` is the same per-media key the seen-ledger uses (CDN content hash or a +synthesized ``:`` key) — hence String(128). UNIQUE +(source_id, filehash) is the upsert key. +""" + +from datetime import datetime + +from sqlalchemy import ForeignKey, Integer, String, Text, UniqueConstraint, func +from sqlalchemy.orm import Mapped, mapped_column +from sqlalchemy.types import DateTime + +from .base import Base + + +class SubscribeStarFailedMedia(Base): + __tablename__ = "subscribestar_failed_media" + __table_args__ = ( + UniqueConstraint( + "source_id", "filehash", name="uq_subscribestar_failed_media_source_id" + ), + ) + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + source_id: Mapped[int] = mapped_column( + ForeignKey("source.id", ondelete="CASCADE"), nullable=False, index=True + ) + filehash: Mapped[str] = mapped_column(String(128), nullable=False) + attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=1) + last_error: Mapped[str | None] = mapped_column(Text, nullable=True) + first_failed_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + last_failed_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) diff --git a/backend/app/models/subscribestar_seen_media.py b/backend/app/models/subscribestar_seen_media.py new file mode 100644 index 0000000..67b1608 --- /dev/null +++ b/backend/app/models/subscribestar_seen_media.py @@ -0,0 +1,40 @@ +"""SubscribeStarSeenMedia — per-source ledger of SubscribeStar media already +downloaded+processed. + +Mirror of PatreonSeenMedia for the SubscribeStar native ingester (replacing +gallery-dl). One queryable row per (source, media) so routine walks skip media +we've already ingested; recovery mode bypasses the ledger to re-walk. + +`filehash` is a CDN content hash when the media URL carries one, else a +synthesized ``:`` key (SubscribeStar URLs aren't always +content-addressed) — hence String(128) rather than 32. +""" + +from datetime import datetime + +from sqlalchemy import ForeignKey, Integer, String, UniqueConstraint, func +from sqlalchemy.orm import Mapped, mapped_column +from sqlalchemy.types import DateTime + +from .base import Base + + +class SubscribeStarSeenMedia(Base): + __tablename__ = "subscribestar_seen_media" + __table_args__ = ( + # Dedup key the downloader upserts against: one ledger row per + # (source, media). A second sighting of the same media is a no-op. + UniqueConstraint( + "source_id", "filehash", name="uq_subscribestar_seen_media_source_id" + ), + ) + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + source_id: Mapped[int] = mapped_column( + ForeignKey("source.id", ondelete="CASCADE"), nullable=False, index=True + ) + filehash: Mapped[str] = mapped_column(String(128), nullable=False) + post_id: Mapped[str | None] = mapped_column(String(64), nullable=True) + seen_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) From 817a002c2b3f1da1c8e9a7456d3b8ac14165f1e0 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 17 Jun 2026 11:06:02 -0400 Subject: [PATCH 2/7] feat(subscribestar): native client + downloader + ingester (post-first) (#890/#891/#892) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase-1 steps 2-4 of moving SubscribeStar off gallery-dl onto the native core ingester. SubscribeStar has no JSON:API, so the client scrapes HTML; the platform-agnostic core (ingest_core) is unchanged. - subscribestar_client.py: HTML-scrape read path. iter_posts pages via the creator page → infinite_scroll-next_page href → JSON {html} fragments (campaign_id = creator URL; no resolver). extract_media reads the per-post data-gallery JSON manifest (id/original_filename/type/url). post_record_key, post_meta, and post_is_gated (best-effort locked-teaser marker, pending a live locked sample). Loud auth/drift taxonomy (SubscribeStar{API,Auth,Drift}Error). Parser validated against the real Step-0 fixtures. - subscribestar_downloader.py: mirrors PatreonDownloader minus the Mux/yt-dlp branch (SubscribeStar serves files directly via /post_uploads). gallery-dl on-disk layout so existing downloads dedup on disk at cutover. Post-first: _post.json owns the body/links; per-media sidecar carries image identity only. - subscribestar_ingester.py: thin adapter wiring client/downloader/the SubscribeStar ledgers into the core; ledger_key = filehash else post_id:media_id; SubscribeStar failure mapping. verify_subscribestar_credential. - tests: client parsing/pagination/media/gating/record-key/dates, downloader layout/sidecar/post-record/skip-seen, ingester ledger_key + failure mapping. Not yet wired into dispatch (Step 5) — these modules are inert until then. Co-Authored-By: Claude Opus 4.8 --- backend/app/services/subscribestar_client.py | 454 ++++++++++++++++++ .../app/services/subscribestar_downloader.py | 317 ++++++++++++ .../app/services/subscribestar_ingester.py | 145 ++++++ tests/test_subscribestar_native.py | 294 ++++++++++++ 4 files changed, 1210 insertions(+) create mode 100644 backend/app/services/subscribestar_client.py create mode 100644 backend/app/services/subscribestar_downloader.py create mode 100644 backend/app/services/subscribestar_ingester.py create mode 100644 tests/test_subscribestar_native.py diff --git a/backend/app/services/subscribestar_client.py b/backend/app/services/subscribestar_client.py new file mode 100644 index 0000000..c4a6031 --- /dev/null +++ b/backend/app/services/subscribestar_client.py @@ -0,0 +1,454 @@ +"""Native SubscribeStar client — the SubscribeStar adapter's read path. + +Unlike Patreon (a clean JSON:API), SubscribeStar has NO public API: gallery-dl +scrapes its HTML, and so do we. The platform-agnostic core (`ingest_core`) only +calls `client.iter_posts` / `extract_media` / the optional seams, so the +HTML-scrape divergence is contained entirely to this module. + +Feed shape (characterized from a live sample 2026-06-17 — Scribe note +"SubscribeStar HTML characterization"): + - Page 1 is the creator page HTML: GET /. + - Each page embeds `data-role="infinite_scroll-next_page" href="/posts?...&page=N + &slug=&sort_by=newest"`. That path returns JSON {"html": "<...posts...>"}; + the fragment carries the NEXT page link, until it runs out. + - A post is `
`; body lives + in `.post-content .trix-content`; date in `.post-date`; media in a `data-gallery` + JSON manifest on `.uploads-images`. + +`campaign_id` for SubscribeStar is the full creator URL (e.g. +https://subscribestar.adult/sabu) — the host (.com vs .adult) and slug both come +from it, so no separate resolver is needed. + +Drift is loud on purpose (mirrors patreon_client): an HTML login/age-gate where +the feed was expected is AUTH (rotate cookies); a feed whose post structure we +can't parse at all is DRIFT (the scraper needs updating). FC runs on a +plain-HTTP homelab; nothing here uses a secure-context Web API. +""" + +from __future__ import annotations + +import http.cookiejar +import json +import logging +import os +import re +import time +from collections.abc import Iterator +from dataclasses import dataclass +from datetime import datetime +from html import unescape +from pathlib import Path +from urllib.parse import urljoin, urlsplit + +import requests + +from ..utils.paths import filehash_from_url, safe_ext +from .patreon_client import _MAX_429_RETRIES, _retry_after_seconds + +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" +) +_TIMEOUT_SECONDS = 30.0 + +# A post block opens with this wrapper; we slice the page between consecutive +# occurrences (regex can't match the balanced close of nested divs, so chunk- +# per-post is the robust approach gallery-dl also uses). +_POST_OPEN = '
([^<]+)
') +# The body: inner of `.post-content .trix-content`. Non-greedy to the LAST +#
in the chunk would over-capture; instead capture from trix-content +# to the close of the post-content wrapper, which the post-body block ends with. +_BODY_RE = re.compile( + r'
]*>(.*?)
\s*\s*', re.DOTALL +) +_GALLERY_RE = re.compile(r'data-gallery="([^"]*)"') +_NEXT_PAGE_RE = re.compile( + r'data-role="infinite_scroll-next_page"\s+href="([^"]+)"' +) +# Signals an auth/age wall served in place of the feed (cookies expired or the +# age cookie missing) rather than a real — possibly empty — creator feed. +_LOGIN_MARKERS = ("/session/new", 'data-role="sign_in"', "age_confirmation_warning") + + +class SubscribeStarAPIError(Exception): + """Base for native SubscribeStar client failures. `status_code` carries the + HTTP status when the failure was an HTTP response (None for transport/parse), + so the ingester can map it to a DownloadResult.error_type.""" + + 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 SubscribeStarAuthError(SubscribeStarAPIError): + """Auth/authorization failure — expired cookies, missing age cookie, or an + HTML login/age wall served where the feed was expected. Fix = rotate the + credential, not update the scraper. Maps to error_type 'auth_error'.""" + + +class SubscribeStarDriftError(SubscribeStarAPIError): + """The feed HTML did not match the structure we scrape (no recognizable post + blocks AND no known empty-feed state). The scrape analog of API drift — fail + loud so the import step flags 'SubscribeStar changed its markup' instead of + silently importing nothing.""" + + +@dataclass +class MediaItem: + """One resolved downloadable item belonging to a SubscribeStar post. + + Fields mirror patreon_client.MediaItem (so the downloader is structurally the + same) plus `media_id` — the stable per-upload gallery id. SubscribeStar's + full-res URL is an opaque `/post_uploads?payload=...` (not content-addressed), + so `filehash` is usually None and the ledger keys on `:`. + """ + + url: str + filename: str + kind: str + filehash: str | None + post_id: str + media_id: str + + +def _load_session(cookies_path: str | Path | None) -> requests.Session: + session = requests.Session() + session.headers.update( + { + "User-Agent": _USER_AGENT, + # HTML feed + JSON "load more"; let the server pick. + "Accept": "text/html,application/xhtml+xml,application/json,*/*", + "X-Requested-With": "XMLHttpRequest", + } + ) + 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 SubscribeStar cookies from %s: %s", cookies_path, exc) + return session + + +def _split_creator_url(campaign_id: str) -> tuple[str, str]: + """`campaign_id` is the creator URL → (base, slug). + + base = scheme://host (preserving .com vs .adult); slug = first path segment. + """ + parts = urlsplit(campaign_id) + base = f"{parts.scheme or 'https'}://{parts.netloc}" + slug = parts.path.strip("/").split("/")[0] if parts.path else "" + return base, slug + + +def _parse_ss_datetime(text: str) -> str | None: + """SubscribeStar renders human dates: 'Jun 17, 2026 03:19 am' (and an + 'Updated on ' variant). Return ISO-8601 (UTC-naive) or None.""" + s = unescape(text or "").strip() + if s.lower().startswith("updated on "): + s = s[len("updated on "):].strip() + for fmt in ("%b %d, %Y %I:%M %p", "%b %d, %Y"): + try: + return datetime.strptime(s, fmt).isoformat() + except ValueError: + continue + return None + + +def _basename_from_url(url: str) -> str: + 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 "file" + + +class SubscribeStarClient: + """Synchronous SubscribeStar HTML-scrape read client. Construct with a path + to a Netscape cookies.txt (the same file CredentialService.get_cookies_path + materializes, already carrying the age cookie via augment_cookies).""" + + 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 = _load_session(cookies_path) + self._request_sleep = request_sleep or 0.0 + self._max_retries = max_retries + + # -- request ----------------------------------------------------------- + + def _get(self, url: str) -> requests.Response: + if self._request_sleep > 0: + time.sleep(self._request_sleep) + attempt = 0 + while True: + try: + resp = self._session.get(url, timeout=_TIMEOUT_SECONDS) + except requests.RequestException as exc: + raise SubscribeStarAPIError( + f"SubscribeStar request failed ({url}): {exc}" + ) from exc + if resp.status_code == 429 and attempt < self._max_retries: + attempt += 1 + delay = _retry_after_seconds(resp, attempt) + log.warning( + "SubscribeStar 429 (%s) — backing off %.1fs (retry %d/%d)", + url, delay, attempt, self._max_retries, + ) + time.sleep(delay) + continue + break + if resp.status_code in (401, 403): + raise SubscribeStarAuthError( + f"SubscribeStar returned HTTP {resp.status_code} — auth rejected " + f"(cookies expired or tier insufficient; {url})", + status_code=resp.status_code, + ) + if resp.status_code != 200: + 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 SubscribeStarAPIError( + f"SubscribeStar returned HTTP {resp.status_code} ({url})", + status_code=resp.status_code, + retry_after=retry_after, + ) + return resp + + def _feed_html(self, url: str) -> str: + """Page 1: the creator page (full HTML document).""" + resp = self._get(url) + text = resp.text or "" + if any(m in text for m in _LOGIN_MARKERS) and _POST_OPEN not in text: + raise SubscribeStarAuthError( + f"SubscribeStar served a login/age wall instead of the feed " + f"(cookies expired or age cookie missing; {url})" + ) + return text + + def _loadmore_html(self, url: str) -> str: + """Subsequent pages: GET returns JSON {"html": ""}.""" + resp = self._get(url) + try: + payload = resp.json() + except ValueError as exc: + raise SubscribeStarAuthError( + f"SubscribeStar 'load more' returned non-JSON (session expired?; " + f"{url}): {exc}" + ) from exc + html = payload.get("html") if isinstance(payload, dict) else None + return html if isinstance(html, str) else "" + + # -- parsing ----------------------------------------------------------- + + @staticmethod + def _post_chunks(html: str) -> list[str]: + """Slice the page into one chunk per post (between consecutive wrapper + opens). Regex can't match a post's balanced close, so each chunk runs to + the next post's open (or end of fragment) — enough to scope per-post + field extraction.""" + starts = [m.start() for m in re.finditer(re.escape(_POST_OPEN), html)] + chunks = [] + for i, start in enumerate(starts): + end = starts[i + 1] if i + 1 < len(starts) else len(html) + chunks.append(html[start:end]) + return chunks + + def _parse_post(self, chunk: str) -> dict | None: + m = _POST_ID_RE.search(chunk) + if not m: + return None + post_id = m.group(1) + date_m = _POST_DATE_RE.search(chunk) + published = _parse_ss_datetime(date_m.group(1)) if date_m else None + body_m = _BODY_RE.search(chunk) + content = body_m.group(1).strip() if body_m else "" + return { + "id": post_id, + "attributes": { + # SubscribeStar has no title field; the importer synthesizes a + # display title from the body's first line (sidecar util). + "title": "", + "content": content, + "published_at": published, + "post_type": "subscribestar", + }, + # Raw chunk retained so extract_media parses the data-gallery manifest + # and post_is_gated can scan for the locked-teaser marker. + "_html": chunk, + } + + def _parse_posts(self, html: str) -> list[dict]: + posts = [] + for chunk in self._post_chunks(html): + post = self._parse_post(chunk) + if post is not None: + posts.append(post) + return posts + + @staticmethod + def _next_page_href(html: str) -> str | None: + m = _NEXT_PAGE_RE.search(html) + return unescape(m.group(1)) if m else None + + def extract_media(self, post: dict, included_index: dict) -> list[MediaItem]: + """Resolve downloadable media from the post's `data-gallery` JSON + manifest. Each item carries a stable `id`, `original_filename`, `type`, + and a full-res `url` (`/post_uploads?payload=...`, relative — joined to the + post's base). `included_index` is unused (HTML carries media inline).""" + chunk = post.get("_html") or "" + base = post.get("_base") or "https://www.subscribestar.com" + post_id = str(post.get("id") or "") + items: list[MediaItem] = [] + for gm in _GALLERY_RE.finditer(chunk): + try: + gallery = json.loads(unescape(gm.group(1))) + except (ValueError, TypeError): + continue + if not isinstance(gallery, list): + continue + for it in gallery: + if not isinstance(it, dict): + continue + rel = it.get("url") + if not isinstance(rel, str) or not rel: + continue + url = urljoin(base + "/", rel) + media_id = str(it.get("id") or "") + name = it.get("original_filename") + filename = name if isinstance(name, str) and name else _basename_from_url(url) + items.append( + MediaItem( + url=url, + filename=filename, + kind=str(it.get("type") or "image"), + filehash=filehash_from_url(url), + post_id=post_id, + media_id=media_id, + ) + ) + return items + + @staticmethod + def post_meta(post: dict) -> dict: + """Title + date for the preview sample. Title is synthesized from the body + (SubscribeStar has no title field).""" + attrs = post.get("attributes") or {} + return {"title": None, "date": attrs.get("published_at")} + + @staticmethod + def post_is_gated(post: dict) -> bool: + """True when the subscriber cannot view this post (locked teaser). #874 + "no stub for gated content": the core skips it ENTIRELY (no media, no + post-record), so we don't replicate a hollow teaser in Curator. + + SubscribeStar does NOT expose downloadable preview media for locked posts + (no `data-gallery`), so the Patreon "downloaded blurred junk" failure + can't happen here — this gate only prevents capturing an empty teaser + stub. Detect the locked marker conservatively (default to NOT gated when + absent, so we never over-filter an accessible text post). The exact marker + is being confirmed against a live locked sample; until then we gate only on + the explicit lock-overlay class SubscribeStar renders on a paywalled post. + """ + chunk = post.get("_html") or "" + return 'class="post-content-locked"' in chunk or "for-locked_content" in chunk + + @staticmethod + def post_record_key(post: dict) -> tuple[str, str] | None: + """`(ledger_key, post_id)` for a post's seen-ledger entry (the `post:` + synthetic key gates post-record capture through the same ledger as media), + or None when the post has no id.""" + 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, {}, page_cursor) for every post in the feed. + + `campaign_id` is the creator URL. `cursor` is the relative "load more" + href that fetches a page (None → page 1, the creator page HTML). The + yielded `page_cursor` is the href that FETCHED this post's page, so the + core checkpoints a value that re-fetches the same page on resume (matching + the Patreon cursor contract). + """ + base, slug = _split_creator_url(campaign_id) + if not slug: + raise SubscribeStarDriftError( + f"Could not extract a creator slug from {campaign_id!r}" + ) + current = cursor + first_page = True + while True: + page_cursor = current + if current is None: + html = self._feed_html(f"{base}/{slug}") + else: + html = self._loadmore_html(urljoin(base + "/", current)) + posts = self._parse_posts(html) + if first_page and not posts and _POST_OPEN not in html: + # Page 1 with no recognizable post wrappers at all: either a brand + # new creator with zero posts, or our scraper is stale. The core's + # body canary catches systematic emptiness across a populated feed; + # here we only raise if the feed container itself is missing. + if 'data-role="posts_container-list"' not in html: + raise SubscribeStarDriftError( + f"SubscribeStar feed for {slug!r} had no posts and no " + "recognizable feed container — markup changed?" + ) + for post in posts: + post["_base"] = base + yield post, {}, page_cursor + next_href = self._next_page_href(html) + if not next_href: + return + current = next_href + first_page = False + + # -- verify ------------------------------------------------------------ + + def verify_auth(self, campaign_id: str) -> tuple[bool | None, str]: + """Cheap auth probe: fetch the first feed page and report whether the + credential authenticated, without downloading anything.""" + base, slug = _split_creator_url(campaign_id) + if not slug: + return None, f"Couldn't parse a SubscribeStar creator from {campaign_id!r}" + try: + html = self._feed_html(f"{base}/{slug}") + except SubscribeStarAuthError as exc: + return False, f"SubscribeStar rejected the credential — {exc}" + except SubscribeStarAPIError as exc: + return None, f"Couldn't verify (network/HTTP issue): {exc}" + if _POST_OPEN in html or 'data-role="posts_container-list"' in html: + return True, "Credentials valid — the SubscribeStar feed loaded." + return None, "Couldn't verify — SubscribeStar feed shape unrecognized." diff --git a/backend/app/services/subscribestar_downloader.py b/backend/app/services/subscribestar_downloader.py new file mode 100644 index 0000000..94982e0 --- /dev/null +++ b/backend/app/services/subscribestar_downloader.py @@ -0,0 +1,317 @@ +"""Native SubscribeStar media downloader — the SubscribeStar counterpart to +patreon_downloader. + +Given a SubscribeStar post and its resolved `MediaItem`s +(subscribestar_client.extract_media), download the media to gallery-dl's on-disk +layout (so existing gallery-dl downloads are recognized on disk and not +re-fetched at cutover), write the post-first sidecars the importer consumes, and +report per-media outcomes. + +Simpler than the Patreon downloader: SubscribeStar serves every upload as a +direct file via `/post_uploads?payload=...` (plain GET — no Mux/HLS, no yt-dlp), +and the full post body is already present in the feed HTML (no detail-endpoint +enrichment). PURE: no DB; the seen-skip is an injected predicate. + +On-disk layout (matches gallery-dl's subscribestar config +`{date:%Y-%m-%d}_{id}_{title[:40]}` / `{num:>02}_{filename}`): SubscribeStar posts +have no title, so the directory is `__` — the display title is +synthesized by the importer from the body, so the bare dir name is on-disk only. + +FC runs on a plain-HTTP homelab; nothing here uses a secure-context Web API. +""" + +from __future__ import annotations + +import contextlib +import json +import logging +import os +import time +from collections.abc import Callable +from datetime import datetime +from pathlib import Path + +import requests + +from .file_validator import is_validatable, quarantine_file, validate_file +from .patreon_downloader import ( + MediaOutcome, + PostRecordOutcome, + _sanitize, +) +from .subscribestar_client import _MAX_429_RETRIES, _load_session, _retry_after_seconds + +log = logging.getLogger(__name__) + +_TITLE_MAX = 40 +_TIMEOUT_SECONDS = 120.0 +_CHUNK = 1 << 16 +_MAX_MEDIA_RETRIES = 3 +_BACKOFF_CAP_SECONDS = 30.0 +_TRANSIENT_TRANSPORT_EXC = ( + requests.ConnectionError, + requests.Timeout, + requests.exceptions.ChunkedEncodingError, +) + + +def _post_dir_name(post: dict) -> str: + """`__` matching gallery-dl's subscribestar + layout (title is empty for SubscribeStar → trailing underscore). Date prefix + omitted when published_at is missing/unparseable.""" + 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: + try: + date_prefix = f"{datetime.fromisoformat(published):%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(raw) + + +class SubscribeStarDownloader: + """Download resolved SubscribeStar media to gallery-dl's on-disk layout. PURE: + no DB. The HTTP session is an injectable seam (pass `session=` or monkeypatch + `_fetch_to_file`) so tests run without network.""" + + def __init__( + self, + images_root: Path, + cookies_path: str | None = None, + *, + validate: bool = True, + rate_limit: float = 0.0, + session: requests.Session | None = None, + max_retries: int = _MAX_429_RETRIES, + ): + self.images_root = Path(images_root) + self.cookies_path = str(cookies_path) if cookies_path else None + self._validate = validate + self._rate_limit = rate_limit or 0.0 + self._max_retries = max_retries + self.session = session if session is not None else _load_session(cookies_path) + + # -- public ------------------------------------------------------------ + + def download_post( + self, + post: dict, + media_items: list, + artist_slug: str, + *, + is_seen: Callable[[object], bool] = lambda m: False, + should_stop: Callable[[], bool] = lambda: False, + recapture: bool = False, + ) -> list[MediaOutcome]: + """Download every media item of one post; return per-item outcomes. + Mirrors PatreonDownloader.download_post (two-tier skip, mid-post time-box, + recapture surfacing) minus the video branch.""" + post_dir = self.images_root / artist_slug / "subscribestar" / _post_dir_name(post) + outcomes: list[MediaOutcome] = [] + for i, media in enumerate(media_items, start=1): + if should_stop(): + break + try: + outcomes.append( + self._download_one( + post, media, post_dir, artist_slug, i, is_seen, + recapture=recapture, + ) + ) + except Exception as exc: # resilient: isolate one item's failure + log.warning( + "SubscribeStar media failed (post %s, item %d): %s", + post.get("id"), i, exc, + ) + outcomes.append( + MediaOutcome(media=media, status="error", path=None, error=str(exc)) + ) + return outcomes + + # -- per-item ---------------------------------------------------------- + + def _download_one( + self, + post: dict, + media, + post_dir: Path, + artist_slug: str, + index: int, + is_seen: Callable[[object], 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) + + nn = f"{index:02d}" + media_path = post_dir / _sanitize(f"{nn}_{media.filename}") + + if media_path.exists(): # tier-2: already on disk + return MediaOutcome( + media=media, status="skipped_disk", path=media_path, error=None + ) + # recapture: a seen item not on disk is NOT re-downloaded (recovery's job). + if seen: + return MediaOutcome(media=media, status="skipped_seen", path=None, error=None) + + post_dir.mkdir(parents=True, exist_ok=True) + if self._rate_limit > 0: + time.sleep(self._rate_limit) + + out_path = self._fetch_get(media.url, media_path) + reason, quarantine_dest = self._validate_path(out_path, artist_slug, media.url) + if reason is not None: + return MediaOutcome( + media=media, status="quarantined", path=quarantine_dest, error=reason, + ) + self._write_sidecar(post, out_path, source_url=media.url) + return MediaOutcome(media=media, status="downloaded", path=out_path, error=None) + + # -- download seams ---------------------------------------------------- + + def _fetch_get(self, url: str, dest: Path) -> Path: + """Stream `url` to a .part file 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, 5xx) with backoff + resume-from-disk (Range), failing fast on + permanent 4xx. Mirrors PatreonDownloader._fetch_to_file.""" + 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( + "SubscribeStar media transient HTTP %d (%s) — backing off " + "%.1fs (retry %d/%d)", + resp.status_code, url, delay, attempt, _MAX_MEDIA_RETRIES, + ) + time.sleep(delay) + continue + if have > 0 and resp.status_code == 416: + return # Range past EOF — we already have the whole file + 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( + "SubscribeStar media transport error (%s) — backing off %.1fs " + "(retry %d/%d): %s", + 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. Mirrors the Patreon + path (shared file_validator).""" + 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, "subscribestar", + url=source_url, result=result, + ) + return (result.reason or "validation failed"), (dest or path) + + # -- sidecar ----------------------------------------------------------- + + def _write_sidecar( + self, post: dict, media_path: Path, *, source_url: str | None = None + ) -> Path: + """Per-media sidecar — post-first (#856): image identity ONLY + (category/id/source_url). The post body/links live solely in _post.json.""" + return self._write_sidecar_data( + post, media_path.with_suffix(".json"), source_url=source_url, minimal=True, + ) + + def _write_sidecar_data( + self, post: dict, sidecar_path: Path, *, source_url: str | None = None, + minimal: bool = False, + ) -> Path: + """Serialize the post's metadata. minimal=True → per-media sidecar (image + identity only); else the full post record (body/title/date/url).""" + if minimal: + data = {"category": "subscribestar", "id": str(post.get("id") or "")} + if source_url: + data["source_url"] = source_url + sidecar_path.write_text(json.dumps(data, indent=2)) + return sidecar_path + attrs = post.get("attributes") or {} + content = attrs.get("content") + # SubscribeStar synthesizes the post permalink from the id (matches the + # platforms/subscribestar.py derive_post_url helper). + pid = str(post.get("id") or "") + data = { + "category": "subscribestar", + "id": pid, + "post_id": pid, # ensures derive_post_url has its key on the native path + "title": attrs.get("title") if isinstance(attrs.get("title"), str) else "", + "content": content if isinstance(content, str) else "", + "published_at": attrs.get("published_at"), + } + if source_url: + data["source_url"] = source_url + sidecar_path.write_text(json.dumps(data, indent=2)) + return sidecar_path + + def write_post_record(self, post: dict, artist_slug: str) -> PostRecordOutcome: + """Write the post-first `_post.json` (body/links/metadata) — the sole + writer of the post record on the native path. SubscribeStar's body is + already in the feed HTML, so no detail-fetch is needed.""" + attrs = post.get("attributes") or {} + title = attrs.get("title") if isinstance(attrs.get("title"), str) else None + post_type = attrs.get("post_type") if isinstance(attrs.get("post_type"), str) else None + pid = str(post.get("id") or "") + if not pid: + return PostRecordOutcome( + path=None, post_type=post_type, title=title, body_chars=0, + ) + post_dir = self.images_root / artist_slug / "subscribestar" / _post_dir_name(post) + post_dir.mkdir(parents=True, exist_ok=True) + path = self._write_sidecar_data(post, post_dir / "_post.json") + body = attrs.get("content") + body_chars = len(body) if isinstance(body, str) else 0 + return PostRecordOutcome( + path=path, post_type=post_type, title=title, body_chars=body_chars, + ) diff --git a/backend/app/services/subscribestar_ingester.py b/backend/app/services/subscribestar_ingester.py new file mode 100644 index 0000000..e28d1d0 --- /dev/null +++ b/backend/app/services/subscribestar_ingester.py @@ -0,0 +1,145 @@ +"""Native SubscribeStar ingester — the SubscribeStar ADAPTER over the +platform-agnostic core (`ingest_core.Ingester`). + +Thin counterpart to patreon_ingester: wires the SubscribeStar client/downloader/ +ledger models/constraints/key into the core and supplies the SubscribeStar +failure mapping. The three modes (tick / backfill / recovery / recapture), the +seen + dead-letter ledgers, cursor checkpointing, and the post-first capture all +live in the core — identical to Patreon. `download_service.download_source` +drives `SubscribeStarIngester.run` exactly as it drives the Patreon one. + +`campaign_id` is the creator URL (the client derives host + slug from it), so no +campaign-id resolver is needed. FC runs on a plain-HTTP homelab; nothing here +uses a secure-context Web API. +""" + +from __future__ import annotations + +import asyncio +import logging +from collections.abc import Callable +from pathlib import Path + +from ..models import SubscribeStarFailedMedia, SubscribeStarSeenMedia +from .gallery_dl import DownloadResult, ErrorType +from .ingest_core import DEAD_LETTER_THRESHOLD, Ingester +from .subscribestar_client import ( + MediaItem, + SubscribeStarAPIError, + SubscribeStarAuthError, + SubscribeStarClient, + SubscribeStarDriftError, +) +from .subscribestar_downloader import SubscribeStarDownloader + +__all__ = [ + "DEAD_LETTER_THRESHOLD", + "SubscribeStarIngester", + "_ledger_key", + "verify_subscribestar_credential", +] + +log = logging.getLogger(__name__) + +_LEDGER_KEY_MAX = 128 + + +def _ledger_key(media: MediaItem) -> str: + """Stable per-media identity for the cross-run seen-ledger. SubscribeStar's + full-res URL is an opaque `/post_uploads?payload=...` (no content hash), so + `media.filehash` is normally None and the stable proxy is the gallery item id + scoped to its post: `:`. Bounded to the column width.""" + if media.filehash: + return media.filehash + return f"{media.post_id}:{media.media_id}"[:_LEDGER_KEY_MAX] + + +class SubscribeStarIngester(Ingester): + """Walk a SubscribeStar creator's posts, download unseen media, return a + `DownloadResult`. A thin adapter over `ingest_core.Ingester`; `client` / + `downloader` are injectable seams so unit tests run without network.""" + + def __init__( + self, + images_root: Path, + cookies_path: str | None, + session_factory: Callable[[], object], + *, + validate: bool = True, + rate_limit: float = 0.0, + request_sleep: float = 0.0, + client: SubscribeStarClient | None = None, + downloader: SubscribeStarDownloader | None = None, + ): + self.images_root = Path(images_root) + self.cookies_path = str(cookies_path) if cookies_path else None + resolved_client = ( + client + if client is not None + else SubscribeStarClient(cookies_path, request_sleep=request_sleep) + ) + resolved_downloader = ( + downloader + if downloader is not None + else SubscribeStarDownloader( + self.images_root, cookies_path, validate=validate, rate_limit=rate_limit, + ) + ) + super().__init__( + client=resolved_client, + downloader=resolved_downloader, + session_factory=session_factory, + seen_model=SubscribeStarSeenMedia, + failed_model=SubscribeStarFailedMedia, + seen_constraint="uq_subscribestar_seen_media_source_id", + failed_constraint="uq_subscribestar_failed_media_source_id", + ledger_key=_ledger_key, + platform="subscribestar", + error_base=SubscribeStarAPIError, + ) + + # -- failure mapping (SubscribeStar exception taxonomy) ---------------- + + def _failure_result(self, exc: Exception, _result) -> DownloadResult: + """Map a client-level exception to a loud, typed failed DownloadResult — + never a silent zero-download "success". SubscribeStarAuthError and + SubscribeStarDriftError both subclass SubscribeStarAPIError, so match them + before the generic HTTP/transport fallthrough.""" + message = str(exc) + if isinstance(exc, SubscribeStarAuthError): + error_type = ErrorType.AUTH_ERROR + elif isinstance(exc, SubscribeStarDriftError): + error_type = ErrorType.API_DRIFT + message = f"SubscribeStar markup changed — scraper needs update: {message}" + else: + status = getattr(exc, "status_code", None) + if status == 429: + error_type = ErrorType.RATE_LIMITED + elif status == 404: + error_type = ErrorType.NOT_FOUND + elif status is not None: + error_type = ErrorType.HTTP_ERROR + else: + error_type = ErrorType.NETWORK_ERROR + log.warning("SubscribeStar ingest failed (%s): %s", error_type.value, message) + result = _result( + success=False, return_code=1, + error_type=error_type, error_message=message, + ) + if error_type == ErrorType.RATE_LIMITED: + result.retry_after_seconds = getattr(exc, "retry_after", None) + return result + + +async def verify_subscribestar_credential( + url: str, + cookies_path: str | None, + overrides: dict | None, +) -> tuple[bool | None, str]: + """Native SubscribeStar credential probe — fetches ONE feed page via + SubscribeStarClient.verify_auth. `campaign_id` is just the creator URL (no + resolver). Returns the uniform `(ok, message)` contract so + download_backends.verify_credential treats it like the gallery-dl probe.""" + client = SubscribeStarClient(cookies_path) + loop = asyncio.get_running_loop() + return await loop.run_in_executor(None, client.verify_auth, url) diff --git a/tests/test_subscribestar_native.py b/tests/test_subscribestar_native.py new file mode 100644 index 0000000..5bc0e24 --- /dev/null +++ b/tests/test_subscribestar_native.py @@ -0,0 +1,294 @@ +"""Unit tests for the native SubscribeStar client / downloader / ingester. + +No network, no DB (PURE modules → fast lane, unmarked). The HTML feed is fed via +monkeypatched `_feed_html` / `_loadmore_html`; the media HTTP layer via a fake +`session` seam. Canned HTML mirrors the real markup characterized in the Step-0 +spike (Scribe note "SubscribeStar HTML characterization"). +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import requests + +from backend.app.services.gallery_dl import DownloadResult, ErrorType +from backend.app.services.subscribestar_client import ( + MediaItem, + SubscribeStarAPIError, + SubscribeStarAuthError, + SubscribeStarClient, + SubscribeStarDriftError, + _parse_ss_datetime, +) +from backend.app.services.subscribestar_downloader import ( + SubscribeStarDownloader, + _post_dir_name, +) +from backend.app.services.subscribestar_ingester import ( + SubscribeStarIngester, + _ledger_key, +) +from backend.app.utils.sidecar import find_sidecar + +_PNG_HEAD = b"\x89PNG\r\n\x1a\n" +_PNG_BYTES = _PNG_HEAD + (b"\x00" * 16) + b"\x00\x00\x00\x00IEND\xaeB`\x82" + + +def _gallery_attr(items: list[dict]) -> str: + # data-gallery is HTML-entity-escaped JSON. + return json.dumps(items).replace("&", "&").replace('"', """) + + +def _post_html(post_id="111", date="May 01, 2026 12:00 pm", body="

hello

", + media: list[dict] | None = None, locked=False): + gallery = "" + if media is not None: + gallery = ( + f'
' + ) + lock = ' for-locked_content' if locked else "" + return ( + f'
' + f'
{date}
' + f'
' + f'
' + f'
{body}
' + f'{gallery}
' + ) + + +def _feed_page(posts_html: str, next_href: str | None = None): + nxt = ( + f'
' + if next_href else "" + ) + return ( + '
' + f'{posts_html}{nxt}
' + ) + + +# -- client: dates ----------------------------------------------------------- + +def test_parse_ss_datetime_variants(): + assert _parse_ss_datetime("Jun 17, 2026 03:19 am") == "2026-06-17T03:19:00" + assert _parse_ss_datetime("Updated on Jul 11, 2025 02:39 pm") == "2025-07-11T14:39:00" + assert _parse_ss_datetime("nonsense") is None + + +# -- client: iteration + pagination ----------------------------------------- + +def test_iter_posts_parses_and_paginates(monkeypatch): + client = SubscribeStarClient(None) + page1 = _feed_page( + _post_html("111") + _post_html("112"), + next_href="/posts?page=2&slug=x&sort_by=newest", + ) + page2 = _feed_page(_post_html("113")) # no next link → stop + monkeypatch.setattr(client, "_feed_html", lambda url: page1) + monkeypatch.setattr(client, "_loadmore_html", lambda url: page2) + + out = list(client.iter_posts("https://subscribestar.adult/x")) + ids = [p["id"] for p, _inc, _cur in out] + assert ids == ["111", "112", "113"] + # page_cursor is the value that FETCHED each post's page: None for page 1, the + # next_page href for page 2. + assert out[0][2] is None + assert out[2][2] == "/posts?page=2&slug=x&sort_by=newest" + # base is stamped for media URL joining. + assert out[0][0]["_base"] == "https://subscribestar.adult" + + +def test_iter_posts_drift_when_no_container(monkeypatch): + client = SubscribeStarClient(None) + monkeypatch.setattr(client, "_feed_html", lambda url: "nothing") + try: + list(client.iter_posts("https://subscribestar.adult/x")) + except SubscribeStarDriftError: + return + raise AssertionError("expected SubscribeStarDriftError") + + +# -- client: media extraction ------------------------------------------------ + +def test_extract_media_from_gallery(): + client = SubscribeStarClient(None) + media = [ + {"id": 9001, "type": "image", "original_filename": "art.jpg", + "url": "/post_uploads?payload=ABC"}, + {"id": 9002, "type": "image", "original_filename": "art2.jpg", + "url": "/post_uploads?payload=DEF"}, + ] + [post] = client._parse_posts(_feed_page(_post_html("111", media=media))) + post["_base"] = "https://subscribestar.adult" + items = client.extract_media(post, {}) + assert [m.media_id for m in items] == ["9001", "9002"] + assert items[0].url == "https://subscribestar.adult/post_uploads?payload=ABC" + assert items[0].filename == "art.jpg" + assert items[0].post_id == "111" + + +def test_text_post_has_no_media_but_keeps_body(): + client = SubscribeStarClient(None) + [post] = client._parse_posts(_feed_page(_post_html("111", body="

text only

"))) + post["_base"] = "https://subscribestar.adult" + assert client.extract_media(post, {}) == [] + assert "text only" in post["attributes"]["content"] + + +# -- client: gating + record key -------------------------------------------- + +def test_post_is_gated(): + client = SubscribeStarClient(None) + [open_post] = client._parse_posts(_feed_page(_post_html("1", media=[]))) + [locked] = client._parse_posts(_feed_page(_post_html("2", locked=True))) + assert client.post_is_gated(open_post) is False + assert client.post_is_gated(locked) is True + + +def test_post_record_key(): + assert SubscribeStarClient.post_record_key({"id": "55"}) == ("post:55", "55") + assert SubscribeStarClient.post_record_key({}) is None + + +# -- downloader -------------------------------------------------------------- + +class _FakeResponse: + def __init__(self, payload=_PNG_BYTES, status_code=200): + self._payload = payload + self.status_code = status_code + self.headers: dict = {} + + def raise_for_status(self): + if self.status_code >= 400: + raise requests.HTTPError(f"HTTP {self.status_code}") + + def iter_content(self, chunk_size=65536): + for i in range(0, len(self._payload), chunk_size): + yield self._payload[i:i + chunk_size] + + +class _FakeSession: + def __init__(self): + self.calls: list[str] = [] + + def get(self, url, stream=False, timeout=None, headers=None): + self.calls.append(url) + return _FakeResponse() + + +def _post(post_id="111", date="May 01, 2026 12:00 pm"): + return { + "id": post_id, + "attributes": { + "title": "", + "content": "
body text
", + "published_at": _parse_ss_datetime(date), + "post_type": "subscribestar", + }, + } + + +def _img(name, post_id="111", media_id="9001"): + return MediaItem( + url=f"https://subscribestar.adult/post_uploads?payload={name}", + filename=name, kind="image", filehash=None, + post_id=post_id, media_id=media_id, + ) + + +def _downloader(tmp_path: Path, session=None, validate=True): + return SubscribeStarDownloader( + images_root=tmp_path, cookies_path=None, validate=validate, + session=session or _FakeSession(), + ) + + +def test_post_dir_name_empty_title_matches_gallery_dl(): + # SubscribeStar has no title → "__" (matches gallery-dl's layout so + # existing downloads dedup on disk at cutover). + assert _post_dir_name(_post("111")) == "2026-05-01_111_" + + +def test_download_post_layout_and_outcomes(tmp_path): + # .png names so the PNG bytes pass the content/extension validator. + dl = _downloader(tmp_path) + outcomes = dl.download_post(_post(), [_img("a.png"), _img("b.png", media_id="9002")], "artist-x") + post_dir = tmp_path / "artist-x" / "subscribestar" / "2026-05-01_111_" + assert (post_dir / "01_a.png").is_file() + assert (post_dir / "02_b.png").is_file() + assert [o.status for o in outcomes] == ["downloaded", "downloaded"] + + +def test_per_media_sidecar_is_minimal(tmp_path): + dl = _downloader(tmp_path) + outcomes = dl.download_post(_post(), [_img("a.png")], "artist-x") + sidecar = find_sidecar(outcomes[0].path) + data = json.loads(sidecar.read_text()) + assert data["category"] == "subscribestar" + assert data["id"] == "111" + assert data["source_url"].endswith("payload=a.png") + # post-first: no body/title in the per-media sidecar. + assert "content" not in data and "title" not in data + + +def test_write_post_record(tmp_path): + dl = _downloader(tmp_path) + rec = dl.write_post_record(_post("111"), "artist-x") + assert rec.path.name == "_post.json" + data = json.loads(rec.path.read_text()) + assert data["category"] == "subscribestar" + assert data["id"] == "111" and data["post_id"] == "111" + assert "body text" in data["content"] + assert rec.body_chars > 0 + + +def test_skip_seen_does_not_download(tmp_path): + session = _FakeSession() + dl = _downloader(tmp_path, session=session) + outcomes = dl.download_post( + _post(), [_img("a.jpg")], "artist-x", is_seen=lambda m: True, + ) + assert [o.status for o in outcomes] == ["skipped_seen"] + assert session.calls == [] + + +# -- ingester ---------------------------------------------------------------- + +def test_ledger_key_prefers_filehash_then_post_media(): + assert _ledger_key(_img("a.jpg")) == "111:9001" # no filehash → post:media + hashed = MediaItem( + url="u", filename="a.jpg", kind="image", filehash="deadbeef", + post_id="111", media_id="9001", + ) + assert _ledger_key(hashed) == "deadbeef" + + +def _ingester(tmp_path): + return SubscribeStarIngester( + images_root=tmp_path, cookies_path=None, session_factory=lambda: None, + ) + + +def _mk_result(**kw) -> DownloadResult: + # _failure_result supplies success/return_code/error_type/error_message in kw; + # the factory only injects the always-required identity fields. + return DownloadResult(url="u", artist_slug="a", platform="subscribestar", **kw) + + +def test_failure_result_maps_exceptions(tmp_path): + ing = _ingester(tmp_path) + assert ing._failure_result(SubscribeStarAuthError("x"), _mk_result).error_type == ErrorType.AUTH_ERROR + assert ing._failure_result(SubscribeStarDriftError("x"), _mk_result).error_type == ErrorType.API_DRIFT + assert ing._failure_result( + SubscribeStarAPIError("x", status_code=429), _mk_result + ).error_type == ErrorType.RATE_LIMITED + assert ing._failure_result( + SubscribeStarAPIError("x", status_code=404), _mk_result + ).error_type == ErrorType.NOT_FOUND + assert ing._failure_result( + SubscribeStarAPIError("x"), _mk_result + ).error_type == ErrorType.NETWORK_ERROR From 7ac5c7e522ba38cddc823c57f07955d52ed62f2c Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 17 Jun 2026 11:32:28 -0400 Subject: [PATCH 3/7] refactor(native-ingest): extract native_ingest_common + BaseNativeDownloader (#899 DRY 1/3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DRY pass commit 1 (process #594). Consolidate the helpers + download plumbing the Patreon and SubscribeStar adapters had duplicated (SubscribeStar was importing patreon privates — wrong owner). New backend/app/services/ native_ingest_common.py is the neutral home for: - make_session (was _load_session ×2), retry_after_seconds + 429 constants, sanitize_segment, basename_from_url, post_dir_name, MediaOutcome / PostRecordOutcome. - BaseNativeDownloader: the shared streaming GET (transient-retry + Range-resume) and validation/quarantine. Patreon + SubscribeStar downloaders now subclass it; each keeps only what differs (Patreon's Mux/yt-dlp video branch + detail-fetch enrichment; SubscribeStar nothing extra). Behavior preserved exactly; the divergence-bug risk (a fix to one _fetch_to_file not reaching the other) is gone. - Folds in #899 L2: a quarantine now log.warning's path+reason (was counted only). post_dir_name merges both date handlers (accepts trailing-Z and pre-parsed ISO). Tests repointed to the single source at every consumer (rule 93 / §8b parity): patreon_client/downloader, subscribestar_native. Exception-trio consolidation + base _failure_result (2/3) and the remaining ingest_core logging (3/3) follow. Co-Authored-By: Claude Opus 4.8 --- backend/app/services/native_ingest_common.py | 310 ++++++++++++++++++ backend/app/services/patreon_client.py | 91 +---- backend/app/services/patreon_downloader.py | 258 ++------------- backend/app/services/subscribestar_client.py | 57 +--- .../app/services/subscribestar_downloader.py | 149 ++------- tests/test_patreon_client.py | 14 +- tests/test_patreon_downloader.py | 19 +- tests/test_subscribestar_native.py | 8 +- 8 files changed, 402 insertions(+), 504 deletions(-) create mode 100644 backend/app/services/native_ingest_common.py diff --git a/backend/app/services/native_ingest_common.py b/backend/app/services/native_ingest_common.py new file mode 100644 index 0000000..4317e9a --- /dev/null +++ b/backend/app/services/native_ingest_common.py @@ -0,0 +1,310 @@ +"""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('<>:"/\\|?*') + + +# -- 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: + """`__` 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 + + +# -- 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=-`; 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 diff --git a/backend/app/services/patreon_client.py b/backend/app/services/patreon_client.py index 907ce73..b5264b7 100644 --- a/backend/app/services/patreon_client.py +++ b/backend/app/services/patreon_client.py @@ -26,9 +26,7 @@ FC runs on a plain-HTTP homelab; nothing here uses a secure-context Web API. from __future__ import annotations -import http.cookiejar import logging -import os import re import time from collections.abc import Iterator @@ -39,46 +37,20 @@ from urllib.parse import parse_qs, urlsplit import requests -from ..utils.paths import filehash_from_url, safe_ext +from ..utils.paths import filehash_from_url from ..utils.prosemirror import post_body_html +from .native_ingest_common import ( + _MAX_429_RETRIES, + basename_from_url, + make_session, + retry_after_seconds, +) log = logging.getLogger(__name__) _POSTS_URL = "https://www.patreon.com/api/posts" -_USER_AGENT = ( - "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " - "(KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36" -) _TIMEOUT_SECONDS = 30.0 -# 429 backoff (plan #703): ride out a transient API rate-limit instead of -# failing the whole walk (which would stamp RATE_LIMITED → platform-wide -# cooldown → every Patreon source dark). Honor the server's `Retry-After`; -# otherwise exponential base·2^(n-1), capped. Only after the retries are -# exhausted does the 429 propagate as terminal RATE_LIMITED. -_MAX_429_RETRIES = 3 -_BACKOFF_BASE_SECONDS = 2.0 -_BACKOFF_CAP_SECONDS = 30.0 - - -def _retry_after_seconds( - resp: requests.Response, - attempt: int, - *, - base: float = _BACKOFF_BASE_SECONDS, - cap: float = _BACKOFF_CAP_SECONDS, -) -> float: - """Backoff delay for a 429: the `Retry-After` seconds header if present and - numeric, else exponential `base·2^(attempt-1)`, both capped. (HTTP-date form - of Retry-After is rare here and falls through to exponential.)""" - 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) - # JSON:API request contract (observed from real traffic — see module plan). _INCLUDE = ( "campaign,access_rules,attachments,attachments_media,audio,images,media," @@ -168,49 +140,12 @@ class MediaItem: post_id: str -def _load_session(cookies_path: str | Path | None) -> requests.Session: - session = requests.Session() - session.headers.update( - { - "User-Agent": _USER_AGENT, - "Accept": "application/vnd.api+json", - } - ) - 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 Patreon cookies from %s: %s", cookies_path, exc) - return session - - 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 _basename_from_url(url: str) -> str: - """Derive a sane filename from a URL when the media has no file_name. - - Strips query/fragment, takes the path basename, and drops a junk - extension (the importer._safe_ext gotcha) so we never write base64 noise - as a name. Falls back to the filehash, then to "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 - # Keep the stem bounded; URL-encoded stems can be enormous. - stem = stem[:120] or "file" - return f"{stem}{ext}" - fh = _filehash(url) - return fh or "file" - - def parse_cursor_from_url(url: str | None) -> str | None: """Extract the `page[cursor]` query param from a links.next URL.""" if not url: @@ -238,7 +173,7 @@ class PatreonClient: max_retries: int = _MAX_429_RETRIES, ): self.cookies_path = str(cookies_path) if cookies_path else None - self._session = _load_session(cookies_path) + 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 @@ -283,7 +218,7 @@ class PatreonClient: # 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) + 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, @@ -416,7 +351,7 @@ class PatreonClient: # 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) + filename = file_name if isinstance(file_name, str) and file_name else basename_from_url(url) return MediaItem( url=url, filename=filename, @@ -462,7 +397,7 @@ class PatreonClient: items.append( MediaItem( url=large_url, - filename=_basename_from_url(large_url), + filename=basename_from_url(large_url), kind="image_large", filehash=_filehash(large_url), post_id=post_id, @@ -481,7 +416,7 @@ class PatreonClient: filename = ( pf_name if isinstance(pf_name, str) and pf_name - else _basename_from_url(pf_url) + else basename_from_url(pf_url) ) items.append( MediaItem( @@ -503,7 +438,7 @@ class PatreonClient: items.append( MediaItem( url=src, - filename=_basename_from_url(src), + filename=basename_from_url(src), kind="content", filehash=_filehash(src), post_id=post_id, diff --git a/backend/app/services/patreon_downloader.py b/backend/app/services/patreon_downloader.py index 2502d14..47132e7 100644 --- a/backend/app/services/patreon_downloader.py +++ b/backend/app/services/patreon_downloader.py @@ -27,46 +27,33 @@ FC runs on a plain-HTTP homelab; nothing here uses a secure-context Web API. from __future__ import annotations -import contextlib import json import logging import os import subprocess 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.prosemirror import post_body_html -from .file_validator import is_validatable, quarantine_file, validate_file -from .patreon_client import ( +from .native_ingest_common import ( _BACKOFF_CAP_SECONDS, - _load_session, - _retry_after_seconds, + _MAX_MEDIA_RETRIES, + BaseNativeDownloader, + MediaOutcome, + PostRecordOutcome, + post_dir_name, + sanitize_segment, ) log = logging.getLogger(__name__) -_TITLE_MAX = 40 +# yt-dlp subprocess wall-clock per attempt (video only; the shared HTTP fetch +# budgets live in BaseNativeDownloader). _TIMEOUT_SECONDS = 120.0 -_CHUNK = 1 << 16 -# Retry a media GET that hits a TRANSIENT failure within the same pass (plan -# #705 #8): a transport blip (connection reset / timeout / truncated stream), a -# 429, or a 5xx. PERMANENT failures (404 gone, 403 forbidden) fail fast straight -# to the error/dead-letter path — no point re-fetching them. Keeps a momentary -# network hiccup from becoming a per-item error that waits for the next walk. -_MAX_MEDIA_RETRIES = 3 -# requests transport errors worth retrying (vs. an HTTPError, which is a real -# server response and is classified by status code). -_TRANSIENT_TRANSPORT_EXC = ( - requests.ConnectionError, - requests.Timeout, - requests.exceptions.ChunkedEncodingError, -) # Referer/Origin yt-dlp must send for Mux-hosted Patreon video. Mux's JWT # playback policy checks Referer/Origin on every request, so yt-dlp must send @@ -78,26 +65,6 @@ _VIDEO_HEADERS = { "Origin": "https://www.patreon.com", } -# Characters Windows/gallery-dl path-restrict forbids, plus path separators. -_FORBIDDEN = set('<>:"/\\|?*') - - -def _sanitize(name: str) -> str: - """Make `name` safe for a single filesystem path segment. - - Replaces path separators, the Windows-forbidden set <>:"/\\|?* and control - characters with `_`, then strips trailing dots/spaces (gallery-dl - path-restrict behavior). Never returns empty (falls back to "_"). - """ - out = [] - for ch in name: - if ch in _FORBIDDEN or ord(ch) < 32: - out.append("_") - else: - out.append(ch) - cleaned = "".join(out).rstrip(". ") - return cleaned or "_" - def _is_video_url(url: str) -> bool: parts = urlsplit(url) @@ -106,73 +73,12 @@ def _is_video_url(url: str) -> bool: return parts.path.lower().endswith(".m3u8") -def _post_dir_name(post: dict) -> str: - """Build the post directory name matching gallery-dl's layout.""" - post_id = str(post.get("id") or "") - attrs = post.get("attributes") or {} - title = attrs.get("title") - title = title if isinstance(title, str) else "" - title40 = title[:_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: - dt = datetime.fromisoformat(s) - except ValueError: - dt = None - if dt is not None: - date_prefix = f"{dt:%Y-%m-%d}" - - if date_prefix: - raw = f"{date_prefix}_{post_id}_{title40}" - else: - raw = f"{post_id}_{title40}" - return _sanitize(raw) - - -@dataclass -class MediaOutcome: - """Per-media result of a download_post pass. - - status is one of: "downloaded", "skipped_seen", "skipped_disk", - "quarantined", "error". `path` is the final on-disk path for "downloaded" - (the actual yt-dlp output for video), the path that already existed for - "skipped_disk", or the _quarantine destination for "quarantined"; None for - "skipped_seen" and (usually) "error". `error` carries the failure/validation - reason for "error"/"quarantined", else None. - """ - - media: object # MediaItem (avoid importing the name for a bare annotation) - status: str - path: Path | None - error: str | None - - -@dataclass -class PostRecordOutcome: - """Result of write_post_record — mirrors the download_post → MediaOutcome - contract so the engine reports per-post handling without re-reading the post. - `path` is the _post.json sidecar (None when the post had no id); the rest is - the captured body's shape (post_type + final char count) for the run log. - """ - - path: Path | None - post_type: str | None - title: str | None - body_chars: int - - -class PatreonDownloader: - """Download resolved Patreon media to gallery-dl's on-disk layout. - - PURE: no DB. The HTTP session and the yt-dlp invocation are injectable seams - so tests run without network or a real subprocess: - - pass `session=` to stub `session.get`, or monkeypatch `_fetch_to_file`. - - monkeypatch `_run_ytdlp` to avoid spawning yt-dlp. +class PatreonDownloader(BaseNativeDownloader): + """Download resolved Patreon media to gallery-dl's on-disk layout. Subclasses + BaseNativeDownloader for the shared streaming GET (transient-retry + + Range-resume) and validation/quarantine; adds the Mux/HLS yt-dlp video branch + and the detail-fetch body enrichment. PURE: no DB. `_run_ytdlp` is + monkeypatchable and the HTTP session is the injectable `session=` seam. """ def __init__( @@ -185,23 +91,16 @@ class PatreonDownloader: session: requests.Session | None = None, content_fetcher: Callable[[str], str | None] | None = None, ): - self.images_root = Path(images_root) - self.cookies_path = str(cookies_path) if cookies_path else None - self._validate = validate + super().__init__( + images_root, cookies_path, platform="patreon", + validate=validate, rate_limit=rate_limit, session=session, + ) # Best-effort enrichment seam: (post_id) -> full HTML body, or None. The # feed endpoint often omits `content`; the adapter wires this to # PatreonClient.fetch_post_detail_content so the sidecar captures the # real body (formatting + inline + external links). # None in unit tests / when enrichment isn't wanted. self._content_fetcher = content_fetcher - # Politeness: seconds to sleep before each actual media download (paces - # the CDN; honors ImportSettings.download_rate_limit_seconds, the same - # value gallery-dl used as its between-downloads `sleep`). 0 = no pacing. - # Applied only to real downloads, not to seen/disk skips. plan #703. - self._rate_limit = rate_limit or 0.0 - # Build a cookie-loaded session the same way patreon_client does, so the - # CDN GETs carry the creator's auth. - self.session = session if session is not None else _load_session(cookies_path) # -- public ------------------------------------------------------------ @@ -234,7 +133,7 @@ class PatreonDownloader: would tier-1 skip it — so the engine can backfill source_filehash for inline-image localization. Genuinely-missing seen media is NOT refetched. """ - post_dir = self.images_root / artist_slug / "patreon" / _post_dir_name(post) + post_dir = self.images_root / artist_slug / "patreon" / post_dir_name(post) outcomes: list[MediaOutcome] = [] for i, media in enumerate(media_items, start=1): @@ -279,7 +178,7 @@ class PatreonDownloader: return MediaOutcome(media=media, status="skipped_seen", path=None, error=None) nn = f"{index:02d}" - final_name = _sanitize(f"{nn}_{media.filename}") + final_name = sanitize_segment(f"{nn}_{media.filename}") media_path = post_dir / final_name # tier-2: already on disk. @@ -333,87 +232,9 @@ class PatreonDownloader: self._write_sidecar(post, out_path, source_url=media.url) return MediaOutcome(media=media, status="downloaded", path=out_path, error=None) - # -- download seams ---------------------------------------------------- - - def _fetch_get(self, url: str, dest: Path) -> Path: - """Stream `url` to a .part file then atomic-rename to `dest`. - - Thin wrapper over `_fetch_to_file` so tests can stub either the whole - GET path (`_fetch_to_file`) or just `session.get`. - """ - 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 non-video URL to `dest` via the (stubbable) session, retrying - TRANSIENT failures within the same pass (plan #705 #8) and RESUMING from - the bytes already on disk via a Range request when a retry follows a - mid-download cut (plan #708 B5). - - Retried (backoff): transport blips (connection reset / timeout / - truncated stream — incl. mid-download), HTTP 429 (honoring Retry-After), - and 5xx. Failed fast (no retry → HTTPError → per-item error → dead-letter - path): 4xx other than 429 (404 gone, 403 forbidden) — re-fetching a - permanent failure is pointless. - - Resume: on a retry, if bytes already landed in `dest`, ask for the rest - with `Range: bytes=-`. A 206 means the server honored it → append; a - 200 means it ignored it (served the whole file) → start clean. The caller - (_fetch_get) stages into a `.part`, so a non-range server never corrupts - the output — the worst case is re-downloading from zero, as before. - """ - 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( - "Patreon media transient HTTP %d (%s) — backing off " - "%.1fs (retry %d/%d)", - resp.status_code, url, delay, attempt, _MAX_MEDIA_RETRIES, - ) - time.sleep(delay) - continue - # A Range that starts at/past EOF (we already have the whole file) - # comes back 416 — the bytes we kept ARE the file. - if have > 0 and resp.status_code == 416: - return - # 2xx → ok; 4xx-non-429 (or an exhausted 429/5xx) → HTTPError - # (permanent for this pass) → not caught below → per-item error. - resp.raise_for_status() - # 206 → server honored the Range; append after the kept bytes. - # Anything else (200) → it served the whole file → start clean. - 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 # exhausted → terminal error outcome - attempt += 1 - delay = min(2.0 * (2 ** (attempt - 1)), _BACKOFF_CAP_SECONDS) - log.warning( - "Patreon media transport error (%s) — backing off %.1fs " - "(retry %d/%d): %s", - url, delay, attempt, _MAX_MEDIA_RETRIES, exc, - ) - time.sleep(delay) + # -- video (Mux/HLS via yt-dlp) ---------------------------------------- + # The plain-GET streaming path (_fetch_get / _fetch_to_file) and + # _validate_path are inherited from BaseNativeDownloader. def _run_ytdlp(self, url: str, dest: Path, headers: dict) -> Path | None: """Invoke yt-dlp to fetch a Mux/HLS stream to (around) `dest`. @@ -495,35 +316,6 @@ class PatreonDownloader: return cand return None - # -- 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. - - Uses the shared `file_validator.quarantine_file` — same move + provenance - sidecar gallery-dl writes (the native path used to skip the sidecar; that - parity gap is closed here). Returns `(reason, quarantine_dest)` when - quarantined (dest is the original path if the move itself failed), else - `(None, None)` (ok / not validatable / disabled). plan #704: the dest is - surfaced so the run reports a real quarantined-paths list. - """ - 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, "patreon", - url=source_url, result=result, - ) - return (result.reason or "validation failed"), (dest or path) - # -- sidecar ----------------------------------------------------------- def _write_sidecar( @@ -614,7 +406,7 @@ class PatreonDownloader: return PostRecordOutcome( path=None, post_type=post_type, title=title, body_chars=0, ) - post_dir = self.images_root / artist_slug / "patreon" / _post_dir_name(post) + post_dir = self.images_root / artist_slug / "patreon" / post_dir_name(post) post_dir.mkdir(parents=True, exist_ok=True) path = self._write_sidecar_data(post, post_dir / "_post.json") # _write_sidecar_data has by now memoized any detail-fetched body onto diff --git a/backend/app/services/subscribestar_client.py b/backend/app/services/subscribestar_client.py index c4a6031..2565089 100644 --- a/backend/app/services/subscribestar_client.py +++ b/backend/app/services/subscribestar_client.py @@ -27,10 +27,8 @@ plain-HTTP homelab; nothing here uses a secure-context Web API. from __future__ import annotations -import http.cookiejar import json import logging -import os import re import time from collections.abc import Iterator @@ -42,16 +40,20 @@ from urllib.parse import urljoin, urlsplit import requests -from ..utils.paths import filehash_from_url, safe_ext -from .patreon_client import _MAX_429_RETRIES, _retry_after_seconds +from ..utils.paths import filehash_from_url +from .native_ingest_common import ( + _MAX_429_RETRIES, + basename_from_url, + make_session, + retry_after_seconds, +) 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" -) _TIMEOUT_SECONDS = 30.0 +# Accept header for the HTML feed + JSON "load more" + the XHR marker. +_ACCEPT = "text/html,application/xhtml+xml,application/json,*/*" +_XHR_HEADERS = {"X-Requested-With": "XMLHttpRequest"} # A post block opens with this wrapper; we slice the page between consecutive # occurrences (regex can't match the balanced close of nested divs, so chunk- @@ -122,26 +124,6 @@ class MediaItem: media_id: str -def _load_session(cookies_path: str | Path | None) -> requests.Session: - session = requests.Session() - session.headers.update( - { - "User-Agent": _USER_AGENT, - # HTML feed + JSON "load more"; let the server pick. - "Accept": "text/html,application/xhtml+xml,application/json,*/*", - "X-Requested-With": "XMLHttpRequest", - } - ) - 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 SubscribeStar cookies from %s: %s", cookies_path, exc) - return session - - def _split_creator_url(campaign_id: str) -> tuple[str, str]: """`campaign_id` is the creator URL → (base, slug). @@ -167,17 +149,6 @@ def _parse_ss_datetime(text: str) -> str | None: return None -def _basename_from_url(url: str) -> str: - 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 "file" - - class SubscribeStarClient: """Synchronous SubscribeStar HTML-scrape read client. Construct with a path to a Netscape cookies.txt (the same file CredentialService.get_cookies_path @@ -191,7 +162,9 @@ class SubscribeStarClient: max_retries: int = _MAX_429_RETRIES, ): self.cookies_path = str(cookies_path) if cookies_path else None - self._session = _load_session(cookies_path) + self._session = make_session( + cookies_path, accept=_ACCEPT, extra_headers=_XHR_HEADERS + ) self._request_sleep = request_sleep or 0.0 self._max_retries = max_retries @@ -210,7 +183,7 @@ class SubscribeStarClient: ) from exc if resp.status_code == 429 and attempt < self._max_retries: attempt += 1 - delay = _retry_after_seconds(resp, attempt) + delay = retry_after_seconds(resp, attempt) log.warning( "SubscribeStar 429 (%s) — backing off %.1fs (retry %d/%d)", url, delay, attempt, self._max_retries, @@ -341,7 +314,7 @@ class SubscribeStarClient: url = urljoin(base + "/", rel) media_id = str(it.get("id") or "") name = it.get("original_filename") - filename = name if isinstance(name, str) and name else _basename_from_url(url) + filename = name if isinstance(name, str) and name else basename_from_url(url) items.append( MediaItem( url=url, diff --git a/backend/app/services/subscribestar_downloader.py b/backend/app/services/subscribestar_downloader.py index 94982e0..9845a3b 100644 --- a/backend/app/services/subscribestar_downloader.py +++ b/backend/app/services/subscribestar_downloader.py @@ -22,62 +22,31 @@ FC runs on a plain-HTTP homelab; nothing here uses a secure-context Web API. from __future__ import annotations -import contextlib import json import logging -import os import time from collections.abc import Callable -from datetime import datetime from pathlib import Path import requests -from .file_validator import is_validatable, quarantine_file, validate_file -from .patreon_downloader import ( +from .native_ingest_common import ( + BaseNativeDownloader, MediaOutcome, PostRecordOutcome, - _sanitize, + post_dir_name, + sanitize_segment, ) -from .subscribestar_client import _MAX_429_RETRIES, _load_session, _retry_after_seconds log = logging.getLogger(__name__) -_TITLE_MAX = 40 -_TIMEOUT_SECONDS = 120.0 -_CHUNK = 1 << 16 -_MAX_MEDIA_RETRIES = 3 -_BACKOFF_CAP_SECONDS = 30.0 -_TRANSIENT_TRANSPORT_EXC = ( - requests.ConnectionError, - requests.Timeout, - requests.exceptions.ChunkedEncodingError, -) - -def _post_dir_name(post: dict) -> str: - """`__` matching gallery-dl's subscribestar - layout (title is empty for SubscribeStar → trailing underscore). Date prefix - omitted when published_at is missing/unparseable.""" - 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: - try: - date_prefix = f"{datetime.fromisoformat(published):%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(raw) - - -class SubscribeStarDownloader: - """Download resolved SubscribeStar media to gallery-dl's on-disk layout. PURE: - no DB. The HTTP session is an injectable seam (pass `session=` or monkeypatch - `_fetch_to_file`) so tests run without network.""" +class SubscribeStarDownloader(BaseNativeDownloader): + """Download resolved SubscribeStar media to gallery-dl's on-disk layout. + Subclasses BaseNativeDownloader for the shared streaming GET (transient-retry + + Range-resume) and validation/quarantine. No video branch (SubscribeStar serves + files directly via /post_uploads) and no detail-fetch (the body is already in + the feed HTML). PURE: no DB.""" def __init__( self, @@ -87,14 +56,11 @@ class SubscribeStarDownloader: validate: bool = True, rate_limit: float = 0.0, session: requests.Session | None = None, - max_retries: int = _MAX_429_RETRIES, ): - self.images_root = Path(images_root) - self.cookies_path = str(cookies_path) if cookies_path else None - self._validate = validate - self._rate_limit = rate_limit or 0.0 - self._max_retries = max_retries - self.session = session if session is not None else _load_session(cookies_path) + super().__init__( + images_root, cookies_path, platform="subscribestar", + validate=validate, rate_limit=rate_limit, session=session, + ) # -- public ------------------------------------------------------------ @@ -111,7 +77,7 @@ class SubscribeStarDownloader: """Download every media item of one post; return per-item outcomes. Mirrors PatreonDownloader.download_post (two-tier skip, mid-post time-box, recapture surfacing) minus the video branch.""" - post_dir = self.images_root / artist_slug / "subscribestar" / _post_dir_name(post) + post_dir = self.images_root / artist_slug / "subscribestar" / post_dir_name(post) outcomes: list[MediaOutcome] = [] for i, media in enumerate(media_items, start=1): if should_stop(): @@ -151,7 +117,7 @@ class SubscribeStarDownloader: return MediaOutcome(media=media, status="skipped_seen", path=None, error=None) nn = f"{index:02d}" - media_path = post_dir / _sanitize(f"{nn}_{media.filename}") + media_path = post_dir / sanitize_segment(f"{nn}_{media.filename}") if media_path.exists(): # tier-2: already on disk return MediaOutcome( @@ -174,85 +140,8 @@ class SubscribeStarDownloader: self._write_sidecar(post, out_path, source_url=media.url) return MediaOutcome(media=media, status="downloaded", path=out_path, error=None) - # -- download seams ---------------------------------------------------- - - def _fetch_get(self, url: str, dest: Path) -> Path: - """Stream `url` to a .part file 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, 5xx) with backoff + resume-from-disk (Range), failing fast on - permanent 4xx. Mirrors PatreonDownloader._fetch_to_file.""" - 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( - "SubscribeStar media transient HTTP %d (%s) — backing off " - "%.1fs (retry %d/%d)", - resp.status_code, url, delay, attempt, _MAX_MEDIA_RETRIES, - ) - time.sleep(delay) - continue - if have > 0 and resp.status_code == 416: - return # Range past EOF — we already have the whole file - 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( - "SubscribeStar media transport error (%s) — backing off %.1fs " - "(retry %d/%d): %s", - 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. Mirrors the Patreon - path (shared file_validator).""" - 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, "subscribestar", - url=source_url, result=result, - ) - return (result.reason or "validation failed"), (dest or path) + # The plain-GET streaming path (_fetch_get / _fetch_to_file) and + # _validate_path are inherited from BaseNativeDownloader. # -- sidecar ----------------------------------------------------------- @@ -307,7 +196,7 @@ class SubscribeStarDownloader: return PostRecordOutcome( path=None, post_type=post_type, title=title, body_chars=0, ) - post_dir = self.images_root / artist_slug / "subscribestar" / _post_dir_name(post) + post_dir = self.images_root / artist_slug / "subscribestar" / post_dir_name(post) post_dir.mkdir(parents=True, exist_ok=True) path = self._write_sidecar_data(post, post_dir / "_post.json") body = attrs.get("content") diff --git a/tests/test_patreon_client.py b/tests/test_patreon_client.py index 2106237..2f4a990 100644 --- a/tests/test_patreon_client.py +++ b/tests/test_patreon_client.py @@ -12,13 +12,13 @@ import pytest import requests from backend.app.services import patreon_client as pc_mod +from backend.app.services.native_ingest_common import retry_after_seconds from backend.app.services.patreon_client import ( MediaItem, PatreonAPIError, PatreonAuthError, PatreonClient, PatreonDriftError, - _retry_after_seconds, parse_cursor_from_url, ) @@ -288,19 +288,19 @@ def test_verify_auth_inconclusive_on_network(monkeypatch): def test_retry_after_seconds_honors_numeric_header(): - assert _retry_after_seconds(_FakeResp(429, headers={"Retry-After": "7"}), 1) == 7.0 + assert retry_after_seconds(_FakeResp(429, headers={"Retry-After": "7"}), 1) == 7.0 def test_retry_after_seconds_exponential_without_header(): resp = _FakeResp(429) - assert _retry_after_seconds(resp, 1) == 2.0 - assert _retry_after_seconds(resp, 2) == 4.0 - assert _retry_after_seconds(resp, 3) == 8.0 + assert retry_after_seconds(resp, 1) == 2.0 + assert retry_after_seconds(resp, 2) == 4.0 + assert retry_after_seconds(resp, 3) == 8.0 def test_retry_after_seconds_caps(): - assert _retry_after_seconds(_FakeResp(429), 10) == 30.0 - assert _retry_after_seconds(_FakeResp(429, headers={"Retry-After": "999"}), 1) == 30.0 + assert retry_after_seconds(_FakeResp(429), 10) == 30.0 + assert retry_after_seconds(_FakeResp(429, headers={"Retry-After": "999"}), 1) == 30.0 def _client_with_sequence(monkeypatch, responses): diff --git a/tests/test_patreon_downloader.py b/tests/test_patreon_downloader.py index 5162364..a26ffb5 100644 --- a/tests/test_patreon_downloader.py +++ b/tests/test_patreon_downloader.py @@ -14,13 +14,14 @@ import pytest import requests from backend.app.services import patreon_downloader as pd_mod -from backend.app.services.patreon_client import MediaItem -from backend.app.services.patreon_downloader import ( +from backend.app.services.native_ingest_common import ( MediaOutcome, - PatreonDownloader, PostRecordOutcome, - _sanitize, + post_dir_name, + sanitize_segment, ) +from backend.app.services.patreon_client import MediaItem +from backend.app.services.patreon_downloader import PatreonDownloader from backend.app.utils.sidecar import find_sidecar # Minimal valid PNG so the file_validator passes on the happy path. @@ -414,10 +415,10 @@ def test_validation_off_keeps_bytes(tmp_path): def test_sanitize_helper(): - assert _sanitize('a/b') == "a_b" - assert _sanitize('x<>:"|?*y') == "x_______y" - assert _sanitize("trail. ") == "trail" - assert _sanitize("") == "_" + assert sanitize_segment('a/b') == "a_b" + assert sanitize_segment('x<>:"|?*y') == "x_______y" + assert sanitize_segment("trail. ") == "trail" + assert sanitize_segment("") == "_" def test_media_outcome_shape(): @@ -613,7 +614,7 @@ def test_media_sidecar_is_minimal_post_first(tmp_path): post["attributes"]["content"] = "" # even an empty feed body isn't fetched here dl.download_post(post, [_img("a.png"), _img("b.png")], "artist-x") - post_dir = tmp_path / "artist-x" / "patreon" / pd_mod._post_dir_name(post) + post_dir = tmp_path / "artist-x" / "patreon" / post_dir_name(post) sc = find_sidecar(post_dir / "01_a.png") assert sc is not None data = json.loads(sc.read_text()) diff --git a/tests/test_subscribestar_native.py b/tests/test_subscribestar_native.py index 5bc0e24..c84cfd7 100644 --- a/tests/test_subscribestar_native.py +++ b/tests/test_subscribestar_native.py @@ -14,6 +14,7 @@ from pathlib import Path import requests from backend.app.services.gallery_dl import DownloadResult, ErrorType +from backend.app.services.native_ingest_common import post_dir_name from backend.app.services.subscribestar_client import ( MediaItem, SubscribeStarAPIError, @@ -22,10 +23,7 @@ from backend.app.services.subscribestar_client import ( SubscribeStarDriftError, _parse_ss_datetime, ) -from backend.app.services.subscribestar_downloader import ( - SubscribeStarDownloader, - _post_dir_name, -) +from backend.app.services.subscribestar_downloader import SubscribeStarDownloader from backend.app.services.subscribestar_ingester import ( SubscribeStarIngester, _ledger_key, @@ -210,7 +208,7 @@ def _downloader(tmp_path: Path, session=None, validate=True): def test_post_dir_name_empty_title_matches_gallery_dl(): # SubscribeStar has no title → "__" (matches gallery-dl's layout so # existing downloads dedup on disk at cutover). - assert _post_dir_name(_post("111")) == "2026-05-01_111_" + assert post_dir_name(_post("111")) == "2026-05-01_111_" def test_download_post_layout_and_outcomes(tmp_path): From ebe6ab9741ab091533f4b543f413c62aaf461713 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 17 Jun 2026 11:40:36 -0400 Subject: [PATCH 4/7] refactor(native-ingest): shared exception trio + base _failure_result (#899 DRY 2/3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DRY pass commit 2. The two adapters re-implemented the same auth→drift→429→404 →http→network mapping in _failure_result; only the exception classes + drift phrasing differed (divergence-bug risk: a new error_type handled in one and not the other). - native_ingest_common gains NativeIngestError / NativeAuthError / NativeDriftError (status_code + retry_after on the base). Patreon{API,Auth,Drift}Error and SubscribeStar{API,Auth,Drift}Error now subclass them via multiple inheritance, keeping their isinstance-distinct platform names. - Ingester._failure_result (base) does the whole mapping via the shared NativeAuthError/NativeDriftError taxonomy + status_code; a new platform gets it free. New drift_label kwarg supplies the per-platform API_DRIFT phrasing ("Patreon API" / "SubscribeStar markup"), preserving the existing message (test asserts "Patreon API changed"). - Both adapters drop their near-identical _failure_result overrides and their now -unused DownloadResult/ErrorType/*Auth/*Drift imports. Verified at every consumer (rule 93/§8b): test_patreon_ingester (auth/drift/429/ 404/network) and test_subscribestar_native (_failure_result mapping) both exercise the base method now. Remaining: ingest_core L1/L3 logging (3/3). Co-Authored-By: Claude Opus 4.8 --- backend/app/services/ingest_core.py | 44 +++++++++++++-- backend/app/services/native_ingest_common.py | 35 ++++++++++++ backend/app/services/patreon_client.py | 31 +++-------- backend/app/services/patreon_ingester.py | 55 ++----------------- backend/app/services/subscribestar_client.py | 25 +++------ .../app/services/subscribestar_ingester.py | 44 ++------------- 6 files changed, 98 insertions(+), 136 deletions(-) diff --git a/backend/app/services/ingest_core.py b/backend/app/services/ingest_core.py index ba628f7..5cfaf5c 100644 --- a/backend/app/services/ingest_core.py +++ b/backend/app/services/ingest_core.py @@ -36,6 +36,7 @@ from sqlalchemy import delete, func, select, text from sqlalchemy.dialects.postgresql import insert as pg_insert from .gallery_dl import DownloadResult, ErrorType, make_run_stats +from .native_ingest_common import NativeAuthError, NativeDriftError log = logging.getLogger(__name__) @@ -88,6 +89,7 @@ class Ingester: ledger_key: Callable[[object], str], platform: str, error_base: type[Exception], + drift_label: str | None = None, ): self.client = client self.downloader = downloader @@ -99,6 +101,10 @@ class Ingester: self._ledger_key = ledger_key self._platform = platform self._error_base = error_base + # Human label for the API_DRIFT message ("