chengma/cmautobuy
2
0
Fork 0
Code Issues 89 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 安全应用采购规格解析结果 (#257)

Browse Source
This commit is contained in:
chengma
2026-08-17 17:43:25 +08:00
parent 484ce4e8ec
commit afc1cdf719
19 changed files with 1669 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/admin_gateway.py
+34 -1
View File
@@ -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:
"""一次性解析当前真机尺码候选；不得用于查询任务状态。"""
client/src/db_schema.py
+43 -1
View File
@@ -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,
}
client/src/http_admin_gateway.py
+145
View File
@@ -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,
client/src/mock_admin_gateway.py
+167
View File
@@ -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,
client/src/pdd_purchase_adapter.py
+13
View File
@@ -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:
"""设置采购数量。"""
client/src/pdd_purchase_reconcile_adapter.py
+2
View File
@@ -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)
client/src/pdd_u2_purchase_adapter.py
+92
View File
@@ -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:
client/src/purchase_reconcile_service.py
+15 -2
View File
@@ -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": {
client/src/purchase_task_service.py
+249 -6
View File
@@ -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": {
client/src/task_models.py
+26
View File
@@ -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:
"""一条非敏感应用设置。"""
client/src/task_repository.py
+190
View File
@@ -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 = (