From afc1cdf719a86ebdf1f1bc65371c620c9e0c3964 Mon Sep 17 00:00:00 2001 From: chengma Date: Mon, 17 Aug 2026 17:43:25 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=89=E5=85=A8=E5=BA=94=E7=94=A8?= =?UTF-8?q?=E9=87=87=E8=B4=AD=E8=A7=84=E6=A0=BC=E8=A7=A3=E6=9E=90=E7=BB=93?= =?UTF-8?q?=E6=9E=9C=20(#257)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- client/src/admin_gateway.py | 35 ++- client/src/db_schema.py | 44 ++- client/src/http_admin_gateway.py | 145 +++++++++ client/src/mock_admin_gateway.py | 167 +++++++++++ client/src/pdd_purchase_adapter.py | 13 + client/src/pdd_purchase_reconcile_adapter.py | 2 + client/src/pdd_u2_purchase_adapter.py | 92 ++++++ client/src/purchase_reconcile_service.py | 17 +- client/src/purchase_task_service.py | 255 +++++++++++++++- client/src/task_models.py | 26 ++ client/src/task_repository.py | 190 ++++++++++++ client/test/test_admin_gateway_contract.py | 92 +++++- client/test/test_db.py | 53 +++- client/test/test_http_admin_gateway.py | 101 +++++++ client/test/test_pdd_u2_purchase_adapter.py | 72 +++++ client/test/test_purchase_task_service.py | 291 ++++++++++++++++++- client/test/test_task_repository.py | 54 ++++ docs/client/03-data-model.md | 38 ++- docs/client/06-quality-security.md | 6 + 19 files changed, 1669 insertions(+), 24 deletions(-) diff --git a/client/src/admin_gateway.py b/client/src/admin_gateway.py index cda7a6f..642d1b4 100644 --- a/client/src/admin_gateway.py +++ b/client/src/admin_gateway.py @@ -1,6 +1,6 @@ """Client 访问 Admin 的稳定边界和简单数据对象。 -AdminGateway 只有登记、领取任务、提交成功结果、提交失败结果四个业务方法。 +AdminGateway 提供登记、领取、结果提交和一次性采购规格解析命令。 业务层不应直接依赖 HTTP 请求或 Mock 的内部实现。 """ @@ -113,6 +113,30 @@ class RegistrationReceipt: registered_at: str +@dataclass(frozen=True) +class SpecResolutionMatch: + """Admin 从本次候选白名单中选中的原始规格。""" + + candidate_id: str + raw_text: str + options: Mapping[str, str] + + +@dataclass(frozen=True) +class SpecResolutionReceipt: + """一次采购规格解析的完整业务响应。""" + + schema_version: int + resolution_id: str + outcome: str + source: Optional[str] + candidate_snapshot_hash: str + match: Optional[SpecResolutionMatch] + confidence_bps: Optional[int] + reason: str + resolved_at: str + + class AdminGatewayError(RuntimeError): """带稳定错误代码和可重试标志的 Admin 边界错误。""" @@ -171,3 +195,12 @@ class AdminGateway(ClientRegistrationGateway, TaskClaimGateway): failure: Mapping[str, Any], ) -> SubmissionReceipt: """幂等提交失败、取消或人工处理结果。""" + + @abstractmethod + def resolve_purchase_spec( + self, + task_id: str, + idempotency_key: str, + observation: Mapping[str, Any], + ) -> SpecResolutionReceipt: + """一次性解析当前真机尺码候选；不得用于查询任务状态。""" diff --git a/client/src/db_schema.py b/client/src/db_schema.py index 4e0dcbf..cee5ffa 100644 --- a/client/src/db_schema.py +++ b/client/src/db_schema.py @@ -4,7 +4,7 @@ ``MIGRATIONS`` 末尾增加版本，不能修改已经发布的迁移。 """ -SCHEMA_VERSION = 5 +SCHEMA_VERSION = 6 MIGRATION_1 = ( @@ -163,6 +163,47 @@ MIGRATION_5 = ( """, ) +MIGRATION_6 = ( + """ + CREATE TABLE purchase_spec_resolutions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id INTEGER NOT NULL, + attempt_id TEXT NOT NULL, + idempotency_key TEXT NOT NULL UNIQUE, + request_json TEXT NOT NULL, + request_hash TEXT NOT NULL, + candidate_snapshot_hash TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending' + CHECK (status IN ('pending', 'resolved')), + resolution_id TEXT, + outcome TEXT CHECK (outcome IS NULL OR outcome IN ( + 'matched', 'uncertain', 'rejected', 'failed' + )), + source TEXT CHECK (source IS NULL OR source IN ( + 'rule', 'ai', 'reused' + )), + confidence_bps INTEGER CHECK ( + confidence_bps IS NULL OR + confidence_bps BETWEEN 0 AND 10000 + ), + match_candidate_id TEXT, + match_raw_text TEXT, + match_options_json TEXT, + reason TEXT, + resolved_at TEXT, + received_at TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + FOREIGN KEY (task_id) REFERENCES pdd_tasks(id) ON DELETE CASCADE, + UNIQUE (task_id, attempt_id, candidate_snapshot_hash) + ) + """, + """ + CREATE INDEX idx_purchase_spec_resolutions_attempt + ON purchase_spec_resolutions(attempt_id, id DESC) + """, +) + MIGRATIONS = { 1: MIGRATION_1, @@ -170,4 +211,5 @@ MIGRATIONS = { 3: MIGRATION_3, 4: MIGRATION_4, 5: MIGRATION_5, + 6: MIGRATION_6, } diff --git a/client/src/http_admin_gateway.py b/client/src/http_admin_gateway.py index 12f84f1..9e44cdb 100644 --- a/client/src/http_admin_gateway.py +++ b/client/src/http_admin_gateway.py @@ -2,6 +2,7 @@ import hashlib import json +import re import socket from http.client import RemoteDisconnected from typing import Any, Callable, Mapping, Optional @@ -17,6 +18,8 @@ from .admin_gateway import ( ClaimCapabilities, ClientInfo, RegistrationReceipt, + SpecResolutionMatch, + SpecResolutionReceipt, SubmissionReceipt, ) from .task_models import TaskType @@ -254,6 +257,148 @@ class HttpAdminGateway(AdminGateway): return self._submit(task_id, idempotency_key, failure, "failure") + def resolve_purchase_spec( + self, + task_id: str, + idempotency_key: str, + observation: Mapping[str, Any], + ) -> SpecResolutionReceipt: + """发送一次已持久化的规格解析命令并严格校验响应。""" + + if not task_id.strip() or not idempotency_key.strip(): + raise ValueError("规格解析的 task_id 和 Idempotency-Key 不能为空") + if not self._client_id: + raise AdminGatewayError( + "CLIENT_ID_MISSING", "规格解析前必须配置 Client 设备号", False + ) + request_id = str(uuid4()) + headers = { + "Content-Type": "application/json; charset=utf-8", + "Accept": "application/json", + "Idempotency-Key": idempotency_key.strip(), + "X-Request-Id": request_id, + "X-Client-Id": self._client_id, + } + if self._token: + headers["Authorization"] = f"Bearer {self._token}" + request = Request( + f"{self._base_url}/api/v1/client/tasks/{task_id.strip()}/spec-resolution", + data=json.dumps(observation, ensure_ascii=False).encode("utf-8"), + headers=headers, + method="POST", + ) + try: + with self._opener(request, timeout=self._timeout_seconds) as response: + status = getattr(response, "status", None) or response.getcode() + body = response.read() + except HTTPError as exc: + self._raise_http_error(exc, request_id, "解析采购规格") + except ( + URLError, + RemoteDisconnected, + ConnectionError, + socket.timeout, + TimeoutError, + ) as exc: + reason = getattr(exc, "reason", exc) + code = ( + "ADMIN_TIMEOUT" + if isinstance(reason, (socket.timeout, TimeoutError)) + else "ADMIN_UNAVAILABLE" + ) + raise AdminGatewayError( + code, + "Admin 规格解析超时" if code == "ADMIN_TIMEOUT" else "无法连接 Admin 规格解析服务", + True, + request_id, + ) from exc + if status != 200: + raise AdminGatewayError( + "ADMIN_UNEXPECTED_RESPONSE", + f"Admin 规格解析返回了未预期的状态码 {status}", + status >= 500, + request_id, + ) + return self._parse_spec_resolution_response(body, request_id) + + @classmethod + def _parse_spec_resolution_response( + cls, body: bytes, request_id: str + ) -> SpecResolutionReceipt: + data = cls._decode_json(body, request_id) + schema_version = data.get("schema_version") + resolution_id = data.get("resolution_id") + outcome = data.get("outcome") + source = data.get("source") + snapshot_hash = data.get("candidate_snapshot_hash") + confidence = data.get("confidence_bps") + reason = data.get("reason") + resolved_at = data.get("resolved_at") + match_data = data.get("match") + valid = ( + schema_version == 1 + and isinstance(resolution_id, str) + and bool(resolution_id.strip()) + and outcome in {"matched", "uncertain", "rejected", "failed"} + and (source is None or source in {"rule", "ai", "reused"}) + and isinstance(snapshot_hash, str) + and re.fullmatch(r"[0-9a-f]{64}", snapshot_hash) is not None + and (confidence is None or ( + isinstance(confidence, int) + and not isinstance(confidence, bool) + and 0 <= confidence <= 10000 + )) + and isinstance(reason, str) + and len(reason) <= 500 + and isinstance(resolved_at, str) + and bool(resolved_at.strip()) + ) + match = None + if outcome == "matched" and isinstance(match_data, Mapping): + candidate_id = match_data.get("candidate_id") + raw_text = match_data.get("raw_text") + options = match_data.get("options") + if ( + isinstance(candidate_id, str) + and bool(candidate_id.strip()) + and isinstance(raw_text, str) + and bool(raw_text) + and isinstance(options, Mapping) + and set(options) == {"color", "size"} + and all( + isinstance(value, str) and bool(value) + for value in options.values() + ) + and options.get("size") == raw_text + ): + match = SpecResolutionMatch( + candidate_id, + raw_text, + {str(key): str(value) for key, value in options.items()}, + ) + if ( + not valid + or (outcome == "matched") != (match is not None) + or (outcome != "matched" and match_data is not None) + ): + raise AdminGatewayError( + "ADMIN_INVALID_RESPONSE", + "Admin 规格解析响应字段不完整或不合法", + False, + request_id, + ) + return SpecResolutionReceipt( + 1, + resolution_id, + outcome, + source, + snapshot_hash, + match, + confidence, + reason, + resolved_at, + ) + def _submit( self, task_id: str, diff --git a/client/src/mock_admin_gateway.py b/client/src/mock_admin_gateway.py index b3cb622..970eab4 100644 --- a/client/src/mock_admin_gateway.py +++ b/client/src/mock_admin_gateway.py @@ -16,6 +16,8 @@ from .admin_gateway import ( ClaimCapabilities, ClientInfo, RegistrationReceipt, + SpecResolutionMatch, + SpecResolutionReceipt, SubmissionReceipt, ) @@ -48,6 +50,10 @@ class MockAdminGateway(AdminGateway): self._submissions: Dict[ str, Tuple[str, SubmissionReceipt] ] = {} + self._spec_resolutions: Dict[ + str, Tuple[str, SpecResolutionReceipt] + ] = {} + self._next_spec_resolution: Optional[SpecResolutionReceipt] = None self._next_error: Optional[AdminGatewayError] = None self._reject_next_submission = False self._registrations: Dict[str, Tuple[ClientInfo, ClaimCapabilities]] = {} @@ -170,6 +176,167 @@ class MockAdminGateway(AdminGateway): ) -> SubmissionReceipt: return self._submit("failure", task_id, idempotency_key, failure) + def set_next_spec_resolution( + self, receipt: SpecResolutionReceipt + ) -> None: + """测试辅助：设置下一次新规格解析请求的业务响应。""" + + self._next_spec_resolution = deepcopy(receipt) + + @property + def spec_resolution_count(self) -> int: + return len(self._spec_resolutions) + + def resolve_purchase_spec( + self, + task_id: str, + idempotency_key: str, + observation: Mapping[str, Any], + ) -> SpecResolutionReceipt: + with self._lock: + self._raise_forced_error() + task = self._validate_submission_target(task_id, idempotency_key) + self._validate_spec_observation(task, idempotency_key, observation) + fingerprint = self._fingerprint( + "spec_resolution", task_id, observation + ) + previous = self._spec_resolutions.get(idempotency_key) + if previous is not None: + old_fingerprint, receipt = previous + if old_fingerprint != fingerprint: + raise AdminGatewayError( + "IDEMPOTENCY_CONFLICT", + "相同规格解析幂等键携带了不同内容", + False, + ) + return deepcopy(receipt) + + receipt = self._next_spec_resolution or SpecResolutionReceipt( + schema_version=1, + resolution_id=str(uuid4()), + outcome="uncertain", + source=None, + candidate_snapshot_hash=str( + observation["candidate_snapshot_hash"] + ), + match=None, + confidence_bps=None, + reason="Mock 未配置唯一匹配", + resolved_at=utc_now_iso(), + ) + self._next_spec_resolution = None + self._spec_resolutions[idempotency_key] = ( + fingerprint, + deepcopy(receipt), + ) + return deepcopy(receipt) + + @staticmethod + def _validate_spec_observation( + task: AdminTask, + idempotency_key: str, + observation: Mapping[str, Any], + ) -> None: + if task.task_type.value != "purchase": + raise AdminGatewayError( + "TASK_NOT_PURCHASE", "当前任务不是采购任务", False + ) + if observation.get("schema_version") != 1: + raise AdminGatewayError( + "INVALID_SPEC_RESOLUTION_SCHEMA", "规格解析版本无效", False + ) + if observation.get("task_version") != task.version: + raise AdminGatewayError( + "TASK_VERSION_CONFLICT", "规格解析任务版本不一致", False + ) + payload = task.payload + if observation.get("pdd_goods_id") != payload.get("goods_id"): + raise AdminGatewayError( + "PDD_GOODS_MISMATCH", "规格解析商品编号不一致", False + ) + if dict(observation.get("original_options") or {}) != dict( + payload.get("options") or {} + ): + raise AdminGatewayError( + "INVALID_SPEC_RESOLUTION_REQUEST", "原始规格不一致", False + ) + candidates = observation.get("candidates") + snapshot_hash = observation.get("candidate_snapshot_hash") + selected_color = observation.get("selected_color") + if ( + not isinstance(candidates, list) + or not 1 <= len(candidates) <= 100 + or not isinstance(snapshot_hash, str) + or len(snapshot_hash) != 64 + or not str(observation.get("attempt_id") or "").strip() + ): + raise AdminGatewayError( + "INVALID_SPEC_RESOLUTION_REQUEST", "规格候选结构无效", False + ) + if any( + not isinstance(candidate, Mapping) + or candidate.get("candidate_id") != f"c{index}" + or not isinstance(candidate.get("raw_text"), str) + or not candidate["raw_text"] + or dict(candidate.get("options") or {}) + != { + "color": selected_color, + "size": candidate.get("raw_text"), + } + for index, candidate in enumerate(candidates, start=1) + ): + raise AdminGatewayError( + "INVALID_SPEC_RESOLUTION_REQUEST", "规格候选字段无效", False + ) + candidate_material = "".join( + ( + MockAdminGateway._frame("spec-resolution-v1"), + MockAdminGateway._frame(str(observation["pdd_goods_id"])), + MockAdminGateway._frame(str(selected_color)), + MockAdminGateway._frame(str(len(candidates))), + *( + "".join( + MockAdminGateway._frame(str(value)) + for value in ( + candidate.get("candidate_id"), + candidate.get("raw_text"), + (candidate.get("options") or {}).get("color"), + (candidate.get("options") or {}).get("size"), + ) + ) + for candidate in candidates + if isinstance(candidate, Mapping) + ), + ) + ) + expected_snapshot_hash = hashlib.sha256( + candidate_material.encode("utf-8") + ).hexdigest() + if snapshot_hash != expected_snapshot_hash: + raise AdminGatewayError( + "SPEC_RESOLUTION_HASH_MISMATCH", "候选快照哈希不一致", False + ) + identity = "".join( + MockAdminGateway._frame(str(value)) + for value in ( + task.task_id, + observation["attempt_id"], + snapshot_hash, + "spec-resolution-v1", + ) + ) + expected_key = "spec-resolution-v1:" + hashlib.sha256( + identity.encode("utf-8") + ).hexdigest() + if idempotency_key != expected_key: + raise AdminGatewayError( + "SPEC_RESOLUTION_HASH_MISMATCH", "规格解析幂等键不一致", False + ) + + @staticmethod + def _frame(value: str) -> str: + return f"{len(value.encode('utf-8'))}:{value}" + def _submit( self, submission_type: str, diff --git a/client/src/pdd_purchase_adapter.py b/client/src/pdd_purchase_adapter.py index 6d572d9..75345aa 100644 --- a/client/src/pdd_purchase_adapter.py +++ b/client/src/pdd_purchase_adapter.py @@ -198,6 +198,19 @@ class PddPurchaseAdapter(ABC): def select_options(self, options: Mapping[str, str]) -> None: """按完整动态规格对象精确选择，不做相似匹配。""" + def apply_resolved_size( + self, + expected_snapshot: PurchaseSpecCandidateSnapshot, + candidate: PurchaseSpecCandidate, + ) -> None: + """重新核对真机候选后应用 Admin 返回的原始尺码。""" + + raise PddPurchaseError( + "PURCHASE_SPEC_RESOLUTION_UNSUPPORTED", + "当前采购 Adapter 不支持运行时规格解析", + step="purchase_resolve_options", + ) + @abstractmethod def set_quantity(self, quantity: int) -> None: """设置采购数量。""" diff --git a/client/src/pdd_purchase_reconcile_adapter.py b/client/src/pdd_purchase_reconcile_adapter.py index 6e42e42..3504104 100644 --- a/client/src/pdd_purchase_reconcile_adapter.py +++ b/client/src/pdd_purchase_reconcile_adapter.py @@ -20,6 +20,8 @@ class PurchaseReconcileQuery: irreversible_action_at: str order_submitted_at: str reconcile_started_at: str + original_options: Mapping[str, str] = field(default_factory=dict) + spec_resolution: Mapping[str, object] = field(default_factory=dict) @dataclass(frozen=True) diff --git a/client/src/pdd_u2_purchase_adapter.py b/client/src/pdd_u2_purchase_adapter.py index e88e820..0e5bc1b 100644 --- a/client/src/pdd_u2_purchase_adapter.py +++ b/client/src/pdd_u2_purchase_adapter.py @@ -54,6 +54,7 @@ from .pdd_purchase_adapter import ( PddPurchaseError, PddPurchaseSpecResolutionRequired, PurchaseSizeObservation, + PurchaseSpecCandidate, PurchaseSpecCandidateSnapshot, PurchasePageState, ) @@ -1262,6 +1263,97 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter): ) return service.collect_second_dimension_candidates(device) + def apply_resolved_size( + self, + expected_snapshot: PurchaseSpecCandidateSnapshot, + candidate: PurchaseSpecCandidate, + ) -> None: + """重读并复算候选快照，只按响应中的页面原文选择一次尺码。""" + + self._check_cancelled("purchase_resolve_options") + if ( + expected_snapshot.goods_id != self._goods_id + or candidate not in expected_snapshot.candidates + or candidate.options.get("color") != expected_snapshot.selected_color + or candidate.options.get("size") != candidate.raw_text + ): + raise PddPurchaseError( + "PURCHASE_SPEC_RESOLUTION_INVALID", + "规格解析结果不属于本次商品和候选快照", + step="purchase_resolve_options", + ) + + current_xml = self._dump_hierarchy() + root = _parse_xml(current_xml) + if _page_kind(root, "") != "order_confirmation" or not _selected( + root, expected_snapshot.selected_color + ): + raise PddPurchaseError( + "PURCHASE_SPEC_CONTEXT_CHANGED", + "等待规格解析期间页面或已选颜色发生变化", + step="purchase_resolve_options", + ) + + current_xml = self._restore_purchase_panel_color_region(current_xml) + try: + dimension = self._size_candidate_collector(self._require_device()) + except PddCollectError as exc: + raise PddPurchaseError( + "PURCHASE_SPEC_CANDIDATES_CHANGED", + f"无法重新完整读取尺码候选：{exc.message}", + step="purchase_resolve_options", + diagnostics={"candidate_scan_error": exc.code}, + ) from exc + if dimension is None: + raise PddPurchaseError( + "PURCHASE_SPEC_CANDIDATES_CHANGED", + "等待规格解析期间尺码候选已经消失", + step="purchase_resolve_options", + ) + self._validate_size_candidate_dimension(dimension) + current_snapshot = PurchaseSpecCandidateSnapshot.build( + goods_id=self._goods_id, + selected_color=expected_snapshot.selected_color, + target_size=expected_snapshot.target_size, + dimension_name=dimension.name, + observations=tuple( + PurchaseSizeObservation(item.text, item.available) + for item in dimension.values + ), + observed_at=expected_snapshot.observed_at, + ) + if ( + current_snapshot.dimension_name != expected_snapshot.dimension_name + or current_snapshot.candidate_snapshot_hash + != expected_snapshot.candidate_snapshot_hash + or current_snapshot.candidates != expected_snapshot.candidates + ): + raise PddPurchaseError( + "PURCHASE_SPEC_CANDIDATES_CHANGED", + "等待规格解析期间真机尺码候选已经变化", + step="purchase_resolve_options", + ) + + latest_xml = self._restore_purchase_panel_color_region( + self._dump_hierarchy() + ) + if not self._select_size( + self._require_device(), latest_xml, candidate.raw_text, action_delay=0.2 + ): + raise PddPurchaseError( + "PURCHASE_RESOLVED_SIZE_NOT_SELECTED", + "解析得到的原始尺码没有可靠选中", + step="purchase_resolve_options", + ) + selected_xml = self._dump_hierarchy() + if not _selected(_parse_xml(selected_xml), candidate.raw_text): + raise PddPurchaseError( + "PURCHASE_RESOLVED_SIZE_NOT_SELECTED", + "点击解析尺码后页面没有确认选中", + step="purchase_resolve_options", + ) + self._requested_options["size"] = candidate.raw_text + def set_quantity(self, quantity: int) -> None: device = self._require_device() if isinstance(quantity, bool) or not isinstance(quantity, int) or quantity <= 0: diff --git a/client/src/purchase_reconcile_service.py b/client/src/purchase_reconcile_service.py index 2758839..018bc42 100644 --- a/client/src/purchase_reconcile_service.py +++ b/client/src/purchase_reconcile_service.py @@ -255,7 +255,15 @@ class PurchaseReconcileService: normalized_confirmed_options = { str(key): str(value) for key, value in confirmed_options.items() } - if normalized_confirmed_options != requested_options: + resolution = snapshot.get("spec_resolution") + resolution = resolution if isinstance(resolution, Mapping) else {} + applied_options = resolution.get("applied_options") + effective_options = ( + {str(key): str(value) for key, value in applied_options.items()} + if isinstance(applied_options, Mapping) + else requested_options + ) + if normalized_confirmed_options != effective_options: raise ValueError("采购运行确认规格与任务不一致") if ( isinstance(confirmed_quantity, bool) @@ -291,6 +299,8 @@ class PurchaseReconcileService: irreversible_action_at=str(run.irreversible_action_at), order_submitted_at=order_submitted_at, reconcile_started_at=_utc_now_iso(), + original_options=requested_options, + spec_resolution=dict(resolution), ) def _result_data( @@ -306,7 +316,7 @@ class PurchaseReconcileService: "purchase": { "mode": "live", "requested": { - "options": dict(query.options), + "options": dict(query.original_options or query.options), "quantity": query.quantity, "max_price_cent": task.price_cent, }, @@ -324,6 +334,9 @@ class PurchaseReconcileService: "ordered_at": candidate.ordered_at, "ordered_at_raw": candidate.ordered_at_raw, "match_status": "matched", + "spec_resolution": dict(query.spec_resolution) + if query.spec_resolution + else None, }, "captured_at": _utc_now_iso(), "source": { diff --git a/client/src/purchase_task_service.py b/client/src/purchase_task_service.py index b1d439f..04de61c 100644 --- a/client/src/purchase_task_service.py +++ b/client/src/purchase_task_service.py @@ -6,15 +6,26 @@ from __future__ import annotations -from dataclasses import dataclass +import hashlib +import json +from dataclasses import dataclass, replace from datetime import datetime, timezone -from typing import Callable, Mapping +from typing import Callable, Mapping, Optional -from .admin_gateway import AdminGateway, AdminGatewayError, ClientInfo +from .admin_gateway import ( + AdminGateway, + AdminGatewayError, + ClientInfo, + SpecResolutionMatch, + SpecResolutionReceipt, +) from .pdd_purchase_adapter import ( PddLivePurchaseAdapter, PddPurchaseAdapter, PddPurchaseError, + PddPurchaseSpecResolutionRequired, + PurchaseSpecCandidate, + PurchaseSpecCandidateSnapshot, PurchasePageState, ) from .task_models import OutboxEventRecord, OutboxEventType, TaskStatus @@ -38,6 +49,8 @@ class PurchaseTarget: options: Mapping[str, str] quantity: int max_price_cent: int # 人民币订单总价上限，单位分 + original_options: Optional[Mapping[str, str]] = None + spec_resolution: Optional[Mapping[str, object]] = None @dataclass(frozen=True) @@ -121,6 +134,7 @@ class PurchaseTaskService: ) self._started(remote_task_id) adapter: PddPurchaseAdapter | None = None + target: PurchaseTarget | None = None step = "purchase_prepare" failure_message = "" failure_outcome_kind = "" @@ -149,7 +163,19 @@ class PurchaseTaskService: step = "purchase_select_options" self._enter_step(remote_task_id, started.attempt_id, step) self._validate_identity(adapter.read_state(), target, "goods") - adapter.select_options(target.options) + try: + adapter.select_options(target.options) + except PddPurchaseSpecResolutionRequired as required: + step = "purchase_resolve_options" + self._enter_step(remote_task_id, started.attempt_id, step) + target = self._resolve_and_apply_spec( + remote_task_id, + started.attempt_id, + started.task.version, + target, + required.snapshot, + adapter, + ) step = "purchase_verify_options" self._enter_step(remote_task_id, started.attempt_id, step) @@ -216,6 +242,9 @@ class PurchaseTaskService: failure_outcome_kind = "global_failed" else: failure_outcome_kind = "task_failed" + diagnostics = dict(exc.diagnostics) + if target is not None and target.spec_resolution is not None: + diagnostics["spec_resolution"] = dict(target.spec_resolution) event = self._save_failure( remote_task_id, started.attempt_id, @@ -223,7 +252,7 @@ class PurchaseTaskService: exc.message, exc.retryable, step, - exc.diagnostics, + diagnostics, ) except Exception as exc: failure_message = f"采购在“{step}”发生未知错误：{exc}" @@ -277,6 +306,9 @@ class PurchaseTaskService: "quantity": state.quantity, "unit_price_cent": state.price_cent, "total_price_cent": state.price_cent * state.quantity, + "spec_resolution": dict(target.spec_resolution) + if target.spec_resolution is not None + else None, }, ) submitted_at = None @@ -314,6 +346,214 @@ class PurchaseTaskService: remote_task_id, ) + def _resolve_and_apply_spec( + self, + remote_task_id: str, + attempt_id: str, + task_version: int, + target: PurchaseTarget, + snapshot: PurchaseSpecCandidateSnapshot, + adapter: PddPurchaseAdapter, + ) -> PurchaseTarget: + """请求一次规格解析，先落库，再重读真机并精确应用结果。""" + + if ( + snapshot.goods_id != target.goods_id + or snapshot.selected_color != target.options.get("color") + or snapshot.target_size != target.options.get("size") + ): + raise PddPurchaseError( + "PURCHASE_SPEC_CONTEXT_CHANGED", + "规格候选的商品、颜色或目标尺码与采购任务不一致", + step="purchase_resolve_options", + ) + request = { + "schema_version": 1, + "task_version": task_version, + "attempt_id": attempt_id, + "pdd_goods_id": snapshot.goods_id, + "original_options": dict(target.options), + "selected_color": snapshot.selected_color, + "target_size": snapshot.target_size, + "candidates": [item.to_dict() for item in snapshot.candidates], + "candidate_snapshot_hash": snapshot.candidate_snapshot_hash, + "observed_at": snapshot.observed_at, + } + encoded = json.dumps( + request, ensure_ascii=False, separators=(",", ":") + ).encode("utf-8") + if len(encoded) > 64 * 1024: + raise PddPurchaseError( + "PURCHASE_SPEC_RESOLUTION_REQUEST_TOO_LARGE", + "规格解析请求超过 64 KiB，已停止采购", + step="purchase_resolve_options", + ) + idempotency_key = self._spec_resolution_idempotency_key( + remote_task_id, attempt_id, snapshot.candidate_snapshot_hash + ) + record = self._repository.prepare_purchase_spec_resolution( + remote_task_id, + attempt_id, + idempotency_key, + request, + ) + if record.status == "resolved": + receipt = self._receipt_from_record(record) + else: + try: + receipt = self._gateway.resolve_purchase_spec( + remote_task_id, idempotency_key, record.request_json + ) + except AdminGatewayError as exc: + raise PddPurchaseError( + "PURCHASE_SPEC_RESOLUTION_ADMIN_FAILED", + f"Admin 规格解析失败：{exc}", + step="purchase_resolve_options", + diagnostics={ + "admin_error_code": exc.code, + "request_id": exc.request_id, + }, + ) from exc + record = self._repository.save_purchase_spec_resolution( + record.id, self._receipt_to_dict(receipt) + ) + + candidate = self._validated_resolution_candidate(snapshot, receipt) + effective_options = dict(target.options) + effective_options["size"] = candidate.raw_text + audit = { + "resolution_id": receipt.resolution_id, + "candidate_snapshot_hash": receipt.candidate_snapshot_hash, + "source": receipt.source, + "confidence_bps": receipt.confidence_bps, + "candidate_id": candidate.candidate_id, + "raw_text": candidate.raw_text, + "applied_options": dict(effective_options), + } + try: + adapter.apply_resolved_size(snapshot, candidate) + except PddPurchaseError as exc: + diagnostics = dict(exc.diagnostics) + diagnostics["spec_resolution"] = dict(audit) + raise PddPurchaseError( + exc.code, + exc.message, + step=exc.step, + retryable=False, + diagnostics=diagnostics, + ) from exc + return replace( + target, + options=effective_options, + original_options=dict(target.options), + spec_resolution=audit, + ) + + @staticmethod + def _validated_resolution_candidate( + snapshot: PurchaseSpecCandidateSnapshot, + receipt: SpecResolutionReceipt, + ) -> PurchaseSpecCandidate: + if receipt.candidate_snapshot_hash != snapshot.candidate_snapshot_hash: + raise PddPurchaseError( + "PURCHASE_SPEC_RESOLUTION_INVALID", + "Admin 返回的候选快照哈希与请求不一致", + step="purchase_resolve_options", + ) + if receipt.outcome != "matched" or receipt.match is None: + code = { + "uncertain": "PURCHASE_SPEC_RESOLUTION_UNCERTAIN", + "rejected": "PURCHASE_SPEC_RESOLUTION_REJECTED", + "failed": "PURCHASE_SPEC_RESOLUTION_FAILED", + }.get(receipt.outcome, "PURCHASE_SPEC_RESOLUTION_INVALID") + raise PddPurchaseError( + code, + f"Admin 未返回唯一可用尺码：{receipt.reason}", + step="purchase_resolve_options", + diagnostics={"resolution_id": receipt.resolution_id}, + ) + match = receipt.match + candidates = tuple( + candidate + for candidate in snapshot.candidates + if candidate.candidate_id == match.candidate_id + and candidate.raw_text == match.raw_text + and dict(candidate.options) == dict(match.options) + ) + if len(candidates) != 1: + raise PddPurchaseError( + "PURCHASE_SPEC_RESOLUTION_INVALID", + "Admin 返回的规格不是本次候选的逐字副本", + step="purchase_resolve_options", + diagnostics={"resolution_id": receipt.resolution_id}, + ) + return candidates[0] + + @staticmethod + def _spec_resolution_idempotency_key( + remote_task_id: str, + attempt_id: str, + snapshot_hash: str, + ) -> str: + def frame(value: str) -> str: + return f"{len(value.encode('utf-8'))}:{value}" + + material = "".join( + frame(value) + for value in ( + remote_task_id, + attempt_id, + snapshot_hash, + "spec-resolution-v1", + ) + ) + return "spec-resolution-v1:" + hashlib.sha256( + material.encode("utf-8") + ).hexdigest() + + @staticmethod + def _receipt_to_dict(receipt: SpecResolutionReceipt) -> dict[str, object]: + return { + "schema_version": receipt.schema_version, + "resolution_id": receipt.resolution_id, + "outcome": receipt.outcome, + "source": receipt.source, + "candidate_snapshot_hash": receipt.candidate_snapshot_hash, + "match": ( + { + "candidate_id": receipt.match.candidate_id, + "raw_text": receipt.match.raw_text, + "options": dict(receipt.match.options), + } + if receipt.match is not None + else None + ), + "confidence_bps": receipt.confidence_bps, + "reason": receipt.reason, + "resolved_at": receipt.resolved_at, + } + + @staticmethod + def _receipt_from_record(record: object) -> SpecResolutionReceipt: + match = None + if getattr(record, "match_candidate_id", None): + match = SpecResolutionMatch( + record.match_candidate_id, + record.match_raw_text, + record.match_options or {}, + ) + return SpecResolutionReceipt( + 1, + record.resolution_id or "", + record.outcome or "failed", + record.source, + record.candidate_snapshot_hash, + match, + record.confidence_bps, + record.reason or "", + record.resolved_at or "", + ) + def _enter_step( self, remote_task_id: str, attempt_id: str, step: str ) -> None: @@ -619,7 +859,7 @@ class PurchaseTaskService: "purchase": { "mode": "dry_run", "requested": { - "options": dict(target.options), + "options": dict(target.original_options or target.options), "quantity": target.quantity, "max_price_cent": target.max_price_cent, }, @@ -636,6 +876,9 @@ class PurchaseTaskService: "ordered_at": None, "ordered_at_raw": None, "match_status": "not_submitted", + "spec_resolution": dict(target.spec_resolution) + if target.spec_resolution is not None + else None, }, "captured_at": utc_now_iso(), "source": { diff --git a/client/src/task_models.py b/client/src/task_models.py index 0ad6482..cacac40 100644 --- a/client/src/task_models.py +++ b/client/src/task_models.py @@ -226,6 +226,32 @@ class StartedTaskRun: attempt_no: int +@dataclass(frozen=True) +class PurchaseSpecResolutionRecord: + """一次采购执行中的规格解析请求和已持久化响应。""" + + id: int + task_id: int + attempt_id: str + idempotency_key: str + request_json: Dict[str, Any] + request_hash: str + candidate_snapshot_hash: str + status: str + resolution_id: Optional[str] + outcome: Optional[str] + source: Optional[str] + confidence_bps: Optional[int] + match_candidate_id: Optional[str] + match_raw_text: Optional[str] + match_options: Optional[Dict[str, str]] + reason: Optional[str] + resolved_at: Optional[str] + received_at: Optional[str] + created_at: str + updated_at: str + + @dataclass(frozen=True) class AppSettingRecord: """一条非敏感应用设置。""" diff --git a/client/src/task_repository.py b/client/src/task_repository.py index 9e2fc0d..ac304ce 100644 --- a/client/src/task_repository.py +++ b/client/src/task_repository.py @@ -3,6 +3,7 @@ Repository 是数据库访问入口。界面和自动化代码不应自行拼接任务 SQL。 """ +import hashlib import json import sqlite3 from datetime import datetime, timezone @@ -16,6 +17,7 @@ from .task_models import ( OutboxEventRecord, OutboxEventType, OutboxStatus, + PurchaseSpecResolutionRecord, RunStatus, StartedTaskRun, TaskDetail, @@ -820,6 +822,159 @@ class TaskRepository: finally: connection.close() + def prepare_purchase_spec_resolution( + self, + remote_task_id: str, + attempt_id: str, + idempotency_key: str, + request: Dict[str, object], + ) -> PurchaseSpecResolutionRecord: + """发送前保存完整请求；相同身份只能复用完全相同的内容。""" + + canonical = json.dumps( + request, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + request_hash = hashlib.sha256(canonical.encode("utf-8")).hexdigest() + snapshot_hash = str(request.get("candidate_snapshot_hash") or "") + if not idempotency_key.strip() or len(snapshot_hash) != 64: + raise ValueError("规格解析幂等键或候选快照哈希无效") + now = utc_now_iso() + connection = open_database(self._db_path) + try: + with connection: + task = connection.execute( + "SELECT id, task_type, status FROM pdd_tasks" + " WHERE remote_task_id = ?", + (remote_task_id,), + ).fetchone() + if task is None or task["task_type"] != TaskType.PURCHASE.value: + raise ValueError("采购任务不存在或类型错误") + run = connection.execute( + "SELECT run_status, irreversible_action_at FROM task_runs" + " WHERE task_id = ? AND attempt_id = ?", + (task["id"], attempt_id), + ).fetchone() + if ( + task["status"] != TaskStatus.RUNNING.value + or run is None + or run["run_status"] != RunStatus.RUNNING.value + or run["irreversible_action_at"] is not None + ): + raise ValueError("采购执行已结束或已进入不可逆阶段") + existing = connection.execute( + "SELECT * FROM purchase_spec_resolutions" + " WHERE task_id = ? AND attempt_id = ?" + " AND candidate_snapshot_hash = ?", + (task["id"], attempt_id, snapshot_hash), + ).fetchone() + if existing is not None: + if ( + existing["idempotency_key"] != idempotency_key + or existing["request_hash"] != request_hash + or existing["request_json"] != canonical + ): + raise ValueError("同一规格解析身份的请求内容发生冲突") + return self._to_spec_resolution(existing) + cursor = connection.execute( + "INSERT INTO purchase_spec_resolutions (task_id, attempt_id," + " idempotency_key, request_json, request_hash," + " candidate_snapshot_hash, created_at, updated_at)" + " VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + ( + task["id"], attempt_id, idempotency_key, canonical, + request_hash, snapshot_hash, now, now, + ), + ) + row = connection.execute( + "SELECT * FROM purchase_spec_resolutions WHERE id = ?", + (cursor.lastrowid,), + ).fetchone() + assert row is not None + return self._to_spec_resolution(row) + finally: + connection.close() + + def save_purchase_spec_resolution( + self, + record_id: int, + response: Dict[str, object], + ) -> PurchaseSpecResolutionRecord: + """先保存 Admin 业务响应，再允许自动化继续选择尺码。""" + + match = response.get("match") + match_data = dict(match) if isinstance(match, dict) else {} + options = match_data.get("options") + now = utc_now_iso() + connection = open_database(self._db_path) + try: + with connection: + current = connection.execute( + "SELECT r.*, tr.run_status, tr.irreversible_action_at" + " FROM purchase_spec_resolutions r" + " JOIN task_runs tr ON tr.attempt_id = r.attempt_id" + " AND tr.task_id = r.task_id" + " WHERE r.id = ?", + (record_id,), + ).fetchone() + if current is None: + raise ValueError("规格解析请求记录不存在") + if current["status"] == "resolved": + saved = self._to_spec_resolution(current) + incoming_match = ( + { + str(key): str(value) + for key, value in options.items() + } + if isinstance(options, dict) + else None + ) + if ( + saved.resolution_id != response.get("resolution_id") + or saved.outcome != response.get("outcome") + or saved.source != response.get("source") + or saved.confidence_bps != response.get("confidence_bps") + or saved.match_candidate_id + != match_data.get("candidate_id") + or saved.match_raw_text != match_data.get("raw_text") + or saved.match_options != incoming_match + or saved.reason != response.get("reason") + or saved.resolved_at != response.get("resolved_at") + ): + raise ValueError("同一规格解析记录收到了不同响应") + return saved + if ( + current["run_status"] != RunStatus.RUNNING.value + or current["irreversible_action_at"] is not None + ): + raise ValueError("采购执行已结束或已进入不可逆阶段") + connection.execute( + "UPDATE purchase_spec_resolutions SET status = 'resolved'," + " resolution_id = ?, outcome = ?, source = ?," + " confidence_bps = ?, match_candidate_id = ?," + " match_raw_text = ?, match_options_json = ?, reason = ?," + " resolved_at = ?, received_at = ?, updated_at = ? WHERE id = ?", + ( + response.get("resolution_id"), response.get("outcome"), + response.get("source"), response.get("confidence_bps"), + match_data.get("candidate_id"), match_data.get("raw_text"), + json.dumps(options, ensure_ascii=False) + if isinstance(options, dict) else None, + response.get("reason"), response.get("resolved_at"), + now, now, record_id, + ), + ) + row = connection.execute( + "SELECT * FROM purchase_spec_resolutions WHERE id = ?", + (record_id,), + ).fetchone() + assert row is not None + return self._to_spec_resolution(row) + finally: + connection.close() + def mark_purchase_irreversible( self, remote_task_id: str, @@ -1775,6 +1930,41 @@ class TaskRepository: sent_at=row["sent_at"], ) + @staticmethod + def _to_spec_resolution( + row: sqlite3.Row, + ) -> PurchaseSpecResolutionRecord: + match_options = None + if row["match_options_json"]: + loaded = TaskRepository._load_json_object( + row["match_options_json"] + ) + match_options = { + str(key): str(value) for key, value in loaded.items() + } + return PurchaseSpecResolutionRecord( + id=row["id"], + task_id=row["task_id"], + attempt_id=row["attempt_id"], + idempotency_key=row["idempotency_key"], + request_json=TaskRepository._load_json_object(row["request_json"]), + request_hash=row["request_hash"], + candidate_snapshot_hash=row["candidate_snapshot_hash"], + status=row["status"], + resolution_id=row["resolution_id"], + outcome=row["outcome"], + source=row["source"], + confidence_bps=row["confidence_bps"], + match_candidate_id=row["match_candidate_id"], + match_raw_text=row["match_raw_text"], + match_options=match_options, + reason=row["reason"], + resolved_at=row["resolved_at"], + received_at=row["received_at"], + created_at=row["created_at"], + updated_at=row["updated_at"], + ) + @staticmethod def _to_task_run(row: sqlite3.Row) -> TaskRunRecord: diagnostics = ( diff --git a/client/test/test_admin_gateway_contract.py b/client/test/test_admin_gateway_contract.py index 0553338..23d6f8c 100644 --- a/client/test/test_admin_gateway_contract.py +++ b/client/test/test_admin_gateway_contract.py @@ -1,5 +1,6 @@ """AdminGateway 边界和 Mock 契约测试。""" +import hashlib import unittest from src.admin_gateway import ( @@ -9,6 +10,8 @@ from src.admin_gateway import ( AndroidDeviceInfo, ClaimCapabilities, ClientInfo, + SpecResolutionMatch, + SpecResolutionReceipt, ) from src.mock_admin_gateway import MockAdminGateway from src.task_models import TaskType @@ -61,7 +64,7 @@ class MockAdminGatewayContractTest(unittest.TestCase): "reported_at": "2026-08-06T08:03:00Z", } - def test_gateway_has_only_four_business_methods(self): + def test_gateway_has_only_expected_business_methods(self): self.assertEqual( AdminGateway.__abstractmethods__, { @@ -69,6 +72,7 @@ class MockAdminGatewayContractTest(unittest.TestCase): "claim_next", "submit_result", "submit_failure", + "resolve_purchase_spec", }, ) for forbidden in ("get_status", "heartbeat", "renew_lease"): @@ -162,6 +166,92 @@ class MockAdminGatewayContractTest(unittest.TestCase): ) self.assertTrue(unavailable_context.exception.retryable) + def test_spec_resolution_is_idempotent_and_returns_configured_match(self): + task = AdminTask( + task_id="PUR-SPEC", + task_type=TaskType.PURCHASE, + version=3, + priority=1, + payload={ + "goods_id": "737116531267", + "options": {"color": "黑色", "size": "60公斤"}, + }, + ) + self.gateway.enqueue_task(task, self.client.client_id) + self.gateway.claim_next(self.client, self.all_capabilities) + frame = lambda value: f"{len(value.encode('utf-8'))}:{value}" + snapshot_material = "".join( + frame(value) + for value in ( + "spec-resolution-v1", + "737116531267", + "黑色", + "1", + "c1", + "120斤", + "黑色", + "120斤", + ) + ) + snapshot_hash = hashlib.sha256( + snapshot_material.encode("utf-8") + ).hexdigest() + observation = { + "schema_version": 1, + "task_version": 3, + "attempt_id": "attempt-001", + "pdd_goods_id": "737116531267", + "original_options": {"color": "黑色", "size": "60公斤"}, + "selected_color": "黑色", + "target_size": "60公斤", + "candidates": [ + { + "candidate_id": "c1", + "raw_text": "120斤", + "options": {"color": "黑色", "size": "120斤"}, + } + ], + "candidate_snapshot_hash": snapshot_hash, + "observed_at": "2026-08-17T08:00:00Z", + } + material = "".join( + frame(value) + for value in ( + "PUR-SPEC", + "attempt-001", + snapshot_hash, + "spec-resolution-v1", + ) + ) + key = "spec-resolution-v1:" + hashlib.sha256( + material.encode("utf-8") + ).hexdigest() + configured = SpecResolutionReceipt( + 1, + "psr-001", + "matched", + "rule", + snapshot_hash, + SpecResolutionMatch( + "c1", "120斤", {"color": "黑色", "size": "120斤"} + ), + 10000, + "唯一重量等价", + "2026-08-17T08:00:01Z", + ) + self.gateway.set_next_spec_resolution(configured) + + first = self.gateway.resolve_purchase_spec( + "PUR-SPEC", key, observation + ) + second = self.gateway.resolve_purchase_spec( + "PUR-SPEC", key, observation + ) + + self.assertEqual(first, configured) + self.assertEqual(second, configured) + self.assertEqual(self.gateway.spec_resolution_count, 1) + def test_same_idempotency_key_and_content_reuses_receipt(self): self.gateway.enqueue_task( self._task("TASK-001", TaskType.COLLECT), "client-001" diff --git a/client/test/test_db.py b/client/test/test_db.py index 68c8c95..272a5f4 100644 --- a/client/test/test_db.py +++ b/client/test/test_db.py @@ -6,7 +6,13 @@ import unittest from pathlib import Path from src.db import DatabaseVersionError, initialize_database, open_database -from src.db_schema import MIGRATION_1, MIGRATION_2, MIGRATION_3, MIGRATION_4 +from src.db_schema import ( + MIGRATION_1, + MIGRATION_2, + MIGRATION_3, + MIGRATION_4, + MIGRATION_5, +) EXPECTED_TABLES = { @@ -14,6 +20,7 @@ EXPECTED_TABLES = { "task_runs", "outbox_events", "app_settings", + "purchase_spec_resolutions", } EXPECTED_INDEXES = { @@ -23,6 +30,7 @@ EXPECTED_INDEXES = { "idx_task_runs_task", "idx_outbox_pending", "idx_pdd_tasks_visible_list", + "idx_purchase_spec_resolutions_attempt", } @@ -64,7 +72,7 @@ class DatabaseInitializationTests(unittest.TestCase): self.assertTrue(EXPECTED_TABLES.issubset(tables)) self.assertTrue(EXPECTED_INDEXES.issubset(indexes)) - self.assertEqual(version, 5) + self.assertEqual(version, 6) def test_v1_database_is_upgraded_without_losing_task_runs(self) -> None: connection = open_database(self.db_path) @@ -107,7 +115,7 @@ class DatabaseInitializationTests(unittest.TestCase): connection.close() self.assertIn("result_data", columns) self.assertEqual(attempt_id, "ATTEMPT-OLD") - self.assertEqual(version, 5) + self.assertEqual(version, 6) def test_initialize_can_run_twice_without_losing_data(self) -> None: initialize_database(self.db_path) @@ -170,7 +178,7 @@ class DatabaseInitializationTests(unittest.TestCase): finally: connection.close() self.assertEqual(row[0], "dry_run") - self.assertEqual(version, 5) + self.assertEqual(version, 6) def test_v3_database_adds_soft_remove_column_without_losing_tasks(self) -> None: connection = open_database(self.db_path) @@ -212,7 +220,7 @@ class DatabaseInitializationTests(unittest.TestCase): connection.close() self.assertIn("removed_at", columns) self.assertEqual(tuple(row), ("COL-V3", None)) - self.assertEqual(version, 5) + self.assertEqual(version, 6) def test_v4_database_backfills_shop_name_from_valid_pdd_data(self) -> None: connection = open_database(self.db_path) @@ -245,7 +253,40 @@ class DatabaseInitializationTests(unittest.TestCase): finally: connection.close() self.assertEqual(row[0], "测试店铺") - self.assertEqual(version, 5) + self.assertEqual(version, 6) + + def test_v5_database_adds_purchase_spec_resolution_audit_table(self): + connection = open_database(self.db_path) + try: + with connection: + for statement in ( + MIGRATION_1 + + MIGRATION_2 + + MIGRATION_3 + + MIGRATION_4 + + MIGRATION_5 + ): + connection.execute(statement) + connection.execute("PRAGMA user_version = 5") + finally: + connection.close() + + initialize_database(self.db_path) + + connection = open_database(self.db_path) + try: + columns = { + row[1] + for row in connection.execute( + "PRAGMA table_info(purchase_spec_resolutions)" + ) + } + version = connection.execute("PRAGMA user_version").fetchone()[0] + finally: + connection.close() + self.assertIn("candidate_snapshot_hash", columns) + self.assertIn("idempotency_key", columns) + self.assertEqual(version, 6) def test_new_connection_uses_required_pragmas(self) -> None: initialize_database(self.db_path) diff --git a/client/test/test_http_admin_gateway.py b/client/test/test_http_admin_gateway.py index d49d02a..2fadce9 100644 --- a/client/test/test_http_admin_gateway.py +++ b/client/test/test_http_admin_gateway.py @@ -206,6 +206,107 @@ class HttpAdminGatewayTest(unittest.TestCase): self.assertTrue(receipt.accepted) self.assertTrue(receipt.result_id.startswith("legacy-failure-")) + def test_resolve_purchase_spec_posts_once_and_parses_whitelisted_match(self): + response = { + "schema_version": 1, + "resolution_id": "psr-001", + "outcome": "matched", + "source": "rule", + "candidate_snapshot_hash": "a" * 64, + "match": { + "candidate_id": "c1", + "raw_text": "120斤", + "options": {"color": "黑色", "size": "120斤"}, + }, + "confidence_bps": 10000, + "reason": "唯一等价", + "resolved_at": "2026-08-17T08:00:01Z", + "future_field": "ignored", + } + opener = RecordingOpener(FakeResponse(200, response)) + gateway = HttpAdminGateway( + opener=opener, client_id="CLIENT-001", timeout_seconds=5 + ) + + receipt = gateway.resolve_purchase_spec( + "PUR-001", "spec-resolution-v1:key", {"schema_version": 1} + ) + + self.assertEqual(receipt.resolution_id, "psr-001") + self.assertEqual(receipt.match.raw_text, "120斤") + self.assertTrue(opener.request.full_url.endswith( + "/api/v1/client/tasks/PUR-001/spec-resolution" + )) + headers = { + key.lower(): value for key, value in opener.request.header_items() + } + self.assertEqual( + headers["idempotency-key"], "spec-resolution-v1:key" + ) + self.assertEqual(opener.timeout, 5) + + def test_resolve_purchase_spec_accepts_all_nonmatched_business_outcomes(self): + for outcome in ("uncertain", "rejected", "failed"): + with self.subTest(outcome=outcome): + gateway = HttpAdminGateway( + opener=RecordingOpener( + FakeResponse( + 200, + { + "schema_version": 1, + "resolution_id": f"psr-{outcome}", + "outcome": outcome, + "source": None, + "candidate_snapshot_hash": "b" * 64, + "match": None, + "confidence_bps": None, + "reason": "没有唯一安全候选", + "resolved_at": "2026-08-17T08:00:01Z", + }, + ) + ), + client_id="CLIENT-001", + ) + + receipt = gateway.resolve_purchase_spec( + "PUR-001", "spec-resolution-v1:key", {} + ) + + self.assertEqual(receipt.outcome, outcome) + self.assertIsNone(receipt.match) + + def test_resolve_purchase_spec_rejects_match_on_nonmatched_outcome(self): + gateway = HttpAdminGateway( + opener=RecordingOpener( + FakeResponse( + 200, + { + "schema_version": 1, + "resolution_id": "psr-invalid", + "outcome": "uncertain", + "source": "ai", + "candidate_snapshot_hash": "c" * 64, + "match": { + "candidate_id": "c1", + "raw_text": "120斤", + "options": {"color": "黑色", "size": "120斤"}, + }, + "confidence_bps": 5000, + "reason": "响应自相矛盾", + "resolved_at": "2026-08-17T08:00:01Z", + }, + ) + ), + client_id="CLIENT-001", + ) + + with self.assertRaises(AdminGatewayError) as raised: + gateway.resolve_purchase_spec( + "PUR-001", "spec-resolution-v1:key", {} + ) + + self.assertEqual(raised.exception.code, "ADMIN_INVALID_RESPONSE") + def test_malformed_2xx_submission_is_retryable_and_not_called_rejected(self): gateway = HttpAdminGateway( opener=RecordingOpener(FakeResponse(200, {"accepted": True})), diff --git a/client/test/test_pdd_u2_purchase_adapter.py b/client/test/test_pdd_u2_purchase_adapter.py index e23410c..54ea157 100644 --- a/client/test/test_pdd_u2_purchase_adapter.py +++ b/client/test/test_pdd_u2_purchase_adapter.py @@ -11,6 +11,8 @@ from src.pdd_collect_service import DimensionValue, PddCollectError, SpecDimensi from src.pdd_purchase_adapter import ( PddPurchaseError, PddPurchaseSpecResolutionRequired, + PurchaseSizeObservation, + PurchaseSpecCandidateSnapshot, ) from src.performance_timing import TaskPerformanceTrace from src.pdd_u2_purchase_adapter import ( @@ -991,6 +993,76 @@ class U2PddPurchaseAdapterTest(unittest.TestCase): ) adapter.close() + def test_apply_resolved_size_rechecks_same_snapshot_before_exact_selection(self): + device = FakeDevice() + selected = [] + snapshot = PurchaseSpecCandidateSnapshot.build( + goods_id="753136429979", + selected_color="黑色", + target_size="XXL", + dimension_name="尺码", + observations=( + PurchaseSizeObservation("3XL【140-165斤】", True), + ), + observed_at="2026-08-17T08:00:00Z", + ) + adapter = U2PddPurchaseAdapter( + "USB-001", + device_service=PddDeviceService(connector=lambda _serial: device), + sleeper=lambda _seconds: None, + select_size_fn=lambda _device, _xml, target, **_kwargs: ( + selected.append(target) or True + ), + size_candidate_collector=lambda _device: SpecDimension( + "size", + "尺码", + (DimensionValue("3XL【140-165斤】", True),), + ), + ) + adapter.open_goods(GOODS_URL) + device.click(790, 2214) + + adapter.apply_resolved_size(snapshot, snapshot.candidates[0]) + + self.assertEqual(selected, ["3XL【140-165斤】"]) + self.assertEqual( + adapter.read_state().selected_options, + {"size": "3XL【140-165斤】"}, + ) + adapter.close() + + def test_apply_resolved_size_stops_when_candidate_snapshot_changed(self): + device = FakeDevice() + selections = [] + snapshot = PurchaseSpecCandidateSnapshot.build( + goods_id="753136429979", + selected_color="黑色", + target_size="XXL", + dimension_name="尺码", + observations=(PurchaseSizeObservation("M", True),), + observed_at="2026-08-17T08:00:00Z", + ) + adapter = U2PddPurchaseAdapter( + "USB-001", + device_service=PddDeviceService(connector=lambda _serial: device), + sleeper=lambda _seconds: None, + select_size_fn=lambda *_args, **_kwargs: selections.append(True), + size_candidate_collector=lambda _device: SpecDimension( + "size", "尺码", (DimensionValue("L", True),) + ), + ) + adapter.open_goods(GOODS_URL) + device.click(790, 2214) + + with self.assertRaises(PddPurchaseError) as raised: + adapter.apply_resolved_size(snapshot, snapshot.candidates[0]) + + self.assertEqual( + raised.exception.code, "PURCHASE_SPEC_CANDIDATES_CHANGED" + ) + self.assertEqual(selections, []) + adapter.close() + def test_candidate_snapshot_is_not_used_when_target_exists_but_click_fails(self): device = FakeDevice() adapter = U2PddPurchaseAdapter( diff --git a/client/test/test_purchase_task_service.py b/client/test/test_purchase_task_service.py index 20d3975..db9fa48 100644 --- a/client/test/test_purchase_task_service.py +++ b/client/test/test_purchase_task_service.py @@ -3,21 +3,30 @@ import tempfile import unittest from pathlib import Path +from types import SimpleNamespace from src.admin_gateway import ( AdminTask, AndroidDeviceInfo, ClaimCapabilities, ClientInfo, + SpecResolutionMatch, + SpecResolutionReceipt, ) from src.mock_admin_gateway import MockAdminGateway +from src.db import open_database from src.pdd_purchase_adapter import ( PddLivePurchaseAdapter, PddPurchaseAdapter, PddPurchaseError, + PddPurchaseSpecResolutionRequired, + PurchaseSizeObservation, + PurchaseSpecCandidateSnapshot, PurchasePageState, ) +from src.pdd_purchase_reconcile_adapter import PurchaseOrderCandidate from src.purchase_task_service import PurchaseTaskService +from src.purchase_reconcile_service import PurchaseReconcileService from src.task_models import NewClaimedTask, TaskStatus, TaskType from src.task_repository import TaskRepository @@ -135,12 +144,65 @@ class UnavailableGoodsAdapter(RecordingDryRunAdapter): ) +class ResolvingDryRunAdapter(RecordingDryRunAdapter): + """首次选择报告尺码不存在，随后记录审计候选的精确应用。""" + + def __init__(self) -> None: + super().__init__() + self.snapshot = PurchaseSpecCandidateSnapshot.build( + goods_id=self.goods_id, + selected_color="黑色", + target_size="L", + dimension_name="尺码", + observations=( + PurchaseSizeObservation("120斤", True), + PurchaseSizeObservation("130斤", True), + ), + observed_at="2026-08-17T08:00:00Z", + ) + + def select_options(self, options) -> None: + self.calls.append(("select_options", dict(options))) + self.options = { + key: value for key, value in options.items() if key != "size" + } + raise PddPurchaseSpecResolutionRequired(self.snapshot) + + def apply_resolved_size(self, expected_snapshot, candidate) -> None: + self.calls.append( + ("apply_resolved_size", expected_snapshot.candidate_snapshot_hash, + candidate.candidate_id, candidate.raw_text) + ) + self.options["size"] = candidate.raw_text + + +class ResolvingLiveAdapter(ResolvingDryRunAdapter, PddLivePurchaseAdapter): + def __init__(self) -> None: + super().__init__() + self.submit_count = 0 + + def update_shipping_address(self, purchase_number: str) -> None: + self.calls.append(("update_shipping_address", purchase_number)) + + def submit_order_once(self) -> None: + self.submit_count += 1 + self.calls.append(("submit_order_once",)) + + +class ChangedAfterResolutionAdapter(ResolvingDryRunAdapter): + def apply_resolved_size(self, expected_snapshot, candidate) -> None: + raise PddPurchaseError( + "PURCHASE_SPEC_CANDIDATES_CHANGED", + "等待期间候选变化", + step="purchase_resolve_options", + ) + + class PurchaseTaskServiceTest(unittest.TestCase): def setUp(self) -> None: self.temp_dir = tempfile.TemporaryDirectory() - self.repository = TaskRepository( - Path(self.temp_dir.name) / "client.db" - ) + self.db_path = Path(self.temp_dir.name) / "client.db" + self.repository = TaskRepository(self.db_path) self.gateway = MockAdminGateway() self.client = ClientInfo("CLIENT-001", "测试客户端") @@ -233,6 +295,229 @@ class PurchaseTaskServiceTest(unittest.TestCase): self.assertFalse(purchase["order_submitted"]) self.assertFalse(purchase["payment_attempted"]) + def test_matched_spec_resolution_is_persisted_and_existing_flow_continues(self): + self._prepare_task(task_id="PUR-RESOLVE") + adapter = ResolvingDryRunAdapter() + candidate = adapter.snapshot.candidates[0] + self.gateway.set_next_spec_resolution( + SpecResolutionReceipt( + 1, + "psr-001", + "matched", + "rule", + adapter.snapshot.candidate_snapshot_hash, + SpecResolutionMatch( + candidate.candidate_id, + candidate.raw_text, + candidate.options, + ), + 10000, + "唯一重量等价", + "2026-08-17T08:00:01Z", + ) + ) + + outcome = self._service(adapter).execute_selected("PUR-RESOLVE") + + self.assertEqual(outcome.kind, "succeeded") + self.assertTrue(any(call[0] == "apply_resolved_size" for call in adapter.calls)) + detail = self.repository.get_task("PUR-RESOLVE") + purchase = detail.pdd_data["purchase"] + self.assertEqual(purchase["requested"]["options"], OPTIONS) + self.assertEqual(purchase["confirmed"]["options"]["size"], "120斤") + self.assertEqual( + purchase["spec_resolution"]["resolution_id"], "psr-001" + ) + connection = open_database(self.db_path) + try: + row = connection.execute( + "SELECT status, resolution_id, outcome" + " FROM purchase_spec_resolutions" + ).fetchone() + finally: + connection.close() + self.assertEqual(tuple(row), ("resolved", "psr-001", "matched")) + + def test_uncertain_spec_resolution_stops_before_quantity_and_submission(self): + self._prepare_task(task_id="PUR-UNCERTAIN") + adapter = ResolvingDryRunAdapter() + self.gateway.set_next_spec_resolution( + SpecResolutionReceipt( + 1, + "psr-uncertain", + "uncertain", + "ai", + adapter.snapshot.candidate_snapshot_hash, + None, + 7000, + "无法唯一判断", + "2026-08-17T08:00:01Z", + ) + ) + + outcome = self._service(adapter).execute_selected("PUR-UNCERTAIN") + + self.assertEqual(outcome.kind, "task_failed") + self.assertFalse(any(call[0] == "apply_resolved_size" for call in adapter.calls)) + self.assertNotIn(("set_quantity", 2), adapter.calls) + detail = self.repository.get_task("PUR-UNCERTAIN") + self.assertEqual( + detail.last_error_code, "PURCHASE_SPEC_RESOLUTION_UNCERTAIN" + ) + + def test_rejected_and_failed_spec_resolution_use_stable_failure_codes(self): + for outcome, expected_code in ( + ("rejected", "PURCHASE_SPEC_RESOLUTION_REJECTED"), + ("failed", "PURCHASE_SPEC_RESOLUTION_FAILED"), + ): + with self.subTest(outcome=outcome): + task_id = f"PUR-{outcome.upper()}" + self._prepare_task(task_id=task_id) + adapter = ResolvingDryRunAdapter() + self.gateway.set_next_spec_resolution( + SpecResolutionReceipt( + 1, + f"psr-{outcome}", + outcome, + None, + adapter.snapshot.candidate_snapshot_hash, + None, + None, + "没有可安全采用的候选", + "2026-08-17T08:00:01Z", + ) + ) + + result = self._service(adapter).execute_selected(task_id) + + self.assertEqual(result.kind, "task_failed") + self.assertEqual( + self.repository.get_task(task_id).last_error_code, + expected_code, + ) + self.assertFalse( + any(call[0] == "apply_resolved_size" for call in adapter.calls) + ) + + def test_live_resolution_keeps_original_request_and_existing_safety_gates(self): + self._prepare_task(task_id="PUR-RESOLVE-LIVE", execution_mode="live") + adapter = ResolvingLiveAdapter() + candidate = adapter.snapshot.candidates[0] + self.gateway.set_next_spec_resolution( + SpecResolutionReceipt( + 1, + "psr-live", + "matched", + "ai", + adapter.snapshot.candidate_snapshot_hash, + SpecResolutionMatch( + candidate.candidate_id, + candidate.raw_text, + candidate.options, + ), + 9300, + "唯一候选", + "2026-08-17T08:00:01Z", + ) + ) + + outcome = self._service(adapter).execute_selected("PUR-RESOLVE-LIVE") + + self.assertEqual(outcome.kind, "reconcile_pending") + self.assertEqual(adapter.submit_count, 1) + task = self.repository.get_task("PUR-RESOLVE-LIVE") + run = self.repository.latest_task_run("PUR-RESOLVE-LIVE") + self.assertIsNotNone(run.irreversible_action_at) + self.assertEqual( + run.diagnostics_json["final_confirmation"]["options"]["size"], + "120斤", + ) + self.assertEqual( + run.diagnostics_json["final_confirmation"]["spec_resolution"] + ["resolution_id"], + "psr-live", + ) + query = PurchaseReconcileService._query(task, run) + self.assertEqual(query.original_options, OPTIONS) + self.assertEqual(query.options["size"], "120斤") + self.assertEqual(query.spec_resolution["resolution_id"], "psr-live") + result = PurchaseReconcileService._result_data( + SimpleNamespace(_device_address="127.0.0.1:5555"), + task, + query, + PurchaseOrderCandidate( + order_no="ORDER-001", + ordered_at="2026-08-17T08:00:03Z", + ordered_at_raw="2026-08-17 16:00:03", + payment_status="unpaid", + ), + ) + self.assertEqual(result["purchase"]["requested"]["options"], OPTIONS) + self.assertEqual( + result["purchase"]["confirmed"]["options"]["size"], "120斤" + ) + self.assertEqual( + result["purchase"]["spec_resolution"]["resolution_id"], + "psr-live", + ) + + def test_spec_resolution_timeout_keeps_persisted_request_and_does_not_retry(self): + self._prepare_task(task_id="PUR-RESOLVE-TIMEOUT") + adapter = ResolvingDryRunAdapter() + self.gateway.timeout_next_call() + + outcome = self._service(adapter).execute_selected( + "PUR-RESOLVE-TIMEOUT" + ) + + self.assertEqual(outcome.kind, "task_failed") + self.assertEqual(self.gateway.spec_resolution_count, 0) + connection = open_database(self.db_path) + try: + row = connection.execute( + "SELECT status, idempotency_key, request_json" + " FROM purchase_spec_resolutions" + ).fetchone() + finally: + connection.close() + self.assertEqual(row["status"], "pending") + self.assertTrue(row["idempotency_key"].startswith("spec-resolution-v1:")) + self.assertIn('"candidate_snapshot_hash"', row["request_json"]) + + def test_matched_resolution_page_change_failure_keeps_resolution_audit(self): + self._prepare_task(task_id="PUR-RESOLVE-CHANGED") + adapter = ChangedAfterResolutionAdapter() + candidate = adapter.snapshot.candidates[0] + self.gateway.set_next_spec_resolution( + SpecResolutionReceipt( + 1, + "psr-changed", + "matched", + "ai", + adapter.snapshot.candidate_snapshot_hash, + SpecResolutionMatch( + candidate.candidate_id, + candidate.raw_text, + candidate.options, + ), + 9000, + "模型匹配", + "2026-08-17T08:00:01Z", + ) + ) + + outcome = self._service(adapter).execute_selected( + "PUR-RESOLVE-CHANGED" + ) + + self.assertEqual(outcome.kind, "task_failed") + run = self.repository.latest_task_run("PUR-RESOLVE-CHANGED") + self.assertEqual( + run.diagnostics_json["spec_resolution"]["resolution_id"], + "psr-changed", + ) + self.assertIsNone(run.irreversible_action_at) + def test_started_callback_runs_after_database_enters_running(self): self._prepare_task(task_id="PUR-STARTED") observed = [] diff --git a/client/test/test_task_repository.py b/client/test/test_task_repository.py index 1b3ffb8..f63d60a 100644 --- a/client/test/test_task_repository.py +++ b/client/test/test_task_repository.py @@ -114,6 +114,60 @@ class TaskRepositoryTests(unittest.TestCase): self.assertEqual(detail.admin_payload["task_id"], "TASK-001") self.assertEqual(detail.pdd_data["schema_version"], 1) + def test_purchase_spec_resolution_request_is_persisted_before_response(self): + self.repository.add_claimed_task( + self._task("PUR-SPEC", TaskType.PURCHASE) + ) + started = self.repository.start_purchase_run("PUR-SPEC", "USB-001") + request = { + "schema_version": 1, + "candidate_snapshot_hash": "a" * 64, + "candidates": [{"candidate_id": "c1", "raw_text": "L"}], + } + + first = self.repository.prepare_purchase_spec_resolution( + "PUR-SPEC", started.attempt_id, "spec-resolution-v1:key", request + ) + second = self.repository.prepare_purchase_spec_resolution( + "PUR-SPEC", started.attempt_id, "spec-resolution-v1:key", request + ) + + self.assertEqual(first.id, second.id) + self.assertEqual(first.status, "pending") + self.assertEqual(first.request_json, request) + saved = self.repository.save_purchase_spec_resolution( + first.id, + { + "resolution_id": "psr-001", + "outcome": "matched", + "source": "rule", + "confidence_bps": 10000, + "match": { + "candidate_id": "c1", + "raw_text": "L", + "options": {"color": "黑色", "size": "L"}, + }, + "reason": "唯一匹配", + "resolved_at": "2026-08-17T08:00:01Z", + }, + ) + self.assertEqual(saved.status, "resolved") + self.assertEqual(saved.resolution_id, "psr-001") + self.assertEqual(saved.match_options["size"], "L") + with self.assertRaisesRegex(ValueError, "不同响应"): + self.repository.save_purchase_spec_resolution( + first.id, + { + "resolution_id": "psr-other", + "outcome": "rejected", + "source": None, + "confidence_bps": None, + "match": None, + "reason": "内容变化", + "resolved_at": "2026-08-17T08:00:02Z", + }, + ) + def test_live_mode_and_irreversible_marker_are_persisted_once(self) -> None: self.repository.add_claimed_task( self._task( diff --git a/docs/client/03-data-model.md b/docs/client/03-data-model.md index 5fde4b7..8713e6f 100644 --- a/docs/client/03-data-model.md +++ b/docs/client/03-data-model.md @@ -278,8 +278,7 @@ CREATE INDEX idx_task_runs_task [接口契约 §7.1](04-admin-api-contract.md) 定义了一个与 `task_runs.attempt_id` 绑定的一次性 规格解析命令。Admin 的 `purchase_spec_resolutions` 是服务端候选观察和最终决策的权威 -审计；Client 本地仍必须在发送前保存以下最小信息，后续由工单 #257 增加 SQLite migration -和 Repository： +审计；Client 本地仍必须在发送前保存以下最小信息： - `task_id`（本地外键）与 `attempt_id`； - 完整且大小受限的请求 JSON、确定性 `Idempotency-Key` 和请求哈希； @@ -292,8 +291,39 @@ CREATE INDEX idx_task_runs_task `task_runs` 的执行期有效规格。进入过 `irreversible_action_at` 的运行不得新建或重放解析 以继续采购，只能核对订单。 -#254 只冻结接口和 Admin 审计模型，当前 Client SQLite schema 不在本工单改动。未实现 -#257 的旧 Client 继续按“规格不匹配即失败”运行，不调用新接口，也不会出现半持久化状态。 +Client 使用 `purchase_spec_resolutions` 保存这些执行期记录： + +```sql +CREATE TABLE purchase_spec_resolutions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id INTEGER NOT NULL, + attempt_id TEXT NOT NULL, + idempotency_key TEXT NOT NULL UNIQUE, + request_json TEXT NOT NULL, + request_hash TEXT NOT NULL, + candidate_snapshot_hash TEXT NOT NULL, + status TEXT NOT NULL, -- pending / resolved + resolution_id TEXT, + outcome TEXT, + source TEXT, + confidence_bps INTEGER, + match_candidate_id TEXT, + match_raw_text TEXT, + match_options_json TEXT, + reason TEXT, + resolved_at TEXT, + received_at TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + FOREIGN KEY (task_id) REFERENCES pdd_tasks(id) ON DELETE CASCADE, + UNIQUE (task_id, attempt_id, candidate_snapshot_hash) +); +``` + +`pending` 表示请求已落库但尚未得到可验证的业务响应，不能据此继续采购；`resolved` +表示完整响应已经落库，仍须通过候选白名单和真机快照复核。网络调用不放在 SQLite +事务中。旧任务原始规格不覆盖，最终结果分别保存原始请求规格、实际采用规格和 +`resolution_id`。 ## 5. `outbox_events` diff --git a/docs/client/06-quality-security.md b/docs/client/06-quality-security.md index 3ea337e..fdfa9a8 100644 --- a/docs/client/06-quality-security.md +++ b/docs/client/06-quality-security.md @@ -115,6 +115,12 @@ Client 不保存或读取 `purchase.live_*` 手工授权设置。`purchase_mode` 当前前台应用，避免在部分真机上重复等待十秒以上。最终下单点击前仍必须单独 查询前台应用。性能日志分别记录状态读取、颜色选择和尺码选择耗时。 - 规格选择后读取最新控件树并验证选中状态。 +- 运行时规格解析请求必须在网络发送前写入独立 SQLite 记录；只调用一次专用命令， + 不轮询 Admin。只有 `matched` 响应的哈希、候选编号、原文和 options 都是本次候选的 + 逐字副本时才继续。响应返回后重新完整遍历真机候选并复算哈希，页面、颜色或候选 + 有任何变化都在可逆阶段停止。 +- 解析记录不能覆盖任务原始规格。成功结果同时保留原始规格、实际采用规格和 + `resolution_id`；解析失败不自动重新采购，已有不可逆标记时永远不调用规格解析。 - 设置采购数量时先读取当前值；数量相同不聚焦输入框，小差值优先使用加减按钮。 只有输入框兜底路径确认输入法已经显示时才允许按一次返回键；输入法关闭后必须 重新核对规格、数量、价格和唯一提交目标，规格面板丢失时立即停止。