MediaCrawler/tools/probe_creator_api.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Phase 0 probe: can the creator backend be reached with a plain signed request?

The question this answers, and why it is worth a whole script: two sources
disagree. The `xhshow` library ships `sign_xyw()` whose docstring says it exists
because the creator data APIs *reject* the main-site signature with HTTP 406 -- but
a field report claims the creator gateway rejects *any* self-made request with 406
regardless of signature. Only a live request settles it.

The response is read as a three-way verdict, because a bare "it failed" is not
useful here:

    406  -> the gateway rejected the signature. The browser-interception route is
            the only way forward.
    401 / not-logged-in
         -> the signature PASSED and only the creator session is missing. That is
            good news: it means Phase 1 is pure-request after all.
    200  -> we are through, and the payload is captured for field mapping.

Cookies are read out of the browser over CDP and never printed -- they are
credentials, and this script has no reason to echo them.
"""

import argparse
import base64
import hashlib
import json
import sys
import urllib.parse
from datetime import datetime, time, timedelta

# The XYW_ scheme, matching both xhshow/config/config.py and the independent
# reverse-engineering in xiaohongshu-cli. Constants are byte-identical in both.
XYW_AES_KEY = b"7cc4adla5ay0701v"
XYW_AES_IV = b"4uzjr7mbsibcaldp"
XYW_ENV_FLAGS = "0|0|0|1|0|0|1|0|0|0|1|0|0|0|0|1|0|0|0"

CREATOR_ORIGIN = "https://creator.xiaohongshu.com"
# The data-analysis note list. This is the page the creator console itself calls,
# and it is where exposure/views live.
NOTE_LIST_PATH = "/api/galaxy/creator/datacenter/note/analyze/list"

USER_AGENT = (
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36"
)


def _aes_encrypt_hex(plaintext: str) -> str:
    from Crypto.Cipher import AES
    from Crypto.Util.Padding import pad

    cipher = AES.new(XYW_AES_KEY, AES.MODE_CBC, XYW_AES_IV)
    return cipher.encrypt(pad(plaintext.encode("utf-8"), AES.block_size)).hex()


def sign_xyw(api: str, a1: str, app_id: str = "ugc", data: dict | None = None) -> dict[str, str]:
    """Mirror of xiaohongshu-cli's creator_signing.sign_creator.

    Written out rather than imported so the probe can vary ``api`` and ``app_id``
    independently -- figuring out which combination the gateway accepts is the
    entire point of running it.
    """
    content = api
    if data is not None:
        content += json.dumps(data, separators=(",", ":"), ensure_ascii=False)

    digest = hashlib.md5(content.encode("utf-8")).hexdigest()
    timestamp_ms = int(datetime.now().timestamp() * 1000)
    plaintext = f"x1={digest};x2={XYW_ENV_FLAGS};x3={a1};x4={timestamp_ms};"
    encoded = base64.b64encode(plaintext.encode("utf-8")).decode("utf-8")

    envelope = {
        "signSvn": "56",
        "signType": "x2",
        "appId": app_id,
        "signVersion": "1",
        "payload": _aes_encrypt_hex(encoded),
    }
    x_s = "XYW_" + base64.b64encode(
        json.dumps(envelope, separators=(",", ":")).encode("utf-8")
    ).decode("utf-8")
    return {"x-s": x_s, "x-t": str(timestamp_ms)}


def build_query(start_days_ago: int, end_days_ago: int, page_size: int = 10) -> str:
    """Query string exactly as the working collector builds it: epoch millis."""
    today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)

    def ms(days_ago: int, at_end: bool) -> int:
        day = (today - timedelta(days=days_ago)).date()
        clock = time(23, 59, 59) if at_end else time(0, 0, 0)
        return int(datetime.combine(day, clock).timestamp() * 1000)

    return urllib.parse.urlencode(
        {
            "post_begin_time": ms(start_days_ago, False),
            "post_end_time": ms(end_days_ago, True),
            "type": 0,
            "page_size": page_size,
            "page_num": 1,
        }
    )


async def cookies_from_cdp() -> dict[str, str]:
    """Session cookies for creator.xiaohongshu.com, read out of the live browser."""
    from playwright.async_api import async_playwright

    playwright = await async_playwright().start()
    try:
        browser = await playwright.chromium.connect_over_cdp(
            "http://127.0.0.1:9222", timeout=15000
        )
        jars = {}
        for context in browser.contexts:
            for cookie in await context.cookies():
                jars[cookie["name"]] = cookie["value"]
        return jars
    finally:
        await playwright.stop()


def classify(status: int, body: str) -> str:
    if status == 406:
        return "406 —— 网关拒了签名（回退浏览器拦截路线）"
    if status == 200:
        return "200 —— 通了，可以考虑解析数据"
    if status in (401, 403):
        return f"{status} —— 签名过了，只差创作者会话（好消息）"
    return f"{status} —— 未知，需要看响应体"


async def main() -> int:
    import httpx

    ap = argparse.ArgumentParser()
    ap.add_argument("--start-days-ago", type=int, default=30)
    ap.add_argument("--end-days-ago", type=int, default=0)
    ap.add_argument("--app-id", default="ugc", help="参考实现用 ugc；xhshow 默认 xhs-pc-web")
    ap.add_argument("--cookie", default="", help="留空则从 CDP 浏览器读取")
    ap.add_argument("--show-body", action="store_true", help="打印响应前 800 字符")
    ap.add_argument(
        "--no-cookie-header",
        action="store_true",
        help="签名照签（仍需 a1）但不发 cookie 头，用来分清"
        "「空数据是缺会话」还是「接口本身就这样」",
    )
    args = ap.parse_args()

    if args.cookie:
        cookies = dict(
            pair.split("=", 1) for pair in args.cookie.split("; ") if "=" in pair
        )
    else:
        cookies = await cookies_from_cdp()

    print(f"  cookie 条数 {len(cookies)}，名字: {sorted(cookies)}")
    if not cookies.get("a1"):
        print("  ✗ 没有 a1 —— 签名必须用它，无法继续")
        return 2

    query = build_query(args.start_days_ago, args.end_days_ago)
    cookie_header = "; ".join(f"{k}={v}" for k, v in cookies.items())
    headers_common = {
        "user-agent": USER_AGENT,
        "accept": "application/json, text/plain, */*",
        "origin": CREATOR_ORIGIN,
        "referer": f"{CREATOR_ORIGIN}/statistics/data-analysis",
        "accept-language": "zh-CN,zh;q=0.9",
    }
    if not args.no_cookie_header:
        headers_common["cookie"] = cookie_header

    # Three signing variants: the reference bakes the query into the signed string,
    # but the exact form is not documented beyond an example with no query at all.
    variants = {
        "path+q（参考实现写法）": f"url={NOTE_LIST_PATH}?{query}",
        "path only": f"url={NOTE_LIST_PATH}",
        "裸 path+query（无 url= 前缀）": f"{NOTE_LIST_PATH}?{query}",
    }

    url = f"{CREATOR_ORIGIN}{NOTE_LIST_PATH}?{query}"
    async with httpx.AsyncClient(timeout=25, follow_redirects=False) as client:
        for label, api in variants.items():
            signature = sign_xyw(api, cookies["a1"], app_id=args.app_id)
            headers = {**headers_common, **signature}
            try:
                response = await client.get(url, headers=headers)
            except Exception as exc:  # noqa: BLE001
                print(f"  [{label}] 请求异常: {exc.__class__.__name__}: {exc}")
                continue

            print(f"\n  [{label}]")
            print(f"      HTTP {response.status_code}  {classify(response.status_code, response.text)}")
            body = response.text or ""
            if body:
                print(f"      响应前 160 字符: {body[:160]!r}")
            if args.show_body and body:
                print(f"      完整响应: {body[:800]}")

    print("\n  提示：若三种都返回 406，再试 --app-id xhs-pc-web。")
    return 0


if __name__ == "__main__":
    import asyncio

    sys.exit(asyncio.run(main()))