"""Rakuten 加购探针：验证抓取端 purchase 块字段 + 官方旗舰店是否可加购 要做的事： 1. 从搜索结果取若干商品 URL（普通 + 旗舰店） 2. 用 Playwright 打开商品页，抽 __INITIAL_STATE__.purchase（与抓取端同样的逻辑） 3. 用同一份 cookie POST basketDomain，看响应 4. 对比抓取端 contract 是否充分、官方店是否走同一端点 输出：.probe/checkout/probe-purchase-block.txt """ 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" PROBE_DIR.mkdir(parents=True, exist_ok=True) # 测试矩阵： # - 普通市场商品（来自搜索） # - 楽天ブックス（books.rakuten.co.jp） # - 楽天ブランドアベニュー（brandavenue.rakuten.co.jp） # - ビックカメラ（biccamera.rakuten.co.jp） NORMAL_URLS = [ "https://item.rakuten.co.jp/livingut/uni437951", "https://item.rakuten.co.jp/waabbit/088-4p", ] BOOKS_URLS = [ "https://books.rakuten.co.jp/rb/17427435/", # 占位：实际跑时换当前有库存的 ] # brandavenue / biccamera 占位（探针跑时若搜索命中会自动用上） 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_purchase_fields(state: dict) -> dict: """与 scraping/parsers/item.py _purchase_info 相同的契约""" 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 {} basket_domain = sell_type.get("basketDomain") or "" 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") variant_id = item.get("variantId") 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", } if inventory_flag == "1" and variant_id: form_fields["variant_id"] = str(variant_id) return { "basket_domain": basket_domain.replace("\\u002F", "/"), "form_fields": form_fields, "quantity_field": "units", "variant_field": "variant_id", "options_field": "choice" if (purchase.get("information") or {}).get("options") else "", "min_price": sell_type.get("minPrice"), "purchase_condition": sell_type.get("purchaseCondition"), "shop_name": shop.get("shopName"), "item_name": item.get("itemName"), } 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) except Exception as exc: log.append(f" goto FAIL: {exc}") return False try: await page.wait_for_function( "() => window.__INITIAL_STATE__ && window.__INITIAL_STATE__.purchase", timeout=10_000, ) except Exception: log.append(" no __INITIAL_STATE__.purchase within 10s") return False html = await page.content() state = parse_state(html) if not state: log.append(" state parse failed") return False fields = extract_purchase_fields(state) log.append(f" basket_domain: {fields['basket_domain']}") log.append(f" form_fields: {fields['form_fields']}") log.append(f" shop: {fields['shop_name']}") log.append(f" item: {(fields['item_name'] or '')[:60]}") log.append(f" price: {fields['min_price']}, condition: {fields['purchase_condition']}") # 跳过卖完的 if fields["purchase_condition"] != "enabled": log.append(f" SKIP: condition={fields['purchase_condition']}") return False payload = dict(fields["form_fields"]) payload[fields["quantity_field"]] = "1" # POST 加购 resp = await context.request.post( fields["basket_domain"], form=payload, max_redirects=5, headers={"Referer": item_url, "Origin": "https://item.rakuten.co.jp"}, ) log.append(f" POST add: final_status={resp.status} url={resp.url}") body = await resp.text() save( f"add-{urlparse(item_url).path.replace('/', '_')}.html", body, ) if "/error" in str(resp.url): log.append(f" -> ERROR page (item may be stale or fields wrong)") return False # 成功标志 success_markers = ["レジに進む", "買い物かごに追加しました", "数量", "cart"] is_success = any(m in body for m in success_markers) log.append(f" success_markers_hit={is_success}") 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, 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() # 先去 sp.cart 建立 session await page.goto(auth_site.RAKUTEN_CART_URL, wait_until="domcontentloaded", timeout=30_000) await page.wait_for_timeout(1500) # 1. 搜普通市场商品（取前 2 个） log.append("=== 搜索普通市场商品 ===") await page.goto( "https://search.rakuten.co.jp/search/mall/?max=500&min=100&s=1&p=1&v=2", wait_until="domcontentloaded", timeout=30_000, ) await page.wait_for_timeout(2000) normal_urls = re.findall( r"https://item\.rakuten\.co\.jp/[\w-]+/[\w-]+/?", await page.content(), ) seen = set() normal_urls = [u for u in normal_urls if not (u in seen or seen.add(u))][:3] log.append(f"候选: {normal_urls}") # 2. 搜 Books log.append("\n=== 搜索楽天ブックス ===") 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 = re.findall( r"https://books\.rakuten\.co\.jp/rb/\d+/?", books_html, ) books_urls = [u for u in books_urls if not (u in seen or seen.add(u))][:2] log.append(f"候选 books: {books_urls}") # 3. 搜 brandavenue log.append("\n=== 搜索楽天ブランドアベニュー ===") await page.goto( "https://search.rakuten.co.jp/search/?sid=brandavenue&max=2000&min=100&s=1&p=1", wait_until="domcontentloaded", timeout=30_000, ) await page.wait_for_timeout(2000) bv_urls = re.findall( r"https://brandavenue\.rakuten\.co\.jp/[\w-]+/[\w-]+/?", await page.content(), ) bv_urls = [u for u in bv_urls if not (u in seen or seen.add(u))][:2] log.append(f"候选 brandavenue: {bv_urls}") # 4. 实际加购测试 log.append("\n========== 加购矩阵测试 ==========") matrix = [ ("NORMAL", normal_urls), ("BOOKS", books_urls), ("BRANDAVENUE", bv_urls), ] results: dict[str, list[tuple[str, bool]]] = {} for kind, urls in matrix: results[kind] = [] for url in urls: ok = await probe_one(page, context, url, log) results[kind].append((url, ok)) # 5. 收尾 log.append("\n========== 汇总 ==========") for kind, lst in results.items(): ok_count = sum(1 for _, ok in lst if ok) log.append(f"{kind}: {ok_count}/{len(lst)} 成功") for url, ok in lst: log.append(f" {'OK' if ok else 'FAIL'} {url}") await browser.close() save("probe-purchase-block.txt", "\n".join(log)) print(f"输出: {PROBE_DIR / 'probe-purchase-block.txt'}") return 0 if __name__ == "__main__": raise SystemExit(asyncio.run(main()))