228 lines
8.0 KiB
Python
228 lines
8.0 KiB
Python
"""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()))
|