"""Rakuten 探针：搜低价商品 → 实测 add-to-cart → 跨 cart/checkout 策略（顺序）： 1. 用 Playwright 打开 search.rakuten.co.jp 限定最低价筛选 2. 从结果里逐个取商品 URL 3. 用 Playwright 实际打开商品页，等 "買い物かごに入れる" 按钮 4. 点按钮（不是手工 POST，确保 form 字段都是当前版本） 5. 跟到 cart 后让 SPA 渲染够久 6. 落证据，尝试清理（删除购物车里这件） 避免 httpx（已被 TLS 指纹拦）。全程 Playwright。 """ from __future__ import annotations import asyncio import json 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" / "checkout" PROBE_DIR.mkdir(parents=True, exist_ok=True) 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 collect_item_urls(page, limit: int = 10) -> list[str]: """从搜索结果抽商品 URL（PC 与 SP 都试）""" html = await page.content() # 候选：/shop-code/item-code/、item.rakuten.co.jp、div[data-itemid] urls: list[str] = [] for m in re.findall(r'https://item\.rakuten\.co\.jp/[\w-]+/[\w-]+/?', html): if m not in urls: urls.append(m) if len(urls) >= limit: break return urls async def try_add_item(page, item_url: str) -> tuple[bool, str]: """尝试加购一个商品。返回 (是否成功, 落地 URL)。""" print(f" open item page: {item_url}") try: await page.goto(item_url, wait_until="domcontentloaded", timeout=30_000) except Exception as exc: return False, f"goto failed: {exc}" # 等 __INITIAL_STATE__ 出现 try: await page.wait_for_function( "() => window.__INITIAL_STATE__ && window.__INITIAL_STATE__.purchase", timeout=10_000, ) except Exception: return False, "no __INITIAL_STATE__ (PC 模板/反爬)" # 找加购按钮 candidates = [ 'input[type="submit"][value*="買い物かご"]', 'input[type="submit"][value*="購入"]', 'button:has-text("買い物かご")', 'button:has-text("カートに入れる")', '[data-testid*="add-cart"]', 'button.add-cart', ] btn = None for sel in candidates: loc = page.locator(sel).first try: await loc.wait_for(state="visible", timeout=2000) btn = loc print(f" found button via {sel!r}") break except Exception: continue if btn is None: return False, "no add-to-cart button visible" # 监听导航 try: async with page.expect_navigation(timeout=15_000, wait_until="domcontentloaded"): await btn.click() except Exception as exc: return False, f"click/navigation failed: {exc}" final_url = page.url body = await page.content() # 成功标志：URL 含 cart/add-item 或 confirm；或正文含 レジに進む success = ( "add-item" in final_url or "cart" in final_url or "レジに進む" in body or "買い物かごに追加" in body ) # 失败标志：URL 含 /error if "/error" in final_url: return False, f"error page: {final_url}" return success, final_url async def main() -> int: settings = get_settings() state_path = settings.auth_state_path / "rakuten_state.json" if not state_path.exists(): print(f"找不到 {state_path}") return 1 from playwright.async_api import async_playwright async with async_playwright() as pw: browser = await pw.chromium.launch( headless=True, 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() # 搜索：100~300 円筛选 print("=== 步骤 1：搜低价商品 ===") search_url = "https://search.rakuten.co.jp/search/mall/?f=2&fs=1000&max=300&min=100&s=1&p=1" # 备用关键词：邮费便宜 / 免邮的便宜商品 await page.goto(search_url, wait_until="domcontentloaded", timeout=30_000) await page.wait_for_timeout(2000) save("search-result.html", await page.content()) item_urls = await collect_item_urls(page, limit=10) print(f" 候选商品 {len(item_urls)} 个") for u in item_urls[:5]: print(f" {u}") # 试加购，最多试 5 个 chosen: str | None = None for i, url in enumerate(item_urls[:5]): print(f"\n=== 步骤 2.{i+1}：试加购 {url} ===") ok, info = await try_add_item(page, url) print(f" result: ok={ok} info={info}") if ok: chosen = url save("03-add-success-landing.html", await page.content()) break # 失败：保存页面用于复盘 save(f"03-add-fail-{i+1}.html", await page.content()) if not chosen: print("\n✗ 5 个商品全部加购失败。") await browser.close() return 1 print(f"\n=== 步骤 3：访问 cart，让 SPA 充分渲染 ===") # 先访问 cart await page.goto(auth_site.RAKUTEN_CART_URL, wait_until="domcontentloaded", timeout=30_000) # 等 SPA 把 cart 数据渲染出来：尝试等 "レジに進む" 出现，最多 15 秒 try: await page.wait_for_selector( 'a:has-text("レジに進む"), button:has-text("レジに進む"), [data-testid*="checkout"]', timeout=15_000, ) print(" 'レジに進む' 出现了！") except Exception: print(" 'レジに進む' 15s 内没出现，继续抓现状") cart_html = await page.content() save("04-cart-after-add.html", cart_html) print(f" cart len={len(cart_html)} url={page.url}") # 找 レジに進む href checkout_links = re.findall(r'href=["\']([^"\']*(?:checkout|step1|order/step)[^"\']*)["\']', cart_html, re.IGNORECASE) print(f" checkout links: {sorted(set(checkout_links))[:3]}") # 看 cart count API resp = await context.request.get( "https://cart-api.step.rakuten.co.jp/rms/mall/cart/count/all/jsonp/?sid=1010", headers={"Referer": "https://sp.cart.step.rakuten.co.jp/cart"}, ) print(f" count API: status={resp.status} body={(await resp.text())[:200]!r}") print("\n=== 步骤 4：尝试清理购物车 ===") # 找削除按钮 del_selectors = [ 'a:has-text("削除")', 'button:has-text("削除")', '[data-testid*="delete"]', '[data-testid*="remove"]', ] for sel in del_selectors: loc = page.locator(sel).first try: await loc.wait_for(state="visible", timeout=3000) print(f" 找到删除按钮 {sel!r}，点击") await loc.click() await page.wait_for_timeout(2000) break except Exception: continue else: print(" 没找到删除按钮，请手动清理购物车") save("05-cart-after-cleanup.html", await page.content()) await browser.close() print(f"\n探针输出目录：{PROBE_DIR}") print(f"测试用商品：{chosen}") return 0 if __name__ == "__main__": raise SystemExit(asyncio.run(main()))