"""Rakuten 加购探针 v2：补 variant_id 与必填选项 修复点： - 多规格商品（inventory_flag=2）：从 sku.variants[] 选第一个非售罄的，把 variant_id 加进 form - 必填选项：从 purchase.information.options[] 选第一个 value，拼成 "名:值" 加进 choice 字段 - 加 Books 测试（普通书籍应该 inventory_flag=1，单一库存） 目的：验证抓取端 PurchaseInfo 契约 + 调用方选 variant/option 后，能否成功加购。 """ from __future__ import annotations import asyncio import json import re import sys from pathlib import Path from urllib.parse import urlparse 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" def save(name: str, content: str) -> Path: path = PROBE_DIR / name path.write_text(content, encoding="utf-8") return path def parse_state(html: str) -> dict | None: m = re.search(r"window\.__INITIAL_STATE__\s*=\s*(.+?);\s*window\.", html, re.DOTALL) if not m: return None try: return json.loads(m.group(1)) except json.JSONDecodeError: return None def extract_full(state: dict) -> dict: purchase = state.get("purchase") or {} sell_type = (purchase.get("sellType") or {}).get("normalPurchase") or {} raw_sku = purchase.get("sku") or {} item = state.get("item") or {} shop = (state.get("shop") or {}).get("information") or {} information = purchase.get("information") or {} basket_domain = (sell_type.get("basketDomain") or "").replace("\\u002F", "/") inventory_type = raw_sku.get("inventoryType") inventory_flag = "2" if inventory_type == "multiple" else "1" shop_id = shop.get("shopId") item_id = item.get("itemId") form_fields: dict[str, str] = { "shop_bid": str(shop_id) if shop_id else "", "item_id": str(item_id) if item_id else "", "inventory_flag": inventory_flag, "__event": "ES01_003_001", } # 多规格：挑第一个非售罄 variant chosen_variant = None variants = raw_sku.get("variants") or [] if inventory_flag == "2" and variants: for v in variants: if not v.get("isSoldOut"): chosen_variant = v break if chosen_variant is None and variants: chosen_variant = variants[0] if chosen_variant: form_fields["variant_id"] = str(chosen_variant.get("variantId") or chosen_variant.get("id") or "") elif inventory_flag == "1" and item.get("variantId"): form_fields["variant_id"] = str(item.get("variantId")) # 必填选项：拼成 "名:値" choice_pairs: list[str] = [] options = information.get("options") or [] required_options = [] for opt in options: if opt.get("isRequired"): required_options.append(opt) values = opt.get("values") or [] if values: first = values[0] choice_pairs.append(f"{opt.get('name')}:{first.get('name')}") if choice_pairs: form_fields["choice"] = ",".join(choice_pairs) return { "basket_domain": basket_domain, "form_fields": form_fields, "quantity_field": "units", "shop_name": shop.get("shopName"), "item_name": (item.get("itemName") or "")[:80], "min_price": sell_type.get("minPrice"), "purchase_condition": sell_type.get("purchaseCondition"), "inventory_type": inventory_type, "variant_count": len(variants), "chosen_variant": chosen_variant.get("variantId") if chosen_variant else None, "required_options": [o.get("name") for o in required_options], "choice_value": form_fields.get("choice"), } async def probe_one(page, context, item_url: str, log: list[str]) -> bool: log.append(f"\n--- {item_url} ---") 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, ) except Exception as exc: log.append(f" load fail: {exc}") return False state = parse_state(await page.content()) if not state: log.append(" state parse fail") return False info = extract_full(state) log.append(f" shop: {info['shop_name']}") log.append(f" item: {info['item_name']}") log.append(f" price: {info['min_price']} condition: {info['purchase_condition']} inv: {info['inventory_type']} variants: {info['variant_count']} chosen_variant: {info['chosen_variant']}") log.append(f" required_options: {info['required_options']} choice: {info['choice_value']}") log.append(f" basket: {info['basket_domain']}") log.append(f" form: {info['form_fields']}") if info["purchase_condition"] != "enabled": log.append(f" SKIP condition={info['purchase_condition']}") return False if info["inventory_type"] == "multiple" and not info["chosen_variant"]: log.append(" SKIP: multi-inventory but no variant chosen") return False if info["required_options"] and not info["choice_value"]: log.append(" SKIP: required options but no value") return False payload = dict(info["form_fields"]) payload[info["quantity_field"]] = "1" resp = await context.request.post( info["basket_domain"], form=payload, max_redirects=5, headers={"Referer": item_url, "Origin": "https://item.rakuten.co.jp"}, ) body = await resp.text() log.append(f" POST status={resp.status} final_url={resp.url}") # 错误页判断 if "/error" in str(resp.url): # 抓错误页里的具体提示 m_msgs = re.findall(r'>([^<>]{20,200})<', body) msgs = [s.strip() for s in m_msgs if s.strip() and not any(c in s for c in ['(', ')', '=', '{', '.', ':'])][:3] log.append(f" ERROR page messages: {msgs}") save(f"err-{urlparse(item_url).path.replace('/', '_')}.html", body) return False # 成功标志 success_markers = ["レジに進む", "買い物かごに追加", "数量", "cart-item", "orderStep"] is_success = any(m in body for m in success_markers) log.append(f" success_markers_hit={is_success}") save(f"ok-{urlparse(item_url).path.replace('/', '_')}.html", body) return is_success async def main() -> int: settings = get_settings() state_path = settings.auth_state_path / "rakuten_state.json" from playwright.async_api import async_playwright log: list[str] = [] async with async_playwright() as pw: browser = await pw.chromium.launch( headless=True, proxy=settings.playwright_proxy, args=["--no-first-run"], ) 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() # 先去 cart 建立 session await page.goto(auth_site.RAKUTEN_CART_URL, wait_until="domcontentloaded", timeout=30_000) await page.wait_for_timeout(1500) # 搜普通市场商品 log.append("=== 搜 NORMAL ===") await page.goto( "https://search.rakuten.co.jp/search/mall/?max=300&min=100&s=1&p=1&v=2", wait_until="domcontentloaded", timeout=30_000, ) await page.wait_for_timeout(2000) normal_urls = list(dict.fromkeys( re.findall(r"https://item\.rakuten\.co\.jp/[\w-]+/[\w-]+/?", await page.content()) ))[:3] log.append(f"normal: {normal_urls}") # 搜 Books log.append("\n=== 搜 BOOKS ===") await page.goto( "https://search.rakuten.co.jp/search/books/?max=500&min=100&s=1&p=1&v=2", wait_until="domcontentloaded", timeout=30_000, ) await page.wait_for_timeout(2000) books_html = await page.content() books_urls = list(dict.fromkeys( re.findall(r"https://books\.rakuten\.co\.jp/rb/\d+/?", books_html) ))[:3] log.append(f"books: {books_urls}") # 加购矩阵 log.append("\n========== 测试 ==========") results: dict[str, list[tuple[str, bool]]] = {"NORMAL": [], "BOOKS": []} for url in normal_urls: ok = await probe_one(page, context, url, log) results["NORMAL"].append((url, ok)) for url in books_urls: ok = await probe_one(page, context, url, log) results["BOOKS"].append((url, ok)) # 汇总 log.append("\n========== 汇总 ==========") for kind, lst in results.items(): ok = sum(1 for _, s in lst if s) log.append(f"{kind}: {ok}/{len(lst)} 成功") for u, s in lst: log.append(f" {'OK' if s else 'FAIL'} {u}") await browser.close() save("probe-v2.txt", "\n".join(log)) return 0 if __name__ == "__main__": raise SystemExit(asyncio.run(main()))