diff --git a/client/scripts/run_t105_quantity_gate2.py b/client/scripts/run_t105_quantity_gate2.py new file mode 100644 index 0000000..f782b9c --- /dev/null +++ b/client/scripts/run_t105_quantity_gate2.py @@ -0,0 +1,132 @@ +"""运行 T-105 数量 1→2、Gate2 截图与一次安全退出。""" + +from __future__ import annotations + +import argparse +from datetime import datetime +from math import isfinite +from pathlib import Path +import sys + + +CLIENT_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(CLIENT_ROOT / "src")) + +from cmbuyer_client.device.adb import AdbClient, DeviceConnectionError, SubprocessAdbRunner +from cmbuyer_client.device.baseline import NoReconnectUiautomatorConnector +from cmbuyer_client.pdd.quantity_gate2 import ( + EXPECTED_GATE1_UNIT_PRICE, + EXPECTED_GOODS_ID, + TASK_COLOR, + TASK_SIZE, + Gate1Observation, + QuantityGate2Error, +) +from cmbuyer_client.pdd.quantity_gate2_runner import QuantityGate2Runner +from cmbuyer_client.pdd.quantity_gate2_spike import Android16TopResumedForegroundReader + + +def parse_arguments(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description="运行 T-105 已取证数量与 Gate2 闭环。") + parser.add_argument("--serial", required=True, help="显式 ADB serial；禁止自动选择。") + parser.add_argument("--goods-id", required=True) + parser.add_argument("--color", required=True) + parser.add_argument("--size", required=True) + parser.add_argument("--target-quantity", required=True, type=int) + parser.add_argument("--gate1-unit-price", required=True) + parser.add_argument("--gate1-screenshot", required=True, type=Path) + parser.add_argument("--gate1-captured-at", required=True) + parser.add_argument("--max-total-price", required=True) + parser.add_argument("--output-dir", required=True, type=Path) + parser.add_argument("--timeout", type=float, default=10.0) + parser.add_argument("--adb", default="adb") + return parser.parse_args(argv) + + +def validate_arguments(arguments: argparse.Namespace) -> datetime: + if type(arguments.serial) is not str or not arguments.serial.strip() or arguments.serial != arguments.serial.strip(): + raise ValueError("必须显式提供非空 --serial。") + if arguments.goods_id != EXPECTED_GOODS_ID: + raise ValueError("--goods-id 不是 T-105 已批准目标。") + if arguments.color != TASK_COLOR or arguments.size != TASK_SIZE: + raise ValueError("颜色或尺码不是 T-105 已批准目标。") + if type(arguments.target_quantity) is not int or arguments.target_quantity != 2: + raise ValueError("本次真机验收只批准 --target-quantity 2。") + if arguments.gate1_unit_price != EXPECTED_GATE1_UNIT_PRICE: + raise ValueError("--gate1-unit-price 与已确认 Gate1 不一致。") + if not isinstance(arguments.gate1_screenshot, Path) or not arguments.gate1_screenshot.is_file(): + raise ValueError("--gate1-screenshot 必须是现有显式文件。") + try: + captured_at = datetime.fromisoformat(arguments.gate1_captured_at) + except (TypeError, ValueError) as error: + raise ValueError("--gate1-captured-at 必须是带时区 ISO 时间。") from error + if captured_at.utcoffset() is None: + raise ValueError("--gate1-captured-at 必须带时区。") + if arguments.max_total_price != "40.00": + raise ValueError("本次真机验收固定 --max-total-price 40.00。") + if not isinstance(arguments.output_dir, Path) or not arguments.output_dir.name or arguments.output_dir.exists(): + raise ValueError("--output-dir 必须是不存在的明确新目录。") + if ( + not isinstance(arguments.timeout, (int, float)) + or isinstance(arguments.timeout, bool) + or arguments.timeout <= 0 + or not isfinite(arguments.timeout) + ): + raise ValueError("--timeout 必须是大于 0 的有限数值。") + return captured_at + + +def main(argv: list[str] | None = None) -> int: + arguments = parse_arguments(argv) + try: + captured_at = validate_arguments(arguments) + except ValueError as error: + print(f"失败：{error}", file=sys.stderr) + return 2 + + try: + import adbutils + import uiautomator2 as u2 + except ImportError: + print("失败：缺少采购工具真机依赖。", file=sys.stderr) + return 2 + + adb_runner = SubprocessAdbRunner(arguments.adb) + runner = QuantityGate2Runner( + AdbClient(adb_runner, timeout_seconds=arguments.timeout), + NoReconnectUiautomatorConnector( + adbutils.AdbClient(socket_timeout=arguments.timeout).device_list, + u2.connect, + ), + Android16TopResumedForegroundReader(adb_runner, arguments.timeout), + timeout_seconds=arguments.timeout, + ) + gate1 = Gate1Observation( + color=arguments.color, + size=arguments.size, + quantity=1, + gate1_unit_price=arguments.gate1_unit_price, + screenshot_path=arguments.gate1_screenshot, + captured_at=captured_at, + ) + try: + runner.run( + arguments.serial, + arguments.goods_id, + gate1, + arguments.target_quantity, + arguments.max_total_price, + arguments.output_dir, + ) + except (DeviceConnectionError, QuantityGate2Error, OSError): + # 不回显第三方异常、serial、本机路径或页面正文。 + print("T-105 数量/Gate2 运行失败：已停止，未发布证据目录。", file=sys.stderr) + return 1 + + print("T-105 数量/Gate2 运行完成。") + print("人工复核：数量 2、目标规格、12.88/32.76、原始截图和一次安全退出。") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/client/src/cmbuyer_client/pdd/quantity_gate2.py b/client/src/cmbuyer_client/pdd/quantity_gate2.py new file mode 100644 index 0000000..abf3b08 --- /dev/null +++ b/client/src/cmbuyer_client/pdd/quantity_gate2.py @@ -0,0 +1,605 @@ +"""T-105：证据绑定的数量读回与 Gate2 面板总额。 + +本模块只允许数量 1 保持不变，或从数量 1 对唯一加号点击一次到数量 2。 +它不包含确认页、提交围栏、提交订单或付款能力。 +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from decimal import Decimal +from functools import partial +from math import isfinite +from pathlib import Path +import re +from time import monotonic, sleep +from typing import Any, Callable, Protocol +from xml.etree import ElementTree + +from ..core.errors import ValidationError +from ..core.validation import require_money +from ..device.baseline import PDD_PACKAGE +from .product_open import EXPECTED_PDD_VERSION +from .sku_selection import SkuSelectionError, _parse_nodes as _parse_sku_nodes +from .sku_selection import _product_exit_projection + + +EXPECTED_DEVICE_MODEL = "PKG110" +EXPECTED_ANDROID_VERSION = "16" +EXPECTED_SCREEN_SIZE = (1080, 2376) +EXPECTED_GOODS_ID = "937122477375" +TASK_COLOR = "黑色CHA(纯棉)" +TASK_SIZE = "M(建议100-115)" +UI_COLOR = "黑色 CHA (纯棉）" +UI_SIZE = "M（建议100-115）" +EXPECTED_GATE1_UNIT_PRICE = "12.88" + +_PDD_ID = "com.xunmeng.pinduoduo:id/pdd" +_QUANTITY_CONTAINER_ID = "com.xunmeng.pinduoduo:id/gnl" +_COLOR_ID = "com.xunmeng.pinduoduo:id/tv_content" +_TARGET_SUMMARY = f"已选： {UI_COLOR} {UI_SIZE}" +_AMOUNT_TEXT = re.compile(r"^快卖完 ¥(?P(?:0|[1-9]\d*)\.\d{2})$") +_BOUNDS = re.compile(r"^\[(\d+),(\d+)\]\[(\d+),(\d+)\]$") + + +class QuantityGate2Error(RuntimeError): + """数量/Gate2 判据不成立时的脱敏安全停止。""" + + +class QuantityGate2TimeoutError(QuantityGate2Error): + """设备调用或后置条件等待超时。""" + + +class QuantityGate2OverCapError(QuantityGate2Error): + """目标数量面板总额超过管理员授权上限。""" + + +class QuantityGate2Device(Protocol): + """T-105 的窄设备能力；没有通用页面动作。""" + + def app_info(self, package_name: str) -> dict[str, Any]: ... + + def current_foreground(self) -> dict[str, str]: ... + + def display_size(self) -> tuple[int, int]: ... + + def dump_window_hierarchy(self) -> str: ... + + def increment_quantity_once(self, bounds: str) -> None: ... + + def capture_screenshot(self) -> str: ... + + def leave_sku_panel_once(self) -> None: ... + + +@dataclass(frozen=True) +class Gate1Observation: + color: str + size: str + quantity: int + gate1_unit_price: str + screenshot_path: Path + captured_at: datetime + + def __post_init__(self) -> None: + if self.color != TASK_COLOR or self.size != TASK_SIZE or type(self.quantity) is not int or self.quantity != 1: + raise QuantityGate2Error("Gate1 规格或数量不是已取证前置，已停止操作。") + if _money(self.gate1_unit_price) != EXPECTED_GATE1_UNIT_PRICE: + raise QuantityGate2Error("Gate1 单价不是已取证值，已停止操作。") + if not isinstance(self.screenshot_path, Path) or not self.screenshot_path.name: + raise QuantityGate2Error("Gate1 截图路径无效，已停止操作。") + if not isinstance(self.captured_at, datetime) or self.captured_at.utcoffset() is None: + raise QuantityGate2Error("Gate1 采集时间必须带时区，已停止操作。") + + +@dataclass(frozen=True) +class Gate2Observation: + requested_color: str + requested_size: str + actual_color: str + actual_size: str + requested_quantity: int + quantity_read: int + gate1_unit_price: str + gate2_panel_total_price: str + max_total_price: str + screenshot_path: Path + captured_at: datetime + + +@dataclass(frozen=True) +class _PanelProfile: + quantity: int + panel_bounds: str + price_row_bounds: str + price_bounds: str + price_text: str + summary_bounds: str + quantity_bounds: str + minus_bounds: str + value_bounds: str + plus_bounds: str + color_bounds: str + + +_INITIAL = _PanelProfile( + 1, + "[0,474][1080,863]", + "[396,498][895,570]", + "[396,503][712,570]", + "快卖完 ¥12.88", + "[396,654][1053,716]", + "[396,752][645,827]", + "[396,752][474,827]", + "[480,752][561,827]", + "[567,752][645,827]", + "[126,1000][438,1024]", +) +_TARGET = _PanelProfile( + 2, + "[0,474][1080,861]", + "[396,498][740,570]", + "[396,503][722,570]", + "快卖完 ¥32.76", + "[396,582][1053,644]", + "[396,750][645,825]", + "[396,750][474,825]", + "[480,750][561,825]", + "[567,750][645,825]", + "[126,998][438,1024]", +) +_PROFILES = {1: _INITIAL, 2: _TARGET} + + +@dataclass(frozen=True) +class _Node: + element: ElementTree.Element + parent: "_Node | None" + + @property + def text(self) -> str: + return self.element.get("text", "") + + @property + def desc(self) -> str: + return self.element.get("content-desc", "") + + @property + def bounds(self) -> str: + return self.element.get("bounds", "") + + +@dataclass(frozen=True) +class _VerifiedPanel: + quantity: int + panel_total_price: str + plus_bounds: str + projection: tuple[object, ...] + + +class QuantityGate2Flow: + """从已确认数量 1 面板推进至获准数量，并安全退出同一商品。""" + + def __init__( + self, + device: QuantityGate2Device, + wait_timeout_seconds: float = 1.0, + poll_interval_seconds: float = 0.2, + monotonic_clock: Callable[[], float] = monotonic, + sleep_function: Callable[[float], None] = sleep, + ) -> None: + if not _positive_finite(wait_timeout_seconds) or not _positive_finite(poll_interval_seconds): + raise ValueError("等待参数必须是大于 0 的有限数值。") + self._device = device + self._timeout = float(wait_timeout_seconds) + self._poll = float(poll_interval_seconds) + self._clock = monotonic_clock + self._sleep = sleep_function + self._increment_attempted = False + self._pending: tuple[str, Callable[[list[_Node]], _VerifiedPanel]] | None = None + self._terminal = False + self._verified_quantity: int | None = None + self._verified_total: str | None = None + self._exited = False + + @property + def increment_attempts(self) -> int: + return int(self._increment_attempted) + + @property + def exited(self) -> bool: + return self._exited + + @property + def can_exit_safely(self) -> bool: + return self._pending is None and self._verified_quantity in _PROFILES and not self._exited + + def set_quantity_and_verify( + self, + gate1: Gate1Observation, + target_quantity: int, + max_total_price: str, + ) -> _VerifiedPanel: + self._require_active() + _validate_request(gate1, target_quantity, max_total_price) + initial = self._read_verified(_INITIAL) + if initial.panel_total_price != gate1.gate1_unit_price: + raise QuantityGate2Error("Gate1 当前面板事实已漂移，已停止操作。") + + if target_quantity == 1: + verified = initial + else: + # 动作前 fresh 读取，不能使用上一次节点或缓存 bounds。 + before = self._read_hierarchy() + precondition = _verified_panel(_parse_nodes(before), _INITIAL) + _require_unique_action_occupants(_parse_nodes(before), precondition.plus_bounds) + self._pending = (before, partial(_verified_panel, profile=_TARGET)) + self._increment_attempted = True + try: + self._device.increment_quantity_once(precondition.plus_bounds) + verified = self._wait_for_pending() + except BaseException: + # 点击超时可能已送达；封存后不允许本 Flow 重试或继续。 + self._terminal = True + raise + + self._verified_quantity = verified.quantity + self._verified_total = verified.panel_total_price + if Decimal(verified.panel_total_price) > Decimal(max_total_price): + raise QuantityGate2OverCapError("Gate2 面板总额超过授权最高总价，已停止操作。") + return verified + + def build_observation( + self, + gate1: Gate1Observation, + target_quantity: int, + max_total_price: str, + screenshot_path: Path, + captured_at: datetime, + ) -> Gate2Observation: + self._require_active() + _validate_request(gate1, target_quantity, max_total_price) + if not isinstance(screenshot_path, Path) or not screenshot_path.name: + raise QuantityGate2Error("Gate2 截图路径无效，已停止操作。") + if not isinstance(captured_at, datetime) or captured_at.utcoffset() is None: + raise QuantityGate2Error("Gate2 采集时间必须带时区，已停止操作。") + verified = self._read_verified(_PROFILES[target_quantity]) + if ( + self._verified_quantity != verified.quantity + or self._verified_total != verified.panel_total_price + or Decimal(verified.panel_total_price) > Decimal(max_total_price) + ): + raise QuantityGate2Error("截图后 Gate2 事实漂移，已停止操作。") + return Gate2Observation( + requested_color=gate1.color, + requested_size=gate1.size, + actual_color=TASK_COLOR, + actual_size=TASK_SIZE, + requested_quantity=target_quantity, + quantity_read=verified.quantity, + gate1_unit_price=gate1.gate1_unit_price, + gate2_panel_total_price=verified.panel_total_price, + max_total_price=max_total_price, + screenshot_path=screenshot_path, + captured_at=captured_at, + ) + + def exit_sku_panel_safely(self) -> None: + self._require_active() + if self._verified_quantity not in _PROFILES: + raise QuantityGate2Error("没有可用于安全退出的数量事实，已停止操作。") + current = self._read_verified(_PROFILES[self._verified_quantity]) + if current.panel_total_price != self._verified_total: + raise QuantityGate2Error("安全退出前 Gate2 事实漂移，已停止操作。") + try: + self._device.leave_sku_panel_once() + except BaseException: + self._terminal = True + raise + + deadline = self._clock() + self._timeout + stable: tuple[object, ...] | None = None + try: + while True: + self._require_environment() + raw = self._read_hierarchy() + try: + projection = _product_exit_projection(_parse_sku_nodes(raw)) + except SkuSelectionError: + stable = None + else: + self._require_environment() + if stable == projection: + self._exited = True + self._terminal = True + return + stable = projection + remaining = deadline - self._clock() + if remaining <= 0: + raise QuantityGate2TimeoutError("安全退出未达到连续稳定同商品判据，未重试返回。") + self._sleep(min(self._poll, remaining)) + except BaseException: + self._terminal = True + raise + + def reconcile_pending_action(self) -> _VerifiedPanel | None: + """结果不明时只读一次待定后置条件；绝不重发加号。""" + + if self._pending is None: + return None + return self._wait_for_pending() + + def _wait_for_pending(self) -> _VerifiedPanel: + if self._pending is None: + raise QuantityGate2Error("没有可调和的数量动作。") + previous, condition = self._pending + deadline = self._clock() + self._timeout + while True: + self._require_environment() + raw = self._read_hierarchy() + if raw != previous: + try: + verified = condition(_parse_nodes(raw)) + except QuantityGate2Error: + pass + else: + self._pending = None + return verified + remaining = deadline - self._clock() + if remaining <= 0: + raise QuantityGate2TimeoutError("数量动作后置条件未确认，未重试加号。") + self._sleep(min(self._poll, remaining)) + + def _read_verified(self, profile: _PanelProfile) -> _VerifiedPanel: + self._require_environment() + return _verified_panel(_parse_nodes(self._read_hierarchy()), profile) + + def _require_environment(self) -> None: + info = self._device.app_info(PDD_PACKAGE) + version = (info.get("versionName") or info.get("version_name")) if isinstance(info, dict) else None + if version != EXPECTED_PDD_VERSION: + raise QuantityGate2Error("拼多多版本与 T-105 证据不一致，已停止操作。") + foreground = self._device.current_foreground() + if ( + not isinstance(foreground, dict) + or foreground.get("package") != PDD_PACKAGE + or not isinstance(foreground.get("activity"), str) + or not foreground["activity"].strip() + ): + raise QuantityGate2Error("拼多多不是唯一前台应用，已停止操作。") + if self._device.display_size() != EXPECTED_SCREEN_SIZE: + raise QuantityGate2Error("屏幕坐标空间与 T-105 证据不一致，已停止操作。") + + def _read_hierarchy(self) -> str: + value = self._device.dump_window_hierarchy() + if not isinstance(value, str) or not value: + raise QuantityGate2Error("节点树读取失败，已停止操作。") + return value + + def _require_active(self) -> None: + if self._terminal: + raise QuantityGate2Error("数量流程已进入不可重入终止态。") + + +def _positive_finite(value: object) -> bool: + return isinstance(value, (int, float)) and not isinstance(value, bool) and value > 0 and isfinite(value) + + +def _validate_request(gate1: object, target_quantity: object, max_total_price: object) -> None: + if not isinstance(gate1, Gate1Observation): + raise QuantityGate2Error("缺少可信 Gate1Observation，已停止操作。") + if type(target_quantity) is not int or target_quantity not in _PROFILES: + raise QuantityGate2Error("目标数量没有本项目真机证据，已停止操作。") + _money(max_total_price) + + +def _money(value: object) -> str: + try: + return require_money(value, "invalid_money") + except ValidationError as error: + raise QuantityGate2Error("金额不是规范十进制字符串，已停止操作。") from error + + +def _parse_nodes(raw: str) -> list[_Node]: + try: + root = ElementTree.fromstring(raw) + except ElementTree.ParseError as error: + raise QuantityGate2Error("节点树格式无效，已停止操作。") from error + if root.tag != "hierarchy": + raise QuantityGate2Error("节点树根节点无效，已停止操作。") + result: list[_Node] = [] + + def visit(element: ElementTree.Element, parent: _Node | None) -> None: + node = _Node(element, parent) + result.append(node) + for child in element: + visit(child, node) + + visit(root, None) + return result + + +def _verified_panel(nodes: list[_Node], profile: _PanelProfile) -> _VerifiedPanel: + panel = _one( + node for node in nodes + if _exact(node, "android.view.ViewGroup", profile.panel_bounds, resource_id=_PDD_ID, clickable="false") + ) + price_frame = _one( + node for node in nodes + if node.parent is panel and _exact(node, "android.widget.FrameLayout", "[396,498][1053,570]", resource_id=_PDD_ID, clickable="false") + ) + price_row = _one( + node for node in nodes + if node.parent is price_frame and _exact(node, "android.widget.LinearLayout", profile.price_row_bounds, resource_id=_PDD_ID, clickable="false") + ) + price = _one( + node for node in nodes + if node.parent is price_row + and _exact(node, "android.widget.TextView", profile.price_bounds, resource_id=_PDD_ID, clickable="false") + and node.text == profile.price_text + and not node.desc + ) + match = _AMOUNT_TEXT.fullmatch(price.text) + if match is None: + raise QuantityGate2Error("Gate2 面板总额角色不可读。") + amount = _money(match.group("amount")) + + _one( + node for node in nodes + if node.parent is panel + and _exact(node, "android.widget.TextView", profile.summary_bounds, resource_id=_PDD_ID, clickable="false") + and node.text == _TARGET_SUMMARY + and not node.desc + ) + quantity_outer = _one( + node for node in nodes + if node.parent is panel + and _exact(node, "android.widget.LinearLayout", profile.quantity_bounds, resource_id=_QUANTITY_CONTAINER_ID, clickable="false") + ) + quantity_inner = _one( + node for node in nodes + if node.parent is quantity_outer + and _exact(node, "android.widget.LinearLayout", profile.quantity_bounds, resource_id="", clickable="false") + ) + _one( + node for node in nodes + if node.parent is quantity_inner + and _exact(node, "android.widget.ImageView", profile.minus_bounds, resource_id=_PDD_ID, clickable="true") + and node.desc == "减少数量" + and not node.text + ) + quantity = _one( + node for node in nodes + if node.parent is quantity_inner + and _exact(node, "android.widget.EditText", profile.value_bounds, resource_id=_PDD_ID, clickable="true") + and node.text == str(profile.quantity) + and not node.desc + ) + plus = _one( + node for node in nodes + if node.parent is quantity_inner + and _exact(node, "android.widget.ImageView", profile.plus_bounds, resource_id=_PDD_ID, clickable="true") + and node.desc == "增加数量" + and not node.text + ) + if quantity.element.get("selected") != "false" or plus.element.get("selected") != "false": + raise QuantityGate2Error("数量控件选中属性漂移。") + + _require_selected_target(nodes, profile) + projection = ( + "quantity_gate2_8_17_0", + profile.quantity, + amount, + tuple(_projection(node) for node in (panel, price_frame, price_row, price, quantity_outer, quantity_inner, quantity, plus)), + ) + return _VerifiedPanel(profile.quantity, amount, plus.bounds, projection) + + +def _require_selected_target(nodes: list[_Node], profile: _PanelProfile) -> None: + color = _one( + node for node in nodes + if _exact(node, "android.widget.TextView", profile.color_bounds, resource_id=_COLOR_ID, clickable="true", selected="true") + and node.text == UI_COLOR + and not node.desc + ) + if color.parent is None or color.parent.element.get("selected") != "true": + raise QuantityGate2Error("目标颜色没有精确选中。") + size = _one( + node for node in nodes + if _exact(node, "android.widget.TextView", "[439,1582][831,1667]", resource_id=_PDD_ID, clickable="true", selected="true") + and node.text == UI_SIZE + and not node.desc + ) + if size.parent is None or size.parent.element.get("clickable") != "true": + raise QuantityGate2Error("目标尺码结构漂移。") + + +def _exact( + node: _Node, + class_name: str, + bounds: str, + *, + resource_id: str, + clickable: str, + selected: str = "false", +) -> bool: + element = node.element + return ( + element.get("package") == PDD_PACKAGE + and element.get("class") == class_name + and node.bounds == bounds + and element.get("resource-id", "") == resource_id + and element.get("clickable") == clickable + and element.get("selected") == selected + and element.get("enabled") == "true" + and element.get("visible-to-user") == "true" + and element.get("scrollable") == "false" + ) + + +def _one(values: Any) -> _Node: + matches = list(values) + if len(matches) != 1: + raise QuantityGate2Error("T-105 页面证据角色缺失或不唯一。") + return matches[0] + + +def _projection(node: _Node) -> tuple[str, ...]: + return ( + node.element.tag, + node.element.get("package", ""), + node.element.get("class", ""), + node.bounds, + node.element.get("resource-id", ""), + node.element.get("clickable", ""), + node.element.get("selected", ""), + node.text, + node.desc, + ) + + +def _bounds_center(bounds: str) -> tuple[int, int]: + match = _BOUNDS.fullmatch(bounds) + if match is None: + raise QuantityGate2Error("数量控件坐标无效。") + left, top, right, bottom = (int(value) for value in match.groups()) + if not (0 <= left < right <= EXPECTED_SCREEN_SIZE[0] and 0 <= top < bottom <= EXPECTED_SCREEN_SIZE[1]): + raise QuantityGate2Error("数量控件坐标超出已取证屏幕。") + return left + (right - left) // 2, top + (bottom - top) // 2 + + +def _require_unique_action_occupants(nodes: list[_Node], bounds: str) -> None: + target = _one(node for node in nodes if node.bounds == bounds and node.desc == "增加数量") + x, y = _bounds_center(bounds) + occupants = [ + node for node in nodes + if node.element.get("clickable") == "true" + and node.element.get("enabled") == "true" + and node.element.get("visible-to-user") == "true" + and _contains(node.bounds, x, y) + ] + if not occupants or any(not _same_branch(node, target) for node in occupants): + raise QuantityGate2Error("数量加号中心存在未知可点击覆盖层，已停止操作。") + + +def _contains(bounds: str, x: int, y: int) -> bool: + match = _BOUNDS.fullmatch(bounds) + if match is None: + return False + left, top, right, bottom = (int(value) for value in match.groups()) + return left <= x < right and top <= y < bottom + + +def _same_branch(candidate: _Node, target: _Node) -> bool: + current: _Node | None = target + while current is not None: + if current.element is candidate.element: + return True + current = current.parent + current = candidate + while current is not None: + if current.element is target.element: + return True + current = current.parent + return False diff --git a/client/src/cmbuyer_client/pdd/quantity_gate2_runner.py b/client/src/cmbuyer_client/pdd/quantity_gate2_runner.py new file mode 100644 index 0000000..d637f8c --- /dev/null +++ b/client/src/cmbuyer_client/pdd/quantity_gate2_runner.py @@ -0,0 +1,397 @@ +"""T-105 真机运行边界：一次数量加号、Gate2 原图与一次安全退出。""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass, replace +from datetime import UTC, datetime +from hashlib import sha256 +import json +from math import isfinite +import os +from pathlib import Path +import shutil +from time import monotonic +from typing import Any, Protocol +from uuid import uuid4 + +from adbutils.errors import AdbTimeout +from PIL import Image, UnidentifiedImageError +from uiautomator2.exceptions import HTTPTimeoutError + +from ..device.adb import AdbClient, DeviceConnectionError, DeviceInspection +from ..device.baseline import PDD_PACKAGE, SCREENSHOT_PARAMS, _save_base64_screenshot, _sha256_file +from .product_open import EXPECTED_PDD_VERSION +from .quantity_gate2 import ( + EXPECTED_ANDROID_VERSION, + EXPECTED_DEVICE_MODEL, + EXPECTED_GOODS_ID, + EXPECTED_SCREEN_SIZE, + Gate1Observation, + Gate2Observation, + QuantityGate2Device, + QuantityGate2Error, + QuantityGate2Flow, + QuantityGate2TimeoutError, + _bounds_center, + _money, +) + + +class ForegroundReader(Protocol): + def read(self, serial: str) -> dict[str, str]: ... + + +class QuantityGate2AdapterError(QuantityGate2Error): + """第三方设备接口失败后的脱敏映射。""" + + +@dataclass(frozen=True) +class QuantityGate2RunResult: + output_directory: Path + screenshot_path: Path + manifest_path: Path + observation: Gate2Observation + + +class UiautomatorQuantityGate2Adapter(QuantityGate2Device): + """只暴露 T-105 已批准的一个加号和一个 Back。""" + + def __init__( + self, + device: Any, + foreground_reader: ForegroundReader, + serial: str, + timeout_seconds: float, + ) -> None: + if type(serial) is not str or not serial.strip() or serial != serial.strip(): + raise ValueError("serial 必须显式且非空。") + if not _positive_finite(timeout_seconds): + raise ValueError("timeout_seconds 必须是大于 0 的有限数值。") + self._device = device + self._foreground_reader = foreground_reader + self._serial = serial + self._timeout = float(timeout_seconds) + self._increment_attempted = False + self._increment_bounds: str | None = None + self._increment_outcome = "not_attempted" + self._back_attempted = False + self._back_outcome = "not_attempted" + + @property + def increment_attempts(self) -> int: + return int(self._increment_attempted) + + @property + def increment_bounds(self) -> str | None: + return self._increment_bounds + + @property + def increment_outcome(self) -> str: + return self._increment_outcome + + @property + def back_attempts(self) -> int: + return int(self._back_attempted) + + @property + def back_outcome(self) -> str: + return self._back_outcome + + def app_info(self, package_name: str) -> dict[str, Any]: + value = self._call("app_info", package_name) + if not isinstance(value, dict): + raise QuantityGate2AdapterError("无法读取应用版本，已停止操作。") + return value + + def current_foreground(self) -> dict[str, str]: + try: + value = self._foreground_reader.read(self._serial) + except QuantityGate2Error: + raise + except Exception as error: + raise QuantityGate2AdapterError("无法读取 Android 前台摘要，已停止操作。") from error + if not isinstance(value, dict): + raise QuantityGate2AdapterError("Android 前台摘要无效，已停止操作。") + return value + + def display_size(self) -> tuple[int, int]: + value = self._call("window_size") + if not isinstance(value, tuple) or len(value) != 2 or any(type(item) is not int for item in value): + raise QuantityGate2AdapterError("无法读取屏幕坐标空间，已停止操作。") + return value + + def dump_window_hierarchy(self) -> str: + value = self._call("jsonrpc_call", "dumpWindowHierarchy", [False, 50], timeout=self._timeout) + if not isinstance(value, str): + raise QuantityGate2AdapterError("节点树读取失败，已停止操作。") + return value + + def increment_quantity_once(self, bounds: str) -> None: + if self._increment_attempted: + raise QuantityGate2AdapterError("数量加号已尝试过，拒绝重试。") + center_x, center_y = _bounds_center(bounds) + # RPC 超时无法证明事件未送达，动作机会必须先持久在内存审计状态中。 + self._increment_attempted = True + self._increment_bounds = bounds + self._increment_outcome = "ambiguous" + self._call("jsonrpc_call", "click", [center_x, center_y], timeout=self._timeout) + self._increment_outcome = "completed" + + def capture_screenshot(self) -> str: + value = self._call("jsonrpc_call", "takeScreenshot", SCREENSHOT_PARAMS, timeout=self._timeout) + if not isinstance(value, str): + raise QuantityGate2AdapterError("Gate2 原始截图读取失败，已停止操作。") + return value + + def leave_sku_panel_once(self) -> None: + if self._back_attempted: + raise QuantityGate2AdapterError("安全返回已尝试过，拒绝重试。") + self._back_attempted = True + self._back_outcome = "ambiguous" + self._call("jsonrpc_call", "pressKey", ["back"], timeout=self._timeout) + self._back_outcome = "completed" + + def _call(self, method: str, *args: Any, **kwargs: Any) -> Any: + try: + return getattr(self._device, method)(*args, **kwargs) + except (AdbTimeout, HTTPTimeoutError, TimeoutError) as error: + raise QuantityGate2TimeoutError("T-105 设备调用超时，已停止操作。") from error + except QuantityGate2Error: + raise + except Exception as error: + raise QuantityGate2AdapterError("T-105 设备调用失败，已停止操作。") from error + + +class QuantityGate2Runner: + """从人工停驻的数量 1 目标面板执行 T-105 已取证闭环。""" + + def __init__( + self, + adb_client: AdbClient, + connector: Callable[[str], Any], + foreground_reader: ForegroundReader, + timeout_seconds: float, + monotonic_clock: Callable[[], float] = monotonic, + ) -> None: + if not _positive_finite(timeout_seconds): + raise ValueError("timeout_seconds 必须是大于 0 的有限数值。") + self._adb_client = adb_client + self._connector = connector + self._foreground_reader = foreground_reader + self._timeout = float(timeout_seconds) + self._clock = monotonic_clock + + def run( + self, + serial: str, + goods_id: str, + gate1: Gate1Observation, + target_quantity: int, + max_total_price: str, + output_directory: Path, + ) -> QuantityGate2RunResult: + target = Path(output_directory) + staging: Path | None = None + adapter: UiautomatorQuantityGate2Adapter | None = None + flow: QuantityGate2Flow | None = None + try: + _validate_preflight(serial, goods_id, gate1, target_quantity, max_total_price, target) + staging = _prepare_staging(target) + inspection = self._adb_client.inspect(serial) + _require_expected_device(inspection) + adapter = UiautomatorQuantityGate2Adapter( + self._connector(serial), + self._foreground_reader, + serial, + self._timeout, + ) + flow = QuantityGate2Flow( + adapter, + wait_timeout_seconds=self._timeout, + monotonic_clock=self._clock, + ) + flow.set_quantity_and_verify(gate1, target_quantity, max_total_price) + + screenshot_path = staging / "gate2_screenshot.png" + _save_base64_screenshot(adapter.capture_screenshot(), screenshot_path) + captured_at = datetime.now(UTC) + _require_screenshot_size(screenshot_path) + observation = flow.build_observation( + gate1, + target_quantity, + max_total_price, + screenshot_path, + captured_at, + ) + flow.exit_sku_panel_safely() + _require_action_audit(adapter, target_quantity) + + manifest_path = staging / "manifest.json" + manifest_path.write_text( + json.dumps( + _manifest(inspection, serial, gate1, observation, screenshot_path, adapter), + ensure_ascii=False, + indent=2, + sort_keys=True, + ) + + "\n", + encoding="utf-8", + ) + os.rename(staging, target) + staging = None + except (DeviceConnectionError, QuantityGate2Error): + _attempt_known_safe_exit(flow) + _clean_staging(staging) + raise + except (AdbTimeout, HTTPTimeoutError, TimeoutError) as error: + _attempt_known_safe_exit(flow) + _clean_staging(staging) + raise QuantityGate2TimeoutError("T-105 真机运行超时，未发布证据。") from error + except OSError as error: + _attempt_known_safe_exit(flow) + _clean_staging(staging) + raise QuantityGate2Error("T-105 证据无法原子发布。") from error + except Exception as error: + _attempt_known_safe_exit(flow) + _clean_staging(staging) + raise QuantityGate2Error("T-105 真机运行未完成。") from error + + published_observation = replace( + observation, + screenshot_path=target / "gate2_screenshot.png", + ) + return QuantityGate2RunResult( + output_directory=target, + screenshot_path=target / "gate2_screenshot.png", + manifest_path=target / "manifest.json", + observation=published_observation, + ) + + +def _positive_finite(value: object) -> bool: + return isinstance(value, (int, float)) and not isinstance(value, bool) and value > 0 and isfinite(value) + + +def _validate_preflight( + serial: object, + goods_id: object, + gate1: object, + target_quantity: object, + max_total_price: object, + target: Path, +) -> None: + if type(serial) is not str or not serial.strip() or serial != serial.strip(): + raise QuantityGate2Error("必须显式提供非空设备通道。") + if type(goods_id) is not str or goods_id != EXPECTED_GOODS_ID: + raise QuantityGate2Error("商品不是 T-105 已批准目标。") + if not isinstance(gate1, Gate1Observation) or not gate1.screenshot_path.is_file(): + raise QuantityGate2Error("Gate1 原始截图不存在，已停止操作。") + if type(target_quantity) is not int or target_quantity not in {1, 2}: + raise QuantityGate2Error("目标数量没有 T-105 真机证据。") + _money(max_total_price) + if target.exists() or not target.name: + raise QuantityGate2Error("输出目录必须是不存在的明确新目录。") + + +def _prepare_staging(target: Path) -> Path: + staging: Path | None = None + try: + target.parent.mkdir(parents=True, exist_ok=True) + staging = target.parent / f".{target.name}.staging-{uuid4().hex}" + staging.mkdir() + return staging + except OSError as error: + _clean_staging(staging) + raise QuantityGate2Error("输出目录不可写，已停止操作。") from error + + +def _clean_staging(staging: Path | None) -> None: + if staging is not None and staging.exists(): + shutil.rmtree(staging) + + +def _require_expected_device(inspection: DeviceInspection) -> None: + if inspection.model != EXPECTED_DEVICE_MODEL or inspection.android_version != EXPECTED_ANDROID_VERSION: + raise QuantityGate2Error("设备型号或 Android 版本与 T-105 证据不一致。") + + +def _require_screenshot_size(path: Path) -> None: + try: + with Image.open(path) as image: + image.load() + if image.size != EXPECTED_SCREEN_SIZE: + raise QuantityGate2Error("Gate2 截图尺寸与 T-105 证据不一致。") + except QuantityGate2Error: + raise + except (OSError, UnidentifiedImageError) as error: + raise QuantityGate2Error("Gate2 截图不是有效图像。") from error + + +def _require_action_audit(adapter: UiautomatorQuantityGate2Adapter, target_quantity: int) -> None: + expected_increment = int(target_quantity == 2) + if ( + adapter.increment_attempts != expected_increment + or (expected_increment and adapter.increment_outcome != "completed") + or (expected_increment and adapter.increment_bounds != "[567,752][645,827]") + or (not expected_increment and adapter.increment_outcome != "not_attempted") + or adapter.back_attempts != 1 + or adapter.back_outcome != "completed" + ): + raise QuantityGate2Error("T-105 动作审计链不完整，拒绝发布。") + + +def _attempt_known_safe_exit(flow: QuantityGate2Flow | None) -> None: + if flow is None or not flow.can_exit_safely: + return + try: + flow.exit_sku_panel_safely() + except Exception: + pass + + +def _manifest( + inspection: DeviceInspection, + serial: str, + gate1: Gate1Observation, + observation: Gate2Observation, + screenshot_path: Path, + adapter: UiautomatorQuantityGate2Adapter, +) -> dict[str, Any]: + return { + "schema_version": 1, + "operation": "t105-quantity-gate2", + "captured_at": observation.captured_at.isoformat(), + "product": {"goods_id": EXPECTED_GOODS_ID}, + "channel": "wifi" if ":" in serial else "usb", + "serial_sha256": sha256(serial.encode("utf-8")).hexdigest(), + "device": { + "model": inspection.model, + "android_version": inspection.android_version, + "pdd_package": PDD_PACKAGE, + "pdd_version": EXPECTED_PDD_VERSION, + }, + "selection": {"color": observation.actual_color, "size": observation.actual_size}, + "quantity": {"requested": observation.requested_quantity, "read": observation.quantity_read}, + "prices": { + "gate1_unit_price": observation.gate1_unit_price, + "gate2_panel_total_price": observation.gate2_panel_total_price, + "max_total_price": observation.max_total_price, + }, + "gate1_evidence": { + "captured_at": gate1.captured_at.isoformat(), + "screenshot_sha256": _sha256_file(gate1.screenshot_path), + }, + "gate2_evidence": { + "path": screenshot_path.name, + "sha256": _sha256_file(screenshot_path), + }, + "action_audit": { + "increment_attempts": adapter.increment_attempts, + "increment_rpc_outcome": adapter.increment_outcome, + "back_attempts": adapter.back_attempts, + "back_rpc_outcome": adapter.back_outcome, + }, + "safe_exit": "completed", + "review_status": "human_review_required", + } diff --git a/client/tests/pdd/fixtures/quantity_gate2_initial_8_17_0.xml b/client/tests/pdd/fixtures/quantity_gate2_initial_8_17_0.xml new file mode 100644 index 0000000..75af6fa --- /dev/null +++ b/client/tests/pdd/fixtures/quantity_gate2_initial_8_17_0.xml @@ -0,0 +1,30 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/client/tests/pdd/fixtures/quantity_gate2_target_8_17_0.xml b/client/tests/pdd/fixtures/quantity_gate2_target_8_17_0.xml new file mode 100644 index 0000000..b57cca9 --- /dev/null +++ b/client/tests/pdd/fixtures/quantity_gate2_target_8_17_0.xml @@ -0,0 +1,30 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/client/tests/pdd/test_quantity_gate2.py b/client/tests/pdd/test_quantity_gate2.py index c1367cf..35b9b17 100644 --- a/client/tests/pdd/test_quantity_gate2.py +++ b/client/tests/pdd/test_quantity_gate2.py @@ -6,6 +6,7 @@ import argparse import ast import base64 from contextlib import redirect_stderr +from datetime import UTC, datetime from functools import lru_cache from importlib.util import module_from_spec, spec_from_file_location from io import BytesIO, StringIO @@ -33,6 +34,14 @@ from cmbuyer_client.pdd.quantity_gate2_spike import ( QuantityGate2ForegroundReader, QuantityGate2ReadDevice, ) +from cmbuyer_client.pdd.quantity_gate2 import ( + Gate1Observation, + QuantityGate2Device, + QuantityGate2Error, + QuantityGate2Flow, + QuantityGate2OverCapError, +) +from cmbuyer_client.pdd.quantity_gate2_runner import QuantityGate2Runner SERIAL = "192.168.0.173:5555" @@ -144,6 +153,15 @@ def _load_script() -> object: return module +def _load_run_script() -> object: + script_path = CLIENT_ROOT / "scripts" / "run_t105_quantity_gate2.py" + spec = spec_from_file_location("run_t105_quantity_gate2_for_test", script_path) + assert spec is not None and spec.loader is not None + module = module_from_spec(spec) + spec.loader.exec_module(module) + return module + + def _namespace(**changes: object) -> argparse.Namespace: values: dict[str, object] = { "serial": SERIAL, @@ -555,5 +573,372 @@ class QuantityGate2CliAndStaticBoundaryTests(unittest.TestCase): self.assertNotIn(forbidden_import, source.read_text(encoding="utf-8")) +_INITIAL_FIXTURE = CLIENT_ROOT / "tests" / "pdd" / "fixtures" / "quantity_gate2_initial_8_17_0.xml" +_TARGET_FIXTURE = CLIENT_ROOT / "tests" / "pdd" / "fixtures" / "quantity_gate2_target_8_17_0.xml" +_EXIT_FIXTURE = CLIENT_ROOT / "tests" / "pdd" / "fixtures" / "product_exit_8_17_0.xml" + + +def _gate1(path: Path = Path("gate1.png")) -> Gate1Observation: + return Gate1Observation( + color="黑色CHA(纯棉)", + size="M(建议100-115)", + quantity=1, + gate1_unit_price="12.88", + screenshot_path=path, + captured_at=datetime(2026, 8, 6, 0, 53, tzinfo=UTC), + ) + + +class FakeQuantityFlowDevice: + def __init__(self, initial: str | None = None, target: str | None = None) -> None: + self.initial = initial or _INITIAL_FIXTURE.read_text(encoding="utf-8") + self.target = target or _TARGET_FIXTURE.read_text(encoding="utf-8") + self.exit = _EXIT_FIXTURE.read_text(encoding="utf-8") + self.current = self.initial + self.version = "8.17.0" + self.foreground = {"package": "com.xunmeng.pinduoduo", "activity": ".activity.NewPageActivity"} + self.screen_size = (1080, 2376) + self.increment_calls: list[str] = [] + self.back_calls = 0 + self.dump_calls = 0 + self.raise_increment: BaseException | None = None + + def app_info(self, package_name: str) -> dict[str, str]: + return {"versionName": self.version} + + def current_foreground(self) -> dict[str, str]: + return dict(self.foreground) + + def display_size(self) -> tuple[int, int]: + return self.screen_size + + def dump_window_hierarchy(self) -> str: + self.dump_calls += 1 + return self.current + + def increment_quantity_once(self, bounds: str) -> None: + self.increment_calls.append(bounds) + if self.raise_increment is not None: + raise self.raise_increment + self.current = self.target + + def capture_screenshot(self) -> str: + return _png_base64() + + def leave_sku_panel_once(self) -> None: + self.back_calls += 1 + self.current = self.exit + + +class QuantityGate2FlowTests(unittest.TestCase): + def test_quantity_two_reads_nonlinear_panel_total_and_exits_once(self) -> None: + device = FakeQuantityFlowDevice() + flow = QuantityGate2Flow(device, wait_timeout_seconds=0.1, poll_interval_seconds=0.01) + verified = flow.set_quantity_and_verify(_gate1(), 2, "40.00") + observation = flow.build_observation( + _gate1(), + 2, + "40.00", + Path("gate2.png"), + datetime(2026, 8, 6, 1, 10, tzinfo=UTC), + ) + flow.exit_sku_panel_safely() + + self.assertEqual(verified.quantity, 2) + self.assertEqual(observation.quantity_read, 2) + self.assertEqual(observation.gate1_unit_price, "12.88") + self.assertEqual(observation.gate2_panel_total_price, "32.76") + self.assertEqual(observation.max_total_price, "40.00") + self.assertEqual(device.increment_calls, ["[567,752][645,827]"]) + self.assertEqual(device.back_calls, 1) + self.assertTrue(flow.exited) + + def test_quantity_one_is_zero_click_and_still_uses_panel_amount(self) -> None: + device = FakeQuantityFlowDevice() + flow = QuantityGate2Flow(device) + verified = flow.set_quantity_and_verify(_gate1(), 1, "12.88") + self.assertEqual((verified.quantity, verified.panel_total_price), (1, "12.88")) + self.assertEqual(device.increment_calls, []) + + def test_only_evidenced_quantities_and_canonical_decimal_cap_are_allowed(self) -> None: + for quantity in (0, 3, -1, True, "2"): + with self.subTest(quantity=quantity): + device = FakeQuantityFlowDevice() + with self.assertRaises(QuantityGate2Error): + QuantityGate2Flow(device).set_quantity_and_verify(_gate1(), quantity, "40.00") # type: ignore[arg-type] + self.assertEqual(device.increment_calls, []) + for cap in ("0.00", "40", "040.00", 40.0, "NaN"): + with self.subTest(cap=cap): + device = FakeQuantityFlowDevice() + with self.assertRaises(QuantityGate2Error): + QuantityGate2Flow(device).set_quantity_and_verify(_gate1(), 2, cap) # type: ignore[arg-type] + self.assertEqual(device.increment_calls, []) + + def test_over_cap_stops_after_exact_readback_without_fabricated_math(self) -> None: + device = FakeQuantityFlowDevice() + flow = QuantityGate2Flow(device) + with self.assertRaises(QuantityGate2OverCapError): + flow.set_quantity_and_verify(_gate1(), 2, "30.00") + self.assertEqual(device.increment_calls, ["[567,752][645,827]"]) + self.assertTrue(flow.can_exit_safely) + + def test_increment_timeout_is_sealed_and_cannot_be_retried(self) -> None: + device = FakeQuantityFlowDevice() + device.raise_increment = TimeoutError("ambiguous delivery") + flow = QuantityGate2Flow(device, wait_timeout_seconds=0.01, poll_interval_seconds=0.005) + with self.assertRaises(TimeoutError): + flow.set_quantity_and_verify(_gate1(), 2, "40.00") + with self.assertRaises(QuantityGate2Error): + flow.set_quantity_and_verify(_gate1(), 2, "40.00") + self.assertEqual(len(device.increment_calls), 1) + self.assertFalse(flow.can_exit_safely) + + def test_missing_duplicate_drift_and_overlay_fail_before_click(self) -> None: + initial = _INITIAL_FIXTURE.read_text(encoding="utf-8") + plus = '' + variants = ( + initial.replace('content-desc="增加数量"', 'content-desc="数量加一"', 1), + initial.replace(plus, plus + plus), + initial.replace('text="M（建议100-115）"', 'text="S（建议80-100）"', 1), + initial.replace( + "", + '', + ), + ) + for hierarchy in variants: + with self.subTest(marker=hierarchy[-180:]): + device = FakeQuantityFlowDevice(initial=hierarchy) + with self.assertRaises(QuantityGate2Error): + QuantityGate2Flow(device).set_quantity_and_verify(_gate1(), 2, "40.00") + self.assertEqual(device.increment_calls, []) + + def test_foreground_version_and_screen_drift_are_zero_click(self) -> None: + devices = (FakeQuantityFlowDevice(), FakeQuantityFlowDevice(), FakeQuantityFlowDevice()) + devices[0].foreground = {"package": "com.android.systemui", "activity": ".Keyguard"} + devices[1].version = "8.17.1" + devices[2].screen_size = (1080, 2400) + for device in devices: + with self.subTest(device=device): + with self.assertRaises(QuantityGate2Error): + QuantityGate2Flow(device).set_quantity_and_verify(_gate1(), 2, "40.00") + self.assertEqual(device.increment_calls, []) + + def test_bottom_submit_amount_is_never_a_gate2_candidate(self) -> None: + target = _TARGET_FIXTURE.read_text(encoding="utf-8").replace("提交订单 ¥32.76", "提交订单 ¥999.99") + device = FakeQuantityFlowDevice(target=target) + result = QuantityGate2Flow(device).set_quantity_and_verify(_gate1(), 2, "40.00") + self.assertEqual(result.panel_total_price, "32.76") + + def test_gate1_target_spec_and_timestamp_are_strict(self) -> None: + invalid = ( + {"color": "白色"}, + {"size": "L"}, + {"quantity": 2}, + {"gate1_unit_price": "12.89"}, + {"captured_at": datetime(2026, 8, 6, 0, 53)}, + ) + base = _gate1().__dict__ + for change in invalid: + with self.subTest(change=change), self.assertRaises(QuantityGate2Error): + Gate1Observation(**{**base, **change}) + + +class FakeRawQuantityDevice: + def __init__(self, *, click_timeout: bool = False) -> None: + self.current = _INITIAL_FIXTURE.read_text(encoding="utf-8") + self.target = _TARGET_FIXTURE.read_text(encoding="utf-8") + self.exit = _EXIT_FIXTURE.read_text(encoding="utf-8") + self.calls: list[tuple[object, ...]] = [] + self.click_timeout = click_timeout + + def app_info(self, package: str) -> dict[str, str]: + self.calls.append(("app_info", package)) + return {"versionName": "8.17.0"} + + def window_size(self) -> tuple[int, int]: + self.calls.append(("window_size",)) + return (1080, 2376) + + def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> object: + self.calls.append(("jsonrpc", method, params, timeout)) + if method == "dumpWindowHierarchy": + return self.current + if method == "click": + self.current = self.target + if self.click_timeout: + raise TimeoutError("may have been delivered") + return None + if method == "takeScreenshot": + return _png_base64() + if method == "pressKey": + self.current = self.exit + return None + raise AssertionError(method) + + +class QuantityGate2RunnerTests(unittest.TestCase): + def test_runner_publishes_gate2_evidence_then_one_safe_exit(self) -> None: + with TemporaryDirectory() as temporary: + base = Path(temporary) + gate1_path = base / "gate1.png" + gate1_path.write_bytes(base64.b64decode(_png_base64())) + target = base / "gate2" + raw = FakeRawQuantityDevice() + result = QuantityGate2Runner( + FakeAdbClient(), + lambda serial: raw, + FakeForegroundReader(), + timeout_seconds=0.1, + ).run(SERIAL, EXPECTED_GOODS_ID, _gate1(gate1_path), 2, "40.00", target) + + manifest = json.loads(result.manifest_path.read_text(encoding="utf-8")) + methods = [call[1] for call in raw.calls if call[0] == "jsonrpc"] + self.assertEqual(methods.count("click"), 1) + self.assertEqual(methods.count("pressKey"), 1) + self.assertEqual(methods.count("takeScreenshot"), 1) + self.assertEqual(result.observation.gate2_panel_total_price, "32.76") + self.assertEqual(manifest["prices"]["gate1_unit_price"], "12.88") + self.assertEqual(manifest["prices"]["gate2_panel_total_price"], "32.76") + self.assertEqual(manifest["prices"]["max_total_price"], "40.00") + self.assertEqual(manifest["action_audit"]["increment_attempts"], 1) + self.assertEqual(manifest["action_audit"]["back_attempts"], 1) + self.assertEqual(manifest["safe_exit"], "completed") + self.assertEqual(manifest["review_status"], "human_review_required") + serialized = result.manifest_path.read_text(encoding="utf-8") + self.assertNotIn(SERIAL, serialized) + self.assertNotIn(str(gate1_path), serialized) + + def test_over_cap_attempts_one_safe_exit_and_publishes_nothing(self) -> None: + with TemporaryDirectory() as temporary: + base = Path(temporary) + gate1_path = base / "gate1.png" + gate1_path.write_bytes(base64.b64decode(_png_base64())) + raw = FakeRawQuantityDevice() + target = base / "gate2" + with self.assertRaises(QuantityGate2OverCapError): + QuantityGate2Runner( + FakeAdbClient(), lambda serial: raw, FakeForegroundReader(), timeout_seconds=0.1 + ).run(SERIAL, EXPECTED_GOODS_ID, _gate1(gate1_path), 2, "30.00", target) + methods = [call[1] for call in raw.calls if call[0] == "jsonrpc"] + self.assertEqual(methods.count("click"), 1) + self.assertEqual(methods.count("pressKey"), 1) + self.assertFalse(target.exists()) + self.assertEqual(list(base.glob(".gate2.staging-*")), []) + + def test_ambiguous_increment_is_never_retried_or_followed_by_back(self) -> None: + with TemporaryDirectory() as temporary: + base = Path(temporary) + gate1_path = base / "gate1.png" + gate1_path.write_bytes(base64.b64decode(_png_base64())) + raw = FakeRawQuantityDevice(click_timeout=True) + with self.assertRaises(QuantityGate2Error): + QuantityGate2Runner( + FakeAdbClient(), lambda serial: raw, FakeForegroundReader(), timeout_seconds=0.1 + ).run(SERIAL, EXPECTED_GOODS_ID, _gate1(gate1_path), 2, "40.00", base / "gate2") + methods = [call[1] for call in raw.calls if call[0] == "jsonrpc"] + self.assertEqual(methods.count("click"), 1) + self.assertEqual(methods.count("pressKey"), 0) + + +class QuantityGate2ProductionStaticBoundaryTests(unittest.TestCase): + def test_protocol_has_only_named_t105_actions(self) -> None: + public = {name for name in QuantityGate2Device.__dict__ if not name.startswith("_")} + self.assertEqual( + public, + { + "app_info", + "current_foreground", + "display_size", + "dump_window_hierarchy", + "increment_quantity_once", + "capture_screenshot", + "leave_sku_panel_once", + }, + ) + + def test_production_sources_have_no_old_gate2_field_or_forbidden_capability(self) -> None: + paths = ( + CLIENT_ROOT / "src/cmbuyer_client/pdd/quantity_gate2.py", + CLIENT_ROOT / "src/cmbuyer_client/pdd/quantity_gate2_runner.py", + CLIENT_ROOT / "scripts/run_t105_quantity_gate2.py", + ) + forbidden_attributes = { + "go_to_order_confirm", + "submit_order_once", + "start_pdd_view_intent", + "pay", + "swipe", + "set_text", + "send_keys", + } + for path in paths: + source = path.read_text(encoding="utf-8") + tree = ast.parse(source) + self.assertNotIn("gate2_unit_price", source) + self.assertFalse( + { + node.attr + for node in ast.walk(tree) + if isinstance(node, ast.Attribute) and node.attr in forbidden_attributes + } + ) + quantity_tree = ast.parse(paths[0].read_text(encoding="utf-8")) + self.assertFalse(any(isinstance(node, ast.Mult) for node in ast.walk(quantity_tree))) + + def test_t103_and_t104_still_do_not_import_t105(self) -> None: + for relative in ( + "src/cmbuyer_client/pdd/sku_selection.py", + "src/cmbuyer_client/pdd/sku_selection_runner.py", + ): + self.assertNotIn("quantity_gate2", (CLIENT_ROOT / relative).read_text(encoding="utf-8")) + + def test_live_cli_fixes_the_human_review_case_and_redacts_runtime_failure(self) -> None: + script = _load_run_script() + with TemporaryDirectory() as temporary: + screenshot = Path(temporary) / "gate1.png" + screenshot.write_bytes(base64.b64decode(_png_base64())) + arguments = script.parse_arguments( # type: ignore[attr-defined] + [ + "--serial", SERIAL, + "--goods-id", EXPECTED_GOODS_ID, + "--color", "黑色CHA(纯棉)", + "--size", "M(建议100-115)", + "--target-quantity", "2", + "--gate1-unit-price", "12.88", + "--gate1-screenshot", str(screenshot), + "--gate1-captured-at", "2026-08-06T00:53:00.023935+00:00", + "--max-total-price", "40.00", + "--output-dir", str(Path(temporary) / "gate2"), + ] + ) + script.validate_arguments(arguments) # type: ignore[attr-defined] + + stderr = StringIO() + fake_runner = unittest.mock.Mock() + fake_runner.run.side_effect = QuantityGate2Error(f"secret {SERIAL} {temporary}") + with ( + patch.object(script, "QuantityGate2Runner", return_value=fake_runner), + patch.dict(sys.modules, {"adbutils": unittest.mock.Mock(), "uiautomator2": unittest.mock.Mock()}), + redirect_stderr(stderr), + ): + result = script.main( # type: ignore[attr-defined] + [ + "--serial", SERIAL, + "--goods-id", EXPECTED_GOODS_ID, + "--color", "黑色CHA(纯棉)", + "--size", "M(建议100-115)", + "--target-quantity", "2", + "--gate1-unit-price", "12.88", + "--gate1-screenshot", str(screenshot), + "--gate1-captured-at", "2026-08-06T00:53:00.023935+00:00", + "--max-total-price", "40.00", + "--output-dir", str(Path(temporary) / "gate2"), + ] + ) + self.assertEqual(result, 1) + self.assertNotIn(SERIAL, stderr.getvalue()) + self.assertNotIn(temporary, stderr.getvalue()) + + if __name__ == "__main__": unittest.main()