实现租约恢复核对：verify_on_site 接真实订单列表反查

新增 SiteInteractor.list_recent_orders（分页拉 order.my.rakuten.co.jp 订单列表，按商品 URL 反查）+ GatewayClient.get_task，替换掉恒返回 UNKNOWN 的桩。 NOT_ORDERED 分支目前只有逻辑验证、没有真实多单数据支撑，刻意仍路由到 needs_human，不自动重新下单。全程只读查询，未触发任何真实付款操作。
2026-08-13 23:58:54 +08:00
parent 51e2538438
commit 93c1a83406
8 changed files with 868 additions and 22 deletions
@@ -0,0 +1,130 @@
+"""订单列表页探针：为实现 verify_on_site（规格 §5 恢复核对）取真实 DOM 结构
+
+只读导航，不做任何加购/下单操作。目标：确认 order.my.rakuten.co.jp 的订单列表页
+（不带 order_number/shop_id 参数）能否按「商品名 + 时间窗口」反查出某个任务是否已
+下单——这是 app/trading/worker/verify.py::verify_on_site 目前的桩要补的实测。
+
+用法：
+    .venv/Scripts/python.exe scripts/probe_order_list.py
+    .venv/Scripts/python.exe scripts/probe_order_list.py --headful
+"""
+from __future__ import annotations
+
+import argparse
+import asyncio
+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" / "order_list"
+PROBE_DIR.mkdir(parents=True, exist_ok=True)
+
+ORDER_LIST_URL = "https://order.my.rakuten.co.jp/purchase-history/"
+KNOWN_ORDER_ID = "306087-20260813-0863947697"
+
+
+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(*, 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
+
+    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()
+
+        api_calls: list[tuple[int, str]] = []
+
+        def on_response(r):
+            url = str(r.url)
+            if any(kw in url for kw in ["purchase-history", "order", "history"]):
+                api_calls.append((r.status, url))
+
+        page.on("response", on_response)
+
+        print("=== 访问订单列表页（不带参数）===")
+        await page.goto(ORDER_LIST_URL, wait_until="domcontentloaded", timeout=30_000)
+        await page.wait_for_timeout(3000)
+        html = await page.content()
+        save("00-list-default.html", html)
+        print(f"  final_url={page.url} body_len={len(html)}")
+        print(f"  含已知订单号 {KNOWN_ORDER_ID}: {KNOWN_ORDER_ID in html}")
+
+        # 尝试常见的日期范围/分页查询参数，看服务端是否支持按时间窗口过滤
+        candidate_qs = [
+            "?period=3months",
+            "?range=3",
+            f"?order_number={KNOWN_ORDER_ID.split('-')[0]}",
+        ]
+        for qs in candidate_qs:
+            url = ORDER_LIST_URL + qs
+            print(f"\n=== 尝试 {url} ===")
+            try:
+                resp = await page.goto(url, wait_until="domcontentloaded", timeout=20_000)
+                await page.wait_for_timeout(2000)
+                html2 = await page.content()
+                status = resp.status if resp else None
+                print(f"  status={status} final_url={page.url} body_len={len(html2)}")
+                safe_name = re.sub(r"[^\w]+", "_", qs) or "root"
+                save(f"01{safe_name}.html", html2)
+            except Exception as exc:
+                print(f"  失败：{type(exc).__name__}: {exc}")
+
+        print("\n=== 捕获的 XHR ===")
+        for status, url in api_calls[-30:]:
+            print(f"  {status} {url[:160]}")
+
+        # 在默认列表页上找商品名/日期相关的文本线索
+        print("\n=== 默认列表页文本线索（商品名/日期候选片段）===")
+        for pat in [
+            r'"itemName"\s*:\s*"([^"]{2,60})"',
+            r'"orderDate"\s*:\s*"([^"]{2,40})"',
+            r'"orderNumber"\s*:\s*"([^"]{2,40})"',
+            r"\d{4}[年/]\d{1,2}[月/]\d{1,2}日?",
+        ]:
+            matches = re.findall(pat, html)
+            print(f"  pattern={pat!r} -> {matches[:5]}")
+
+        await browser.close()
+
+    print(f"\n探针输出目录：{PROBE_DIR}")
+    return 0
+
+
+async def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--headful", action="store_true")
+    args = parser.parse_args()
+    return await run(headful=args.headful)
+
+
+if __name__ == "__main__":
+    raise SystemExit(asyncio.run(main()))