132 lines
4.8 KiB
Python
132 lines
4.8 KiB
Python
"""订单列表页探针：为实现 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,
|
|
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):
|
|
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()))
|