feat(subscribestar): native client + downloader + ingester (post-first) (#890/#891/#892)
CI / lint (push) Successful in 2s
CI / frontend-build (push) Successful in 21s
CI / backend-lint-and-test (push) Successful in 38s
CI / integration (push) Successful in 3m23s

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 <noreply@anthropic.com>
This commit is contained in:
2026-06-17 11:06:02 -04:00
parent f678819093
commit 817a002c2b
4 changed files with 1210 additions and 0 deletions
@@ -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 <base>/<slug>.
- Each page embeds `data-role="infinite_scroll-next_page" href="/posts?...&page=N
&slug=<slug>&sort_by=newest"`. That path returns JSON {"html": "<...posts...>"};
the fragment carries the NEXT page link, until it runs out.
- A post is `<div class="post is-shown ..." data-id="<post_id>" ...>`; 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 = '<div class="post is-shown'
_POST_ID_RE = re.compile(r'data-id="(\d+)"')
_POST_DATE_RE = re.compile(r'<div class="post-date">([^<]+)</div>')
# The body: inner of `.post-content .trix-content`. Non-greedy to the LAST
# </div></div> 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'<div class="post-content"[^>]*>(.*?)</div>\s*</div>\s*</div>', 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 `<post_id>:<media_id>`.
"""
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 <date>' 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": "<fragment>"}."""
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:<id>`
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."
@@ -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 `<YYYY-MM-DD>_<post_id>_` — 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:
"""`<YYYY-MM-DD>_<post_id>_<title40>` 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,
)
@@ -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: `<post_id>:<media_id>`. 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)
+294
View File
@@ -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("&", "&amp;").replace('"', "&quot;")
def _post_html(post_id="111", date="May 01, 2026 12:00 pm", body="<p>hello</p>",
media: list[dict] | None = None, locked=False):
gallery = ""
if media is not None:
gallery = (
f'<div class="uploads-images" data-gallery="{_gallery_attr(media)}"></div>'
)
lock = ' for-locked_content' if locked else ""
return (
f'<div class="post is-shown false false{lock}" data-id="{post_id}" '
f'data-infinite-scroll-id="{post_id}">'
f'<div class="post-date">{date}</div>'
f'<div class="post-body" data-role="post-body">'
f'<div class="post-content" data-role="post_content-text">'
f'<div class="trix-content">{body}</div></div></div>'
f'{gallery}</div>'
)
def _feed_page(posts_html: str, next_href: str | None = None):
nxt = (
f'<div data-role="infinite_scroll-next_page" href="{next_href}"></div>'
if next_href else ""
)
return (
'<html><body><div data-role="posts_container-list">'
f'{posts_html}{nxt}</div></body></html>'
)
# -- 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: "<html>nothing</html>")
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="<p>text only</p>")))
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": "<div>body text</div>",
"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 → "<date>_<id>_" (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