"""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()))