rakuten-api/scripts/probe_address_confirm.py

"""收货地址确认页真实 HTML 探针

走 SiteInteractor 的真实生产代码路径（add_to_cart → verify_cart → enter_checkout），
在 _confirm_default_address 被调用前把该页 HTML 落盘，验证/纠正选择器是否与真实页面
一致。**不调用 submit_order / pay**——enter_checkout 落地到确认页就返回，这里直接
丢弃该页，不产生下单动作。

用法：
    .venv/Scripts/python.exe scripts/probe_address_confirm.py
    .venv/Scripts/python.exe scripts/probe_address_confirm.py --item-url https://item.rakuten.co.jp/...
"""
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" / "address_confirm"
PROBE_DIR.mkdir(parents=True, exist_ok=True)

# fafachai/10000033 已下架/不可购买；改用 2026-08-13 真实下单时用过的商品
# （data/evidence/checkout-live-20260813/i00-cart-initial.html 里的 itemUrl）
DEFAULT_ITEM_URL = "https://item.rakuten.co.jp/moccasin/ds001iwrgesaaa2/"


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

    不改动生产代码（生产 start() 硬编码 headless=True），直接绕过它、把结果
    塞进 site 的私有属性——只在这个一次性诊断脚本里这么干。
    """
    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)

    captured: dict[str, str] = {}
    orig_confirm = site._confirm_default_address

    async def spy_confirm(page, *, task_id: str) -> None:
        html = await page.content()
        captured["html"] = html
        captured["url"] = page.url
        path = PROBE_DIR / "address-confirm-page.html"
        path.write_text(html, encoding="utf-8")
        print(f"[捕获] 收货地址确认页 -> {path} ({len(html)} bytes), url={page.url}")
        await orig_confirm(page, task_id=task_id)

    site._confirm_default_address = spy_confirm

    orig_session_upgrade = site._complete_session_upgrade

    async def spy_session_upgrade(page, *, task_id: str) -> None:
        html = await page.content()
        path = PROBE_DIR / "session-upgrade-before.html"
        path.write_text(html, encoding="utf-8")
        print(f"[捕获] session upgrade 页（点提交前）-> {path} ({len(html)} bytes), url={page.url}")
        try:
            await orig_session_upgrade(page, task_id=task_id)
        finally:
            html_after = await page.content()
            (PROBE_DIR / "session-upgrade-after.html").write_text(html_after, encoding="utf-8")

    site._complete_session_upgrade = spy_session_upgrade

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

    try:
        print("=== add_to_cart ===")
        cart_add_html = await site.add_to_cart(task)
        if cart_add_html:
            (PROBE_DIR / "01-cart-add.html").write_text(cart_add_html, encoding="utf-8")
            print(f"  cart-add 页已存盘 ({len(cart_add_html)} bytes)")
        print("=== verify_cart ===")
        verify_html = await site.verify_cart(task)
        if verify_html:
            (PROBE_DIR / "02-cart-verify.html").write_text(verify_html, encoding="utf-8")
            print(f"  cart-verify 页已存盘 ({len(verify_html)} bytes)")
        print("=== enter_checkout（落地到确认页或撞上未捕获的中间步骤为止）===")
        try:
            html = await site.enter_checkout(task)
            print(f"enter_checkout 返回，长度={len(html)}")
            (PROBE_DIR / "03-enter-checkout-final.html").write_text(html, encoding="utf-8")
        except Exception as exc:
            print(f"enter_checkout 抛出：{type(exc).__name__}: {exc}")
        finally:
            # 不调用 submit_order/pay；enter_checkout 若成功会把 page 留存在
            # _checkout_pages，这里直接丢弃，不产生下单动作
            page = site._checkout_pages.pop(task.task_id, None)
            if page is not None:
                print(f"  最终落地 url={page.url}")
                await page.close()

        if "html" not in captured:
            print("\n没有触发 _confirm_default_address——账号可能没走到这一步"
                  "（比如没有电话补录/地址确认步骤，或选择器判断条件本身有问题）")
    finally:
        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()))