255 lines
9.2 KiB
Python
255 lines
9.2 KiB
Python
"""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()))
|