From f76c0a15181b2e8bd8809a8510edf287fa01c111 Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Mon, 31 Aug 2026 11:41:39 +0800 Subject: [PATCH] feat(trading): support multi-item purchase intents --- README.md | 2 +- app/gateway/models.py | 12 ++- app/trading/worker/site_interact.py | 142 +++++++++++++++++++++------- app/trading/worker/verify.py | 51 +++++++--- docs/order-gateway.md | 15 ++- tests/test_site_interact.py | 78 +++++++++++++++ tests/test_verify.py | 48 ++++++++++ 7 files changed, 291 insertions(+), 57 deletions(-) diff --git a/README.md b/README.md index 9e81611..5ef5938 100644 --- a/README.md +++ b/README.md @@ -139,7 +139,7 @@ PC UA 在搜索页、详情页、店铺页上都能拿到完整模板。因此 | 接口 | 说明 | | --- | --- | | `GET /health` | 健康检查，含 worker 心跳、长时间无人领任务告警 | -| `POST /api/orders` | 上游提交下单意图（幂等） | +| `POST /api/orders` | 上游提交下单意图（幂等）；`intent.items` 支持一次购买多个商品，旧版 `intent.item_url` 仍兼容 | | `GET /api/orders/lease` | 本地 worker 长轮询领取（全局并发度 1） | | `POST /api/orders/{id}/renew` | 续租（worker 在长任务里每 60s 调一次） | | `POST /api/orders/{id}/report` | 本地回报订单状态（同 state 重复上报幂等） | diff --git a/app/gateway/models.py b/app/gateway/models.py index 89fafd8..aa40032 100644 --- a/app/gateway/models.py +++ b/app/gateway/models.py @@ -37,16 +37,20 @@ class SubmitOrderRequest(BaseModel): intent: dict[str, Any] = Field( description=( "下单意图原文。网关不解释内容，原样存库并透传给本地 worker，结构由 trading 侧定义：" - "item_url（必填，商品页 URL）；quantity（可选，默认 1）；" + "推荐使用 items（非空数组，每项含 item_url，及可选 quantity/variant_id/choice）；" + "为兼容已发布客户端，也支持单商品 item_url（及同级 quantity/variant_id/choice）；" + "quantity（可选，默认 1）；" "variant_id（多规格商品必填，取自 /api/item_detail 的 variants，不传时 worker 自动选第一个非售罄规格）；" "choice（可选，商品选项，如 \"颜色:赤\"，可传字符串或字符串列表）；" "max_total_yen（可选，本次金额上限：确认页实际应付超过即中止并报 needs_human，" "缺省用服务端 RAKUTEN_ORDER_MAX_TOTAL_YEN）" ), examples=[{ - "item_url": "https://item.rakuten.co.jp/shop/code/", - "quantity": 1, - "variant_id": "1001", + "items": [{ + "item_url": "https://item.rakuten.co.jp/shop/code/", + "quantity": 1, + "variant_id": "1001", + }], "max_total_yen": 30000, }], ) diff --git a/app/trading/worker/site_interact.py b/app/trading/worker/site_interact.py index 4ef34a2..dbb116a 100644 --- a/app/trading/worker/site_interact.py +++ b/app/trading/worker/site_interact.py @@ -94,7 +94,7 @@ import logging import re from dataclasses import dataclass, field from datetime import datetime -from typing import TYPE_CHECKING, TypeVar +from typing import TYPE_CHECKING, Any, TypeVar from opentelemetry.trace import SpanKind @@ -649,9 +649,9 @@ class SiteInteractor: clear_cart() / remove_item(item_id) """ - # 每任务保留的临时状态：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]] + # 每任务保留的临时状态：task_id → 首个商品字段及全部 item_ids。 + # 旧调用方仍读取 item_id，新调用方由 item_ids 校验整组商品。 + _per_task_state: dict[str, dict[str, Any]] # 每任务保留的下单确认页 Page：enter_checkout 落地后不关闭页面，存在这里， # submit_order / pay 复用同一个页面继续操作——下单确认页是服务端会话态， @@ -984,40 +984,36 @@ class SiteInteractor: async def add_to_cart(self, task: LeaseTask) -> PageSnapshot: """加购（worker 入口）：从 task.intent 取字段，调 _add_to_cart_with_fields - 调用方需在 task.intent 提供： - - item_url: 商品详情页 URL（必填） - - quantity: 数量，默认 1 - - variant_id: 多规格商品的 variant_id；不传则从 sku.variants[] 自动选第一个非售罄 - - choice: 必填选项的取值列表；不传则每个必填选项用第一个候选值（站点不严格校验） + 调用方可在 task.intent 提供 `items` 商品数组；数组元素字段与旧版单商品 + 字段相同（item_url / quantity / variant_id / choice）。为兼容已发布的 + 调用方，也接受顶层 item_url 等旧字段并自动包装成单元素数组。 Returns: PageSnapshot：加购响应的落地页 HTML + 商品页整页截图，供 runner 落证据。 Raises: - InvalidRequestError: intent.item_url 缺失 + InvalidRequestError: intent.item_url 缺失或 intent.items 格式非法 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}") + items = _normalize_intent_items(task.intent or {}) async with self._lock: - result = await self._add_to_cart_with_fields( - item_url=item_url, - quantity=quantity, - variant_id=intent.get("variant_id"), - choice=intent.get("choice"), - ) - self._per_task_state[task.task_id] = { - "item_id": result["item_id"], - "shop_bid": result["shop_bid"], - "basket_domain": result["basket_domain"], + results = [] + for item in items: + results.append(await self._add_to_cart_with_fields(**item)) + first = results[0] + state: dict[str, Any] = { + # 保留旧字段，避免已有 verify/监控代码及外部桩失效。 + "item_id": first["item_id"], + "shop_bid": first["shop_bid"], + "basket_domain": first["basket_domain"], } + if len(results) > 1: + state["item_ids"] = [result["item_id"] for result in results] + state["items"] = results + self._per_task_state[task.task_id] = state + result = results[-1] return PageSnapshot( html=result.get("response_html") or "", screenshot=result.get("screenshot") or b"", @@ -1199,8 +1195,15 @@ class SiteInteractor: await self._ensure_context_ready() 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: + intent_item_id = (task.intent or {}).get("item_id") + item_ids = ( + [intent_item_id] + if intent_item_id + else (per_task.get("item_ids") or []) + ) + if not item_ids and per_task.get("item_id"): + item_ids = [per_task["item_id"]] + if not item_ids: raise CartOperationError( "无法确定 item_id：intent 未提供且 add_to_cart 未记录" ) @@ -1211,8 +1214,15 @@ class SiteInteractor: raise CartOperationError("购物车为空，加购可能未生效") logger.info("cart count=%s task_id=%s", count, task.task_id) - # 2. 渲染 cart 页确认 item_id 在里面 - return await self._verify_item_in_cart_html(item_id, label=f"task_id={task.task_id}") + # 2. 渲染一次 cart 页确认本任务的全部商品都在里面。 + # 单商品继续走旧 helper，保留原有测试桩与内部调用契约。 + if len(item_ids) == 1: + return await self._verify_item_in_cart_html( + str(item_ids[0]), label=f"task_id={task.task_id}" + ) + return await self._verify_items_in_cart_html( + [str(item_id) for item_id in item_ids], label=f"task_id={task.task_id}" + ) @traced("site.cart_status", kind=SpanKind.CLIENT) async def cart_status(self) -> dict: @@ -1436,6 +1446,12 @@ class SiteInteractor: 返回渲染后的 cart 页 HTML + 整页截图（校验通过时），供调用方落证据。 """ + return await self._verify_items_in_cart_html([item_id], label=label) + + async def _verify_items_in_cart_html( + self, item_ids: list[str], *, label: str + ) -> PageSnapshot: + """渲染一次 cart SPA，确认多个 item_id 都存在，避免多商品任务重复开页。""" page = await self._new_page() try: await page.goto(_CART_PAGE, wait_until="domcontentloaded", timeout=30_000) @@ -1448,11 +1464,12 @@ class SiteInteractor: site="rakuten", detail="购物车页出现旧版未登录 marker", ) - if str(item_id) not in html: + missing = [item_id for item_id in item_ids if str(item_id) not in html] + if missing: raise CartOperationError( - f"购物车页未找到 item_id={item_id}（加购可能被服务端静默丢弃）" + f"购物车页未找到 item_id={missing}（加购可能被服务端静默丢弃）" ) - logger.info("cart 校验通过：%s item_id=%s in cart HTML", label, item_id) + logger.info("cart 校验通过：%s item_ids=%s in cart HTML", label, item_ids) # 校验通过的 cart 页整页截图随结果带出；失败不掩盖校验结果 screenshot = b"" try: @@ -2370,6 +2387,63 @@ class SiteInteractor: # ---- 模块级辅助函数（纯函数，便于单测）---- +def _normalize_intent_items(intent: dict[str, Any]) -> list[dict[str, Any]]: + """把新旧下单意图统一成加购参数列表。 + + 新格式是 ``{"items": [{"item_url": ..., "quantity": ...}, ...]}``； + 旧格式的 ``item_url/quantity/variant_id/choice`` 仍直接支持。字符串元素 + 也接受，方便只传多个 URL 的调用方。 + """ + raw_items = intent.get("items") + legacy_single = raw_items is None + if raw_items is None: + if not intent.get("item_url"): + # 保持已发布的单商品错误契约不变。 + raise InvalidRequestError("intent.item_url 必填") + raw_items = [intent] + if not isinstance(raw_items, list) or not raw_items: + raise InvalidRequestError("intent.items 必须是非空数组") + + normalized: list[dict[str, Any]] = [] + for index, raw in enumerate(raw_items): + if isinstance(raw, str): + raw = {"item_url": raw} + if not isinstance(raw, dict): + raise InvalidRequestError(f"intent.items[{index}] 必须是对象") + item_url = raw.get("item_url") + if not item_url: + raise InvalidRequestError(f"intent.items[{index}].item_url 必填") + try: + quantity = int(raw.get("quantity") or 1) + except (TypeError, ValueError) as exc: + field_name = ( + "intent.quantity" + if legacy_single + else f"intent.items[{index}].quantity" + ) + raise InvalidRequestError( + f"{field_name} 必须为正整数" + ) from exc + if quantity <= 0: + field_name = ( + "intent.quantity" + if legacy_single + else f"intent.items[{index}].quantity" + ) + raise InvalidRequestError( + f"{field_name} 必须为正整数，收到 {quantity}" + ) + normalized.append( + { + "item_url": str(item_url), + "quantity": quantity, + "variant_id": raw.get("variant_id"), + "choice": raw.get("choice"), + } + ) + return normalized + + def _parse_initial_state(html: str) -> dict | None: """从商品页 HTML 抽 window.__INITIAL_STATE__ 并解析为 dict""" m = re.search( diff --git a/app/trading/worker/verify.py b/app/trading/worker/verify.py index e22cd4a..4a26981 100644 --- a/app/trading/worker/verify.py +++ b/app/trading/worker/verify.py @@ -15,7 +15,7 @@ from __future__ import annotations from dataclasses import dataclass from enum import StrEnum -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any from urllib.parse import urlsplit from app.shared.errors import AppError @@ -55,24 +55,44 @@ def _normalize_item_url(url: str | None) -> str | None: return f"{parts.scheme}://{parts.netloc}{parts.path.rstrip('/')}" +def _intent_item_urls(intent: dict[str, Any]) -> list[str]: + """读取新旧意图中的商品 URL，供恢复核对使用。""" + raw_items = intent.get("items") + if raw_items is None: + raw_items = [intent] + if not isinstance(raw_items, list): + return [] + urls: list[str] = [] + for item in raw_items: + if isinstance(item, str): + url = item + elif isinstance(item, dict): + url = item.get("item_url") + else: + url = None + normalized = _normalize_item_url(url) + if normalized: + urls.append(normalized) + return urls + + async def verify_on_site( task: LeaseTask, *, gateway: "GatewayClient", site: "SiteInteractor" ) -> VerifyResult: """核对一笔任务是否已在站点上下过单 - 核对链路：intent.item_url → 查任务创建时间（GET /api/orders/{task_id}， - LeaseTask 本身不带 created_at）→ 拉「创建时间之后」的订单列表 → 按商品 URL - 比对。任何一环拿不到足够信息都返回 UNKNOWN，不猜——尤其是 NOT_ORDERED， + 核对链路：intent.items（或兼容的 intent.item_url）→ 查任务创建时间 + （GET /api/orders/{task_id}，LeaseTask 本身不带 created_at）→ 拉「创建时间之后」 + 的订单列表 → 按商品 URL 比对。任何一环拿不到足够信息都返回 UNKNOWN，不猜——尤其是 NOT_ORDERED， 只有在确认翻完了窗口内的全部订单后才允许返回，否则「没找到」可能只是没翻 到那一页。 """ intent = task.intent or {} - item_url = intent.get("item_url") - if not item_url: + targets = set(_intent_item_urls(intent)) + if not targets: return VerifyResult( - VerifyVerdict.UNKNOWN, detail="intent 缺 item_url，无法比对商品" + VerifyVerdict.UNKNOWN, detail="intent 缺商品 URL（item_url/items），无法比对商品" ) - target = _normalize_item_url(item_url) try: task_detail = await gateway.get_task(task.task_id) @@ -96,11 +116,16 @@ async def verify_on_site( detail=f"订单列表查询失败：{type(exc).__name__}: {exc}", ) - matches = [ - entry - for entry in window.entries - if any(_normalize_item_url(it.item_url) == target for it in entry.items) - ] + matches = [] + for entry in window.entries: + entry_urls = { + normalized + for normalized in (_normalize_item_url(it.item_url) for it in entry.items) + if normalized + } + # 多商品任务必须在同一笔订单中全部命中，避免部分匹配误判为已下单。 + if targets.issubset(entry_urls): + matches.append(entry) if len(matches) == 1: return VerifyResult( VerifyVerdict.ALREADY_ORDERED, diff --git a/docs/order-gateway.md b/docs/order-gateway.md index 2a5bee6..fe37790 100644 --- a/docs/order-gateway.md +++ b/docs/order-gateway.md @@ -112,16 +112,22 @@ CREATE TABLE workers ( "task_id": "po-20260727-0001", // 可选，上游自带的幂等键；不传则服务端生成 "site": "rakuten", "intent": { // gateway 原样透传，结构由 trading 侧定义 - "item_url": "https://item.rakuten.co.jp/shop/code/", - "quantity": 1, - "variant_id": "...", // 多规格商品必填，取自 /api/item_detail 的 sku.variants[] - "choice": ["名入れ:希望する"], // 店铺自定义必填选项，格式「选项名:取值名」，取自 /api/item_detail 的 options[] + "items": [{ // 新格式：一次购买多个商品 + "item_url": "https://item.rakuten.co.jp/shop/code/", + "quantity": 1, + "variant_id": "...", // 多规格商品必填，取自 /api/item_detail 的 sku.variants[] + "choice": ["名入れ:希望する"] // 店铺自定义必填选项，格式「选项名:取值名」 + }], "max_total_yen": 30000 // 可选，覆盖本次的金额上限 }, "callback_url": "https://upstream.example.com/hooks/rakuten-order" // 可选，终结类事件通知，见 §4.8 } ``` +`items` 必须是非空数组，worker 会按顺序将每项加入同一购物车后再进入结算。 +为兼容已发布客户端，也可继续使用旧格式：`intent.item_url` 加同级的 +`quantity` / `variant_id` / `choice`，其语义等同于只有一个元素的 `items`。 + 响应 `data`：`{"task_id": "...", "status": "queued", "created": true}` **幂等**：同一个 `task_id` 重复提交不新建任务，返回既有任务且 `created=false`。 @@ -614,4 +620,3 @@ collector 本身不回写「已经存在于下单任务表 `tasks` 的订单」 - [x] 手动 trigger 立即派一轮 list，返回的单在查询队列里可见 - [x] `GET /api/account/orders` 列编目；单笔不存在返回 6005 - [x] collector 只读规范化字段，不解析 raw/raw_pages 站点原始 JSON - diff --git a/tests/test_site_interact.py b/tests/test_site_interact.py index 6f220ef..8645cb4 100644 --- a/tests/test_site_interact.py +++ b/tests/test_site_interact.py @@ -11,6 +11,7 @@ clear_cart 与 _dump_debug_snapshot 用不依赖 Playwright 的 fake page 覆盖 """ from __future__ import annotations +import asyncio import json from pathlib import Path from typing import Any @@ -39,6 +40,7 @@ from app.trading.worker.site_interact import ( _DELETE_BUTTON_SELECTOR, _extract_error_message, _extract_purchase_fields, + _normalize_intent_items, _OrderListAccumulator, _parse_checkout_summary, _parse_initial_state, @@ -70,6 +72,82 @@ def _wrap_state(state: dict[str, Any]) -> str: ) +# ---- 下单意图兼容 ---- + + +def test_normalize_intent_items_accepts_legacy_single_item(): + assert _normalize_intent_items({ + "item_url": "https://item.rakuten.co.jp/shop/x/", + "quantity": 2, + }) == [{ + "item_url": "https://item.rakuten.co.jp/shop/x/", + "quantity": 2, + "variant_id": None, + "choice": None, + }] + + +def test_normalize_intent_items_accepts_multiple_items_and_string_urls(): + assert _normalize_intent_items({ + "items": [ + {"item_url": "https://item.rakuten.co.jp/shop/x/", "quantity": 2}, + "https://item.rakuten.co.jp/shop/y/", + ] + }) == [ + { + "item_url": "https://item.rakuten.co.jp/shop/x/", + "quantity": 2, + "variant_id": None, + "choice": None, + }, + { + "item_url": "https://item.rakuten.co.jp/shop/y/", + "quantity": 1, + "variant_id": None, + "choice": None, + }, + ] + + +def test_normalize_intent_items_rejects_empty_items(): + with pytest.raises(InvalidRequestError): + _normalize_intent_items({"items": []}) + + +async def test_add_to_cart_processes_all_items_and_keeps_legacy_state(): + site = SiteInteractor.__new__(SiteInteractor) + site._lock = asyncio.Lock() + site._per_task_state = {} + calls: list[dict[str, Any]] = [] + + async def fake_add(**kwargs): + calls.append(kwargs) + index = len(calls) + return { + "item_id": str(index), + "shop_bid": "shop", + "basket_domain": "https://basket", + "response_html": f"html-{index}", + "screenshot": b"", + } + + site._add_to_cart_with_fields = fake_add + snapshot = await site.add_to_cart(_make_task(intent={ + "items": [ + {"item_url": "https://item.rakuten.co.jp/shop/x/"}, + {"item_url": "https://item.rakuten.co.jp/shop/y/", "quantity": 3}, + ] + })) + assert [call["item_url"] for call in calls] == [ + "https://item.rakuten.co.jp/shop/x/", + "https://item.rakuten.co.jp/shop/y/", + ] + assert calls[1]["quantity"] == 3 + assert site._per_task_state["t1"]["item_id"] == "1" + assert site._per_task_state["t1"]["item_ids"] == ["1", "2"] + assert snapshot.html == "html-2" + + # ---- _parse_initial_state ---- diff --git a/tests/test_verify.py b/tests/test_verify.py index aabe775..8e0bab9 100644 --- a/tests/test_verify.py +++ b/tests/test_verify.py @@ -134,6 +134,54 @@ async def test_match_ignores_query_string_difference(): assert result.verdict == verify.VerifyVerdict.ALREADY_ORDERED +async def test_multiple_item_intent_requires_all_items_in_same_order(): + gateway = FakeGateway() + site = FakeSite( + window=OrderListWindow( + entries=[ + OrderListEntry( + order_number="o1", + order_date="2026-08-10T00:00:00Z", + items=[ + OrderListItem(item_url="https://item.rakuten.co.jp/shop/x/"), + OrderListItem(item_url="https://item.rakuten.co.jp/shop/y/"), + ], + ) + ], + window_fully_covered=True, + ) + ) + task = _make_task() + task.intent = { + "items": [ + {"item_url": "https://item.rakuten.co.jp/shop/x/"}, + {"item_url": "https://item.rakuten.co.jp/shop/y/"}, + ] + } + result = await verify.verify_on_site(task, gateway=gateway, site=site) + assert result.verdict == verify.VerifyVerdict.ALREADY_ORDERED + assert result.site_order_id == "o1" + + +async def test_multiple_item_intent_partial_order_match_is_not_ordered(): + gateway = FakeGateway() + site = FakeSite( + window=OrderListWindow( + entries=[_entry("o1", "https://item.rakuten.co.jp/shop/x/")], + window_fully_covered=True, + ) + ) + task = _make_task() + task.intent = { + "items": [ + {"item_url": "https://item.rakuten.co.jp/shop/x/"}, + {"item_url": "https://item.rakuten.co.jp/shop/y/"}, + ] + } + result = await verify.verify_on_site(task, gateway=gateway, site=site) + assert result.verdict == verify.VerifyVerdict.NOT_ORDERED + + # ---- 命中 0 笔且窗口确认覆盖完：未下单 ----