`; 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 json
import logging
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
from .native_ingest_common import (
_MAX_429_RETRIES,
NativeAuthError,
NativeDriftError,
NativeIngestError,
basename_from_url,
make_session,
retry_after_seconds,
)
log = logging.getLogger(__name__)
_TIMEOUT_SECONDS = 30.0
# Match gallery-dl's default (cookies-only) request profile EXACTLY — the proven
# way to read SubscribeStar without tripping its /verify_subscriber gate. In that
# mode gallery-dl's base Extractor._init_session sends a Firefox UA, Accept: */*,
# Accept-Language, and a same-site Referer (root/) on EVERY request — including
# the first creator-page GET — and uses NO X-Requested-With on either the creator
# page or the "load more" JSON endpoint (it GETs and parses the body as JSON).
# Our prior Chrome UA + missing Referer + XHR toggling looked enough unlike a
# browser that SubscribeStar 302'd the adult-creator page to /
/verify_
# subscriber even with valid cookies (cheunart, 2026-06-17).
_FIREFOX_UA = (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:140.0) "
"Gecko/20100101 Firefox/140.0"
)
_GDL_HEADERS = {
"User-Agent": _FIREFOX_UA,
"Accept-Language": "en-US,en;q=0.5",
}
# 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).
#
# Delimiter is the GENERIC ` regex
# either returned empty or over-captured into sibling upload divs AND the
# "View next posts (N / M)" pagination counter (the "264 / 265" body bug,
# cheunart 2026-06-17). The trix editor wraps rich bodies in a full
# `…` document, so strip to the body inner when present.
_CONTENT_OPEN = ''
_CONTENT_CLOSE = '
str:
"""A short, log-safe description of an unexpected page: its
+ which
known interstitial it resembles (bot challenge / age gate / login / captcha)."""
m = re.search(r"<title[^>]*>(.*?)", html, re.IGNORECASE | re.DOTALL)
title = unescape(m.group(1).strip())[:120] if m else "(no )"
low = html.lower()
hits = [name for name, needles in _INTERSTITIAL_MARKERS
if any(n.lower() in low for n in needles)]
return f"title={title!r}; resembles: {'/'.join(hits) if hits else 'unrecognized'}"
class SubscribeStarAPIError(NativeIngestError):
"""Base for native SubscribeStar client failures. status_code / retry_after
are inherited from NativeIngestError."""
class SubscribeStarAuthError(SubscribeStarAPIError, NativeAuthError):
"""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, NativeDriftError):
"""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 _extr(text: str, start: str, end: str) -> str:
"""Substring between the first `start` and the next `end` after it (gallery-
dl's `text.extr`); '' when either marker is absent."""
i = text.find(start)
if i < 0:
return ""
i += len(start)
j = text.find(end, i)
if j < 0:
return ""
return text[i:j]
def _extract_content(chunk: str) -> str:
"""The post body HTML — gallery-dl's `_data_from_post` content rule: between
the post_content-text wrapper and the youtube-uploads div, with the trix
editor's `<html><body>…</body></html>` wrapper stripped to its inner."""
content = _extr(chunk, _CONTENT_OPEN, _CONTENT_CLOSE)
if "<html><body>" in content:
content = _extr(content, "<body>", "</body>")
return content.strip()
def _attachment_item(
frag: str, base: str, post_id: str, *, kind: str, title_marker: str, url_attr: str
) -> MediaItem | None:
"""One doc/audio attachment from its preview block — gallery-dl's
`_media_from_post` attachment/audio fields. `url_attr` is the URL-bearing
attribute (`href="` for docs, `src="` for audio); `title_marker` precedes the
display name. Returns None when the block carries no URL."""
rel = unescape(_extr(frag, url_attr, '"'))
if not rel:
return None
url = urljoin(base + "/", rel)
upload_id = _extr(frag, 'data-upload-id="', '"')
name = unescape(_extr(frag, title_marker, "<")).strip()
return MediaItem(
url=url,
filename=name or basename_from_url(url),
kind=kind,
filehash=filehash_from_url(url),
post_id=post_id,
media_id=str(upload_id or ""),
)
def _normalize_ss_host(netloc: str) -> str:
"""Rewrite the `subscribestar.art` host to `subscribestar.adult`.
The age wall on the `.art` domain does not clear with the
`18_plus_agreement_generic` cookie (unlike `.com`/`.adult`): a `.art`
creator page keeps 302'ing to `/age_confirmation_warning` even with the
cookie set (Elasid, event #54116). The same creator is reachable on
`.adult`, where the cookie works — so `.art` behaves as an alias that
doesn't honor the age gate. Normalize it to `.adult` at request time (the
stored Source.url is left untouched). `.com`/`.adult` pass through.
"""
host = netloc.lower()
if host == "subscribestar.art" or host.endswith(".subscribestar.art"):
rewritten = netloc[: -len("art")] + "adult"
log.info(
"SubscribeStar: rewrote age-gated .art host %r → %r", netloc, rewritten
)
return rewritten
return netloc
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; .art → .adult, see
_normalize_ss_host); slug = first path segment.
"""
parts = urlsplit(campaign_id)
base = f"{parts.scheme or 'https'}://{_normalize_ss_host(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
_OG_TITLE_RE = re.compile(
r'<meta[^>]+property=["\']og:title["\'][^>]+content=["\']([^"\']+)["\']',
re.IGNORECASE,
)
_TITLE_RE = re.compile(r"<title[^>]*>(.*?)", re.IGNORECASE | re.DOTALL)
# Trailing " | SubscribeStar" / " on SubscribeStar" the profile carries.
_SS_TITLE_SUFFIX_RE = re.compile(
r"\s*[|·]\s*SubscribeStar.*$|\s+on\s+SubscribeStar.*$", re.IGNORECASE
)
def _extract_creator_name(html: str) -> str | None:
"""The creator's display name from a SubscribeStar profile page: prefer the
og:title meta (it's the bare creator name), else the <title> with the
SubscribeStar suffix stripped. None when neither yields anything (#130)."""
m = _OG_TITLE_RE.search(html)
name = unescape(m.group(1)).strip() if m else ""
if not name:
t = _TITLE_RE.search(html)
raw = unescape(t.group(1)).strip() if t else ""
name = _SS_TITLE_SUFFIX_RE.sub("", raw).strip()
return name or None
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
# gallery-dl-parity request profile (Firefox UA, Accept: */*, Accept-
# Language; Referer is stamped per-walk once the creator base is known).
self._session = make_session(
cookies_path, accept="*/*", extra_headers=_GDL_HEADERS
)
self._request_sleep = request_sleep or 0.0
self._max_retries = max_retries
# -- request -----------------------------------------------------------
def _get(self, url: str, *, headers: dict | None = None) -> 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, headers=headers)
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
# gallery-dl's own gating signal: SubscribeStar 302-redirects an
# unauthenticated / age-unconfirmed request to /verify_subscriber or
# /age_confirmation_warning (lands as a 200 on that URL). Treat it as auth,
# not drift — the fix is to rotate cookies / set the age cookie.
if resp.history and (
"/verify_subscriber" in resp.url
or "/age_confirmation_warning" in resp.url
):
raise SubscribeStarAuthError(
f"SubscribeStar redirected to {resp.url} — auth/age wall "
f"(rotate cookies or set the 18+ age cookie; requested {url})",
status_code=resp.status_code,
)
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: a plain GET (gallery-dl uses no XHR header) to the
`/posts?...` endpoint, which 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)
# gallery-dl date method: text up to first '</', then after the last '>'
# — handles both plain and <a>-wrapped dates (see _DATE_OPEN comment).
raw_date = _extr(chunk, _DATE_OPEN, "</").rpartition(">")[2]
published = _parse_ss_datetime(raw_date) if raw_date else None
content = _extract_content(chunk)
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)
if posts:
dated = sum(1 for p in posts if p["attributes"].get("published_at"))
bodied = sum(1 for p in posts if p["attributes"].get("content"))
log.info(
"SubscribeStar parsed %d posts (%d dated, %d with body)",
len(posts), dated, bodied,
)
# Canary for this exact failure class: posts parsed but NONE got a
# date or a body, while the raw markers ARE present → our extraction
# diverged from the live markup (e.g. the <a>-wrapped date bug). Log
# the marker counts so the cause is diagnosable from the worker log
# alone, without re-fetching the authed page.
if dated == 0 or bodied == 0:
log.warning(
"SubscribeStar parse canary: %d posts but dated=%d bodied=%d; "
"raw markers post-date=%d post_content-text=%d data-gallery=%d "
"— extraction likely diverged from the live markup",
len(posts), dated, bodied,
html.count('class="post-date"'),
html.count("post_content-text"),
html.count("data-gallery"),
)
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 (gallery-dl's `_media_from_post`): the
per-post `data-gallery` JSON manifest (images/videos), PLUS document
attachments (`uploads-docs`) and audio (`uploads-audios`) — some posts
deliver content only through the latter two. `/previews` (locked teaser)
gallery items are skipped. `included_index` is unused (media is 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
# gallery-dl's _media_from_post: a gallery item whose URL is under
# /previews is a locked/blurred TEASER, not the real file — skip it
# (the SubscribeStar analog of the Patreon gated-preview bug #874).
# This is why a locked post yields no downloadable media.
if "/previews" in 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,
)
)
# Document attachments (uploads-docs → doc_preview blocks): href URL.
docs = _extr(chunk, 'class="uploads-docs"', 'class="post-edit_form"')
for frag in _DOC_SPLIT_RE.split(docs)[1:]:
item = _attachment_item(
frag, base, post_id, kind="attachment",
title_marker='doc_preview-title">', url_attr='href="',
)
if item is not None:
items.append(item)
# Audio attachments (uploads-audios → audio_preview-data blocks): src URL.
audios = _extr(chunk, 'class="uploads-audios"', 'class="post-edit_form"')
for frag in _AUDIO_SPLIT_RE.split(audios)[1:]:
item = _attachment_item(
frag, base, post_id, kind="audio",
title_marker='audio_preview-title">', url_attr='src="',
)
if item is not None:
items.append(item)
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}"
)
# Same-site Referer (gallery-dl sends root/ on every request).
self._session.headers["Referer"] = f"{base}/"
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:
# Report what we actually got — length + the page's identity —
# so a served interstitial (bot challenge / age gate / login)
# is named instead of mislabeled "markup changed".
looks_json = html.lstrip()[:1] in ("{", "[")
raise SubscribeStarDriftError(
f"SubscribeStar feed for {slug!r} had no posts and no "
f"recognizable feed container ({len(html)} bytes"
f"{', looks like JSON' if looks_json else ''}; "
f"{_describe_page(html)})"
)
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
# -- display name -------------------------------------------------------
def resolve_display_name(self, campaign_id: str) -> str | None:
"""The creator's display name from their profile page, used to name the
Artist at add-time (#130). `campaign_id` is the creator URL. None on any
failure — the caller falls back to the URL handle. Sync: run in an
executor."""
base, slug = _split_creator_url(campaign_id)
if not slug:
return None
self._session.headers["Referer"] = f"{base}/"
try:
html = self._feed_html(f"{base}/{slug}")
except SubscribeStarAPIError:
return None
return _extract_creator_name(html)
# -- 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}"
self._session.headers["Referer"] = f"{base}/"
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."