From 86ae39691400ea2f18c74030863c0ba3a7638a58 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Fri, 3 Jul 2026 09:36:54 -0400 Subject: [PATCH] =?UTF-8?q?feat(pixiv):=20native=20app-API=20client=20?= =?UTF-8?q?=E2=80=94=20gallery-dl-parity=20profile,=20post-first=20seams?= =?UTF-8?q?=20(#129=20step=201)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PixivClient mirrors gallery-dl 1.32.5's PixivAppAPI request profile exactly (iOS app headers, OAuth refresh with X-Client-Time/X-Client-Hash, /v1/user/illusts pagination via next_url — whose query string doubles as the resumable page cursor). Post-first seams (post_record_key / post_is_gated / post_meta) + extract_media covering multi-page, single-page, ugoira zip (600x600→1920x1080 swap, frame delays memoized for the post record), and limit_* placeholder gating. No PHPSESSID web fallback: FC holds only the refresh token, same effective coverage as the gallery-dl path. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM --- backend/app/services/pixiv_client.py | 531 +++++++++++++++++++ tests/fixtures/pixiv_user_illusts_page1.json | 129 +++++ tests/test_pixiv_client.py | 369 +++++++++++++ 3 files changed, 1029 insertions(+) create mode 100644 backend/app/services/pixiv_client.py create mode 100644 tests/fixtures/pixiv_user_illusts_page1.json create mode 100644 tests/test_pixiv_client.py diff --git a/backend/app/services/pixiv_client.py b/backend/app/services/pixiv_client.py new file mode 100644 index 0000000..3635ca4 --- /dev/null +++ b/backend/app/services/pixiv_client.py @@ -0,0 +1,531 @@ +"""Native Pixiv client — the Pixiv adapter's read path. + +Pixiv has a real (if unofficial) API: the mobile app API gallery-dl drives +(`PixivAppAPI`). Per the downloader ground rule — gallery-dl is the +known-working base — this client mirrors gallery-dl 1.32.5's request profile +EXACTLY: the same iOS app headers on every request, the same OAuth +refresh-token dance against oauth.secure.pixiv.net (X-Client-Time + +X-Client-Hash), and the same `/v1/user/illusts` walk paginated by `next_url`. +Deviating from that profile is how the SubscribeStar/Patreon spikes broke, so +any change here should be diffed against gallery-dl's extractor first. + +Feed shape (characterized from gallery-dl 1.32.5, extractor/pixiv.py): + - `GET /v1/user/illusts?user_id=` returns `{"illusts": [work...], + "next_url": "https://app-api...?user_id=..&offset=30" | null}`. + - Pagination: re-issue the SAME endpoint with `next_url`'s query params. The + query string doubles as our resumable page cursor (re-fetching it re-serves + the same page — the ingest-core resume contract). + - A work carries id/title/type(illust|manga|ugoira)/caption(HTML)/ + create_date(ISO+09:00)/tags[{name,translated_name}]/user/page_count/ + x_restrict/series/total_view/total_bookmarks/meta_single_page/meta_pages. + - Files: multi-page → meta_pages[].image_urls.original; single page → + meta_single_page.original_image_url; ugoira → `/v1/ugoira/metadata` zip + (600x600 → 1920x1080 URL swap, gallery-dl's default non-original mode). + +`campaign_id` for Pixiv is the numeric user id (extracted from the source URL +by `user_id_from_url` — no network resolver needed). + +Gated works: pixiv serves a `https://s.pximg.net/common/images/limit_*.png` +placeholder as the "original" when a work is blocked for this account +(sanity-level filter, my-pixiv lock, deleted). gallery-dl's fallback for those +is a web-AJAX scrape that needs PHPSESSID browser cookies — FC stores only the +OAuth refresh token, so (exactly like our previous gallery-dl configuration, +which warned "No PHPSESSID cookie set") those works are skipped, via the +post_is_gated seam. Auth failures are loud (rotate the refresh token); a +response missing the fields we depend on is DRIFT (update this client). + +FC runs on a plain-HTTP homelab; nothing here uses a secure-context Web API. +""" + +from __future__ import annotations + +import hashlib +import logging +import time +from collections.abc import Iterator +from dataclasses import dataclass +from datetime import UTC, datetime +from urllib.parse import parse_qsl, urlsplit + +import requests + +from ..utils.paths import safe_ext +from .native_ingest_common import ( + _MAX_429_RETRIES, + NativeAuthError, + NativeDriftError, + NativeIngestError, + make_session, + retry_after_seconds, +) + +log = logging.getLogger(__name__) + +_TIMEOUT_SECONDS = 30.0 +_API_ROOT = "https://app-api.pixiv.net" +_OAUTH_URL = "https://oauth.secure.pixiv.net/auth/token" + +# gallery-dl's public Pixiv-app credentials (PixivAppAPI, also pixivpy's) — +# these identify the official iOS app to the API, NOT the operator; the +# operator's identity is the OAuth refresh token. +_CLIENT_ID = "MOBrBDS8blbauoSck0ZfDbtuzpyT" +_CLIENT_SECRET = "lsACyCD94FhDUtGTXi3QzcFE2uU1hqtDaKeqrdwj" +_HASH_SECRET = ( + "28c1fdd170a5204386cb1313c7077b34" + "f83e4aaf4aa829ce78c231e05b0bae2c" +) + +# The exact header set gallery-dl 1.32.5 installs on its session — the proven +# app-API request profile. The Referer also unlocks i.pximg.net media GETs +# (403 without it), so the downloader reuses this constant. +PIXIV_APP_HEADERS = { + "App-OS": "ios", + "App-OS-Version": "16.7.2", + "App-Version": "7.19.1", + "User-Agent": "PixivIOSApp/7.19.1 (iOS 16.7.2; iPhone12,8)", + "Referer": "https://app-api.pixiv.net/", +} + +# Placeholder image prefix pixiv serves instead of a blocked work's original +# (limit_sanity_level / limit_mypixiv / limit_unknown variants). +_LIMIT_URL = "https://s.pximg.net/common/images/limit_" + +# The app API reports rate-limiting as an error MESSAGE (often on HTTP 403), +# not only as HTTP 429. gallery-dl sleeps 300s in-walk; sleeping that long +# inside our time-boxed chunk would eat the whole budget, so we surface it as +# a typed 429 and let download_service's cooldown machinery honor the wait. +_RATE_LIMIT_RETRY_AFTER = 300.0 + +_TITLE_MAX = 50 # gallery-dl pixiv filename template: {title[:50]} + +_RATINGS = {0: "General", 1: "R-18", 2: "R-18G"} + + +class PixivAPIError(NativeIngestError): + """Base for native Pixiv client failures. status_code / retry_after are + inherited from NativeIngestError.""" + + +class PixivAuthError(PixivAPIError, NativeAuthError): + """Auth failure — missing/expired/revoked OAuth refresh token. Fix = + rotate the credential (Settings → Credentials → Pixiv), not update the + client. Maps to error_type 'auth_error'.""" + + +class PixivDriftError(PixivAPIError, NativeDriftError): + """A response did not match the shape this client depends on (missing + `illusts`, un-parseable JSON where JSON was promised). Fail loud so the + run flags 'the Pixiv app API changed' instead of silently importing + nothing. Maps to API_DRIFT.""" + + +@dataclass +class MediaItem: + """One resolved downloadable file belonging to a Pixiv work. + + Fields mirror the other native clients' MediaItem so the downloader and + ledger are structurally the same. Pixiv original URLs carry no content + hash, so `filehash` is always None and the ledger keys on + `:` where media_id is `p` (page) or `ugoira` + (the frame zip) — stable across URL-shape drift. + """ + + url: str + filename: str + kind: str + filehash: str | None + post_id: str + media_id: str + + +def user_id_from_url(url: str) -> str | None: + """The numeric pixiv user id from a source URL, or None. + + Handles the modern forms FC accepts as sources + (https://www.pixiv.net/users/, /en/users/) plus the legacy + member.php?id=. This IS the campaign id — no network resolver. + """ + parts = urlsplit(url or "") + if "pixiv.net" not in parts.netloc: + return None + segs = [s for s in parts.path.split("/") if s] + if segs and segs[0] == "en": + segs = segs[1:] + if len(segs) >= 2 and segs[0] == "users" and segs[1].isdigit(): + return segs[1] + if segs and segs[0] == "member.php": + qid = dict(parse_qsl(parts.query)).get("id", "") + if qid.isdigit(): + return qid + return None + + +def _work_filename(work: dict, num: int, url: str) -> str: + """gallery-dl layout parity: `{id}_{title[:50]}_{num:>02}.{extension}` + (the downloader sanitizes the final segment).""" + title = work.get("title") + title50 = (title if isinstance(title, str) else "")[:_TITLE_MAX] + ext = safe_ext(urlsplit(url).path.rsplit("/", 1)[-1]) + return f"{work.get('id')}_{title50}_{num:02d}{ext}" + + +class PixivClient: + """Synchronous Pixiv app-API read client. Construct with the operator's + OAuth refresh token (the same token-type Credential the gallery-dl path + consumed as `extractor.pixiv.refresh-token`).""" + + def __init__( + self, + refresh_token: str | None, + *, + request_sleep: float = 0.0, + max_retries: int = _MAX_429_RETRIES, + session: requests.Session | None = None, + ): + self.refresh_token = refresh_token + self._request_sleep = request_sleep or 0.0 + self._max_retries = max_retries + # No cookies — the app API authenticates via the Bearer token _login + # installs. make_session still supplies the retry/UA plumbing; the + # extra_headers overwrite its browser UA with the app profile. + self._session = ( + session if session is not None + else make_session(None, extra_headers=PIXIV_APP_HEADERS) + ) + self._authed_user: dict = {} + # Monotonic deadline after which the access token must be refreshed; + # 0 forces a refresh on first use. + self._token_deadline = 0.0 + + # -- auth ---------------------------------------------------------------- + + def _login(self) -> None: + """Exchange the refresh token for a Bearer access token (gallery-dl's + `_login_impl`, including the X-Client-Time/X-Client-Hash pair the + endpoint validates). No-op while the current token is still fresh.""" + if time.monotonic() < self._token_deadline: + return + if not self.refresh_token: + raise PixivAuthError( + "No Pixiv refresh token configured — add the OAuth refresh " + "token as the Pixiv credential (token type)." + ) + # gallery-dl stamps naive-UTC with a literal +00:00 suffix. + now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S+00:00") + headers = { + "X-Client-Time": now, + "X-Client-Hash": hashlib.md5( + (now + _HASH_SECRET).encode() + ).hexdigest(), + } + data = { + "client_id": _CLIENT_ID, + "client_secret": _CLIENT_SECRET, + "grant_type": "refresh_token", + "refresh_token": self.refresh_token, + "get_secure_url": "1", + } + try: + resp = self._session.post( + _OAUTH_URL, data=data, headers=headers, + timeout=_TIMEOUT_SECONDS, + ) + except requests.RequestException as exc: + raise PixivAPIError(f"Pixiv OAuth request failed: {exc}") from exc + if resp.status_code >= 400: + raise PixivAuthError( + "Pixiv rejected the refresh token (HTTP " + f"{resp.status_code}) — rotate the Pixiv credential.", + status_code=resp.status_code, + ) + try: + payload = resp.json()["response"] + access = payload["access_token"] + except (ValueError, KeyError, TypeError) as exc: + raise PixivDriftError( + f"Pixiv OAuth response shape changed: {exc}" + ) from exc + self._authed_user = payload.get("user") or {} + self._session.headers["Authorization"] = f"Bearer {access}" + # expires_in is 3600 today; refresh 60s early so a long walk never + # rides an expiring token into a spurious 400. + expires_in = payload.get("expires_in") + lifetime = float(expires_in) if isinstance(expires_in, (int, float)) else 3600.0 + self._token_deadline = time.monotonic() + max(60.0, lifetime - 60.0) + + # -- request ------------------------------------------------------------- + + def _call(self, endpoint: str, params: dict) -> dict: + """Authenticated app-API GET → parsed JSON body, with the shared 429 + backoff and the loud auth/drift/rate-limit mapping.""" + self._login() + if self._request_sleep > 0: + time.sleep(self._request_sleep) + url = _API_ROOT + endpoint + attempt = 0 + while True: + try: + resp = self._session.get( + url, params=params, timeout=_TIMEOUT_SECONDS + ) + except requests.RequestException as exc: + raise PixivAPIError( + f"Pixiv request failed ({endpoint}): {exc}" + ) from exc + if resp.status_code == 429 and attempt < self._max_retries: + attempt += 1 + delay = retry_after_seconds(resp, attempt) + log.warning( + "Pixiv 429 (%s) — backing off %.1fs (retry %d/%d)", + endpoint, delay, attempt, self._max_retries, + ) + time.sleep(delay) + continue + break + + try: + body = resp.json() + except ValueError as exc: + raise PixivDriftError( + f"Pixiv returned non-JSON for {endpoint} " + f"(HTTP {resp.status_code})" + ) from exc + + error = body.get("error") if isinstance(body, dict) else None + message = "" + if isinstance(error, dict): + message = str( + error.get("user_message") or error.get("message") or "" + ) + # Rate limiting first: the app API reports it as an error MESSAGE + # (often on HTTP 403), which must not be mistaken for an auth failure. + if resp.status_code == 429 or "rate limit" in message.lower(): + raise PixivAPIError( + f"Pixiv rate limit hit ({endpoint}): {message or 'HTTP 429'}", + status_code=429, + retry_after=_RATE_LIMIT_RETRY_AFTER, + ) + if resp.status_code in (400, 401, 403): + # Invalid/expired access token surfaces as 400 invalid_grant-style + # errors on the app API; 401/403 are straight auth rejections. + raise PixivAuthError( + f"Pixiv rejected the request ({endpoint}, HTTP " + f"{resp.status_code}): {message or 'auth rejected'} — " + "rotate the Pixiv refresh token.", + status_code=resp.status_code, + ) + if resp.status_code >= 400: + raise PixivAPIError( + f"Pixiv API error ({endpoint}, HTTP {resp.status_code}): " + f"{message or 'unknown error'}", + status_code=resp.status_code, + ) + if error: + # HTTP 200 carrying an error object — unexpected, but never + # silently treat it as data. + raise PixivAPIError( + f"Pixiv API error ({endpoint}): {message or error}", + status_code=resp.status_code, + ) + return body + + # -- normalization ------------------------------------------------------- + + @staticmethod + def _normalize(work: dict) -> dict: + """Wrap an app-API work in the `{"id", "attributes", ...}` post shape + the platform-agnostic core and shared helpers read. The raw work rides + along under `_work` for extract_media / the post record.""" + title = work.get("title") + caption = work.get("caption") + wtype = work.get("type") + return { + "id": work.get("id"), + "attributes": { + "title": title if isinstance(title, str) else "", + "content": caption if isinstance(caption, str) else "", + "published_at": work.get("create_date"), + "post_type": wtype if isinstance(wtype, str) else "illust", + }, + "_work": work, + } + + # -- post-first seams ---------------------------------------------------- + + @staticmethod + def post_record_key(post: dict) -> tuple[str, str] | None: + """`(ledger_key, post_id)` gating post-record capture through the seen + ledger (`post:` synthetic key), or None when the work 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) + + @staticmethod + def post_meta(post: dict) -> dict: + attrs = post.get("attributes") or {} + return {"title": attrs.get("title") or None, "date": attrs.get("published_at")} + + @staticmethod + def post_is_gated(post: dict) -> bool: + """True when this account cannot fetch the work's real files: pixiv + substitutes a `limit_*` placeholder for the original (sanity-level + filter / my-pixiv lock / deleted), or zeroes the author (deleted + account). Mirrors #874 semantics: gated content leaves NO trace — a + placeholder thumbnail and an empty stub would only pollute the + archive. (gallery-dl's PHPSESSID web-scrape fallback for these is out + of scope: FC holds no pixiv browser cookies — module docstring.)""" + work = post.get("_work") or {} + user = work.get("user") or {} + if not user.get("id"): + return True + if work.get("meta_pages"): + return False + single = work.get("meta_single_page") or {} + original = single.get("original_image_url") + return isinstance(original, str) and original.startswith(_LIMIT_URL) + + # -- media --------------------------------------------------------------- + + def extract_media(self, post: dict, included_index: dict) -> list[MediaItem]: + """Resolve a work's downloadable files (gallery-dl's `_extract_files`): + multi-page originals, the single-page original, or the ugoira frame + zip. `included_index` is unused (pixiv works are self-contained).""" + work = post.get("_work") or {} + pid = str(post.get("id") or "") + if not pid or self.post_is_gated(post): + return [] + + if work.get("type") == "ugoira": + return self._ugoira_media(work, pid) + + meta_pages = work.get("meta_pages") or [] + if meta_pages: + items = [] + for num, page in enumerate(meta_pages): + urls = page.get("image_urls") or {} + url = urls.get("original") + if not isinstance(url, str) or not url: + continue + items.append( + MediaItem( + url=url, + filename=_work_filename(work, num, url), + kind="image", + filehash=None, + post_id=pid, + media_id=f"p{num}", + ) + ) + return items + + single = work.get("meta_single_page") or {} + url = single.get("original_image_url") + if not isinstance(url, str) or not url or url.startswith(_LIMIT_URL): + return [] + return [ + MediaItem( + url=url, + filename=_work_filename(work, 0, url), + kind="image", + filehash=None, + post_id=pid, + media_id="p0", + ) + ] + + def _ugoira_media(self, work: dict, pid: str) -> list[MediaItem]: + """The ugoira frame zip (gallery-dl's default non-original mode): + `/v1/ugoira/metadata` → zip_urls.medium with the 600x600→1920x1080 + swap. Frame delays are memoized onto the work so the post record + captures them (a future ugoira→video conversion needs the timings — + the zip alone has none). A metadata failure downgrades to 'no media' + with a warning (matching gallery-dl) instead of failing the walk — + except auth failures, which stay loud.""" + try: + body = self._call("/v1/ugoira/metadata", {"illust_id": pid}) + meta = body["ugoira_metadata"] + zip_url = meta["zip_urls"]["medium"] + except PixivAuthError: + raise + except (PixivAPIError, KeyError, TypeError) as exc: + log.warning("Pixiv ugoira metadata failed for %s: %s", pid, exc) + return [] + work["_ugoira_frames"] = meta.get("frames") or [] + url = zip_url.replace("_ugoira600x600", "_ugoira1920x1080", 1) + return [ + MediaItem( + url=url, + filename=_work_filename(work, 0, url), + kind="ugoira", + filehash=None, + post_id=pid, + media_id="ugoira", + ) + ] + + # -- iteration ----------------------------------------------------------- + + def iter_posts( + self, campaign_id: str, cursor: str | None = None + ) -> Iterator[tuple[dict, dict, str | None]]: + """Yield (post, {}, page_cursor) for every work in the user's feed. + + `campaign_id` is the numeric pixiv user id. `cursor` is the query + string of the app API's `next_url` (offset pagination); None fetches + page 1. The yielded `page_cursor` is the cursor that FETCHED this + work's page, so the core checkpoints a value that re-serves the same + page on resume (the shared cursor contract).""" + if not str(campaign_id or "").isdigit(): + raise PixivDriftError( + f"Pixiv campaign id must be a numeric user id, got " + f"{campaign_id!r}" + ) + current = cursor + while True: + page_cursor = current + if current is None: + params: dict = {"user_id": campaign_id} + else: + params = dict(parse_qsl(current)) + data = self._call("/v1/user/illusts", params) + works = data.get("illusts") + if not isinstance(works, list): + raise PixivDriftError( + "Pixiv user-illusts response had no 'illusts' list " + f"(keys: {sorted(data)[:8]})" + ) + for work in works: + if not isinstance(work, dict): + continue + yield self._normalize(work), {}, page_cursor + next_url = data.get("next_url") + if not next_url: + return + current = str(next_url).rpartition("?")[2] + + # -- verify -------------------------------------------------------------- + + def verify_auth(self) -> tuple[bool | None, str]: + """Cheap credential probe: run the OAuth refresh (the thing that fails + when the token is bad) without walking any feed.""" + try: + self._token_deadline = 0.0 # force a real refresh + self._login() + except PixivAuthError as exc: + return False, f"Pixiv rejected the credential — {exc}" + except PixivAPIError as exc: + return None, f"Couldn't verify (network/HTTP issue): {exc}" + account = self._authed_user.get("account") or self._authed_user.get("name") + suffix = f" as {account}" if account else "" + return True, f"Credentials valid — Pixiv OAuth refresh succeeded{suffix}." + + +def rating_label(x_restrict) -> str | None: + """Human rating from pixiv's x_restrict (0/1/2) — written into the post + record so the archive keeps the R-18 flag without the reader needing to + know pixiv's numeric scheme.""" + if isinstance(x_restrict, bool) or not isinstance(x_restrict, int): + return None + return _RATINGS.get(x_restrict) diff --git a/tests/fixtures/pixiv_user_illusts_page1.json b/tests/fixtures/pixiv_user_illusts_page1.json new file mode 100644 index 0000000..a515089 --- /dev/null +++ b/tests/fixtures/pixiv_user_illusts_page1.json @@ -0,0 +1,129 @@ +{ + "illusts": [ + { + "id": 111, + "title": "Multi Page Adventure", + "type": "illust", + "caption": "Two-page set. WIP thread", + "create_date": "2026-06-20T18:00:00+09:00", + "user": {"id": 99, "name": "Example Artist", "account": "exartist"}, + "tags": [ + {"name": "オリジナル", "translated_name": "original"}, + {"name": "女の子", "translated_name": "girl"} + ], + "page_count": 2, + "width": 1200, + "height": 1600, + "x_restrict": 0, + "series": {"id": 4242, "title": "Adventure Series"}, + "total_view": 1000, + "total_bookmarks": 250, + "is_bookmarked": false, + "illust_ai_type": 1, + "meta_single_page": {}, + "meta_pages": [ + { + "image_urls": { + "square_medium": "https://i.pximg.net/c/360x360_70/img-master/img/2026/06/20/18/00/00/111_p0_square1200.jpg", + "original": "https://i.pximg.net/img-original/img/2026/06/20/18/00/00/111_p0.png" + } + }, + { + "image_urls": { + "square_medium": "https://i.pximg.net/c/360x360_70/img-master/img/2026/06/20/18/00/00/111_p1_square1200.jpg", + "original": "https://i.pximg.net/img-original/img/2026/06/20/18/00/00/111_p1.png" + } + } + ] + }, + { + "id": 222, + "title": "Single Piece", + "type": "illust", + "caption": "", + "create_date": "2026-06-18T12:30:00+09:00", + "user": {"id": 99, "name": "Example Artist", "account": "exartist"}, + "tags": [{"name": "落書き", "translated_name": "doodle"}], + "page_count": 1, + "width": 900, + "height": 900, + "x_restrict": 1, + "series": null, + "total_view": 500, + "total_bookmarks": 60, + "is_bookmarked": true, + "illust_ai_type": 0, + "meta_single_page": { + "original_image_url": "https://i.pximg.net/img-original/img/2026/06/18/12/30/00/222_p0.jpg" + }, + "meta_pages": [] + }, + { + "id": 333, + "title": "Wiggle Loop", + "type": "ugoira", + "caption": "animated", + "create_date": "2026-06-15T09:00:00+09:00", + "user": {"id": 99, "name": "Example Artist", "account": "exartist"}, + "tags": [{"name": "うごイラ", "translated_name": "ugoira"}], + "page_count": 1, + "width": 600, + "height": 600, + "x_restrict": 0, + "series": null, + "total_view": 300, + "total_bookmarks": 40, + "is_bookmarked": false, + "illust_ai_type": 0, + "meta_single_page": { + "original_image_url": "https://i.pximg.net/img-original/img/2026/06/15/09/00/00/333_ugoira0.jpg" + }, + "meta_pages": [] + }, + { + "id": 444, + "title": "Blocked Work", + "type": "illust", + "caption": "", + "create_date": "2026-06-10T00:00:00+09:00", + "user": {"id": 99, "name": "Example Artist", "account": "exartist"}, + "tags": [], + "page_count": 1, + "width": 0, + "height": 0, + "x_restrict": 2, + "series": null, + "total_view": 0, + "total_bookmarks": 0, + "is_bookmarked": false, + "illust_ai_type": 0, + "meta_single_page": { + "original_image_url": "https://s.pximg.net/common/images/limit_sanity_level_360.png" + }, + "meta_pages": [] + }, + { + "id": 555, + "title": "Ghost Work", + "type": "illust", + "caption": "", + "create_date": "2026-06-01T00:00:00+09:00", + "user": {"id": 0, "name": "", "account": ""}, + "tags": [], + "page_count": 1, + "width": 0, + "height": 0, + "x_restrict": 0, + "series": null, + "total_view": 0, + "total_bookmarks": 0, + "is_bookmarked": false, + "illust_ai_type": 0, + "meta_single_page": { + "original_image_url": "https://i.pximg.net/img-original/img/2026/06/01/00/00/00/555_p0.png" + }, + "meta_pages": [] + } + ], + "next_url": "https://app-api.pixiv.net/v1/user/illusts?user_id=99&offset=30" +} diff --git a/tests/test_pixiv_client.py b/tests/test_pixiv_client.py new file mode 100644 index 0000000..8419159 --- /dev/null +++ b/tests/test_pixiv_client.py @@ -0,0 +1,369 @@ +"""PixivClient tests — parsing + iteration against canned pages, no network. + +The fixture mirrors a real `/v1/user/illusts` page (multi-page work, single +page, ugoira, sanity-limited placeholder, deleted-author ghost). HTTP is +stubbed at the requests-session seam (oauth POST + API GET), so the exact +gallery-dl-parity request profile — headers, oauth form, pagination params — +is asserted rather than assumed. +""" + +import json +from pathlib import Path + +import pytest + +from backend.app.services.pixiv_client import ( + PIXIV_APP_HEADERS, + MediaItem, + PixivAPIError, + PixivAuthError, + PixivClient, + PixivDriftError, + rating_label, + user_id_from_url, +) + +_FIXTURE = Path(__file__).parent / "fixtures" / "pixiv_user_illusts_page1.json" + + +class FakeResponse: + def __init__(self, status_code=200, json_data=None, headers=None): + self.status_code = status_code + self._json = json_data + self.headers = headers or {} + + def json(self): + if self._json is None: + raise ValueError("no JSON") + return self._json + + +class FakeSession: + """Minimal requests.Session stand-in: canned responses per (method, url + fragment), recording every call for profile assertions.""" + + def __init__(self): + self.headers = dict(PIXIV_APP_HEADERS) + self.responses = [] + self.calls = [] + + def queue(self, response): + self.responses.append(response) + return self + + def _next(self): + if not self.responses: + raise AssertionError("FakeSession ran out of queued responses") + return self.responses.pop(0) + + def post(self, url, data=None, headers=None, timeout=None): + self.calls.append(("POST", url, data, headers)) + return self._next() + + def get(self, url, params=None, timeout=None, headers=None): + self.calls.append(("GET", url, params, headers)) + return self._next() + + +def _oauth_ok(): + return FakeResponse(200, { + "response": { + "access_token": "acc-token", + "expires_in": 3600, + "user": {"id": "77", "account": "operator", "name": "Op"}, + } + }) + + +@pytest.fixture +def page1(): + return json.loads(_FIXTURE.read_text()) + + +@pytest.fixture +def client(): + # Never issues a request in pure-parsing tests. + return PixivClient("refresh-tok", session=FakeSession()) + + +def _post_for(client, page1, work_id): + for work in page1["illusts"]: + if work["id"] == work_id: + return client._normalize(work) + raise AssertionError(f"no work {work_id} in fixture") + + +# -- URL → user id ---------------------------------------------------------- + +def test_user_id_from_url_variants(): + assert user_id_from_url("https://www.pixiv.net/users/12345678") == "12345678" + assert user_id_from_url("https://www.pixiv.net/en/users/42") == "42" + assert user_id_from_url("https://pixiv.net/users/7/artworks") == "7" + assert user_id_from_url("https://www.pixiv.net/member.php?id=99") == "99" + + +def test_user_id_from_url_rejects_non_matches(): + assert user_id_from_url("https://www.pixiv.net/artworks/111") is None + assert user_id_from_url("https://www.pixiv.net/users/notdigits") is None + assert user_id_from_url("https://example.com/users/5") is None + assert user_id_from_url("") is None + + +# -- normalization ------------------------------------------------------------ + +def test_normalize_maps_attributes(client, page1): + post = _post_for(client, page1, 111) + attrs = post["attributes"] + assert post["id"] == 111 + assert attrs["title"] == "Multi Page Adventure" + assert "Two-page set." in attrs["content"] + assert attrs["published_at"] == "2026-06-20T18:00:00+09:00" + assert attrs["post_type"] == "illust" + assert post["_work"]["total_bookmarks"] == 250 + + +def test_post_record_key_and_meta(client, page1): + post = _post_for(client, page1, 222) + assert client.post_record_key(post) == ("post:222", "222") + meta = client.post_meta(post) + assert meta["title"] == "Single Piece" + assert meta["date"] == "2026-06-18T12:30:00+09:00" + assert client.post_record_key({"id": None}) is None + + +# -- gating ------------------------------------------------------------------- + +def test_post_is_gated_limit_placeholder(client, page1): + assert client.post_is_gated(_post_for(client, page1, 444)) is True + + +def test_post_is_gated_deleted_author(client, page1): + assert client.post_is_gated(_post_for(client, page1, 555)) is True + + +def test_post_is_gated_normal_works(client, page1): + assert client.post_is_gated(_post_for(client, page1, 111)) is False + assert client.post_is_gated(_post_for(client, page1, 222)) is False + + +# -- extract_media -------------------------------------------------------------- + +def test_extract_media_multi_page(client, page1): + items = client.extract_media(_post_for(client, page1, 111), {}) + assert len(items) == 2 + assert all(isinstance(m, MediaItem) for m in items) + assert [m.media_id for m in items] == ["p0", "p1"] + assert items[0].url.endswith("/111_p0.png") + assert items[1].url.endswith("/111_p1.png") + # gallery-dl filename parity: {id}_{title[:50]}_{num:>02}.{extension} + assert items[0].filename == "111_Multi Page Adventure_00.png" + assert items[1].filename == "111_Multi Page Adventure_01.png" + assert all(m.post_id == "111" for m in items) + + +def test_extract_media_single_page(client, page1): + items = client.extract_media(_post_for(client, page1, 222), {}) + assert len(items) == 1 + assert items[0].media_id == "p0" + assert items[0].url.endswith("/222_p0.jpg") + assert items[0].filename == "222_Single Piece_00.jpg" + + +def test_extract_media_gated_yields_nothing(client, page1): + assert client.extract_media(_post_for(client, page1, 444), {}) == [] + assert client.extract_media(_post_for(client, page1, 555), {}) == [] + + +def test_extract_media_ugoira_zip_swap(client, page1, monkeypatch): + frames = [{"file": "000000.jpg", "delay": 90}, {"file": "000001.jpg", "delay": 90}] + + def fake_call(endpoint, params): + assert endpoint == "/v1/ugoira/metadata" + assert params == {"illust_id": "333"} + return {"ugoira_metadata": { + "zip_urls": {"medium": ( + "https://i.pximg.net/img-zip-ugoira/img/2026/06/15/09/00/00/" + "333_ugoira600x600.zip" + )}, + "frames": frames, + }} + + monkeypatch.setattr(client, "_call", fake_call) + post = _post_for(client, page1, 333) + items = client.extract_media(post, {}) + assert len(items) == 1 + assert items[0].media_id == "ugoira" + assert items[0].kind == "ugoira" + assert items[0].url.endswith("333_ugoira1920x1080.zip") + assert items[0].filename == "333_Wiggle Loop_00.zip" + # Frame delays memoized for the post record (future video conversion). + assert post["_work"]["_ugoira_frames"] == frames + + +def test_extract_media_ugoira_metadata_failure_downgrades( + client, page1, monkeypatch +): + def fake_call(endpoint, params): + raise PixivAPIError("boom", status_code=500) + + monkeypatch.setattr(client, "_call", fake_call) + assert client.extract_media(_post_for(client, page1, 333), {}) == [] + + +def test_extract_media_ugoira_auth_failure_stays_loud( + client, page1, monkeypatch +): + def fake_call(endpoint, params): + raise PixivAuthError("token dead", status_code=400) + + monkeypatch.setattr(client, "_call", fake_call) + with pytest.raises(PixivAuthError): + client.extract_media(_post_for(client, page1, 333), {}) + + +# -- iteration ------------------------------------------------------------------ + +def test_iter_posts_paginates_and_carries_cursor(page1, monkeypatch): + client = PixivClient("refresh-tok", session=FakeSession()) + page2 = { + "illusts": [dict(page1["illusts"][1], id=666, title="Older")], + "next_url": None, + } + seen_params = [] + + def fake_call(endpoint, params): + assert endpoint == "/v1/user/illusts" + seen_params.append(dict(params)) + return page1 if len(seen_params) == 1 else page2 + + monkeypatch.setattr(client, "_call", fake_call) + rows = list(client.iter_posts("99")) + + assert seen_params[0] == {"user_id": "99"} + # Page 2 params come verbatim from next_url's query string. + assert seen_params[1] == {"user_id": "99", "offset": "30"} + # Page-1 posts carry cursor None; page-2 posts carry the fetching cursor. + assert [c for _, _, c in rows[:5]] == [None] * 5 + assert rows[5][2] == "user_id=99&offset=30" + assert rows[5][0]["id"] == 666 + + +def test_iter_posts_resumes_from_cursor(page1, monkeypatch): + client = PixivClient("refresh-tok", session=FakeSession()) + captured = {} + + def fake_call(endpoint, params): + captured["params"] = dict(params) + return {"illusts": [], "next_url": None} + + monkeypatch.setattr(client, "_call", fake_call) + list(client.iter_posts("99", cursor="user_id=99&offset=60")) + assert captured["params"] == {"user_id": "99", "offset": "60"} + + +def test_iter_posts_rejects_non_numeric_campaign(): + client = PixivClient("refresh-tok", session=FakeSession()) + with pytest.raises(PixivDriftError): + next(client.iter_posts("https://www.pixiv.net/users/99")) + + +def test_iter_posts_drift_on_missing_illusts(monkeypatch): + client = PixivClient("refresh-tok", session=FakeSession()) + monkeypatch.setattr(client, "_call", lambda e, p: {"error": None, "body": 1}) + with pytest.raises(PixivDriftError): + next(client.iter_posts("99")) + + +# -- auth ------------------------------------------------------------------------ + +def test_login_sends_gallery_dl_profile_and_sets_bearer(): + session = FakeSession().queue(_oauth_ok()) + client = PixivClient("refresh-tok", session=session) + client._login() + + method, url, data, headers = session.calls[0] + assert method == "POST" + assert url == "https://oauth.secure.pixiv.net/auth/token" + assert data["grant_type"] == "refresh_token" + assert data["refresh_token"] == "refresh-tok" + assert data["client_id"] == "MOBrBDS8blbauoSck0ZfDbtuzpyT" + assert data["get_secure_url"] == "1" + # X-Client-Time/-Hash pair: ISO + literal +00:00, md5 hex digest. + assert headers["X-Client-Time"].endswith("+00:00") + assert len(headers["X-Client-Hash"]) == 32 + assert session.headers["Authorization"] == "Bearer acc-token" + # Token is fresh → a second login is a no-op (no extra POST). + client._login() + assert len(session.calls) == 1 + + +def test_login_maps_rejection_to_auth_error(): + session = FakeSession().queue(FakeResponse(400, {"has_error": True})) + client = PixivClient("refresh-tok", session=session) + with pytest.raises(PixivAuthError): + client._login() + + +def test_login_without_token_is_auth_error(): + client = PixivClient(None, session=FakeSession()) + with pytest.raises(PixivAuthError): + client._login() + + +def test_call_maps_rate_limit_message(): + session = FakeSession().queue(_oauth_ok()).queue(FakeResponse(403, { + "error": {"message": "Rate Limit", "user_message": ""}, + })) + client = PixivClient("refresh-tok", session=session) + with pytest.raises(PixivAPIError) as exc_info: + client._call("/v1/user/illusts", {"user_id": "1"}) + err = exc_info.value + assert not isinstance(err, PixivAuthError) + assert err.status_code == 429 + assert err.retry_after == 300.0 + + +def test_call_maps_403_to_auth_error(): + session = FakeSession().queue(_oauth_ok()).queue(FakeResponse(403, { + "error": {"message": "invalid access token", "user_message": ""}, + })) + client = PixivClient("refresh-tok", session=session) + with pytest.raises(PixivAuthError): + client._call("/v1/user/illusts", {"user_id": "1"}) + + +def test_call_maps_404_status(): + session = FakeSession().queue(_oauth_ok()).queue(FakeResponse(404, { + "error": {"message": "Not Found", "user_message": ""}, + })) + client = PixivClient("refresh-tok", session=session) + with pytest.raises(PixivAPIError) as exc_info: + client._call("/v1/user/illusts", {"user_id": "1"}) + assert exc_info.value.status_code == 404 + + +def test_verify_auth_reports_account(): + session = FakeSession().queue(_oauth_ok()) + client = PixivClient("refresh-tok", session=session) + ok, message = client.verify_auth() + assert ok is True + assert "operator" in message + + +def test_verify_auth_bad_token(): + session = FakeSession().queue(FakeResponse(400, {"has_error": True})) + client = PixivClient("bad", session=session) + ok, message = client.verify_auth() + assert ok is False + assert "rotate" in message.lower() or "rejected" in message.lower() + + +# -- rating ------------------------------------------------------------------------ + +def test_rating_label(): + assert rating_label(0) == "General" + assert rating_label(1) == "R-18" + assert rating_label(2) == "R-18G" + assert rating_label(None) is None + assert rating_label(True) is None + assert rating_label(9) is None