rakuten-api/scripts/probe_payment_method.py

"""支付方式选择页真实 HTML 探针

走真实 add_to_cart → verify_cart → enter_checkout 落地到 order-confirmation 后，
该账号是「默认地址+默认卡」直落确认页，不会自然经过 /pay 选择页（与 /ship 地址
确认页被跳过是同一原因）。本探针从确认页上真实存在的「支払い方法」区块「変更」
按钮手动跳回选择页，在这一页上真实调用生产代码 SiteInteractor._select_payment_method，
验证「已有匹配已保存卡 → 点『次へ』继续」这条分支在真实站点上是否成立
（重点验证 _PAYMENT_NEXT_BUTTON_SELECTOR = 'button:has-text("次へ")' ——
这跟 session upgrade 页那个已证实猜错的选择器是同一种写法，未经真实验证）。

**不调用 submit_order / pay**——validate 完成后直接丢弃 page，不产生下单动作。
不主动触发「新しいカードを追加する」代填/提交分支（会在真实账号上注册一张新卡，
需要单独决定是否执行）。

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

import argparse
import asyncio
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.services.auth_session import AuthSession  # noqa: E402
from app.trading.worker.models import LeaseTask  # noqa: E402
from app.trading.worker.site_interact import SiteInteractor  # noqa: E402

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

DEFAULT_ITEM_URL = "https://item.rakuten.co.jp/moccasin/ds001iwrgesaaa2/"


async def _start_visible(site: SiteInteractor, settings) -> None:
    """复制 SiteInteractor.start() 但 headless=False，供本地诊断肉眼看浏览器

    生产 start() 本身已经改成 headless=False 了（2026-08-14 实测确认无头模式
    会让结算 SPA 表现异常），这里保留独立实现只是为了不依赖生产 start() 的
    内部签名，跟 probe_address_confirm.py 保持一致的写法。
    """
    from playwright.async_api import async_playwright

    from app.trading.core import auth_site

    state_path = settings.auth_state_path / auth_site.profile("rakuten").state_filename
    storage_state = str(state_path) if state_path.exists() else None

    site._playwright = await async_playwright().start()
    site._browser = await site._playwright.chromium.launch(
        headless=False,
        channel=settings.browser_channel or None,
        proxy=settings.playwright_proxy,
        args=["--no-first-run", "--disable-blink-features=AutomationControlled"],
    )
    site._context = await site._browser.new_context(
        storage_state=storage_state,
        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,
    )
    print(f"[visible] 浏览器已启动（headless=False），storage_state={storage_state or '(none)'}")


async def run(item_url: str) -> int:
    settings = get_settings()
    auth = AuthSession(settings)
    await auth.start()
    site = SiteInteractor(auth_session=auth, settings=settings)
    await _start_visible(site, settings)

    task = LeaseTask(
        task_id="probe-payment-1",
        site="rakuten",
        intent={"item_url": item_url, "quantity": 1},
    )

    try:
        print("=== add_to_cart ===")
        await site.add_to_cart(task)
        print("=== verify_cart ===")
        await site.verify_cart(task)
        print("=== enter_checkout（预期直落 order-confirmation）===")
        try:
            html = await site.enter_checkout(task)
            print(f"enter_checkout 返回，长度={len(html)}")
        except Exception as exc:
            print(f"enter_checkout 抛出：{type(exc).__name__}: {exc}")
            return 1

        page = site._checkout_pages.get(task.task_id)
        if page is None:
            print("enter_checkout 没有留存 page，无法继续")
            return 1
        print(f"  落地 url={page.url}")
        (PROBE_DIR / "01-order-confirmation.html").write_text(
            await page.content(), encoding="utf-8"
        )

        print('=== 点击「支払い方法」区块的「変更」按钮，跳回支付方式选择页 ===')
        try:
            change_btn = page.locator('text="支払い方法"').first.locator(
                'xpath=following::button[@aria-label="変更"][1]'
            )
            await change_btn.click(timeout=10_000)
            await page.wait_for_timeout(2_000)
        except Exception as exc:
            print(f"点击「変更」失败：{type(exc).__name__}: {exc}")
            return 1
        print(f"  跳转后 url={page.url}")
        html_pay_page = await page.content()
        (PROBE_DIR / "02-payment-method-page.html").write_text(
            html_pay_page, encoding="utf-8"
        )
        print(f"  支付方式选择页已存盘 ({len(html_pay_page)} bytes)")

        print("=== 真实调用 SiteInteractor._select_payment_method ===")
        try:
            await site._select_payment_method(page, task_id=task.task_id)
            print(f"_select_payment_method 成功返回，当前 url={page.url}")
            (PROBE_DIR / "03-after-select-payment-method.html").write_text(
                await page.content(), encoding="utf-8"
            )
        except Exception as exc:
            print(f"_select_payment_method 抛出：{type(exc).__name__}: {exc}")
            (PROBE_DIR / "03-select-payment-method-error.html").write_text(
                await page.content(), encoding="utf-8"
            )
            print("  失败时页面 HTML 已存盘，供排查真实「次へ」按钮结构")

    finally:
        page = site._checkout_pages.pop(task.task_id, None)
        if page is not None:
            print(f"  丢弃 checkout page（不调用 submit_order/pay），最终 url={page.url}")
            await page.close()
        try:
            await site.clear_cart()
        except Exception as exc:  # noqa: BLE001
            print(f"清理购物车失败（忽略）：{type(exc).__name__}: {exc}")
        await site.close()
        await auth.close()

    print(f"\n探针输出目录：{PROBE_DIR}")
    return 0


async def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--item-url", default=DEFAULT_ITEM_URL)
    args = parser.parse_args()
    return await run(args.item_url)


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