127 lines
4.8 KiB
Python
127 lines
4.8 KiB
Python
"""cart count API 探针：实测「空购物车」与「获取失败」两种响应形态
|
|
|
|
背景：_query_cart_count 目前只认 status=="100"，其余一律 CartOperationError。
|
|
需要真账号确认：购物车为空时 cart-api.step.rakuten.co.jp 到底返回什么——
|
|
如果空车响应与「请求被拒」长的不一样却被同一错误吞掉，就无法区分
|
|
「车是空的」与「获取失败」。
|
|
|
|
只读探测，不做任何加购/删除操作。
|
|
|
|
用法：
|
|
.venv/Scripts/python.exe scripts/probe_cart_count.py
|
|
.venv/Scripts/python.exe scripts/probe_cart_count.py --headful
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
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" / "cart_count"
|
|
PROBE_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
CART_COUNT_API = "https://cart-api.step.rakuten.co.jp/rms/mall/cart/count/all/jsonp/"
|
|
CART_PAGE = auth_site.RAKUTEN_CART_URL
|
|
|
|
|
|
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 call_count_api(context, *, referer: str | None, tag: str) -> None:
|
|
headers = {"Referer": referer} if referer else {}
|
|
resp = await context.request.get(CART_COUNT_API + "?sid=1010", headers=headers)
|
|
body = await resp.text()
|
|
print(f" [{tag}] http_status={resp.status} body={body!r}")
|
|
save(f"{tag}.txt", f"http_status={resp.status}\n{body}")
|
|
|
|
|
|
async def run(*, headful: bool) -> int:
|
|
from playwright.async_api import async_playwright
|
|
|
|
settings = get_settings()
|
|
state_path = settings.auth_state_path / "rakuten_state.json"
|
|
if not state_path.exists():
|
|
print(f"找不到登录态文件：{state_path}")
|
|
return 1
|
|
|
|
async with async_playwright() as pw:
|
|
browser = await pw.chromium.launch(
|
|
headless=not headful,
|
|
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,
|
|
)
|
|
|
|
print("=== 1. 带正确 Referer 调 cart count API（生产路径）===")
|
|
await call_count_api(context, referer=CART_PAGE, tag="10-with-referer")
|
|
|
|
print("=== 2. 不带 Referer 调 cart count API（观测「被拒」形态）===")
|
|
await call_count_api(context, referer=None, tag="11-no-referer")
|
|
|
|
print("=== 3. 错误 sid 调 cart count API（另一种异常入参）===")
|
|
resp = await context.request.get(
|
|
CART_COUNT_API + "?sid=999999", headers={"Referer": CART_PAGE}
|
|
)
|
|
body = await resp.text()
|
|
print(f" [bad-sid] http_status={resp.status} body={body!r}")
|
|
save("12-bad-sid.txt", f"http_status={resp.status}\n{body}")
|
|
|
|
print("=== 4. 打开购物车页，看当前车状态（只读）===")
|
|
page = await context.new_page()
|
|
await page.goto(CART_PAGE, wait_until="domcontentloaded", timeout=30_000)
|
|
await page.wait_for_timeout(5000)
|
|
state = await page.evaluate(
|
|
"""() => {
|
|
const s = window.__INITIAL_STATE__;
|
|
if (!s || !s.cart) return {has_state: !!s, keys: s ? Object.keys(s) : []};
|
|
return {
|
|
has_state: true,
|
|
cart_keys: Object.keys(s.cart),
|
|
shopUrlList_len: Array.isArray(s.cart.shopUrlList) ? s.cart.shopUrlList.length : null,
|
|
};
|
|
}"""
|
|
)
|
|
print(f" __INITIAL_STATE__.cart: {state}")
|
|
html = await page.content()
|
|
save("20-cart-page.html", html)
|
|
delete_btns = await page.locator('button[aria-label^="削除"]').count()
|
|
print(f" 削除按钮数={delete_btns} final_url={page.url}")
|
|
|
|
print("=== 5. 车页打开后再调一次 count API（贴近 clear_cart 末尾场景）===")
|
|
await call_count_api(context, referer=CART_PAGE, tag="21-after-cart-page")
|
|
|
|
await browser.close()
|
|
|
|
print(f"\n探针输出目录：{PROBE_DIR}")
|
|
return 0
|
|
|
|
|
|
async def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--headful", action="store_true")
|
|
args = parser.parse_args()
|
|
return await run(headful=args.headful)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(asyncio.run(main()))
|