rakuten-api/scripts/probe_checkout_playwright.py

"""Rakuten 下单流程探针（Playwright 版）

httpx 因 TLS/HTTP2 指纹问题被 Rakuten 拒认（cookie 有效但站点不识别 session）。
这个探针改用 Playwright（headless Chromium + storage_state）跑 cart → 加购 → 校验
→ 确认页，把每步 SPA 渲染后的 HTML 落到 .probe/checkout/，作为 site_interact 实现
的事实依据。

加购是写操作。默认商品是 fixture 里的小额商品（fafachai/10000033，2190 円），
跑完会尝试从购物车删除。可用 --item-url 覆盖。

用法：
    .venv/Scripts/python.exe scripts/probe_checkout_playwright.py
    .venv/Scripts/python.exe scripts/probe_checkout_playwright.py --item-url https://item.rakuten.co.jp/...
    .venv/Scripts/python.exe scripts/probe_checkout_playwright.py --headful  # 调试时看浏览器
"""
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))

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" / "checkout"
PROBE_DIR.mkdir(parents=True, exist_ok=True)

# 默认测试商品：来自 tests/fixtures/item_state.json，2190 円小额商品
DEFAULT_ITEM_URL = "https://item.rakuten.co.jp/fafachai/10000033/"


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


async def run(item_url: str, *, headful: bool) -> 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

    print(f"加载 storage_state: {state_path}")

    async with async_playwright() as pw:
        browser = await pw.chromium.launch(
            headless=not headful,
            channel=settings.browser_channel or None,
            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()

        # 抓所有 XHR，便于看到 SPA 调用了哪些 API
        api_calls: list[tuple[int, str]] = []

        def on_response(r):
            url = str(r.url)
            if any(kw in url for kw in ["cart", "step", "checkout", "basket", "member", "order"]):
                if "r.r10s.jp" not in url:  # 排除静态资源
                    api_calls.append((r.status, url))

        page.on("response", on_response)

        print("\n=== 步骤 1：先去 myrakuten 触发完整 session 建立 ===")
        await page.goto("https://www.rakuten.co.jp/", wait_until="domcontentloaded", timeout=30_000)
        await page.wait_for_timeout(1500)

        print("\n=== 步骤 2：商品详情页 ===")
        await page.goto(item_url, wait_until="domcontentloaded", timeout=30_000)
        await page.wait_for_timeout(2500)
        item_html = await page.content()
        save("00-item-page.html", item_html)

        # 从 __INITIAL_STATE__ 抽 purchase 块
        m = re.search(r"window\.__INITIAL_STATE__\s*=\s*(.+?);\s*window\.", item_html, re.DOTALL)
        if not m:
            print("  ✗ 抽不出 __INITIAL_STATE__（可能 PC 模板或被反爬）")
        else:
            try:
                state = json.loads(m.group(1))
                pu = state.get("purchase", {}).get("sellType", {}).get("normalPurchase", {})
                print(f"  basketDomain: {pu.get('basketDomain')}")
                print(f"  minPrice: {pu.get('minPrice')}")
                save("00-item-state.json", json.dumps(state, ensure_ascii=False, indent=2))
            except json.JSONDecodeError as exc:
                print(f"  state JSON 解析失败：{exc}")

        print("\n=== 步骤 3：点 '買い物かごに入れる' ===")
        # 商品页直接点按钮；fallback 是直接 POST basketDomain
        btn = page.locator('input[type="submit"][value*="買い物かご"], button:has-text("買い物かごに入れる")').first
        try:
            await btn.wait_for(state="visible", timeout=5000)
            print(f"  找到加购按钮，点击")
            await btn.click()
            await page.wait_for_timeout(3000)
            after_add_url = page.url
            print(f"  点击后落地：{after_add_url}")
            after_add_html = await page.content()
            save("03-cart-add-landing.html", after_add_html)
        except Exception as exc:
            print(f"  没找到加购按钮或点击失败：{type(exc).__name__}: {exc}")
            # 直接 POST basketDomain
            print("  fallback：直接构造表单 POST 到 basketDomain")
            # 这块留给 site_interact 实现，探针只看页面行为

        print("\n=== 步骤 4：访问购物车页 ===")
        await page.goto(auth_site.RAKUTEN_CART_URL, wait_until="networkidle", timeout=30_000)
        await page.wait_for_timeout(3000)
        cart_html = await page.content()
        save("04-cart-after-add.html", cart_html)
        print(f"  final_url={page.url} body_len={len(cart_html)}")

        # 检查 SPA 状态
        for kw in ["レジに進む", "注文手続き", "小計", "商品の金額", "itemId", "shopName"]:
            print(f"  contains {kw!r}: {kw in cart_html}")
        m = re.search(r'"isLoggedIn"\s*:\s*(true|false)', cart_html)
        if m:
            print(f"  isLoggedIn={m.group(1)}")

        # 抓 レジ / checkout 链接
        checkout_links = re.findall(r'href=["\']([^"\']*(?:checkout|step1|order)[^"\']*)["\']', cart_html, re.IGNORECASE)
        print(f"  checkout links: {sorted(set(checkout_links))[:3]}")

        print("\n=== 步骤 5：捕获 SPA 调用的 API ===")
        for status, url in api_calls[-30:]:
            print(f"  {status} {url[:140]}")

        # 不在探针里跑 checkout——避免真的进入付款流程
        # 真实 site_interact 实现时，需要点 レジに進む 后看跳转

        print("\n=== 步骤 6：尝试清理购物车 ===")
        # 找删除按钮
        try:
            del_btn = page.locator('a:has-text("削除"), button:has-text("削除"), [data-testid*="delete"]').first
            if await del_btn.count() > 0:
                print("  找到削除按钮，点击")
                await del_btn.click()
                await page.wait_for_timeout(2000)
                print(f"  清理后 url={page.url}")
            else:
                print("  没找到削除按钮（购物车可能本就为空或按钮选择器要调）")
        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("--item-url", default=DEFAULT_ITEM_URL)
    parser.add_argument("--headful", action="store_true")
    args = parser.parse_args()
    return await run(args.item_url, headful=args.headful)


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