rakuten-api/scripts/probe_option_choice.py

"""必填选项 choice 取值的真账号验证：占位项修复是否真的被站点接受

背景：2026-08-28 发现 trading 自动填 choice 时取 `values[0]`，而必填 select 的
values[0] 恒为 id=0 的「選択してください」占位项，等于把「请选择」当答案提交。
修复后取第一个非占位候选（见 app/shared/purchase_contract.py::auto_choice_for）。

离线测试只能证明「我们填的值变了」，证明不了「站点接受这个值」——后者必须真账号
实测。本探针就为这一件事。

**刻意走生产代码路径**（SiteInteractor.add_to_cart_payload），不重写字段构造逻辑：
上一版探针 scripts/probe_purchase_block_v2.py 自己抄了一遍 form 构造，结果那份
抄写与生产代码一起用了 values[0]，两边同错就测不出问题。实际提交的 payload 从
生产代码自己的 INFO 日志里抓（「加购请求：basket=... payload=...」），确保记录的
就是真正发出去的东西。

安全边界：
- 只做「清空购物车 → 加购 → 校验 → 清空购物车」，**绝不进入结算/支付**
- 加购完立即清空，账号购物车恢复原状（加购本身可逆，不产生订单、不扣款）
- 不触碰 enter_checkout / submit_order / pay

用法：
    .venv/Scripts/python.exe scripts/probe_option_choice.py
"""
from __future__ import annotations

import asyncio
import json
import logging
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.shared.errors import AppError  # noqa: E402
from app.shared.purchase_contract import auto_choice_for, parse_options  # noqa: E402
from app.trading.services.auth_session import AuthSession  # noqa: E402
from app.trading.worker.site_interact import (  # noqa: E402
    SiteInteractor,
    _parse_initial_state,
)

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

# 目标商品：从 .probe/checkout/ 的历史样本里挑「有必填选项且占位项在 values[0]」的两个。
# 这两个正是旧实现填「選択してください」也被站点收下的商品——它们不能证明旧值是对的
# （店铺没做校验），但可以验证新值同样被接受，且是站点自己给的唯一非占位候选。
TARGETS = (
    "https://item.rakuten.co.jp/waabbit/088-4p",
    "https://item.rakuten.co.jp/aoimorinomise/121",
)


class _PayloadCapture(logging.Handler):
    """截获生产代码 `加购请求：basket=%s payload=%s` 这条 INFO 日志

    目的是记录**真正发出去的 payload**，而不是探针自己再算一遍。
    """

    def __init__(self) -> None:
        super().__init__(level=logging.INFO)
        self.payloads: list[dict] = []

    def emit(self, record: logging.LogRecord) -> None:
        if record.msg == "加购请求：basket=%s payload=%s":
            basket, payload = record.args
            self.payloads.append({"basket_domain": basket, "payload": dict(payload)})


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


async def inspect_options(site: SiteInteractor, item_url: str) -> list[dict]:
    """只读打开商品页，把解析出的选项结构落盘，供人工核对占位项判定是否符合真实页面"""
    page = await site._context.new_page()
    try:
        await page.goto(item_url, wait_until="domcontentloaded", timeout=30_000)
        await page.wait_for_function(
            "() => window.__INITIAL_STATE__ && window.__INITIAL_STATE__.purchase",
            timeout=10_000,
        )
        html = await page.content()
    finally:
        await page.close()

    state = _parse_initial_state(html) or {}
    information = (state.get("purchase") or {}).get("information") or {}
    options = parse_options(information)
    auto_choice, unfillable = auto_choice_for(options)

    slug = item_url.rstrip("/").replace("https://item.rakuten.co.jp/", "").replace("/", "_")
    save(
        f"options-{slug}.json",
        json.dumps(
            {
                "item_url": item_url,
                "raw_options": information.get("options"),
                "parsed": [
                    {
                        "name": option.name,
                        "type": option.type,
                        "is_required": option.is_required,
                        "values": [
                            {
                                "value_id": value.value_id,
                                "name": value.name,
                                "is_placeholder": value.is_placeholder,
                            }
                            for value in option.values
                        ],
                    }
                    for option in options
                ],
                "auto_choice": auto_choice,
                "unfillable_required_options": unfillable,
            },
            ensure_ascii=False,
            indent=1,
        ),
    )

    for option in options:
        if not option.is_required:
            continue
        names = [
            f"{value.name}{'(占位)' if value.is_placeholder else ''}" for value in option.values
        ]
        print(f"    必填项「{option.name[:30]}」候选={names}")
    print(f"    auto_choice = {auto_choice!r}")
    if unfillable:
        print(f"    !! 自动填不了的必填项：{unfillable}")
    return [{"name": o.name, "is_required": o.is_required} for o in options]


async def run() -> int:
    settings = get_settings()
    capture = _PayloadCapture()
    logging.basicConfig(level=logging.WARNING)
    # 必须显式把这个 logger 抬到 INFO：basicConfig 把 root 设成 WARNING，而目标
    # logger 自身是 NOTSET，有效级别继承 root——INFO 记录会在到达 handler 之前
    # 就被丢掉，capture 全程收不到东西（第一次跑就是这样，choice 打印成 None）
    target_logger = logging.getLogger("app.trading.worker.site_interact")
    target_logger.setLevel(logging.INFO)
    target_logger.addHandler(capture)

    auth = AuthSession(settings)
    await auth.start()

    print("=== 0. 登录态探测（storage_state 落盘于 2026-08-11，可能已过期）===")
    status = await auth.check("rakuten")
    print(f"    logged_in={status.logged_in} detail={status.detail}")
    if not status.logged_in:
        print("    登录态已失效——需要先跑 scripts/login.py 重新登录（可能要人工过验证码）")
        await auth.close()
        return 2

    site = SiteInteractor(auth_session=auth, settings=settings)
    await site.start()

    results: list[dict] = []
    try:
        print("\n=== 1. 清空购物车（排除历史残留污染校验）===")
        cleared = await site.clear_cart()
        print(f"    removed={cleared['removed_count']} cart_count={cleared['cart_count']}")

        for index, item_url in enumerate(TARGETS, start=1):
            print(f"\n=== 2.{index} {item_url} ===")
            print("  -- 只读核对选项结构 --")
            try:
                await inspect_options(site, item_url)
            except Exception as exc:  # noqa: BLE001
                print(f"    选项结构读取失败（不影响加购验证）：{type(exc).__name__}: {exc}")

            print("  -- 走生产路径加购（add_to_cart_payload）--")
            before = len(capture.payloads)
            entry: dict = {"item_url": item_url}
            try:
                added = await site.add_to_cart_payload(item_url=item_url, quantity=1)
                entry["ok"] = True
                entry["item_id"] = added["item_id"]
                entry["cart_count"] = added["cart_count"]
                print(
                    f"    加购成功 item_id={added['item_id']} cart_count={added['cart_count']}"
                )
                save(f"add-response-{index}.html", added.get("response_html") or "")
            except AppError as exc:
                entry["ok"] = False
                entry["error"] = exc.message
                print(f"    加购失败：{exc.message}")
            except Exception as exc:  # noqa: BLE001
                entry["ok"] = False
                entry["error"] = f"{type(exc).__name__}: {exc}"
                print(f"    加购异常：{type(exc).__name__}: {exc}")

            # 真正发出去的 payload（从生产代码日志截获）
            sent = capture.payloads[before:]
            if not sent:
                # 抓不到就等于这次验证什么也没证明——必须显眼报出来，不能让
                # 「加购成功」把它盖过去（第一次跑就是 logger 级别没抬导致静默为 None）
                entry["capture_failed"] = True
                print("    !! 未截获到 payload 日志，本次无法确认实际提交的 choice")
            else:
                entry["sent_payload"] = sent[-1]["payload"]
                choice = sent[-1]["payload"].get("choice")
                print(f"    实际提交 choice = {choice!r}")
                if choice is None:
                    print("    !! payload 里没有 choice 字段")
                elif "選択してください" in choice:
                    print("    !! 提交值里仍含占位项——修复未生效")
            results.append(entry)

        print("\n=== 3. 购物车状态（确认商品真的进车了）===")
        try:
            status_after = await site.cart_status()
            print(f"    {status_after}")
        except AppError as exc:
            print(f"    查询失败：{exc.message}")

        print("\n=== 4. 清空购物车（恢复账号原状，不进入结算）===")
        final = await site.clear_cart()
        print(f"    removed={final['removed_count']} cart_count={final['cart_count']}")
    finally:
        await site.close()
        await auth.close()

    save("summary.json", json.dumps(results, ensure_ascii=False, indent=1))
    print("\n========== 汇总 ==========")
    for entry in results:
        mark = "OK  " if entry.get("ok") else "FAIL"
        choice = (entry.get("sent_payload") or {}).get("choice")
        print(f"  {mark} {entry['item_url']}")
        print(f"       choice={choice!r}")
        if not entry.get("ok"):
            print(f"       error={entry.get('error')}")
    print(f"\n探针输出目录：{PROBE_DIR}")
    return 0


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