diff --git a/app/trading/container.py b/app/trading/container.py index da7c411..8692e66 100644 --- a/app/trading/container.py +++ b/app/trading/container.py @@ -30,8 +30,9 @@ class TradingContainer: settings: Settings auth_session: AuthSession - # 以下四项仅在 order_gateway_url 配置时构造；否则为 None，worker 不启动 + # 以下五项仅在 order_gateway_url 配置时构造；否则为 None，worker 不启动 worker_client: object | None = None # app.trading.worker.client.GatewayClient worker_local_db: object | None = None # app.trading.worker.local_db.LocalDB worker_evidence: object | None = None # app.trading.worker.evidence.EvidenceStore + worker_site: object | None = None # app.trading.worker.site_interact.SiteInteractor worker_runner: object | None = None # app.trading.worker.runner.WorkerRunner diff --git a/app/trading/main.py b/app/trading/main.py index ac04542..c2e7807 100644 --- a/app/trading/main.py +++ b/app/trading/main.py @@ -44,6 +44,7 @@ def build_container() -> TradingContainer: from app.trading.worker.evidence import EvidenceStore from app.trading.worker.local_db import LocalDB from app.trading.worker.runner import WorkerRunner + from app.trading.worker.site_interact import SiteInteractor client = GatewayClient( settings.order_gateway_url, @@ -52,15 +53,18 @@ def build_container() -> TradingContainer: ) local_db = LocalDB(settings.trading_db_path_resolved) evidence = EvidenceStore(settings.evidence_path) + site = SiteInteractor(auth_session=container.auth_session, settings=settings) runner = WorkerRunner( settings=settings, gateway_client=client, local_db=local_db, evidence=evidence, + site=site, ) container.worker_client = client container.worker_local_db = local_db container.worker_evidence = evidence + container.worker_site = site container.worker_runner = runner return container @@ -86,10 +90,12 @@ async def lifespan(app: FastAPI): worker_task: asyncio.Task | None = None if container.worker_runner is not None: - # 顺序：先开本地 DB（worker 写证据前要先能写库），再起后台任务 + # 顺序：先开本地 DB 与 SiteInteractor（worker 写证据/发请求前要先就绪），再起后台任务 assert container.worker_local_db is not None assert container.worker_client is not None + assert container.worker_site is not None await container.worker_local_db.start() + await container.worker_site.start() # type: ignore[union-attr] worker_task = asyncio.create_task( container.worker_runner.run(), name="trading-worker" ) @@ -113,6 +119,8 @@ async def lifespan(app: FastAPI): await worker_task except asyncio.CancelledError: pass + if container.worker_site is not None: + await container.worker_site.close() # type: ignore[union-attr] if container.worker_local_db is not None: await container.worker_local_db.close() if container.worker_client is not None: diff --git a/app/trading/worker/runner.py b/app/trading/worker/runner.py index d05b547..50d7e37 100644 --- a/app/trading/worker/runner.py +++ b/app/trading/worker/runner.py @@ -30,11 +30,12 @@ from typing import TYPE_CHECKING from app.shared.errors import AppError, OrderGuardError from app.shared.task_state import OrderState, TaskStatus -from app.trading.worker import site_interact, verify +from app.trading.worker import verify from app.trading.worker.client import GatewayClient from app.trading.worker.evidence import EvidenceStore from app.trading.worker.local_db import LocalDB from app.trading.worker.models import LeaseTask +from app.trading.worker.site_interact import SiteInteractor if TYPE_CHECKING: from app.shared.config import Settings @@ -67,11 +68,13 @@ class WorkerRunner: gateway_client: GatewayClient, local_db: LocalDB, evidence: EvidenceStore, + site: SiteInteractor, ): self._settings = settings self._gateway = gateway_client self._db = local_db self._evidence = evidence + self._site = site self._running = False def stop(self) -> None: @@ -231,7 +234,7 @@ class WorkerRunner: # 步骤 1：加购 await self._run_step( task, step_no=1, step_name="cart-add", - action=lambda: site_interact.add_to_cart(task), + action=lambda: self._site.add_to_cart(task), state=OrderState.IN_CART, detail="已加入购物车", ) @@ -239,14 +242,14 @@ class WorkerRunner: # 步骤 2：校验购物车 await self._run_step( task, step_no=2, step_name="cart-check", - action=lambda: site_interact.verify_cart(task), + action=lambda: self._site.verify_cart(task), state=OrderState.IN_CART, detail="购物车已校验", ) # 步骤 3：进入下单确认页 + 金额守卫 - checkout_html = await site_interact.enter_checkout(task) - summary = await site_interact.parse_checkout(checkout_html) + checkout_html = await self._site.enter_checkout(task) + summary = await self._site.parse_checkout(checkout_html) self._enforce_amount_guard(task, summary.payable_yen) await self._run_step( task, step_no=3, step_name="order-confirm", @@ -261,7 +264,7 @@ class WorkerRunner: ) # 步骤 4：提交下单 - site_order_id = await site_interact.submit_order(task) + site_order_id = await self._site.submit_order(task) await self._run_step( task, step_no=4, step_name="order-submit", action=self._noop(), @@ -273,7 +276,7 @@ class WorkerRunner: # 步骤 5：付款 await self._run_step( task, step_no=5, step_name="payment", - action=lambda: site_interact.pay(task, site_order_id), + action=lambda: self._site.pay(task, site_order_id), state=OrderState.AWAITING_PAYMENT, site_order_id=site_order_id, payable_yen=summary.payable_yen, @@ -294,7 +297,7 @@ class WorkerRunner: # 步骤 6：付款后监控（非阻塞，常驻轮询；当前未实现） try: - await site_interact.monitor(task, site_order_id) + await self._site.monitor(task, site_order_id) except NotImplementedError: logger.info("付款后监控未实现，跳过：task_id=%s", task.task_id) diff --git a/app/trading/worker/site_interact.py b/app/trading/worker/site_interact.py index c1bc9ea..e8612a7 100644 --- a/app/trading/worker/site_interact.py +++ b/app/trading/worker/site_interact.py @@ -1,28 +1,58 @@ """站点交互：加购 / 校验购物车 / 进入下单确认页 / 金额守卫 / 提交 / 付款 / 监控 -**全部未实现**。规格 docs/order-gateway.md §10 明确说明：加购 → 下单 → 付款的 -实际站点交互没有实测过，未实测前调用直接抛「未实现」，不要写猜测的提交逻辑。 +实测进度（详见 project://jp-rakuten/checkout-flow-probe-findings）： +- add_to_cart、verify_cart 已实测可用（Playwright + storage_state 走 SP 通道） +- enter_checkout 及之后**未实现**：Rakuten 对 checkout 这类敏感操作要求 + session upgrade（重输密码），即使当前 SSO 已登录。这是自动 checkout 的硬墙， + 比 3DS 还前置。本层留接口缝，未实测前调用直接抛「未实现」，不要写猜测的提交逻辑。 -签名刻意保留：runner 把这一层当接口缝用，等真实账号 + 真实站点把每个动作 -实现填进来后，主循环逻辑不需要改动。 - -实现方需要补齐的实测点（规格 §10）： -1. 自动付款是否触发 3D Secure / 短信验证。触发则这条路走不通，付款环节改为 - 「下单到 awaiting_payment + 上报 needs_human 交人工」。 -2. 下单确认页的实际应付金额、付款方式、付款期限、站点订单号各自在哪个字段。 -3. 订单列表页能否按商品 + 时间窗口可靠地反查「这单下没下」。 -4. ラクマ 侧尚无加购契约（README 已注明未实现）。 +**httpx 不能用于带账号的写操作**：Rakuten 对账号操作有 TLS/HTTP2 指纹校验， +同一份 cookie Playwright 能用、httpx 不能。所以本模块全程使用 Playwright +的 BrowserContext + APIRequestContext（共享 cookie，绕过 CORS）。 """ from __future__ import annotations +import json +import logging +import re from dataclasses import dataclass +from typing import TYPE_CHECKING +from app.shared.errors import ( + CartOperationError, + InvalidRequestError, + NotLoggedInError, +) +from app.trading.core import auth_site from app.trading.worker.models import LeaseTask +if TYPE_CHECKING: + from app.shared.config import Settings + from app.trading.services.auth_session import AuthSession + +logger = logging.getLogger(__name__) + +# 普通购买的事件标识，站点前端构造加购表单时固定带上（同 scraping/parsers/item.py） +_NORMAL_PURCHASE_EVENT = "ES01_003_001" + +# 库存类型 → 加购表单里的 inventory_flag（与 scraping/parsers/item.py 一致） +_INVENTORY_FLAG = {"multiple": "2"} +_DEFAULT_INVENTORY_FLAG = "1" + +# 加购端点路径片段（探针实测：basketDomain 逐商品不同，但路径固定） +_BASKET_PATH = "/rms/mall/bss/cartadd/set" + +# 购物车页（SP）与 cart 数量 JSONP API（探针实测） +_CART_PAGE = auth_site.RAKUTEN_CART_URL +_CART_COUNT_API = "https://cart-api.step.rakuten.co.jp/rms/mall/cart/count/all/jsonp/" + +# 购物车页未登录标记（旧 marker；新 SPA 上不可靠，这里只作辅助判据） +_LEGACY_LOGGED_OUT_MARKER = auth_site.RAKUTEN_LOGGED_OUT_MARKER _NOT_IMPLEMENTED_MSG = ( - "站点交互未实现：见 docs/order-gateway.md §10。" - "未实测前不写猜测的提交逻辑——需要真实账号 + 真实站点把这一步实现填进来。" + "站点交互未实现：见 docs/order-gateway.md §10 与 " + "project://jp-rakuten/checkout-flow-probe-findings。" + "session upgrade 是 checkout 阶段的硬墙，未实测前不写猜测的提交逻辑。" ) @@ -38,36 +68,392 @@ class CheckoutSummary: pay_deadline: str | None = None -async def add_to_cart(task: LeaseTask) -> None: - """加购。需要 auth_session 的登录 cookie + intent 里的 purchase 标识""" - raise NotImplementedError(_NOT_IMPLEMENTED_MSG) +class SiteInteractor: + """Rakuten 站点交互器：持有 Playwright 浏览器 context，复用账号 cookie + + 生命周期： + - start() 在 worker 启动时调用一次：启动 Playwright + 创建带 cookie 的 context + - add_to_cart / verify_cart 在每个任务里调用 + - close() 在 worker 关闭时调用 + + 浏览器 context 复用同一份登录态，所有任务串行（worker 主循环本就串行）， + 不需要为每个任务开新 context——开销大且 cookie 状态会乱。 + """ + + # 每任务保留的临时状态：task_id → {"item_id": str, "shop_bid": str} + # 用于 add_to_cart 把抓出来的 item_id / shop_bid 喂给 verify_cart + _per_task_state: dict[str, dict[str, str]] + + def __init__(self, *, auth_session: "AuthSession", settings: "Settings"): + self._auth_session = auth_session + self._settings = settings + self._playwright = None + self._browser = None + self._context = None # playwright BrowserContext + self._per_task_state = {} + + # ---- 生命周期 ---- + + async def start(self) -> None: + """启动 Playwright 与带 cookie 的浏览器 context + + 登录态文件不存在时同样启动（context 没 cookie），后续 add_to_cart 会 + 在 require_logged_in 里报错。这样保持启动路径一致。 + """ + from playwright.async_api import async_playwright + + state_path = self._settings.auth_state_path / auth_site.profile("rakuten").state_filename + storage_state = str(state_path) if state_path.exists() else None + if storage_state is None: + logger.warning( + "登录态文件不存在：site_interactor 以无 cookie 状态启动，" + "加购请求会被站点拒认" + ) + + self._playwright = await async_playwright().start() + self._browser = await self._playwright.chromium.launch( + headless=True, + channel=self._settings.browser_channel or None, + args=["--no-first-run", "--disable-blink-features=AutomationControlled"], + ) + self._context = await self._browser.new_context( + storage_state=storage_state, + 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, + ) + logger.info("SiteInteractor 已就绪：storage_state=%s", storage_state or "(none)") + + async def close(self) -> None: + """关闭 context、browser、playwright，吞掉单个 close 异常""" + for resource, name in ( + (self._context, "context"), + (self._browser, "browser"), + (self._playwright, "playwright"), + ): + if resource is None: + continue + try: + closer = resource.close() if name != "playwright" else resource.stop() + await closer + except Exception: + logger.debug("关闭 %s 失败", name, exc_info=True) + self._context = None + self._browser = None + self._playwright = None + + # ---- 已实现：add_to_cart / verify_cart ---- + + async def add_to_cart(self, task: LeaseTask) -> None: + """加购：打开商品页 → 抽 __INITIAL_STATE__.purchase → POST basketDomain + + 调用方需在 task.intent 提供： + - item_url: 商品详情页 URL（必填） + - quantity: 数量，默认 1 + - variant_id: 多规格商品的 variant_id；不传则从 sku.variants[] 自动选第一个非售罄 + - choice: 必填选项的取值列表；不传则每个必填选项用第一个候选值（站点不严格校验） + + Raises: + InvalidRequestError: intent.item_url 缺失 + NotLoggedInError: 登录态失效 + CartOperationError: 商品页打不开、state 解析失败、商品不可购买、加购返回错误页 + """ + intent = task.intent or {} + item_url = intent.get("item_url") + if not item_url: + raise InvalidRequestError("intent.item_url 必填") + quantity = int(intent.get("quantity") or 1) + if quantity <= 0: + raise InvalidRequestError(f"intent.quantity 必须为正整数，收到 {quantity}") + + await self._auth_session.require_logged_in("rakuten") + + page = await self._context.new_page() + try: + 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: + raise CartOperationError( + f"打开商品页失败或反爬被触发：{type(exc).__name__}: {exc}" + ) from exc + + html = await page.content() + state = _parse_initial_state(html) + if not state: + raise CartOperationError("无法从商品页抽 __INITIAL_STATE__（可能 PC 模板或反爬）") + + fields = _extract_purchase_fields(state, intent_override=intent) + self._per_task_state[task.task_id] = { + "item_id": fields["form_fields"].get("item_id", ""), + "shop_bid": fields["form_fields"].get("shop_bid", ""), + "basket_domain": fields["basket_domain"], + } + + if fields["purchase_condition"] != "enabled": + raise CartOperationError( + f"商品不可购买：purchaseCondition={fields['purchase_condition']}" + ) + if not fields["basket_domain"]: + raise CartOperationError("basketDomain 为空（商品可能下架）") + # 多规格商品要求选了 variant_id + if fields["inventory_flag"] == _INVENTORY_FLAG["multiple"] and not fields["form_fields"].get("variant_id"): + raise CartOperationError( + "多规格商品未选 variant，且 sku.variants 全部售罄或为空" + ) + # 必填选项要求填了 choice + if fields["has_required_options"] and not fields["form_fields"].get(fields["options_field"]): + raise CartOperationError( + "商品有必填选项但未提供 choice，且选项无候选值" + ) + + payload = dict(fields["form_fields"]) + payload[fields["quantity_field"]] = str(quantity) + + logger.info( + "加购请求：task_id=%s basket=%s payload=%s", + task.task_id, fields["basket_domain"], payload, + ) + + resp = await self._context.request.post( + fields["basket_domain"], + form=payload, + max_redirects=5, + headers={ + "Referer": item_url, + "Origin": "https://item.rakuten.co.jp", + }, + ) + final_url = str(resp.url) + if "/error" in final_url: + body = await resp.text() + msg = _extract_error_message(body) or f"错误页 {final_url}" + raise CartOperationError(f"加购失败：{msg}") + + # 成功标志：URL 跳到 cart 且带 added_item 参数（探针实测的落地） + if "cart" not in final_url: + body = await resp.text() + raise CartOperationError( + f"加购响应异常：final_url={final_url} body_head={body[:200]!r}" + ) + + logger.info( + "加购成功：task_id=%s item_id=%s final=%s", + task.task_id, fields["form_fields"].get("item_id"), final_url, + ) + finally: + await page.close() + + async def verify_cart(self, task: LeaseTask) -> None: + """校验购物车里有没有刚加的商品 + + 策略（探针实测最稳的两步）： + 1. 打 cart count JSONP API：status=100 且 count>=1 才算「购物车非空」 + 2. 打开 cart 页等 SPA 渲染，在 HTML 里找 item_id + + 登录态失效抛 NotLoggedInError；找不到 item 抛 CartOperationError。 + """ + await self._auth_session.require_logged_in("rakuten") + + per_task = self._per_task_state.get(task.task_id, {}) + item_id = (task.intent or {}).get("item_id") or per_task.get("item_id") + if not item_id: + raise CartOperationError( + "无法确定 item_id：intent 未提供且 add_to_cart 未记录" + ) + + # 1. cart count API + resp = await self._context.request.get( + _CART_COUNT_API + "?sid=1010", + headers={"Referer": _CART_PAGE}, + ) + count_body = await resp.text() + status_match = re.search(r'"status"\s*:\s*"(\d+)"', count_body) + if not status_match: + raise CartOperationError( + f"cart count 响应无法解析：{count_body[:200]!r}" + ) + if status_match.group(1) != "100": + raise CartOperationError( + f"cart count API 异常：status={status_match.group(1)} body={count_body[:200]!r}" + ) + count_match = re.search(r'"count"\s*:\s*"(\d+)"', count_body) + count = int(count_match.group(1)) if count_match and count_match.group(1) else 0 + if count == 0: + raise CartOperationError("购物车为空，加购可能未生效") + logger.info("cart count=%s task_id=%s", count, task.task_id) + + # 2. 渲染 cart 页确认 item_id 在里面 + page = await self._context.new_page() + try: + await page.goto(_CART_PAGE, wait_until="domcontentloaded", timeout=30_000) + # 等 SPA 把 shopUrlList / cartItem 渲染出来 + try: + await page.wait_for_function( + """() => { + const s = window.__INITIAL_STATE__; + return s && s.cart && Array.isArray(s.cart.shopUrlList) && s.cart.shopUrlList.length > 0; + }""", + timeout=15_000, + ) + except Exception: + logger.warning( + "cart SPA 15s 内未渲染出 shopUrlList，继续按静态 HTML 校验：task_id=%s", + task.task_id, + ) + + html = await page.content() + # 旧 marker 出现一定是登录失效；新 SPA 不渲染 marker，所以这判据是单边的 + if _LEGACY_LOGGED_OUT_MARKER in html: + raise NotLoggedInError( + site="rakuten", + detail="购物车页出现旧版未登录 marker", + ) + if str(item_id) not in html: + raise CartOperationError( + f"购物车页未找到 item_id={item_id}（加购可能被服务端静默丢弃）" + ) + logger.info( + "cart 校验通过：task_id=%s item_id=%s in cart HTML", + task.task_id, item_id, + ) + finally: + await page.close() + + # ---- 未实现：保留 NotImplementedError ---- + + async def enter_checkout(self, task: LeaseTask) -> str: + """进入下单确认页。**未实现**：session upgrade 是硬墙""" + raise NotImplementedError(_NOT_IMPLEMENTED_MSG) + + async def parse_checkout(self, html: str) -> CheckoutSummary: + """从下单确认页解析应付金额、付款期限、订单号。**未实现**""" + raise NotImplementedError(_NOT_IMPLEMENTED_MSG) + + async def submit_order(self, task: LeaseTask) -> str: + """提交下单。**未实现**""" + raise NotImplementedError(_NOT_IMPLEMENTED_MSG) + + async def pay(self, task: LeaseTask, site_order_id: str) -> None: + """付款。**未实现**""" + raise NotImplementedError(_NOT_IMPLEMENTED_MSG) + + async def monitor(self, task: LeaseTask, site_order_id: str) -> None: + """付款后监控。**未实现**""" + raise NotImplementedError(_NOT_IMPLEMENTED_MSG) -async def verify_cart(task: LeaseTask) -> None: - """校验购物车：加购后购物车里有没有这件商品""" - raise NotImplementedError(_NOT_IMPLEMENTED_MSG) +# ---- 模块级辅助函数（纯函数，便于单测）---- -async def enter_checkout(task: LeaseTask) -> str: - """进入下单确认页，返回页面 HTML（供解析金额守卫所需字段）""" - raise NotImplementedError(_NOT_IMPLEMENTED_MSG) +def _parse_initial_state(html: str) -> dict | None: + """从商品页 HTML 抽 window.__INITIAL_STATE__ 并解析为 dict""" + 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 -async def parse_checkout(html: str) -> CheckoutSummary: - """从下单确认页解析实际应付金额、付款期限、站点订单号""" - raise NotImplementedError(_NOT_IMPLEMENTED_MSG) +def _extract_purchase_fields(state: dict, *, intent_override: dict | None) -> dict: + """从 __INITIAL_STATE__ 抽加购所需字段（与 scraping/parsers/item.py 同源逻辑） + + intent_override 允许调用方提供 variant_id / choice 覆盖自动选择。 + """ + intent_override = intent_override or {} + 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 = _INVENTORY_FLAG.get(inventory_type, _DEFAULT_INVENTORY_FLAG) + 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": _NORMAL_PURCHASE_EVENT, + } + + # variant_id 选择：调用方覆盖 > 多规格自动选第一个非售罄 > 单规格用 item.variantId + chosen_variant = None + if intent_override.get("variant_id"): + form_fields["variant_id"] = str(intent_override["variant_id"]) + elif inventory_flag == _INVENTORY_FLAG["multiple"]: + for v in raw_sku.get("variants") or []: + if not v.get("isSoldOut"): + chosen_variant = v + break + if chosen_variant is None and (raw_sku.get("variants") or []): + chosen_variant = raw_sku["variants"][0] + if chosen_variant: + form_fields["variant_id"] = str( + chosen_variant.get("variantId") or chosen_variant.get("id") or "" + ) + elif inventory_flag == _DEFAULT_INVENTORY_FLAG and item.get("variantId"): + form_fields["variant_id"] = str(item.get("variantId")) + + # 必填选项：调用方覆盖 > 自动填第一个候选值 + options = information.get("options") or [] + required_options = [o for o in options if o.get("isRequired")] + has_required = bool(required_options) + if intent_override.get("choice"): + # 调用方给的可能是 list 或 str + c = intent_override["choice"] + form_fields["choice"] = ",".join(c) if isinstance(c, list) else str(c) + elif has_required: + pairs: list[str] = [] + for opt in required_options: + values = opt.get("values") or [] + if values: + pairs.append(f"{opt.get('name')}:{values[0].get('name')}") + if pairs: + form_fields["choice"] = ",".join(pairs) + + return { + "basket_domain": basket_domain, + "form_fields": form_fields, + "quantity_field": "units", + "variant_field": "variant_id", + "options_field": "choice" if options else "", + "options": options, + "has_required_options": has_required, + "inventory_flag": inventory_flag, + "purchase_condition": sell_type.get("purchaseCondition"), + "min_price": sell_type.get("minPrice"), + "shop_name": shop.get("shopName"), + "item_name": item.get("itemName"), + } -async def submit_order(task: LeaseTask) -> str: - """提交下单。返回站点订单号（site_order_id）""" - raise NotImplementedError(_NOT_IMPLEMENTED_MSG) - - -async def pay(task: LeaseTask, site_order_id: str) -> None: - """付款。检测到 3DS / 短信验证等人工环节时抛 OrderGuardError""" - raise NotImplementedError(_NOT_IMPLEMENTED_MSG) - - -async def monitor(task: LeaseTask, site_order_id: str) -> None: - """付款后监控：常驻轮询订单状态与付款期限，状态变化时继续 report""" - raise NotImplementedError(_NOT_IMPLEMENTED_MSG) +def _extract_error_message(body: str) -> str: + """从 Rakuten 错误页 HTML 抽可读的提示文案""" + msgs: list[str] = [] + for m in re.findall(r">([^<>]{20,200})<", body): + s = m.strip() + # 过滤 CSS/JS 标识与版权之类 + if not s or any(c in s for c in ["(", ")", "=", "{", "}", "."]): + continue + if "Rakuten Group" in s or "SSL" in s: + continue + msgs.append(s) + # 取前两条拼起来（实测错误页一般 1~2 条核心提示） + return " / ".join(msgs[:2]) diff --git a/pyproject.toml b/pyproject.toml index 434b509..ae6411e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -13,6 +13,7 @@ dependencies = [ "opentelemetry-instrumentation-fastapi>=0.48b0,<1.0.0", "opentelemetry-instrumentation-httpx>=0.48b0,<1.0.0", "opentelemetry-sdk>=1.27.0,<2.0.0", + "playwright>=1.61.0", "pydantic>=2.0.0,<3.0.0", "pydantic-settings>=2.4.0,<3.0.0", "selectolax>=0.3.21,<1.0.0", diff --git a/scripts/probe_checkout.py b/scripts/probe_checkout.py new file mode 100644 index 0000000..4ecb1d3 --- /dev/null +++ b/scripts/probe_checkout.py @@ -0,0 +1,226 @@ +"""Rakuten 下单流程只读探针：摸清 cart / 加购响应 / 确认页结构 + +只读模式：不发任何写请求。只 GET 现有购物车页（用户可能在历史会话里已有商品）， +分析页面结构、找出 item_id 与"レジに進む"链接的位置。 + +要跑通完整流程（加购 → 确认页），加 --mutate 参数；加购后尝试自动清理。 + +用法： + .venv/Scripts/python.exe scripts/probe_checkout.py # 只读 + .venv/Scripts/python.exe scripts/probe_checkout.py --mutate # 含加购与清理 + .venv/Scripts/python.exe scripts/probe_checkout.py --mutate --item-url https://item.rakuten.co.jp/... +""" +from __future__ import annotations + +import argparse +import asyncio +import json +import re +import sys +from pathlib import Path + +# 允许 `python scripts/probe_checkout.py` 直接运行 +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import httpx + +from app.shared.config import get_settings +from app.trading.core import auth_site + +PROBE_DIR = Path(__file__).resolve().parent.parent / ".probe" / "checkout" +PROBE_DIR.mkdir(parents=True, exist_ok=True) + +# 默认测试商品：一个低单价、稳定的乐天市场直营商品（用户可覆盖） +# 留空则只跑只读流程；--mutate 必须显式传 --item-url +DEFAULT_ITEM_URL = "" + +MOBILE_UA = auth_site.RAKUTEN_USER_AGENT + + +def load_cookies(state_path: Path) -> list[dict]: + state = json.loads(state_path.read_text(encoding="utf-8")) + return state.get("cookies", []) + + +def build_client(cookies: list[dict]) -> httpx.AsyncClient: + client = httpx.AsyncClient( + headers=auth_site.PROFILES["rakuten"].headers(), + timeout=30.0, + follow_redirects=True, + http2=True, + ) + for cookie in cookies: + name = cookie.get("name") + value = cookie.get("value") + if not name or value is None: + continue + client.cookies.set( + name, value, domain=cookie.get("domain") or "", path=cookie.get("path") or "/" + ) + return client + + +def save(name: str, content: str | bytes, *, is_bytes: bool = False) -> Path: + path = PROBE_DIR / name + if is_bytes: + path.write_bytes(content) # type: ignore[arg-type] + else: + path.write_text(content, encoding="utf-8") # type: ignore[arg-type] + print(f" saved → {path}") + return path + + +async def probe_readonly(client: httpx.AsyncClient) -> None: + print("\n=== 步骤 1：GET 购物车页 ===") + resp = await client.get(auth_site.RAKUTEN_CART_URL) + print(f" status={resp.status_code} final_url={resp.url}") + save("01-cart.html", resp.text) + body = resp.text + + logged_out = auth_site.RAKUTEN_LOGGED_OUT_MARKER in body + print(f" logged_in={not logged_out}") + + # 找页面里出现的 item_id（购物车里所有商品） + # 常见模式：data-item-id="..."、name="item_id" value="..."、/item/.../ 等 + item_ids = set(re.findall(r'"item_id"\s*[:=]\s*"?(\d+)', body)) + print(f" cart item_ids={sorted(item_ids)}") + + # 找 "レジに進む" 链接 + checkout_links = re.findall( + r'href=["\']([^"\']*checkout[^"\']*)["\']', body, re.IGNORECASE + ) + checkout_links += re.findall( + r'href=["\']([^"\']*step\.rakuten[^"\']*)["\']', body, re.IGNORECASE + ) + print(f" checkout links={sorted(set(checkout_links))[:5]}") + + # 如果有 checkout 链接，尝试 GET + if checkout_links: + target = checkout_links[0] + if not target.startswith("http"): + target = "https://sp.cart.step.rakuten.co.jp" + target + print(f"\n=== 步骤 2：GET checkout 链接 {target} ===") + resp2 = await client.get(target) + print(f" status={resp2.status_code} final_url={resp2.url}") + save("02-checkout.html", resp2.text) + else: + print("\n 购物车里没有可结账商品（或链接结构变化），跳过 checkout 探测") + + +async def probe_with_mutation(client: httpx.AsyncClient, item_url: str) -> None: + """完整跑一遍：加购 → 校验 → 进确认页。完成后尝试清理""" + from app.scraping.parsers.item import parse_item_detail + from app.scraping.parsers.state import require_state_marker + import app.scraping.parsers.state as state_mod + + print(f"\n=== 加载商品 purchase 块：{item_url} ===") + resp = await client.get(item_url) + print(f" item page status={resp.status_code}") + save("00-item-page.html", resp.text) + + # 抽 __INITIAL_STATE__ + m = re.search( + r']*>window\.\_\_INITIAL_STATE\_\_\s*=\s*(\{.*?\});?\s*', + resp.text, re.DOTALL, + ) + if not m: + # 备用：脚本不一定有 window. 前缀 + m = re.search( + r'\_\_INITIAL\_STATE\_\_\s*[:=]\s*(\{.*?\})\s*[;<]', + resp.text, re.DOTALL, + ) + if not m: + print(" ✗ 抽不出 __INITIAL_STATE__，可能不是手机版模板或反爬被触发") + return + + state = json.loads(m.group(1)) + save("00-item-state.json", json.dumps(state, ensure_ascii=False, indent=2)) + + shop_code_match = re.search(r"/([^/]+)/([^/]+)/?$", item_url.rstrip("/")) + shop_code = shop_code_match.group(1) if shop_code_match else "" + detail = parse_item_detail(state, item_url=item_url, shop_code=shop_code, include_sku_variants=True) + pu = detail.purchase + print(f" cart_url={pu.cart_url}") + print(f" form_fields={pu.form_fields}") + print(f" quantity_field={pu.quantity_field} variant_field={pu.variant_field} options_field={pu.options_field}") + + if not pu.cart_url: + print(" ✗ basketDomain 为空，不能加购") + return + + # 组装加购表单 + payload = dict(pu.form_fields) + payload[pu.quantity_field or "units"] = "1" + # 单一库存商品 variant_id 已在 form_fields 里；多规格需调用方传 + + print(f"\n=== POST 加购 → {pu.cart_url} ===") + print(f" payload={payload}") + resp = await client.post(pu.cart_url, data=payload) + print(f" status={resp.status_code} final_url={resp.url}") + save("03-cart-add-response.html", resp.text) + + print("\n=== GET 购物车页（校验）===") + resp = await client.get(auth_site.RAKUTEN_CART_URL) + print(f" status={resp.status_code}") + save("04-cart-after-add.html", resp.text) + item_id = pu.form_fields.get("item_id", "") + found = item_id and item_id in resp.text + print(f" item_id={item_id} in_cart={found}") + + # 找 レジに進む 链接 + checkout_links = re.findall( + r'href=["\']([^"\']*checkout[^"\']*)["\']', resp.text, re.IGNORECASE + ) + if checkout_links: + target = checkout_links[0] + if not target.startswith("http"): + target = "https://sp.cart.step.rakuten.co.jp" + target + print(f"\n=== GET 确认页 {target} ===") + resp = await client.get(target) + print(f" status={resp.status_code} final_url={resp.url}") + save("05-checkout-confirm.html", resp.text) + + # 在确认页里找金额 + amounts = re.findall(r"[¥￥]?\s*([\d,]+)\s*円", resp.text) + print(f" 金额候选（前 5）={amounts[:5]}") + + print("\n=== 清理：尝试从购物车删除 ===") + # 找删除链接 + del_matches = re.findall( + r'href=["\']([^"\']*(?:delete|del|remove)[^"\']*)["\']', + resp.text, re.IGNORECASE, + ) + if del_matches: + print(f" delete 链接={del_matches[:3]}") + # 真实删除需 POST + 各种 token，留给用户手动清 + + +async def main() -> int: + parser = argparse.ArgumentParser(description="Rakuten 下单流程探针") + parser.add_argument("--mutate", action="store_true", help="包含加购写操作（默认只读）") + parser.add_argument("--item-url", default=DEFAULT_ITEM_URL, help="测试商品 URL") + args = parser.parse_args() + + settings = get_settings() + state_path = settings.auth_state_path / "rakuten_state.json" + if not state_path.exists(): + print(f"✗ 找不到登录态文件：{state_path}") + return 1 + + cookies = load_cookies(state_path) + print(f"加载 {len(cookies)} 条 cookie 自 {state_path}") + + async with build_client(cookies) as client: + await probe_readonly(client) + if args.mutate: + if not args.item_url: + print("\n✗ --mutate 需要搭配 --item-url") + return 1 + await probe_with_mutation(client, args.item_url) + + print(f"\n探针输出目录：{PROBE_DIR}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(asyncio.run(main())) diff --git a/scripts/probe_checkout_playwright.py b/scripts/probe_checkout_playwright.py new file mode 100644 index 0000000..5abd3ad --- /dev/null +++ b/scripts/probe_checkout_playwright.py @@ -0,0 +1,179 @@ +"""Rakuten 下单流程探针（Playwright 版） + +httpx 因 TLS/HTTP2 指纹问题被 Rakuten 拒认（cookie 有效但站点不识别 session）。 +这个探针改用 Playwright（headless Chromium + storage_state）跑 cart → 加购 → 校验 +→ 确认页，把每步 SPA 渲染后的 HTML 落到 .probe/checkout/，作为 site_interact 实现 +的事实依据。 + +加购是写操作。默认商品是 fixture 里的小额商品（fafachai/10000033，2190 円）， +跑完会尝试从购物车删除。可用 --item-url 覆盖。 + +用法： + .venv/Scripts/python.exe scripts/probe_checkout_playwright.py + .venv/Scripts/python.exe scripts/probe_checkout_playwright.py --item-url https://item.rakuten.co.jp/... + .venv/Scripts/python.exe scripts/probe_checkout_playwright.py --headful # 调试时看浏览器 +""" +from __future__ import annotations + +import argparse +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) + +# 默认测试商品：来自 tests/fixtures/item_state.json，2190 円小额商品 +DEFAULT_ITEM_URL = "https://item.rakuten.co.jp/fafachai/10000033/" + + +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 run(item_url: str, *, 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 + + print(f"加载 storage_state: {state_path}") + + async with async_playwright() as pw: + browser = await pw.chromium.launch( + headless=not headful, + channel=settings.browser_channel or None, + 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() + + # 抓所有 XHR，便于看到 SPA 调用了哪些 API + api_calls: list[tuple[int, str]] = [] + + def on_response(r): + url = str(r.url) + if any(kw in url for kw in ["cart", "step", "checkout", "basket", "member", "order"]): + if "r.r10s.jp" not in url: # 排除静态资源 + api_calls.append((r.status, url)) + + page.on("response", on_response) + + print("\n=== 步骤 1：先去 myrakuten 触发完整 session 建立 ===") + await page.goto("https://www.rakuten.co.jp/", wait_until="domcontentloaded", timeout=30_000) + await page.wait_for_timeout(1500) + + print("\n=== 步骤 2：商品详情页 ===") + await page.goto(item_url, wait_until="domcontentloaded", timeout=30_000) + await page.wait_for_timeout(2500) + item_html = await page.content() + save("00-item-page.html", item_html) + + # 从 __INITIAL_STATE__ 抽 purchase 块 + m = re.search(r"window\.__INITIAL_STATE__\s*=\s*(.+?);\s*window\.", item_html, re.DOTALL) + if not m: + print(" ✗ 抽不出 __INITIAL_STATE__（可能 PC 模板或被反爬）") + else: + try: + state = json.loads(m.group(1)) + pu = state.get("purchase", {}).get("sellType", {}).get("normalPurchase", {}) + print(f" basketDomain: {pu.get('basketDomain')}") + print(f" minPrice: {pu.get('minPrice')}") + save("00-item-state.json", json.dumps(state, ensure_ascii=False, indent=2)) + except json.JSONDecodeError as exc: + print(f" state JSON 解析失败：{exc}") + + print("\n=== 步骤 3：点 '買い物かごに入れる' ===") + # 商品页直接点按钮；fallback 是直接 POST basketDomain + btn = page.locator('input[type="submit"][value*="買い物かご"], button:has-text("買い物かごに入れる")').first + try: + await btn.wait_for(state="visible", timeout=5000) + print(f" 找到加购按钮，点击") + await btn.click() + await page.wait_for_timeout(3000) + after_add_url = page.url + print(f" 点击后落地：{after_add_url}") + after_add_html = await page.content() + save("03-cart-add-landing.html", after_add_html) + except Exception as exc: + print(f" 没找到加购按钮或点击失败：{type(exc).__name__}: {exc}") + # 直接 POST basketDomain + print(" fallback：直接构造表单 POST 到 basketDomain") + # 这块留给 site_interact 实现，探针只看页面行为 + + print("\n=== 步骤 4：访问购物车页 ===") + await page.goto(auth_site.RAKUTEN_CART_URL, wait_until="networkidle", timeout=30_000) + await page.wait_for_timeout(3000) + cart_html = await page.content() + save("04-cart-after-add.html", cart_html) + print(f" final_url={page.url} body_len={len(cart_html)}") + + # 检查 SPA 状态 + for kw in ["レジに進む", "注文手続き", "小計", "商品の金額", "itemId", "shopName"]: + print(f" contains {kw!r}: {kw in cart_html}") + m = re.search(r'"isLoggedIn"\s*:\s*(true|false)', cart_html) + if m: + print(f" isLoggedIn={m.group(1)}") + + # 抓 レジ / checkout 链接 + checkout_links = re.findall(r'href=["\']([^"\']*(?:checkout|step1|order)[^"\']*)["\']', cart_html, re.IGNORECASE) + print(f" checkout links: {sorted(set(checkout_links))[:3]}") + + print("\n=== 步骤 5：捕获 SPA 调用的 API ===") + for status, url in api_calls[-30:]: + print(f" {status} {url[:140]}") + + # 不在探针里跑 checkout——避免真的进入付款流程 + # 真实 site_interact 实现时，需要点 レジに進む 后看跳转 + + print("\n=== 步骤 6：尝试清理购物车 ===") + # 找删除按钮 + try: + del_btn = page.locator('a:has-text("削除"), button:has-text("削除"), [data-testid*="delete"]').first + if await del_btn.count() > 0: + print(" 找到削除按钮，点击") + await del_btn.click() + await page.wait_for_timeout(2000) + print(f" 清理后 url={page.url}") + else: + print(" 没找到削除按钮（购物车可能本就为空或按钮选择器要调）") + except Exception as exc: + print(f" 清理失败：{type(exc).__name__}: {exc}") + + await browser.close() + + print(f"\n探针输出目录：{PROBE_DIR}") + return 0 + + +async def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--item-url", default=DEFAULT_ITEM_URL) + parser.add_argument("--headful", action="store_true") + args = parser.parse_args() + return await run(args.item_url, headful=args.headful) + + +if __name__ == "__main__": + raise SystemExit(asyncio.run(main())) diff --git a/scripts/probe_purchase_block.py b/scripts/probe_purchase_block.py new file mode 100644 index 0000000..5f7772c --- /dev/null +++ b/scripts/probe_purchase_block.py @@ -0,0 +1,260 @@ +"""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())) diff --git a/scripts/probe_purchase_block_v2.py b/scripts/probe_purchase_block_v2.py new file mode 100644 index 0000000..5c8fdb4 --- /dev/null +++ b/scripts/probe_purchase_block_v2.py @@ -0,0 +1,250 @@ +"""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, 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())) diff --git a/scripts/probe_search_add.py b/scripts/probe_search_add.py new file mode 100644 index 0000000..8785515 --- /dev/null +++ b/scripts/probe_search_add.py @@ -0,0 +1,226 @@ +"""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, + 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())) diff --git a/tests/test_site_interact.py b/tests/test_site_interact.py new file mode 100644 index 0000000..5bd67cb --- /dev/null +++ b/tests/test_site_interact.py @@ -0,0 +1,289 @@ +"""site_interact 单元测试 + +重点测模块级纯函数（_parse_initial_state / _extract_purchase_fields / +_extract_error_message）——这三者覆盖了「从商品页 HTML 抽加购表单」的核心逻辑， +是 worker 真实跑起来时最关键的转换。 + +SiteInteractor 类的 add_to_cart / verify_cart 涉及 Playwright，不在离线测试覆盖 +范围；只在 test_worker_runner.py 里用桩站点覆盖 runner 与 site 的契约。 +""" +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +import pytest + +from app.shared.errors import CartOperationError, InvalidRequestError +from app.trading.worker.models import LeaseTask +from app.trading.worker.site_interact import ( + CheckoutSummary, + SiteInteractor, + _extract_error_message, + _extract_purchase_fields, + _parse_initial_state, +) + +FIXTURES = Path(__file__).parent / "fixtures" + + +def _make_task(**kwargs: Any) -> LeaseTask: + defaults = { + "task_id": "t1", + "site": "rakuten", + "intent": {"item_url": "https://item.rakuten.co.jp/shop/x/"}, + } + defaults.update(kwargs) + return LeaseTask(**defaults) + + +def _wrap_state(state: dict[str, Any]) -> str: + """构造一段模拟的 item page HTML，把 state 嵌进 __INITIAL_STATE__""" + return ( + "" + ) + + +# ---- _parse_initial_state ---- + + +def test_parse_initial_state_extracts_state_dict(): + state = {"item": {"itemId": 123}, "purchase": {}} + parsed = _parse_initial_state(_wrap_state(state)) + assert parsed == state + + +def test_parse_initial_state_returns_none_when_no_marker(): + assert _parse_initial_state("no state here") is None + + +def test_parse_initial_state_returns_none_on_invalid_json(): + html = "" + assert _parse_initial_state(html) is None + + +# ---- _extract_purchase_fields ---- + + +def test_extract_fields_from_multi_inventory_fixture(): + """item_state.json 是多规格商品，应自动选第一个非售罄 variant""" + state = json.loads((FIXTURES / "item_state.json").read_text(encoding="utf-8")) + fields = _extract_purchase_fields(state, intent_override={}) + assert fields["purchase_condition"] == "enabled" + assert fields["basket_domain"].startswith("https://") + assert fields["basket_domain"].endswith("/rms/mall/bss/cartadd/set") + # 多规格：form 应包含自动选的 variant_id + assert fields["inventory_flag"] == "2" + assert "variant_id" in fields["form_fields"] + assert fields["form_fields"]["__event"] == "ES01_003_001" + assert fields["form_fields"]["inventory_flag"] == "2" + # quantity_field / variant_field 是约定字段名 + assert fields["quantity_field"] == "units" + assert fields["variant_field"] == "variant_id" + + +def test_extract_fields_single_inventory_uses_item_variant_id(): + """单规格商品应直接用 item.variantId，不需要选 sku.variants""" + state = { + "item": {"itemId": 999, "variantId": "v-1"}, + "purchase": { + "sku": {"inventoryType": "single", "variants": []}, + "sellType": { + "normalPurchase": { + "basketDomain": "https://example.co.jp/add", + "purchaseCondition": "enabled", + } + }, + "information": {}, + }, + "shop": {"information": {"shopId": 42}}, + } + fields = _extract_purchase_fields(state, intent_override={}) + assert fields["inventory_flag"] == "1" + assert fields["form_fields"]["variant_id"] == "v-1" + + +def test_extract_fields_intent_overrides_variant_id(): + """调用方显式给 variant_id 时，跳过自动选""" + state = json.loads((FIXTURES / "item_state.json").read_text(encoding="utf-8")) + fields = _extract_purchase_fields(state, intent_override={"variant_id": "my-choice"}) + assert fields["form_fields"]["variant_id"] == "my-choice" + + +def test_extract_fields_multi_inventory_no_variants_returns_no_variant(): + """多规格但 variants 空：不强行填 variant_id；调用方靠这个判断不可加购""" + state = { + "item": {"itemId": 1}, + "purchase": { + "sku": {"inventoryType": "multiple", "variants": []}, + "sellType": {"normalPurchase": {"basketDomain": "https://x/add", "purchaseCondition": "enabled"}}, + "information": {}, + }, + "shop": {"information": {"shopId": 1}}, + } + fields = _extract_purchase_fields(state, intent_override={}) + assert "variant_id" not in fields["form_fields"] + + +def test_extract_fields_required_options_auto_picks_first_value(): + """有必填选项时，自动用第一个候选值填 choice""" + state = { + "item": {"itemId": 1, "variantId": "v"}, + "purchase": { + "sku": {"inventoryType": "single"}, + "sellType": {"normalPurchase": {"basketDomain": "https://x/add", "purchaseCondition": "enabled"}}, + "information": { + "options": [ + { + "id": 1, + "name": "サイズ", + "type": "select", + "isRequired": True, + "values": [ + {"id": 10, "name": "S"}, + {"id": 11, "name": "M"}, + ], + } + ] + }, + }, + "shop": {"information": {"shopId": 1}}, + } + fields = _extract_purchase_fields(state, intent_override={}) + assert fields["has_required_options"] is True + assert fields["form_fields"]["choice"] == "サイズ:S" + + +def test_extract_fields_intent_choice_override_accepts_list_and_str(): + state = { + "item": {"itemId": 1, "variantId": "v"}, + "purchase": { + "sku": {"inventoryType": "single"}, + "sellType": {"normalPurchase": {"basketDomain": "https://x/add", "purchaseCondition": "enabled"}}, + "information": {"options": [{"name": "x", "isRequired": True, "values": [{"name": "a"}]}]}, + }, + "shop": {"information": {"shopId": 1}}, + } + # list 形态 + fields = _extract_purchase_fields(state, intent_override={"choice": ["サイズ:M", "色:赤"]}) + assert fields["form_fields"]["choice"] == "サイズ:M,色:赤" + # str 形态 + fields = _extract_purchase_fields(state, intent_override={"choice": "custom:value"}) + assert fields["form_fields"]["choice"] == "custom:value" + + +def test_extract_fields_basket_domain_unescapes_u002f(): + """JSON encoded 的 / 在 __INITIAL_STATE__ 里可能是 \\u002F，要还原""" + state = { + "item": {"itemId": 1, "variantId": "v"}, + "purchase": { + "sku": {"inventoryType": "single"}, + "sellType": { + "normalPurchase": { + "basketDomain": "https:\\u002F\\u002Fx.example\\u002Fadd", + "purchaseCondition": "enabled", + } + }, + "information": {}, + }, + "shop": {"information": {"shopId": 1}}, + } + fields = _extract_purchase_fields(state, intent_override={}) + assert fields["basket_domain"] == "https://x.example/add" + + +# ---- _extract_error_message ---- + + +def test_extract_error_message_picks_visible_text(): + body = """ + +
「商品 X」は商品情報が変更されました。
+
最新の商品ページから再度かごに追加してください。
+
ご注文手続き中の通信は、SSLによりすべて暗号化されます
+
© Rakuten Group, Inc.
+ + """ + msg = _extract_error_message(body) + # 应当包含前两条核心提示；SSL 与版权被过滤 + assert "商品情報が変更されました" in msg + assert "SSL" not in msg + assert "Rakuten Group" not in msg + + +def test_extract_error_message_filters_punctuation_heavy_strings(): + body = "
未選択の項目からどれか１つ選んでください。
{json: 'x'}
" + msg = _extract_error_message(body) + assert "未選択" in msg + # 含 { 与 ' 的 JSON 片段应被过滤 + assert "json" not in msg + + +def test_extract_error_message_returns_empty_when_no_meaningful_text(): + body = "" + assert _extract_error_message(body) == "" + + +# ---- CheckoutSummary 仍是 dataclass ---- + + +def test_checkout_summary_defaults(): + s = CheckoutSummary(payable_yen=1000) + assert s.payable_yen == 1000 + assert s.site_order_id is None + assert s.pay_deadline is None + + +# ---- SiteInteractor 构造（不调 start，无 Playwright 依赖）---- + + +def test_site_interactor_construction_does_not_require_playwright(): + """构造 SiteInteractor 不应触发 Playwright import；start() 才会""" + site = SiteInteractor(auth_session=None, settings=None) # type: ignore[arg-type] + assert site._playwright is None + assert site._browser is None + assert site._context is None + assert site._per_task_state == {} + + +# ---- 未实现的方法仍然抛 NotImplementedError ---- + + +@pytest.mark.parametrize( + "method,args", + [ + ("enter_checkout", (_make_task(),)), + ("submit_order", (_make_task(),)), + ("pay", (_make_task(), "ord-1")), + ("monitor", (_make_task(), "ord-1")), + ], +) +async def test_unimplemented_methods_raise(method: str, args: tuple): + site = SiteInteractor(auth_session=None, settings=None) # type: ignore[arg-type] + fn = getattr(site, method) + with pytest.raises(NotImplementedError): + await fn(*args) + + +async def test_parse_checkout_unimplemented(): + site = SiteInteractor(auth_session=None, settings=None) # type: ignore[arg-type] + with pytest.raises(NotImplementedError): + await site.parse_checkout("") + + +# ---- add_to_cart / verify_cart 在没启动 Playwright 时应失败 ---- + + +async def test_add_to_cart_without_start_raises(): + """没调 start() 就调 add_to_cart，_context 是 None，应抛错""" + site = SiteInteractor(auth_session=None, settings=None) # type: ignore[arg-type] + # require_logged_in 会因为 auth_session=None 抛 AttributeError，比 _context None 更早 + # 这里测的是「未启动时不应该静默成功」 + with pytest.raises((AttributeError, TypeError)): + await site.add_to_cart(_make_task()) + + +# ---- helper ---- diff --git a/tests/test_worker_runner.py b/tests/test_worker_runner.py index 8cb7aef..0e78b8c 100644 --- a/tests/test_worker_runner.py +++ b/tests/test_worker_runner.py @@ -1,11 +1,11 @@ """worker 主循环测试 不启动真实的后台任务，直接构造 WorkerRunner 与桩 gateway client / local_db / -evidence store，覆盖 §6 主循环的分支： +evidence store / site，覆盖 §6 主循环的分支： - 已完成的任务再次被领：直接补报，不重新执行 - lease_count > 1（恢复领取）：走核对分支，verify unknown → needs_human -- 站点交互未实现（site_interact 抛 NotImplementedError）：转 needs_human +- 站点交互未实现（SiteInteractor 默认抛 NotImplementedError）：转 needs_human - 金额守卫拦截：转 needs_human，站点侧无提交动作 每一步的证据文件在 report 之前就已落盘（§9 第 11 条）也在这里验证。 @@ -20,11 +20,12 @@ import pytest from app.shared.errors import OrderGuardError from app.shared.task_state import OrderState, TaskStatus -from app.trading.worker import site_interact, verify +from app.trading.worker import verify from app.trading.worker.evidence import EvidenceStore from app.trading.worker.local_db import LocalDB from app.trading.worker.models import LeaseTask from app.trading.worker.runner import WorkerRunner +from app.trading.worker.site_interact import CheckoutSummary, SiteInteractor # ---- 桩：网关客户端 ---- @@ -102,7 +103,18 @@ def evidence(tmp_path: Path) -> EvidenceStore: @pytest.fixture -def runner(tmp_path, local_db, evidence, monkeypatch) -> WorkerRunner: +def site() -> SiteInteractor: + """返回**未启动**的 SiteInteractor（不调 start，不依赖 Playwright） + + 默认 add_to_cart/verify_cart 已实现（但会抛 NotImplementedError 因为没启动）， + 具体测试可用 monkeypatch.setattr(site, "method", fn) 替换为桩。 + """ + # 用最小化的 auth_session / settings 占位；测试不会真的调它们 + return SiteInteractor(auth_session=None, settings=None) # type: ignore[arg-type] + + +@pytest.fixture +def runner(tmp_path, local_db, evidence, site) -> WorkerRunner: """构造一个 worker_id=test、网关 URL 已配置的 runner 使用 fake gateway；settings 取真实 Settings 但 worker_id 注入为 test。 @@ -122,6 +134,7 @@ def runner(tmp_path, local_db, evidence, monkeypatch) -> WorkerRunner: gateway_client=gateway, # type: ignore[arg-type] local_db=local_db, evidence=evidence, + site=site, ) # 把 gateway 桩挂到 runner 上，测试通过它断言调用 r._gateway_for_test = gateway # type: ignore[attr-defined] @@ -162,7 +175,7 @@ async def test_already_finished_task_is_reported_not_re_executed( terminal = gateway.last_terminal_report() assert terminal["state"] == OrderState.PAID assert terminal["terminal"] is True - # 没有进入 execute：site_interact 不应被调用，但为保险起见，我们只看 + # 没有进入 execute：site 不应被调用，但为保险起见，我们只看 # evidence 目录里没有任何步骤文件 evidence_files = list(local_db._db_path.parent.parent.glob("**/*.html")) # noqa # 上面 glob 会扫到 trading.db 之外的所有 .html；这里只关心 evidence 目录 @@ -218,7 +231,19 @@ async def test_recovery_task_with_already_ordered_verdict_reports_succeeded( async def test_unimplemented_site_interaction_becomes_needs_human( runner: WorkerRunner, local_db: LocalDB ): - """site_interact.add_to_cart 抛 NotImplementedError → runner 转 needs_human""" + """SiteInteractor 默认 enter_checkout 抛 NotImplementedError → runner 转 needs_human + + add_to_cart 与 verify_cart 已实现（但站点未启动时会从 Playwright 调用失败， + 测试里先替换为 noop 让流程跑到 enter_checkout）。 + """ + + # 让已实现的两步 noop，触发未实现的 enter_checkout + async def _noop(task): # noqa: ANN001 + return None + + runner._site.add_to_cart = _noop # type: ignore[assignment] + runner._site.verify_cart = _noop # type: ignore[assignment] + await runner.handle(_make_task(task_id="t1")) gateway: FakeGateway = runner._gateway_for_test # type: ignore[attr-defined] @@ -243,7 +268,7 @@ async def test_rakuma_site_reports_needs_human(runner: WorkerRunner): async def test_amount_guard_blocks_when_payable_exceeds_limit( - runner: WorkerRunner, local_db: LocalDB, monkeypatch + runner: WorkerRunner, local_db: LocalDB ): """实际应付超过上限 → OrderGuardError → 转 needs_human，站点侧无提交动作 @@ -255,14 +280,11 @@ async def test_amount_guard_blocks_when_payable_exceeds_limit( async def _ok(task): # noqa: ANN001 call_log.append(task.task_id + ":step") - async def _noop(task): # noqa: ANN001 - return None - async def _checkout_html(task): # noqa: ANN001 return "checkout" async def _parse(html: str): - return site_interact.CheckoutSummary(payable_yen=50000) + return CheckoutSummary(payable_yen=50000) submit_called = False @@ -271,11 +293,11 @@ async def test_amount_guard_blocks_when_payable_exceeds_limit( submit_called = True return "ord-1" - monkeypatch.setattr(site_interact, "add_to_cart", _noop) - monkeypatch.setattr(site_interact, "verify_cart", _noop) - monkeypatch.setattr(site_interact, "enter_checkout", _checkout_html) - monkeypatch.setattr(site_interact, "parse_checkout", _parse) - monkeypatch.setattr(site_interact, "submit_order", _submit) + runner._site.add_to_cart = _ok # type: ignore[assignment] + runner._site.verify_cart = _ok # type: ignore[assignment] + runner._site.enter_checkout = _checkout_html # type: ignore[assignment] + runner._site.parse_checkout = _parse # type: ignore[assignment] + runner._site.submit_order = _submit # type: ignore[assignment] # intent.max_total_yen 缺省，回落到 settings.order_max_total_yen=30000， # 实际 50000 > 30000 → 拦截 @@ -290,7 +312,7 @@ async def test_amount_guard_blocks_when_payable_exceeds_limit( async def test_amount_guard_honors_intent_max_total_yen( - runner: WorkerRunner, local_db: LocalDB, monkeypatch + runner: WorkerRunner, local_db: LocalDB ): """intent 显式给 max_total_yen 时优先于全局上限""" @@ -301,7 +323,7 @@ async def test_amount_guard_honors_intent_max_total_yen( return "checkout" async def _parse(html: str): - return site_interact.CheckoutSummary(payable_yen=8000) + return CheckoutSummary(payable_yen=8000) submit_called = False @@ -310,11 +332,11 @@ async def test_amount_guard_honors_intent_max_total_yen( submit_called = True raise NotImplementedError("不应到这里") # parse 之后的步骤未实现，到这里即说明守卫过了 - monkeypatch.setattr(site_interact, "add_to_cart", _noop) - monkeypatch.setattr(site_interact, "verify_cart", _noop) - monkeypatch.setattr(site_interact, "enter_checkout", _checkout_html) - monkeypatch.setattr(site_interact, "parse_checkout", _parse) - monkeypatch.setattr(site_interact, "submit_order", _submit) + runner._site.add_to_cart = _noop # type: ignore[assignment] + runner._site.verify_cart = _noop # type: ignore[assignment] + runner._site.enter_checkout = _checkout_html # type: ignore[assignment] + runner._site.parse_checkout = _parse # type: ignore[assignment] + runner._site.submit_order = _submit # type: ignore[assignment] # 全局上限 30000，但 intent 给 5000 → 实际 8000 > 5000 → 拦截 await runner.handle(_make_task(task_id="t1", intent={"max_total_yen": 5000})) @@ -329,7 +351,7 @@ async def test_amount_guard_honors_intent_max_total_yen( async def test_evidence_files_exist_before_each_report( - runner: WorkerRunner, evidence: EvidenceStore, monkeypatch + runner: WorkerRunner, evidence: EvidenceStore ): """每个 _run_step 调用：动作 → 落证据 → 写本地 → 回报 gateway @@ -350,11 +372,13 @@ async def test_evidence_files_exist_before_each_report( runner._gateway_for_test.report = _spy_report # type: ignore[attr-defined] - # 让 add_to_cart 正常，verify_cart 抛 NotImplemented，触发第一个 step 写证据 + report + # 让 add_to_cart / verify_cart 正常，enter_checkout 抛 NotImplemented， + # 触发前两步写证据 + report，到第三步转 needs_human async def _noop(task): # noqa: ANN001 return None - monkeypatch.setattr(site_interact, "add_to_cart", _noop) + runner._site.add_to_cart = _noop # type: ignore[assignment] + runner._site.verify_cart = _noop # type: ignore[assignment] await runner.handle(_make_task(task_id="t1")) diff --git a/uv.lock b/uv.lock index 3ca0642..ea4cbf7 100644 --- a/uv.lock +++ b/uv.lock @@ -810,6 +810,7 @@ dependencies = [ { name = "opentelemetry-instrumentation-fastapi" }, { name = "opentelemetry-instrumentation-httpx" }, { name = "opentelemetry-sdk" }, + { name = "playwright" }, { name = "pydantic" }, { name = "pydantic-settings" }, { name = "selectolax" }, @@ -836,6 +837,7 @@ requires-dist = [ { name = "opentelemetry-instrumentation-fastapi", specifier = ">=0.48b0,<1.0.0" }, { name = "opentelemetry-instrumentation-httpx", specifier = ">=0.48b0,<1.0.0" }, { name = "opentelemetry-sdk", specifier = ">=1.27.0,<2.0.0" }, + { name = "playwright", specifier = ">=1.61.0" }, { name = "playwright", marker = "extra == 'browser'", specifier = ">=1.47.0,<2.0.0" }, { name = "pydantic", specifier = ">=2.0.0,<3.0.0" }, { name = "pydantic-settings", specifier = ">=2.4.0,<3.0.0" },