rakuten-api/scripts/probe_order_detail.py

"""订单详情页探针：为账号只读查询通道（docs/order-gateway.md §11）取详情页真实 DOM 结构

目标：`fetch_order_detail` 目前只把详情页 `window.__INITIAL_STATE__` 原样透传
（从未有过真实样本，只做「解析得动就带走」，见 site_interact.py::_parse_initial_state
上方与 OrderDetailSnapshot 文档）。本探针用真账号登录态打开一条真实订单的详情页
（`_ORDER_DETAIL_URL_TEMPLATE` 那条 `act=detail_page_view` 路径），把整页 HTML、
以及解析出的 `__INITIAL_STATE__` JSON 都落盘，作为后续补规范化字段抽取的第一份
真实样本。

只读导航，不做任何加购/下单操作。

用法：
    .venv/Scripts/python.exe scripts/probe_order_detail.py
    .venv/Scripts/python.exe scripts/probe_order_detail.py --headful
    .venv/Scripts/python.exe scripts/probe_order_detail.py --order 306087-20260813-0863947697
"""
from __future__ import annotations

import argparse
import asyncio
import json
import re
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

# Windows 控制台默认 GBK，打印含日文/波浪符（如 U+301C 〜）的中文会抛
# UnicodeEncodeError——统一改 stdout 为 UTF-8，跟 save() 落盘的编码保持一致
# （记忆 project://jp-rakuten/index 里「Git Bash 打中文 body 乱码」是同一问题的另一面）
if hasattr(sys.stdout, "reconfigure"):
    try:
        sys.stdout.reconfigure(encoding="utf-8")
    except Exception:
        pass

from app.shared.config import get_settings  # noqa: E402
from app.trading.core import auth_site  # noqa: E402

PROBE_DIR = Path(__file__).resolve().parent.parent / ".probe" / "order_detail"
PROBE_DIR.mkdir(parents=True, exist_ok=True)

# 与 site_interact.py::_ORDER_DETAIL_URL_TEMPLATE 保持一致
ORDER_DETAIL_URL_TEMPLATE = (
    "https://order.my.rakuten.co.jp/purchase-history/"
    "?order_number={order_number}&shop_id={shop_id}&act=detail_page_view"
)
DEFAULT_ORDER_ID = "306087-20260813-0863947697"


def save(name: str, content: str | bytes) -> Path:
    path = PROBE_DIR / name
    if isinstance(content, bytes):
        path.write_bytes(content)
    else:
        path.write_text(content, encoding="utf-8")
    print(f"  saved -> {path} ({len(content)} bytes)")
    return path


def parse_initial_state(html: str) -> dict | None:
    m = re.search(
        r"window\.__INITIAL_STATE__\s*=\s*(.+?);\s*window\.",
        html,
        re.DOTALL,
    )
    if not m:
        return None
    try:
        return json.loads(m.group(1))
    except json.JSONDecodeError:
        return None


async def run(*, headful: bool, order_id: str) -> int:
    from playwright.async_api import async_playwright

    settings = get_settings()
    state_path = settings.auth_state_path / "rakuten_state.json"
    if not state_path.exists():
        print(f"找不到登录态文件：{state_path}")
        return 1

    shop_id = order_id.split("-", 1)[0]
    url = ORDER_DETAIL_URL_TEMPLATE.format(order_number=order_id, shop_id=shop_id)

    async with async_playwright() as pw:
        browser = await pw.chromium.launch(
            headless=not headful,
            channel=settings.browser_channel or None,
            proxy=settings.playwright_proxy,
            args=["--no-first-run", "--disable-blink-features=AutomationControlled"],
        )
        context = await browser.new_context(
            storage_state=state_path,
            user_agent=auth_site.RAKUTEN_USER_AGENT,
            locale="ja-JP",
            timezone_id="Asia/Tokyo",
            viewport={"width": 390, "height": 844},
            is_mobile=True,
            has_touch=True,
        )
        page = await context.new_page()

        api_calls: list[tuple[int, str]] = []

        def on_response(r):
            api_calls.append((r.status, str(r.url)))

        page.on("response", on_response)

        print(f"=== 访问订单详情页（order_id={order_id}）===")
        try:
            await page.goto(url, wait_until="domcontentloaded", timeout=30_000)
            await page.wait_for_timeout(3000)
        except Exception as exc:
            print(f"导航失败：{type(exc).__name__}: {exc}")

        html = await page.content()
        final_url = page.url
        print(f"  final_url={final_url} body_len={len(html)}")
        print(f"  含订单号 {order_id}: {order_id in html}")
        logged_out = auth_site.looks_logged_out(
            "rakuten", final_url=final_url, body=html
        )
        print(f"  looks_logged_out={logged_out}")

        save("00-detail-default.html", html)

        state = parse_initial_state(html)
        if state is None:
            print("  !! 解析不到 __INITIAL_STATE__（可能 PC 模板 / 掉登录 / 反爬）")
        else:
            print(f"  __INITIAL_STATE__ 顶层键：{sorted(state.keys())}")
            print(f"  pageType={state.get('pageType')!r}")

            # 打印与订单/阶段相关的子结构概览，方便人眼核对
            print("\n=== 与订单字段相关的线索 ===")
            flat_search = {}
            def walk(node, prefix=""):
                if isinstance(node, dict):
                    for k, v in node.items():
                        p = f"{prefix}.{k}" if prefix else k
                        if isinstance(v, (dict, list)):
                            walk(v, p)
                        elif isinstance(v, (str, int, float)) and 0 < len(str(v)) <= 80:
                            flat_search[p] = v
                elif isinstance(node, list):
                    for i, v in enumerate(node[:5]):
                        walk(v, f"{prefix}[{i}]")
            walk(state)
            for pat in ["order", "stage", "配送", "出荷", "配達", "金額", "amount",
                       "address", "payment", "shop", "item", "status"]:
                hits = {k: v for k, v in flat_search.items() if pat in k.lower()}
                if hits:
                    print(f"  键含 {pat!r}（前 15 条）：")
                    for k, v in list(hits.items())[:15]:
                        print(f"    {k} = {v!r}")

            save("01-detail-initial-state.json", json.dumps(
                state, ensure_ascii=False, indent=2,
            ))
            # 顶层平铺一份，方便快速看
            save("02-detail-initial-state-top.json", json.dumps(
                {k: (v if not isinstance(v, (dict, list)) else "…") for k, v in state.items()},
                ensure_ascii=False, indent=2,
            ))

        # 配送阶段进度条（_ORDER_STEPPER_ITEM_PATTERN 那套），确认详情页同样适用
        steps = re.findall(
            r'<li class="([^"]*)">.*?<div class="title--2uGVi">([^<]*)</div></li>',
            html,
            re.DOTALL,
        )
        print(f"\n=== 配送阶段进度条（_ORDER_STEPPER_ITEM_PATTERN）===")
        if steps:
            for cls, label in steps:
                print(f"  stage={label!r} active={'-active--' in cls}")
        else:
            print("  （未匹配到进度条 li 结构）")

        print("\n=== 捕获的 XHR（末 20 条）===")
        for status, u in api_calls[-20:]:
            print(f"  {status} {u[:160]}")

        # 截图留底（无头模式下可能是白屏/异常渲染，正好暴露反爬）
        try:
            await page.screenshot(path=str(PROBE_DIR / "03-detail-screenshot.png"), full_page=True)
            print("  截图 ->", PROBE_DIR / "03-detail-screenshot.png")
        except Exception as exc:
            print(f"  截图失败：{type(exc).__name__}: {exc}")

        await browser.close()

    print(f"\n探针输出目录：{PROBE_DIR}")
    return 0


async def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--headful", action="store_true")
    parser.add_argument(
        "--order",
        default=DEFAULT_ORDER_ID,
        help="要探测的订单号（默认已知订单 2026-08-13 实测那单）",
    )
    args = parser.parse_args()
    return await run(headful=args.headful, order_id=args.order)


if __name__ == "__main__":
    raise SystemExit(asyncio.run(main()))