feat(pixiv): native app-API client — gallery-dl-parity profile, post-first seams (#129 step 1)
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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM
This commit is contained in:
@@ -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=<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
|
||||
`<post_id>:<media_id>` where media_id is `p<num>` (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/<id>, /en/users/<id>) plus the legacy
|
||||
member.php?id=<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:<id>` 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)
|
||||
+129
@@ -0,0 +1,129 @@
|
||||
{
|
||||
"illusts": [
|
||||
{
|
||||
"id": 111,
|
||||
"title": "Multi Page Adventure",
|
||||
"type": "illust",
|
||||
"caption": "Two-page set. <a href=\"https://example.com/wip\">WIP thread</a>",
|
||||
"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"
|
||||
}
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user