diff --git a/client/src/pdd_purchase_reconcile_adapter.py b/client/src/pdd_purchase_reconcile_adapter.py index be24f00..4ebcb43 100644 --- a/client/src/pdd_purchase_reconcile_adapter.py +++ b/client/src/pdd_purchase_reconcile_adapter.py @@ -5,7 +5,7 @@ from abc import ABC, abstractmethod from dataclasses import dataclass, field -from typing import Mapping, Optional +from typing import Mapping, Sequence @dataclass(frozen=True) @@ -15,33 +15,42 @@ class PurchaseReconcileQuery: goods_id: str options: Mapping[str, str] quantity: int + unit_price_cent: int + total_price_cent: int irreversible_action_at: str + reconcile_started_at: str @dataclass(frozen=True) -class PurchaseReconcileObservation: - """只读核对看到的结果。""" +class PurchaseOrderCandidate: + """订单列表中一个不含个人信息的只读候选。""" - match_status: str - order_no: Optional[str] = None - ordered_at: Optional[str] = None + order_no: str = "" + goods_id: str = "" + options: Mapping[str, str] = field(default_factory=dict) + quantity: int = 0 + total_price_cent: int = 0 + ordered_at: str = "" + ordered_at_raw: str = "" + payment_status: str = "" + + +@dataclass(frozen=True) +class PurchaseReconcileScan: + """一次只读扫描结果；diagnostics 只能包含计数和错误摘要。""" + + candidates: Sequence[PurchaseOrderCandidate] = field(default_factory=tuple) diagnostics: Mapping[str, object] = field(default_factory=dict) - def __post_init__(self) -> None: - if self.match_status not in { - "matched", "not_found", "ambiguous", "unknown" - }: - raise ValueError("采购核对结果无效") - class PddPurchaseReconcileAdapter(ABC): """已进入不可逆阶段后使用的只读订单核对会话。""" @abstractmethod - def read_order_match( + def read_order_candidates( self, query: PurchaseReconcileQuery - ) -> PurchaseReconcileObservation: - """读取并核对订单候选；不得点击下单或付款。""" + ) -> PurchaseReconcileScan: + """只读扫描订单候选；不得提交、取消或付款。""" @abstractmethod def close(self) -> None: diff --git a/client/src/pdd_u2_purchase_reconcile_adapter.py b/client/src/pdd_u2_purchase_reconcile_adapter.py new file mode 100644 index 0000000..8aa33e3 --- /dev/null +++ b/client/src/pdd_u2_purchase_reconcile_adapter.py @@ -0,0 +1,388 @@ +"""uiautomator2 只读订单核对 Adapter。 + +只允许启动 PDD、切换到“个人中心/我的订单/待付款”、返回和滚动读取。 +代码中没有提交订单、取消订单或付款入口。 +""" + +from __future__ import annotations + +import re +import time +import xml.etree.ElementTree as ET +from datetime import datetime, timedelta, timezone +from decimal import Decimal, InvalidOperation, ROUND_HALF_UP +from typing import Any, Callable, Optional + +from .pdd_device_service import PDD_PACKAGE_NAME, PddDeviceService +from .pdd_purchase_reconcile_adapter import ( + PddPurchaseReconcileAdapter, + PurchaseOrderCandidate, + PurchaseReconcileQuery, + PurchaseReconcileScan, +) + + +Bounds = tuple[int, int, int, int] +_BOUNDS_PATTERN = re.compile(r"^\[(\d+),(\d+)\]\[(\d+),(\d+)\]$") +_ORDER_NO_PATTERN = re.compile( + r"(?:订单编号|订单号)\s*[:：]?\s*([A-Za-z0-9-]{6,64})" +) +_GOODS_ID_PATTERN = re.compile( + r"(?:goods_id=|商品编号\s*[:：]?\s*)(\d{6,})", + re.IGNORECASE, +) +_QUANTITY_PATTERN = re.compile(r"(?:共\s*(\d+)\s*件|[xX×]\s*(\d+))") +_TOTAL_PATTERN = re.compile( + r"(?:需付款|应付款|合计|实付款)\s*[:：]?\s*[¥￥]?\s*(\d+(?:\.\d{1,2})?)" +) +_ORDER_TIME_PATTERN = re.compile( + r"(?:下单时间|创建时间)\s*[:：]?\s*" + r"(20\d{2}[-/.年]\d{1,2}[-/.月]\d{1,2}日?\s+\d{1,2}:\d{2}(?::\d{2})?)" +) +_LOGIN_MARKERS = ("手机号登录", "登录后继续", "验证码登录", "账号登录") +_CAPTCHA_MARKERS = ("请完成验证", "拖动滑块", "安全验证", "点击图中") +_RISK_MARKERS = ("操作频繁", "异常请求", "风险提示", "账号异常") +_PAYMENT_ACTION_MARKERS = ("立即支付", "确认支付", "输入支付密码") +_UNPAID_MARKERS = ("待付款", "待支付") +_NON_UNPAID_MARKERS = ("已付款", "交易成功", "交易完成", "已取消", "退款") +_SAFE_NAVIGATION_LABELS = ("个人中心", "我的订单", "待付款") + + +class PddPurchaseReconcileError(RuntimeError): + """只读核单无法安全继续。""" + + def __init__(self, code: str, message: str) -> None: + super().__init__(message) + self.code = code + + +def _parse_xml(xml_data: str | bytes) -> ET.Element: + try: + return ET.fromstring(xml_data) + except (ET.ParseError, TypeError) as exc: + raise PddPurchaseReconcileError( + "RECONCILE_XML_INVALID", "订单页无障碍控件树无效" + ) from exc + + +def _label(node: ET.Element) -> str: + return " ".join( + value.strip() + for value in (node.get("text", ""), node.get("content-desc", "")) + if value.strip() + ) + + +def _subtree_text(node: ET.Element) -> str: + return " ".join( + label for item in node.iter("node") if (label := _label(item)) + ) + + +def _parse_bounds(value: str) -> Optional[Bounds]: + match = _BOUNDS_PATTERN.fullmatch(str(value or "").strip()) + if match is None: + return None + left, top, right, bottom = map(int, match.groups()) + if right <= left or bottom <= top: + return None + return left, top, right, bottom + + +def _parse_money_cent(text: str) -> int: + match = _TOTAL_PATTERN.search(text) + if match is None: + return 0 + try: + value = Decimal(match.group(1)).quantize( + Decimal("0.01"), rounding=ROUND_HALF_UP + ) + except InvalidOperation: + return 0 + return int(value * 100) + + +def _parse_quantity(text: str) -> int: + match = _QUANTITY_PATTERN.search(text) + if match is None: + return 0 + value = match.group(1) or match.group(2) or "0" + return int(value) + + +def _parse_ordered_at(text: str) -> tuple[str, str]: + match = _ORDER_TIME_PATTERN.search(text) + if match is None: + return "", "" + raw = match.group(1) + normalized = ( + raw.replace("年", "-") + .replace("月", "-") + .replace("日", "") + .replace("/", "-") + .replace(".", "-") + ) + formats = ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M") + for time_format in formats: + try: + local = datetime.strptime(normalized, time_format).replace( + tzinfo=timezone(timedelta(hours=8)) + ) + ordered_at = local.astimezone(timezone.utc).isoformat( + timespec="seconds" + ).replace("+00:00", "Z") + return ordered_at, raw + except ValueError: + continue + return "", raw + + +def parse_order_candidates( + xml_data: str | bytes, query: PurchaseReconcileQuery +) -> tuple[PurchaseOrderCandidate, ...]: + """从脱敏控件树提取候选，只保留核单必需字段。""" + + root = _parse_xml(xml_data) + parents = { + child: parent + for parent in root.iter("node") + for child in parent + if child.tag == "node" + } + parsed: dict[str, PurchaseOrderCandidate] = {} + for node in root.iter("node"): + node_label = _label(node) + order_match = _ORDER_NO_PATTERN.search(node_label) + if order_match is None: + continue + order_no = order_match.group(1) + current = node + card_text = node_label + for _ in range(8): + text = _subtree_text(current) + if any(marker in text for marker in _UNPAID_MARKERS): + card_text = text + if ( + _TOTAL_PATTERN.search(text) + and _QUANTITY_PATTERN.search(text) + and _ORDER_TIME_PATTERN.search(text) + ): + break + parent = parents.get(current) + if parent is None: + break + current = parent + + goods_match = _GOODS_ID_PATTERN.search(card_text) + ordered_at, ordered_at_raw = _parse_ordered_at(card_text) + options = { + key: value + for key, value in query.options.items() + if str(value).strip() and str(value).strip() in card_text + } + payment_status = ( + "unpaid" + if any(marker in card_text for marker in _UNPAID_MARKERS) + and not any(marker in card_text for marker in _NON_UNPAID_MARKERS) + else "other" + ) + candidate = PurchaseOrderCandidate( + order_no=order_no, + goods_id=goods_match.group(1) if goods_match else "", + options=options, + quantity=_parse_quantity(card_text), + total_price_cent=_parse_money_cent(card_text), + ordered_at=ordered_at, + ordered_at_raw=ordered_at_raw, + payment_status=payment_status, + ) + previous = parsed.get(order_no) + if previous is None or previous == candidate: + parsed[order_no] = candidate + return tuple(parsed.values()) + + +class U2PddPurchaseReconcileAdapter(PddPurchaseReconcileAdapter): + """只导航和滚动读取待付款订单，不包含任何订单写操作。""" + + def __init__( + self, + device_address: str, + *, + device_service: Optional[PddDeviceService] = None, + cancelled: Callable[[], bool] = lambda: False, + settle_seconds: float = 0.4, + max_pages: int = 5, + ) -> None: + self._device_address = str(device_address or "").strip() + self._device_service = device_service or _RECONCILE_DEVICE_SERVICE + self._cancelled = cancelled + self._settle_seconds = max(0.0, float(settle_seconds)) + self._max_pages = max(1, int(max_pages)) + self._session = None + self._device = None + + def read_order_candidates( + self, query: PurchaseReconcileQuery + ) -> PurchaseReconcileScan: + self._check_cancelled() + device = self._connect() + self._open_unpaid_orders(device) + found: dict[str, PurchaseOrderCandidate] = {} + previous_signature = "" + pages_scanned = 0 + for _ in range(self._max_pages): + self._check_cancelled() + xml_data = device.dump_hierarchy(compressed=False) + self._raise_for_special_page(_parse_xml(xml_data)) + pages_scanned += 1 + for candidate in parse_order_candidates(xml_data, query): + found.setdefault(candidate.order_no, candidate) + signature = "|".join(sorted(found)) + f":{len(str(xml_data))}" + if signature == previous_signature: + break + previous_signature = signature + width, height = device.window_size() + device.swipe( + width // 2, + int(height * 0.78), + width // 2, + int(height * 0.32), + 0.35, + ) + self._settle() + return PurchaseReconcileScan( + candidates=tuple(found.values()), + diagnostics={ + "pages_scanned": pages_scanned, + "candidate_count": len(found), + }, + ) + + def close(self) -> None: + session = self._session + self._session = None + self._device = None + if session is not None: + session.__exit__(None, None, None) + + def _connect(self) -> Any: + if self._device is None: + self._session = self._device_service.connect(self._device_address) + self._device = self._session.__enter__() + return self._device + + def _open_unpaid_orders(self, device: Any) -> None: + current = device.app_current() + if str(current.get("package") or "") != PDD_PACKAGE_NAME: + device.app_start(PDD_PACKAGE_NAME) + self._settle() + + opened_order_area = False + for _ in range(6): + self._check_cancelled() + current = device.app_current() + if str(current.get("package") or "") != PDD_PACKAGE_NAME: + raise PddPurchaseReconcileError( + "RECONCILE_WRONG_APP", "核单时 PDD 不在前台" + ) + root = _parse_xml(device.dump_hierarchy(compressed=False)) + self._raise_for_special_page(root) + combined = _subtree_text(root) + if opened_order_area and ( + _ORDER_NO_PATTERN.search(combined) + or "暂无订单" in combined + or "暂无相关订单" in combined + ): + return + if any(marker in combined for marker in _PAYMENT_ACTION_MARKERS): + device.press("back") + self._settle() + continue + if "我的订单" in combined: + if self._click_unique_label(device, root, "待付款"): + opened_order_area = True + self._settle() + continue + if self._click_unique_label(device, root, "我的订单"): + opened_order_area = True + self._settle() + continue + if self._click_unique_label(device, root, "个人中心"): + self._settle() + continue + raise PddPurchaseReconcileError( + "RECONCILE_ORDER_ENTRY_NOT_FOUND", + "没有找到唯一的“个人中心/我的订单/待付款”只读入口", + ) + raise PddPurchaseReconcileError( + "RECONCILE_ORDER_PAGE_TIMEOUT", "打开待付款订单列表超时" + ) + + @staticmethod + def _click_unique_label( + device: Any, root: ET.Element, target: str + ) -> bool: + if target not in _SAFE_NAVIGATION_LABELS: + raise ValueError("只允许使用白名单只读导航入口") + targets = [] + for node in root.iter("node"): + labels = { + node.get("text", "").strip(), + node.get("content-desc", "").strip(), + } + if target not in labels: + continue + if ( + node.get("visible-to-user") != "true" + or node.get("enabled") != "true" + ): + continue + bounds = _parse_bounds(node.get("bounds", "")) + if bounds is not None: + targets.append(bounds) + unique = list(dict.fromkeys(targets)) + if len(unique) != 1: + return False + left, top, right, bottom = unique[0] + device.click((left + right) // 2, (top + bottom) // 2) + return True + + @staticmethod + def _raise_for_special_page(root: ET.Element) -> None: + combined = _subtree_text(root) + markers = ( + ("RECONCILE_CAPTCHA", _CAPTCHA_MARKERS), + ("RECONCILE_LOGIN_REQUIRED", _LOGIN_MARKERS), + ("RECONCILE_RISK_CONTROL", _RISK_MARKERS), + ) + for code, values in markers: + if any(value in combined for value in values): + raise PddPurchaseReconcileError( + code, "PDD 核单遇到登录、验证或风控页面，请人工处理" + ) + + def _check_cancelled(self) -> None: + if self._cancelled(): + raise PddPurchaseReconcileError( + "RECONCILE_CANCELLED", "用户已停止只读订单核对" + ) + + def _settle(self) -> None: + if self._settle_seconds: + time.sleep(self._settle_seconds) + + +_RECONCILE_DEVICE_SERVICE = PddDeviceService() + + +def create_u2_purchase_reconcile_adapter( + device_address: str, cancelled: Callable[[], bool] +) -> PddPurchaseReconcileAdapter: + """为正式 Client 创建只读订单核对会话。""" + + return U2PddPurchaseReconcileAdapter( + device_address, + device_service=_RECONCILE_DEVICE_SERVICE, + cancelled=cancelled, + ) diff --git a/client/src/purchase_reconcile_service.py b/client/src/purchase_reconcile_service.py index 5189526..0982b41 100644 --- a/client/src/purchase_reconcile_service.py +++ b/client/src/purchase_reconcile_service.py @@ -1,14 +1,16 @@ -"""不可逆阶段中断后的只读采购结果核对。""" +"""不可逆阶段后的只读采购订单核对与结果入队。""" from dataclasses import dataclass -from typing import Callable, Mapping +from datetime import datetime, timedelta, timezone +from typing import Callable, Mapping, Optional from .pdd_purchase_reconcile_adapter import ( PddPurchaseReconcileAdapter, - PurchaseReconcileObservation, + PurchaseOrderCandidate, PurchaseReconcileQuery, + PurchaseReconcileScan, ) -from .task_models import TaskDetail +from .task_models import TaskDetail, TaskRunRecord from .task_repository import TaskRepository @@ -17,6 +19,22 @@ PurchaseReconcileFactory = Callable[ ] +def _utc_now_iso() -> str: + return datetime.now(timezone.utc).isoformat(timespec="seconds").replace( + "+00:00", "Z" + ) + + +def _parse_iso_time(value: str) -> datetime: + checked = str(value or "").strip() + if checked.endswith("Z"): + checked = checked[:-1] + "+00:00" + parsed = datetime.fromisoformat(checked) + if parsed.tzinfo is None: + raise ValueError("订单时间缺少时区") + return parsed.astimezone(timezone.utc) + + @dataclass(frozen=True) class PurchaseReconcileOutcome: """只读核对的简短结果。""" @@ -27,7 +45,11 @@ class PurchaseReconcileOutcome: class PurchaseReconcileService: - """只读核对一条已进入不可逆阶段的采购运行。""" + """严格匹配未付款订单；唯一匹配才生成采购成功 Outbox。""" + + _DIAGNOSTIC_KEYS = frozenset( + {"pages_scanned", "candidate_count", "error_code", "error_message"} + ) def __init__( self, @@ -45,7 +67,7 @@ class PurchaseReconcileService: def execute_selected( self, remote_task_id: str ) -> PurchaseReconcileOutcome: - """执行一次只读核对，任何结果都交给人工最终确认。""" + """只读核对一次；唯一未付款订单入 Outbox，其余转人工。""" if not self._device_address: raise ValueError("请先在设置页选择并保存 Android 设备") @@ -60,17 +82,82 @@ class PurchaseReconcileService: if run.irreversible_action_at is None: raise ValueError("未进入不可逆阶段，不应启动订单核对") - query = self._query(task, run.irreversible_action_at) - adapter = None + try: + query = self._query(task, run) + except Exception as exc: + return self._save_manual_review( + task, + run, + "unknown", + {"error_code": "RECONCILE_QUERY_INVALID", "error_message": str(exc)}, + ) + + scan, close_error = self._read_candidates(query) + candidates = tuple(scan.candidates) + matching = tuple( + candidate + for candidate in candidates + if self._candidate_matches(candidate, query) + ) + diagnostics = { + key: value + for key, value in scan.diagnostics.items() + if key in self._DIAGNOSTIC_KEYS + } + diagnostics.update( + { + "candidate_count": len(candidates), + "matching_candidate_count": len(matching), + "mode": "reconcile_only", + } + ) + if close_error: + diagnostics["close_error"] = close_error[:300] + + if len(matching) == 1: + candidate = matching[0] + result = self._result_data(task, query, candidate) + event = self._repository.save_matched_purchase_reconciliation( + task.remote_task_id, + run.attempt_id, + result, + diagnostics, + ) + return PurchaseReconcileOutcome( + "result_pending", + f"任务 {task.remote_task_id} 已核对到唯一未付款订单，等待提交 Admin", + task.remote_task_id, + ) + + if len(matching) > 1: + return self._save_manual_review( + task, run, "ambiguous", diagnostics + ) + match_status = "not_found" if not candidates else "unknown" + return self._save_manual_review(task, run, match_status, diagnostics) + + def _read_candidates( + self, query: PurchaseReconcileQuery + ) -> tuple[PurchaseReconcileScan, str]: + adapter: Optional[PddPurchaseReconcileAdapter] = None close_error = "" + scan = PurchaseReconcileScan() try: adapter = self._factory(self._device_address, self._cancelled) - observation = adapter.read_order_match(query) - if not isinstance(observation, PurchaseReconcileObservation): + scan = adapter.read_order_candidates(query) + if not isinstance(scan, PurchaseReconcileScan): raise TypeError("采购核对 Adapter 返回值无效") + if any( + not isinstance(candidate, PurchaseOrderCandidate) + for candidate in scan.candidates + ): + raise TypeError("采购核对候选数据无效") except Exception as exc: - observation = PurchaseReconcileObservation( - "unknown", diagnostics={"error": str(exc)} + scan = PurchaseReconcileScan( + diagnostics={ + "error_code": "RECONCILE_READ_FAILED", + "error_message": str(exc)[:300], + } ) finally: if adapter is not None: @@ -78,43 +165,63 @@ class PurchaseReconcileService: adapter.close() except Exception as exc: close_error = str(exc) + return scan, close_error - diagnostics = dict(observation.diagnostics) - diagnostics.update( - { - "order_no": observation.order_no, - "ordered_at": observation.ordered_at, - "mode": "reconcile_only", - } - ) - if close_error: - diagnostics["close_error"] = close_error + def _save_manual_review( + self, + task: TaskDetail, + run: TaskRunRecord, + match_status: str, + diagnostics: Mapping[str, object], + ) -> PurchaseReconcileOutcome: self._repository.save_purchase_reconciliation( - remote_task_id, + task.remote_task_id, run.attempt_id, - observation.match_status, - diagnostics, + match_status, + dict(diagnostics), ) - if observation.match_status == "matched": - message = ( - f"任务 {remote_task_id} 仅核对到唯一候选订单；" - "请人工确认，程序不会重新下单" - ) - else: - message = ( - f"任务 {remote_task_id} 核对结果不确定；" - "需人工处理，程序不会重新下单" - ) + messages = { + "not_found": "未找到符合时间范围的订单", + "ambiguous": "找到多个完全匹配的未付款订单", + "unknown": "订单字段不完整、不一致或读取失败", + } return PurchaseReconcileOutcome( - "manual_review", message, remote_task_id + "manual_review", + f"任务 {task.remote_task_id} {messages[match_status]}；需人工处理，绝不重新下单", + task.remote_task_id, ) + @staticmethod + def _candidate_matches( + candidate: PurchaseOrderCandidate, + query: PurchaseReconcileQuery, + ) -> bool: + if ( + not candidate.order_no.strip() + or candidate.goods_id.strip() != query.goods_id + or dict(candidate.options) != dict(query.options) + or candidate.quantity != query.quantity + or candidate.total_price_cent != query.total_price_cent + or candidate.payment_status != "unpaid" + ): + return False + try: + ordered_at = _parse_iso_time(candidate.ordered_at) + lower = _parse_iso_time(query.irreversible_action_at) - timedelta( + minutes=5 + ) + upper = _parse_iso_time(query.reconcile_started_at) + timedelta( + minutes=5 + ) + except (TypeError, ValueError): + return False + return lower <= ordered_at <= upper + @staticmethod def _query( - task: TaskDetail, irreversible_action_at: str + task: TaskDetail, run: TaskRunRecord ) -> PurchaseReconcileQuery: - payload_root = task.admin_payload - payload = payload_root.get("payload") + payload = task.admin_payload.get("payload") if not isinstance(payload, Mapping): raise ValueError("采购任务缺少 payload") options = payload.get("options") @@ -122,6 +229,11 @@ class PurchaseReconcileService: raise ValueError("采购任务缺少 options") goods_id = str(payload.get("goods_id") or "").strip() quantity = payload.get("quantity") + snapshot = run.diagnostics_json.get("final_confirmation") + if not isinstance(snapshot, Mapping): + raise ValueError("采购运行缺少提交前确认快照") + total_price_cent = snapshot.get("total_price_cent") + unit_price_cent = snapshot.get("unit_price_cent") if not goods_id: raise ValueError("采购任务缺少 goods_id") if ( @@ -130,9 +242,63 @@ class PurchaseReconcileService: or quantity <= 0 ): raise ValueError("采购任务缺少有效 quantity") + if ( + isinstance(total_price_cent, bool) + or not isinstance(total_price_cent, int) + or total_price_cent <= 0 + ): + raise ValueError("采购运行缺少有效确认总价") + if ( + isinstance(unit_price_cent, bool) + or not isinstance(unit_price_cent, int) + or unit_price_cent <= 0 + ): + raise ValueError("采购运行缺少有效确认单价") return PurchaseReconcileQuery( goods_id=goods_id, - options={str(k): str(v) for k, v in options.items()}, + options={str(key): str(value) for key, value in options.items()}, quantity=quantity, - irreversible_action_at=irreversible_action_at, + unit_price_cent=unit_price_cent, + total_price_cent=total_price_cent, + irreversible_action_at=str(run.irreversible_action_at), + reconcile_started_at=_utc_now_iso(), ) + + def _result_data( + self, + task: TaskDetail, + query: PurchaseReconcileQuery, + candidate: PurchaseOrderCandidate, + ) -> dict[str, object]: + return { + "schema_version": 1, + "goods_id": query.goods_id, + "goods_url": task.goods_url, + "purchase": { + "mode": "live", + "requested": { + "options": dict(query.options), + "quantity": query.quantity, + "max_price_cent": task.price_cent, + }, + "confirmed": { + "options": dict(candidate.options), + "quantity": candidate.quantity, + "unit_price_cent": query.unit_price_cent, + "total_price_cent": candidate.total_price_cent, + }, + "confirmation_reached": True, + "order_submitted": True, + "payment_attempted": False, + "payment_status": "unpaid", + "order_no": candidate.order_no, + "ordered_at": candidate.ordered_at, + "ordered_at_raw": candidate.ordered_at_raw, + "match_status": "matched", + }, + "captured_at": _utc_now_iso(), + "source": { + "device_address": self._device_address, + "mode": "reconcile_only", + }, + } diff --git a/client/src/purchase_task_service.py b/client/src/purchase_task_service.py index 2cd72f9..0bd91da 100644 --- a/client/src/purchase_task_service.py +++ b/client/src/purchase_task_service.py @@ -225,7 +225,14 @@ class PurchaseTaskService: ) irreversible_at = self._repository.mark_purchase_irreversible( - remote_task_id, attempt_id + remote_task_id, + attempt_id, + { + "options": dict(state.selected_options), + "quantity": state.quantity, + "unit_price_cent": state.price_cent, + "total_price_cent": state.price_cent * state.quantity, + }, ) submitted_at = None message = "订单提交点击已执行，结果待只读核对；绝不重新下单" diff --git a/client/src/task_detail_view.py b/client/src/task_detail_view.py index c8a0cc4..cf89ef9 100644 --- a/client/src/task_detail_view.py +++ b/client/src/task_detail_view.py @@ -73,9 +73,10 @@ CURRENT_STEP_TEXT = { "purchase_submit_once": "已单击一次提交订单", "purchase_dry_run_stopped": "采购演练已在提交前停止", "purchase_recovery_ready": "上次演练中断，已安全等待恢复", - "reconcile_purchase": "只允许核对订单", - "reconcile_completed": "只读核对完成，等待人工确认", - "reconcile_manual_review": "核对结果不确定，需人工处理", + "reconcile_purchase": "订单已提交，只允许核对未付款订单", + "purchase_order_matched_pending_report": "已提交待付款，等待向 Admin 上报", + "reconcile_completed": "已核对到订单，等待人工确认", + "reconcile_manual_review": "订单结果不确定或存在多个候选，需人工处理", } diff --git a/client/src/task_repository.py b/client/src/task_repository.py index 3c1c1bd..64fc10a 100644 --- a/client/src/task_repository.py +++ b/client/src/task_repository.py @@ -602,12 +602,19 @@ class TaskRepository: connection.close() def mark_purchase_irreversible( - self, remote_task_id: str, attempt_id: str + self, + remote_task_id: str, + attempt_id: str, + final_confirmation: Optional[Dict[str, object]] = None, ) -> str: - """事务写入不可逆时间；成功返回后才允许点击提交订单。""" + """同一事务保存最终确认快照和不可逆时间，再允许提交订单。""" now = utc_now_iso() step = "purchase_irreversible_step_entered" + diagnostics_json = json.dumps( + {"final_confirmation": dict(final_confirmation or {})}, + ensure_ascii=False, + ) connection = open_database(self._db_path) try: with connection: @@ -626,11 +633,18 @@ class TaskRepository: raise ValueError("采购任务当前不在执行中") cursor = connection.execute( "UPDATE task_runs SET irreversible_action_at = ?," - " current_step = ?, updated_at = ?" + " current_step = ?, diagnostics_json = ?, updated_at = ?" " WHERE task_id = ? AND attempt_id = ?" " AND run_status = 'running'" " AND irreversible_action_at IS NULL", - (now, step, now, task["id"], attempt_id), + ( + now, + step, + diagnostics_json, + now, + task["id"], + attempt_id, + ), ) if cursor.rowcount != 1: raise ValueError("不可逆标记写入失败或已经存在") @@ -719,9 +733,16 @@ class TaskRepository: "matched": "只读核对发现唯一候选订单，请人工确认", "not_found": "只读核对未找到订单，不得重新下单", "ambiguous": "只读核对发现多个候选订单，请人工确认", - "unknown": "无法确定采购结果，不得重新下单", + "unknown": "订单字段不完整、不一致或读取失败，不得重新下单", + } + error_codes = { + "matched": "PURCHASE_RECONCILED", + "not_found": "ORDER_NOT_FOUND", + "ambiguous": "AMBIGUOUS_ORDER_MATCH", + "unknown": "ORDER_MATCH_UNCERTAIN", } message = messages[match_status] + error_code = error_codes[match_status] saved_diagnostics = dict(diagnostics) saved_diagnostics["match_status"] = match_status connection = open_database(self._db_path) @@ -742,7 +763,8 @@ class TaskRepository: ): raise ValueError("采购任务当前不在待核对状态") run = connection.execute( - "SELECT irreversible_action_at, run_status, current_step" + "SELECT irreversible_action_at, run_status, current_step," + " diagnostics_json" " FROM task_runs" " WHERE task_id = ? AND attempt_id = ?", (task["id"], attempt_id), @@ -754,15 +776,22 @@ class TaskRepository: or run["current_step"] != "reconcile_purchase" ): raise ValueError("采购执行记录已经核对或状态已变更") + run_diagnostics = ( + self._load_json_object(run["diagnostics_json"]) + if run["diagnostics_json"] + else {} + ) + run_diagnostics["reconciliation"] = saved_diagnostics run_cursor = connection.execute( "UPDATE task_runs SET run_status = 'manual_review'," - " current_step = ?, error_code = 'PURCHASE_RECONCILED'," + " current_step = ?, error_code = ?," " error_message = ?, diagnostics_json = ?, updated_at = ?" " WHERE task_id = ? AND attempt_id = ?", ( step, + error_code, message, - json.dumps(saved_diagnostics, ensure_ascii=False), + json.dumps(run_diagnostics, ensure_ascii=False), now, task["id"], attempt_id, @@ -772,17 +801,150 @@ class TaskRepository: raise ValueError("采购执行记录核对保存失败") task_cursor = connection.execute( "UPDATE pdd_tasks SET status = 'manual_review'," - " current_step = ?, last_error_code = 'PURCHASE_RECONCILED'," + " current_step = ?, last_error_code = ?," " last_error_message = ?, updated_at = ? WHERE id = ?" " AND status = 'manual_review'" " AND current_step = 'reconcile_purchase'", - (step, message, now, task["id"]), + (step, error_code, message, now, task["id"]), ) if task_cursor.rowcount != 1: raise ValueError("采购任务状态已变更，核对结果未保存") finally: connection.close() + def save_matched_purchase_reconciliation( + self, + remote_task_id: str, + attempt_id: str, + pdd_data: Dict[str, object], + diagnostics: Dict[str, object], + ) -> OutboxEventRecord: + """唯一未付款订单核对成功后，原子保存结果并创建幂等 Outbox。""" + + purchase = pdd_data.get("purchase") + if not isinstance(purchase, dict) or ( + purchase.get("mode") != "live" + or purchase.get("match_status") != "matched" + or purchase.get("order_submitted") is not True + or purchase.get("payment_attempted") is not False + or purchase.get("payment_status") != "unpaid" + or not str(purchase.get("order_no") or "").strip() + ): + raise ValueError("真实采购核对结果结构无效") + + now = utc_now_iso() + idempotency_key = f"{remote_task_id}:{attempt_id}:result-v1" + connection = open_database(self._db_path) + try: + with connection: + existing = connection.execute( + "SELECT id FROM outbox_events WHERE idempotency_key = ?", + (idempotency_key,), + ).fetchone() + if existing is not None: + event_id = int(existing["id"]) + else: + task = connection.execute( + "SELECT id, version, task_type, status, current_step" + " FROM pdd_tasks WHERE remote_task_id = ?", + (remote_task_id,), + ).fetchone() + if task is None: + raise ValueError(f"任务 {remote_task_id} 不存在") + if task["task_type"] != TaskType.PURCHASE.value: + raise ValueError("当前任务不是采购任务") + if ( + task["status"] != TaskStatus.MANUAL_REVIEW.value + or task["current_step"] != "reconcile_purchase" + ): + raise ValueError("采购任务当前不在待核对状态") + run = connection.execute( + "SELECT run_status, current_step, diagnostics_json," + " irreversible_action_at FROM task_runs" + " WHERE task_id = ? AND attempt_id = ?", + (task["id"], attempt_id), + ).fetchone() + if ( + run is None + or run["irreversible_action_at"] is None + or run["run_status"] != RunStatus.MANUAL_REVIEW.value + or run["current_step"] != "reconcile_purchase" + ): + raise ValueError("不可逆采购执行记录不在待核对状态") + + run_diagnostics = ( + self._load_json_object(run["diagnostics_json"]) + if run["diagnostics_json"] + else {} + ) + run_diagnostics["reconciliation"] = dict(diagnostics) + result_json = json.dumps(pdd_data, ensure_ascii=False) + payload = { + "task_version": task["version"], + "attempt_id": attempt_id, + "result_type": "purchase", + "completed_at": now, + "pdd_data": pdd_data, + } + task_cursor = connection.execute( + "UPDATE pdd_tasks SET status = 'result_pending'," + " current_step = 'purchase_order_matched_pending_report'," + " pdd_data = ?," + " price_cent = ?, last_error_code = NULL," + " last_error_message = NULL, finished_at = ?, updated_at = ?" + " WHERE id = ? AND status = 'manual_review'" + " AND current_step = 'reconcile_purchase'", + ( + result_json, + self._purchase_result_price(pdd_data), + now, + now, + task["id"], + ), + ) + if task_cursor.rowcount != 1: + raise ValueError("采购任务状态已变化，未保存核对结果") + run_cursor = connection.execute( + "UPDATE task_runs SET run_status = 'succeeded'," + " current_step = 'purchase_order_matched_pending_report'," + " result_data = ?," + " diagnostics_json = ?, error_code = NULL," + " error_message = NULL, finished_at = ?, updated_at = ?" + " WHERE task_id = ? AND attempt_id = ?" + " AND run_status = 'manual_review'" + " AND current_step = 'reconcile_purchase'" + " AND irreversible_action_at IS NOT NULL", + ( + result_json, + json.dumps(run_diagnostics, ensure_ascii=False), + now, + now, + task["id"], + attempt_id, + ), + ) + if run_cursor.rowcount != 1: + raise ValueError("采购执行记录状态已变化，未保存核对结果") + event_cursor = connection.execute( + "INSERT INTO outbox_events (task_id, event_type," + " idempotency_key, payload_json, status, created_at," + " updated_at)" + " VALUES (?, 'purchase_result', ?, ?, 'pending', ?, ?)", + ( + task["id"], + idempotency_key, + json.dumps(payload, ensure_ascii=False), + now, + now, + ), + ) + event_id = int(event_cursor.lastrowid) + event = self.get_outbox_event(event_id) + assert event is not None + return event + finally: + connection.close() + def save_purchase_result( self, remote_task_id: str, diff --git a/client/src/ui_main.py b/client/src/ui_main.py index 5496758..19f0422 100644 --- a/client/src/ui_main.py +++ b/client/src/ui_main.py @@ -33,6 +33,9 @@ from .pdd_u2_purchase_adapter import ( create_u2_live_purchase_adapter, create_u2_purchase_adapter, ) +from .pdd_u2_purchase_reconcile_adapter import ( + create_u2_purchase_reconcile_adapter, +) from .settings_ui import SettingsPage from .task_repository import TaskRepository from .update_service import mark_current_version_healthy @@ -62,6 +65,7 @@ class MainWindow(FluentWindow): self, purchase_adapter_factory=create_u2_purchase_adapter, live_purchase_adapter_factory=create_u2_live_purchase_adapter, + purchase_reconcile_factory=create_u2_purchase_reconcile_adapter, ) self.pddTaskPage.openSettingsRequested.connect( diff --git a/client/test/test_pdd_u2_purchase_reconcile_adapter.py b/client/test/test_pdd_u2_purchase_reconcile_adapter.py new file mode 100644 index 0000000..27224ba --- /dev/null +++ b/client/test/test_pdd_u2_purchase_reconcile_adapter.py @@ -0,0 +1,75 @@ +"""只读订单核对 XML 解析测试；不连接真实手机。""" + +import unittest + +from src.pdd_purchase_reconcile_adapter import PurchaseReconcileQuery +from src.pdd_u2_purchase_reconcile_adapter import parse_order_candidates + + +def query() -> PurchaseReconcileQuery: + return PurchaseReconcileQuery( + goods_id="737116531267", + options={"color": "黑色", "size": "L"}, + quantity=2, + unit_price_cent=4200, + total_price_cent=8400, + irreversible_action_at="2026-08-10T08:00:00Z", + reconcile_started_at="2026-08-10T08:05:00Z", + ) + + +def xml_card(text: str) -> str: + return ( + '' + '' + f'' + "" + ) + + +class PddU2PurchaseReconcileAdapterTest(unittest.TestCase): + def test_parser_extracts_only_required_unpaid_order_fields(self): + candidates = parse_order_candidates( + xml_card( + "待付款 订单编号：ORDER-20260810 商品编号：737116531267 " + "黑色 L 共2件 合计 ¥84.00 下单时间：2026-08-10 16:03:00" + ), + query(), + ) + + self.assertEqual(len(candidates), 1) + candidate = candidates[0] + self.assertEqual(candidate.order_no, "ORDER-20260810") + self.assertEqual(candidate.goods_id, "737116531267") + self.assertEqual(candidate.options, {"color": "黑色", "size": "L"}) + self.assertEqual(candidate.quantity, 2) + self.assertEqual(candidate.total_price_cent, 8400) + self.assertEqual(candidate.ordered_at, "2026-08-10T08:03:00Z") + self.assertEqual(candidate.payment_status, "unpaid") + self.assertFalse(hasattr(candidate, "recipient")) + self.assertFalse(hasattr(candidate, "address")) + + def test_paid_or_cancelled_text_never_becomes_unpaid(self): + candidates = parse_order_candidates( + xml_card( + "待付款 已付款 订单编号：ORDER-PAID 商品编号：737116531267 " + "黑色 L 共2件 合计 ¥84.00 下单时间：2026-08-10 16:03:00" + ), + query(), + ) + + self.assertEqual(candidates[0].payment_status, "other") + + def test_missing_goods_id_and_time_are_kept_incomplete_for_manual_review(self): + candidates = parse_order_candidates( + xml_card("待付款 订单编号：ORDER-INCOMPLETE 黑色 L 共2件 合计 ¥84.00"), + query(), + ) + + self.assertEqual(candidates[0].goods_id, "") + self.assertEqual(candidates[0].ordered_at, "") + + +if __name__ == "__main__": + unittest.main() diff --git a/client/test/test_purchase_recovery.py b/client/test/test_purchase_recovery.py index c85e0ee..7b8bb41 100644 --- a/client/test/test_purchase_recovery.py +++ b/client/test/test_purchase_recovery.py @@ -2,10 +2,10 @@ import tempfile import unittest +from dataclasses import replace from pathlib import Path from src.admin_gateway import AdminTask, ClaimCapabilities, ClientInfo -from src.db import open_database from src.mock_admin_gateway import MockAdminGateway from src.pdd_purchase_adapter import ( PddLivePurchaseAdapter, @@ -15,7 +15,8 @@ from src.pdd_purchase_adapter import ( ) from src.pdd_purchase_reconcile_adapter import ( PddPurchaseReconcileAdapter, - PurchaseReconcileObservation, + PurchaseOrderCandidate, + PurchaseReconcileScan, ) from src.purchase_task_service import PurchaseTaskService from src.task_dispatcher import TaskDispatcher, admin_task_to_new_claimed_task @@ -26,7 +27,9 @@ from src.task_repository import TaskRepository OPTIONS = {"color": "黑色", "size": "L"} -def purchase_admin_task(task_id: str = "PUR-RECOVER") -> AdminTask: +def purchase_admin_task( + task_id: str = "PUR-RECOVER", execution_mode: str = "dry_run" +) -> AdminTask: return AdminTask( task_id, TaskType.PURCHASE, @@ -39,6 +42,7 @@ def purchase_admin_task(task_id: str = "PUR-RECOVER") -> AdminTask: "quantity": 2, "max_price_cent": 5000, }, + execution_mode=execution_mode, ) @@ -95,13 +99,37 @@ class FaultAdapter(PddPurchaseAdapter): class ReadOnlyReconcileAdapter(PddPurchaseReconcileAdapter): - def __init__(self, calls) -> None: + def __init__(self, calls, candidates=None, error=None) -> None: self.calls = calls + self.candidates = candidates + self.error = error - def read_order_match(self, query): + def read_order_candidates(self, query): self.calls.append(("reconcile", query.goods_id)) - return PurchaseReconcileObservation( - "matched", "ORDER-001", "2026-08-10T08:00:00Z" + if self.error is not None: + raise self.error + candidates = self.candidates + if candidates is None: + candidates = ( + PurchaseOrderCandidate( + order_no="ORDER-001", + goods_id=query.goods_id, + options=dict(query.options), + quantity=query.quantity, + total_price_cent=query.total_price_cent, + ordered_at=query.irreversible_action_at, + ordered_at_raw="2026-08-10 16:00:00", + payment_status="unpaid", + ), + ) + else: + candidates = tuple( + replace(candidate, ordered_at=query.irreversible_action_at) + for candidate in candidates + ) + return PurchaseReconcileScan( + candidates=candidates, + diagnostics={"pages_scanned": 1}, ) def close(self) -> None: @@ -119,12 +147,19 @@ class PurchaseRecoveryTest(unittest.TestCase): def tearDown(self) -> None: self.temporary.cleanup() - def _add(self, task_id: str = "PUR-RECOVER") -> None: - task = purchase_admin_task(task_id) + def _add( + self, task_id: str = "PUR-RECOVER", execution_mode: str = "dry_run" + ) -> None: + task = purchase_admin_task(task_id, execution_mode) self.gateway.enqueue_task(task, self.client.client_id) claimed = self.gateway.claim_next( self.client, - ClaimCapabilities(supported_types=(TaskType.PURCHASE,)), + ClaimCapabilities( + supported_types=(TaskType.PURCHASE,), + purchase_mode=( + "live" if execution_mode == "live" else "dry_run" + ), + ), ) assert claimed is not None self.repository.add_claimed_task( @@ -142,18 +177,18 @@ class PurchaseRecoveryTest(unittest.TestCase): ) def _interrupt_after_irreversible(self, task_id: str) -> None: - self._add(task_id) + self._add(task_id, "live") started = self.repository.start_purchase_run(task_id, "USB-001") - connection = open_database(self.db_path) - try: - with connection: - connection.execute( - "UPDATE task_runs SET irreversible_action_at = ?" - " WHERE attempt_id = ?", - ("2026-08-10T08:00:00Z", started.attempt_id), - ) - finally: - connection.close() + self.repository.mark_purchase_irreversible( + task_id, + started.attempt_id, + { + "options": dict(OPTIONS), + "quantity": 2, + "unit_price_cent": 4200, + "total_price_cent": 8400, + }, + ) self.repository.recover_interrupted_work() def test_critical_action_failure_keeps_last_persisted_step(self): @@ -240,17 +275,36 @@ class PurchaseRecoveryTest(unittest.TestCase): ) first = dispatcher.execute_one() + pending = self.repository.next_pending_outbox() + matched_detail = self.repository.get_task("PUR-RECOVER") + matched_run = self.repository.latest_task_run("PUR-RECOVER") + duplicate = self.repository.save_matched_purchase_reconciliation( + "PUR-RECOVER", + matched_run.attempt_id, + matched_detail.pdd_data, + {"mode": "reconcile_only"}, + ) + self.assertEqual(duplicate.id, pending.id) second = dispatcher.execute_one() + third = dispatcher.execute_one() - self.assertEqual(first.kind, "manual_review") - self.assertEqual(second.kind, "no_task") + self.assertEqual(first.kind, "result_pending") + self.assertEqual(second.kind, "succeeded") + self.assertEqual(third.kind, "no_task") self.assertNotIn(("purchase",), calls) self.assertEqual(calls.count(("reconcile", "737116531267")), 1) detail = self.repository.get_task("PUR-RECOVER") run = self.repository.latest_task_run("PUR-RECOVER") assert detail is not None and run is not None - self.assertEqual(detail.current_step, "reconcile_completed") - self.assertEqual(run.diagnostics_json["mode"], "reconcile_only") + self.assertEqual(detail.current_step, "completed") + self.assertEqual( + run.diagnostics_json["reconciliation"]["mode"], + "reconcile_only", + ) + self.assertEqual( + detail.pdd_data["purchase"]["payment_status"], "unpaid" + ) + self.assertEqual(detail.pdd_data["purchase"]["order_no"], "ORDER-001") def test_reconcile_device_failure_is_recorded_as_unknown(self): task_id = "PUR-RECONCILE-OFFLINE" @@ -278,7 +332,136 @@ class PurchaseRecoveryTest(unittest.TestCase): run = self.repository.latest_task_run(task_id) assert detail is not None and run is not None self.assertEqual(detail.current_step, "reconcile_manual_review") - self.assertIn("核对设备已断开", run.diagnostics_json["error"]) + self.assertIn( + "核对设备已断开", + run.diagnostics_json["reconciliation"]["error_message"], + ) + + def test_no_multiple_mismatched_and_paid_candidates_need_manual_review(self): + cases = { + "EMPTY": ((), "ORDER_NOT_FOUND"), + "MULTIPLE": ( + ( + PurchaseOrderCandidate( + "ORDER-A", + "737116531267", + OPTIONS, + 2, + 8400, + "2026-08-10T08:00:00Z", + "2026-08-10 16:00:00", + "unpaid", + ), + PurchaseOrderCandidate( + "ORDER-B", + "737116531267", + OPTIONS, + 2, + 8400, + "2026-08-10T08:00:00Z", + "2026-08-10 16:00:00", + "unpaid", + ), + ), + "AMBIGUOUS_ORDER_MATCH", + ), + "MISMATCH": ( + ( + PurchaseOrderCandidate( + "ORDER-C", + "OTHER-GOODS", + OPTIONS, + 2, + 8400, + "2026-08-10T08:00:00Z", + "2026-08-10 16:00:00", + "unpaid", + ), + ), + "ORDER_MATCH_UNCERTAIN", + ), + "PAID": ( + ( + PurchaseOrderCandidate( + "ORDER-D", + "737116531267", + OPTIONS, + 2, + 8400, + "2026-08-10T08:00:00Z", + "2026-08-10 16:00:00", + "paid", + ), + ), + "ORDER_MATCH_UNCERTAIN", + ), + } + for suffix, (candidates, error_code) in cases.items(): + with self.subTest(suffix=suffix): + task_id = f"PUR-{suffix}" + self._interrupt_after_irreversible(task_id) + calls = [] + outcome = TaskDispatcher( + self.gateway, + self.repository, + self.client, + "USB-001", + purchase_adapter_factory=lambda *_args: calls.append( + "purchase" + ), + purchase_reconcile_factory=( + lambda _address, _cancelled, values=candidates: + ReadOnlyReconcileAdapter(calls, values) + ), + device_connection_checker=lambda _serial: None, + ).execute_one() + + self.assertEqual(outcome.kind, "manual_review") + self.assertNotIn("purchase", calls) + detail = self.repository.get_task(task_id) + self.assertEqual(detail.status, TaskStatus.MANUAL_REVIEW) + self.assertEqual(detail.last_error_code, error_code) + self.assertIsNone(self.repository.next_pending_outbox()) + + def test_restart_after_match_only_submits_outbox_without_reading_phone(self): + task_id = "PUR-OUTBOX-RESTART" + self._interrupt_after_irreversible(task_id) + first_calls = [] + first = TaskDispatcher( + self.gateway, + self.repository, + self.client, + "USB-001", + purchase_reconcile_factory=( + lambda _address, _cancelled: ReadOnlyReconcileAdapter( + first_calls + ) + ), + device_connection_checker=lambda _serial: None, + ).execute_one() + self.assertEqual(first.kind, "result_pending") + + restarted_repository = TaskRepository(self.db_path) + restarted_calls = [] + second = TaskDispatcher( + self.gateway, + restarted_repository, + self.client, + "", + purchase_adapter_factory=lambda *_args: restarted_calls.append( + "purchase" + ), + purchase_reconcile_factory=lambda *_args: restarted_calls.append( + "reconcile" + ), + ).execute_one() + + self.assertEqual(second.kind, "succeeded") + self.assertEqual(restarted_calls, []) + self.assertEqual( + restarted_repository.get_task(task_id).status, + TaskStatus.SUCCEEDED, + ) def test_live_submit_is_isolated_and_payment_methods_are_unavailable(self): self.assertEqual(ClaimCapabilities(purchase_mode="live").purchase_mode, "live") diff --git a/client/test/test_task_detail_view.py b/client/test/test_task_detail_view.py index bcb5345..c92293c 100644 --- a/client/test/test_task_detail_view.py +++ b/client/test/test_task_detail_view.py @@ -122,8 +122,13 @@ class TaskDetailViewTest(unittest.TestCase): def test_purchase_recovery_steps_are_clear_chinese(self): cases = { "purchase_dry_run_stopped": "采购演练已在提交前停止", - "reconcile_purchase": "只允许核对订单", - "reconcile_manual_review": "核对结果不确定，需人工处理", + "reconcile_purchase": "订单已提交，只允许核对未付款订单", + "purchase_order_matched_pending_report": ( + "已提交待付款，等待向 Admin 上报" + ), + "reconcile_manual_review": ( + "订单结果不确定或存在多个候选，需人工处理" + ), } for step, expected in cases.items(): with self.subTest(step=step): diff --git a/client/test/test_task_repository.py b/client/test/test_task_repository.py index 8b5581e..2015943 100644 --- a/client/test/test_task_repository.py +++ b/client/test/test_task_repository.py @@ -113,10 +113,21 @@ class TaskRepositoryTests(unittest.TestCase): "PURCHASE-LIVE", "192.168.0.173:5555" ) marked_at = self.repository.mark_purchase_irreversible( - "PURCHASE-LIVE", started.attempt_id + "PURCHASE-LIVE", + started.attempt_id, + { + "options": {"color": "黑色", "size": "L"}, + "quantity": 2, + "unit_price_cent": 3990, + "total_price_cent": 7980, + }, ) run = self.repository.latest_task_run("PURCHASE-LIVE") self.assertEqual(run.irreversible_action_at, marked_at) + self.assertEqual( + run.diagnostics_json["final_confirmation"]["total_price_cent"], + 7980, + ) with self.assertRaisesRegex(ValueError, "已经存在"): self.repository.mark_purchase_irreversible( "PURCHASE-LIVE", started.attempt_id diff --git a/docs/client/02-architecture.md b/docs/client/02-architecture.md index 8f6e735..da6e566 100644 --- a/docs/client/02-architecture.md +++ b/docs/client/02-architecture.md @@ -371,7 +371,9 @@ Admin 侧必须无条件接受，见 [04](04-admin-api-contract.md) §6.1。 `claimed`；恢复执行必须创建新的 `attempt_id`，不会存在两条并发运行。 - `irreversible_action_at` 有值时，启动恢复立即转为 `manual_review / reconcile_purchase`。 `PurchaseReconcileService` 只能调用独立的只读 Adapter，不会调用采购 Adapter。 -- 核对到唯一候选也仍需人工最终确认；未找到、多候选或结果不确定均保持人工处理。 +- 核单按商品编号、完整规格、数量、确认总价、提交时间范围和未付款状态严格匹配。 + 只有唯一候选会原子写入采购结果与 Outbox；未找到、多候选、字段不符、非未付款 + 或结果不确定均保持人工处理。 - Admin 提交失败只重试 Outbox，不再次操作拼多多。 ## 9. PDD 适配边界 @@ -392,9 +394,14 @@ reconcile_purchase(task, run) -> PurchaseResult | ManualReview `irreversible_action_at` 事务，再允许 Adapter 点击一次；之后无论点击结果是否明确， 都只进入订单核对。该接口不提供付款或取消订单方法。 采购规格使用完整 `options` 对象精确比较，不假定只有颜色和尺码两个维度。 -只读核对另用 `PddPurchaseReconcileAdapter`，只暴露 `read_order_match` +只读核对另用 `PddPurchaseReconcileAdapter`，只暴露 `read_order_candidates` 和 `close`，不暴露选规格、设数量、下单或付款方法。 +正式只读核单由 `pdd_u2_purchase_reconcile_adapter.py` 实现，只允许使用白名单 +导航进入个人中心、我的订单和待付款列表，以及返回和滚动。它只提取核单所需的 +订单编号、商品编号、规格、数量、总价、下单时间和付款状态，不保存收货人、地址 +或电话。最终匹配由领域服务完成，不能让页面解析层单独决定采购成功。 + 正式 Client 由 `pdd_u2_purchase_adapter.py` 分别实现演练和 live 接口，并在 `ui_main.py` 注入工厂。Adapter 通过 `PddDeviceService` 独占连接，每次判断 都重新读取当前包名和控件树。当前 Admin 下发的 `color` 和 `size` diff --git a/docs/client/03-data-model.md b/docs/client/03-data-model.md index 3245924..4b268ef 100644 --- a/docs/client/03-data-model.md +++ b/docs/client/03-data-model.md @@ -444,6 +444,7 @@ CREATE TABLE app_settings ( | `running` | `cancelled` | 用户停止，且已到安全点、未进入不可逆阶段 | 任务协调器 | | `result_pending` 结果待提交 | `succeeded` | Admin 返回 `accepted: true` | 结果提交服务 | | `result_pending` | `manual_review` | 重试次数超上限仍提交不上去 | 结果提交服务 | +| `manual_review / reconcile_purchase` | `result_pending` | 严格核对到唯一未付款订单，结果和 Outbox 已在同一事务落库 | 只读核单服务 | | `retry_wait` 重试等待 | `running` | 重新启动获取任务，或用户确认“重新执行”；**本地直接重跑**并新建 `task_runs` 记录 | 任务协调器 / 人 | | `retry_wait` | `failed` | 超过最大重试次数，且从未进入不可逆阶段 | 任务协调器 | | `retry_wait` | `manual_review` | 超过最大重试次数，但**曾经进入过不可逆阶段** | 任务协调器 | @@ -588,6 +589,7 @@ CREATE TABLE app_settings ( "confirmation_reached": true, "order_submitted": false, "payment_attempted": false, + "payment_status": null, "order_no": null, "ordered_at": null, "ordered_at_raw": null, @@ -601,7 +603,10 @@ CREATE TABLE app_settings ( `order_submitted=false`、`payment_attempted=false` 和 `match_status=not_submitted`；只表示已经安全到达最终提交前确认页。 -真实下单后 `mode` 为 `live`，`match_status` 只能是 `matched`、`ambiguous` 或 `not_found`。只有 `matched` 可以自动报告采购成功。 +真实下单成功核对后 `mode` 为 `live`、`payment_status` 为 `unpaid`、 +`match_status` 为 `matched`。只有商品编号、完整规格、数量、总价、时间范围和未付款 +状态全部一致，且候选唯一时才生成采购结果。无候选、多候选、字段不符或非未付款 +订单只保存脱敏诊断并进入人工处理，不生成成功 `pdd_data`。 ## 9. JSON 兼容规则 diff --git a/docs/client/04-admin-api-contract.md b/docs/client/04-admin-api-contract.md index 2a18f8d..ace08d9 100644 --- a/docs/client/04-admin-api-contract.md +++ b/docs/client/04-admin-api-contract.md @@ -314,7 +314,11 @@ Idempotency-Key: task-id:attempt-id:result-v1 } ``` -采购结果把 `result_type` 换成 `purchase`，其余结构相同。 +采购结果把 `result_type` 换成 `purchase`，并按 [03 数据模型 §8.2](03-data-model.md) +携带 `purchase`。真实采购成功必须同时满足 `mode=live`、 +`order_submitted=true`、`payment_attempted=false`、`payment_status=unpaid`、 +`match_status=matched`，并包含非空订单编号和带时区下单时间。0 个或多个候选、 +字段不符及非未付款订单改走 §7 人工处理，不得提交采购成功结果。 响应： diff --git a/docs/client/05-ui-specification.md b/docs/client/05-ui-specification.md index 7b9b7ce..4301531 100644 --- a/docs/client/05-ui-specification.md +++ b/docs/client/05-ui-specification.md @@ -222,6 +222,8 @@ class TaskTableModel(QAbstractTableModel): 详情面向采购人员，任务类型、状态、当前步骤和错误说明优先使用中文。数据库中的 `current_step` 等内部值只用于业务判断和日志，不直接显示；遇到尚未适配的新步骤时显示“未知步骤”。 采购演练和恢复至少要明确显示“演练已在提交前停止”、“结果待提交”、 “只允许核对订单”和“需要人工处理”；不能只用颜色表达安全状态。 +真实提交后的详情还要区分“订单已提交，只允许核对未付款订单”、 +“已提交待付款，等待向 Admin 上报”和“订单结果不确定或存在多个候选”。 采集规格按下面顺序展示： diff --git a/docs/client/06-quality-security.md b/docs/client/06-quality-security.md index 57b28c6..d8dbd32 100644 --- a/docs/client/06-quality-security.md +++ b/docs/client/06-quality-security.md @@ -92,6 +92,8 @@ PDD 解析测试优先使用脱敏的 XML 固件，不要求每次连接真实 - 关键手机动作前已持久化 `current_step`。 - 无不可逆标记的中断会先关闭旧运行，新运行使用新 `attempt_id`。 - 有不可逆标记的中断只调用只读核对，采购 Adapter 调用次数为 0。 +- 只读核单按商品、完整规格、数量、总价、时间和未付款状态全部精确匹配；只有唯一候选生成成功 Outbox。 +- 无候选、多候选、字段缺失或不符、非未付款状态只转人工，不产生采购成功结果。 - 已落库 Outbox 只补交，不重跑手机流程。 - 停止、设备断开、验证码、登录失效和结果不确定都保留稳定错误和诊断。 - `execution_mode=live` 只允许采购任务，且 Client 不得自行升级 Admin 下发的模式。