From e8ca738b33b116ee0ea33c322820595f7a9fc3ef Mon Sep 17 00:00:00 2001 From: QiuSW <105186638@qq.com> Date: Wed, 5 Aug 2026 12:34:59 +0800 Subject: [PATCH] feat(client): capture one-shot sku reveal evidence --- client/scripts/capture_sku_reveal_spike.py | 89 +++ .../cmbuyer_client/pdd/sku_reveal_spike.py | 436 ++++++++++++ .../src/cmbuyer_client/pdd/sku_selection.py | 621 ++++++++++++++---- ...anel_color_selected_size_hidden_8_17_0.xml | 52 ++ .../pdd/fixtures/sku_panel_empty_8_17_0.xml | 52 ++ .../sku_panel_size_m_restored_8_17_0.xml | 58 ++ .../sku_panel_size_s_selected_8_17_0.xml | 58 ++ client/tests/pdd/test_sku_reveal_spike.py | 396 +++++++++++ client/tests/pdd/test_sku_selection.py | 234 +++++-- docs/tasks/T-103.md | 6 +- 10 files changed, 1822 insertions(+), 180 deletions(-) create mode 100644 client/scripts/capture_sku_reveal_spike.py create mode 100644 client/src/cmbuyer_client/pdd/sku_reveal_spike.py create mode 100644 client/tests/pdd/fixtures/sku_panel_color_selected_size_hidden_8_17_0.xml create mode 100644 client/tests/pdd/fixtures/sku_panel_empty_8_17_0.xml create mode 100644 client/tests/pdd/fixtures/sku_panel_size_m_restored_8_17_0.xml create mode 100644 client/tests/pdd/fixtures/sku_panel_size_s_selected_8_17_0.xml create mode 100644 client/tests/pdd/test_sku_reveal_spike.py diff --git a/client/scripts/capture_sku_reveal_spike.py b/client/scripts/capture_sku_reveal_spike.py new file mode 100644 index 0000000..2b40e90 --- /dev/null +++ b/client/scripts/capture_sku_reveal_spike.py @@ -0,0 +1,89 @@ +"""采集 T-103 一次性尺码 reveal 的 before/after 本机证据。""" + +from __future__ import annotations + +import argparse +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.sku_reveal_spike import SkuRevealSpikeCapturer, SkuRevealSpikeError +from cmbuyer_client.pdd.sku_selection import EXPECTED_GOODS_ID, SkuSelectionError +from cmbuyer_client.pdd.sku_selection_runner import SkuSelectionRunError + + +def parse_arguments(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description="采集 T-103 一次性尺码 reveal 的本机证据。") + parser.add_argument("--serial", required=True, help="ADB device serial；禁止自动选择。") + parser.add_argument("--goods-id", required=True, help="固定 T-103 已取证 goods_id。") + parser.add_argument("--output-dir", required=True, type=Path, help="全新本机证据目录；不得覆盖。") + parser.add_argument("--timeout", type=float, default=30.0, help="整趟动作与只读调和时限（秒）。") + parser.add_argument("--adb", default="adb", help="adb 可执行文件路径。") + return parser.parse_args(argv) + + +def validate_arguments(arguments: argparse.Namespace) -> None: + if type(arguments.serial) is not str or not arguments.serial.strip(): + raise ValueError("必须显式提供非空 --serial。") + if type(arguments.goods_id) is not str or arguments.goods_id != EXPECTED_GOODS_ID: + raise ValueError("--goods-id 不是 T-103 已取证商品。") + 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 的有限数值。") + + +def main(argv: list[str] | None = None) -> int: + arguments = parse_arguments(argv) + try: + 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("失败：缺少 uiautomator2；请在采购工具虚拟环境中运行。", file=sys.stderr) + return 2 + + capturer = SkuRevealSpikeCapturer( + AdbClient(SubprocessAdbRunner(arguments.adb), timeout_seconds=arguments.timeout), + NoReconnectUiautomatorConnector( + adbutils.AdbClient(socket_timeout=arguments.timeout).device_list, + u2.connect, + ), + timeout_seconds=arguments.timeout, + ) + try: + result = capturer.capture( + arguments.serial, + arguments.goods_id, + arguments.output_dir, + ) + except (DeviceConnectionError, SkuSelectionError, SkuSelectionRunError, SkuRevealSpikeError): + # 不回显页面正文、节点、serial、坐标、路径或第三方异常。 + print("规格 reveal 取证失败：已停止，未发布本地证据目录。", file=sys.stderr) + return 1 + except OSError: + print("规格 reveal 取证失败：无法创建或发布本地证据目录。", file=sys.stderr) + return 1 + + print(f"规格 reveal 取证完成：{result.output_directory}") + print(f"manifest：{result.manifest_path}") + print("人工复核：请保持手机不动，确认颜色仍选中、S/M 均未选且未进入提交页。") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/client/src/cmbuyer_client/pdd/sku_reveal_spike.py b/client/src/cmbuyer_client/pdd/sku_reveal_spike.py new file mode 100644 index 0000000..589fca6 --- /dev/null +++ b/client/src/cmbuyer_client/pdd/sku_reveal_spike.py @@ -0,0 +1,436 @@ +"""T-103 尺码显示动作的一次性真机取证；不属于生产采购 Flow。""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +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, sleep +from typing import Any +from uuid import uuid4 + +from ..device.adb import AdbClient, DeviceConnectionError, DeviceInspection +from ..device.baseline import PDD_PACKAGE, _save_base64_screenshot, _sha256_file +from .product_url import parse_product_url +from .sku_selection import ( + EXPECTED_GOODS_ID, + _COLOR_ONLY_SUMMARY, + _PANEL_SURFACE, + _PanelProfile, + _REVEAL_NOT_PROVEN, + _S_SIZE_UI, + _TARGET_COLOR_UI, + _TARGET_SIZE_UI, + _action_bounds, + _clickable_before, + _descendant, + _exact_color_action, + _exact_inert, + _exact_live_layout, + _exact_readonly_text, + _exact_recycler, + _is_size_action_text, + _one, + _parse_nodes, + _require_color_only_panel, + _require_color_action_chain, + _require_color_subtree, + _require_exact_selected_set, + _require_panel_chain, + _require_size_wrapper, + _require_size_action_chain, + _spec_for, + resolve_task_selection, + SkuSelectionError, + SkuSelectionFlow, +) +from .sku_selection_runner import ( + EXPECTED_ANDROID_VERSION, + EXPECTED_DEVICE_MODEL, + EXPECTED_SCREEN_SIZE, + SkuSelectionRunError, + UiautomatorSkuPanelAdapter, + _require_expected_device, + _require_expected_version, + _require_screenshot_size, +) + + +_TARGET_URL = f"https://mobile.yangkeduo.com/goods.html?goods_id={EXPECTED_GOODS_ID}" +_REVEAL_START = (360, 1900) +_REVEAL_END = (360, 1300) +_REVEAL_STEPS = 30 + + +class SkuRevealSpikeError(RuntimeError): + """一次性 reveal 取证未形成可发布证据。""" + + +@dataclass(frozen=True) +class SkuRevealSpikeResult: + output_directory: Path + manifest_path: Path + + +class _RevealEvidenceAdapter(UiautomatorSkuPanelAdapter): + """只为 spike 增加一个无参数、固定 profile 的一次性手势。""" + + def __init__(self, device: Any, timeout_seconds: float) -> None: + super().__init__(device, timeout_seconds) + self._reveal_attempted = False + + @property + def reveal_attempted(self) -> bool: + return self._reveal_attempted + + def reveal_size_options_once(self) -> None: + if self._reveal_attempted: + raise SkuRevealSpikeError("规格显示动作已经尝试过，拒绝重试。") + # RPC 超时也可能表示手势已经送达，必须在调用前封存唯一机会。 + self._reveal_attempted = True + self._call( + "jsonrpc_call", + "swipe", + [*_REVEAL_START, *_REVEAL_END, _REVEAL_STEPS], + timeout=self._timeout_seconds, + ) + + +class SkuRevealSpikeCapturer: + """打开已取证面板、选一次目标颜色，再采集唯一 reveal 的前后证据。""" + + def __init__( + self, + adb_client: AdbClient, + connector: Callable[[str], Any], + timeout_seconds: float, + monotonic_clock: Callable[[], float] = monotonic, + sleep_function: Callable[[float], None] = sleep, + ) -> None: + if not _positive_finite(timeout_seconds): + raise ValueError("timeout_seconds 必须是大于 0 的有限数值") + self._adb_client = adb_client + self._connector = connector + self._timeout_seconds = timeout_seconds + self._clock = monotonic_clock + self._sleep = sleep_function + + def capture( + self, + serial: str, + goods_id: str, + output_directory: Path, + ) -> SkuRevealSpikeResult: + if type(goods_id) is not str or goods_id != EXPECTED_GOODS_ID: + raise SkuRevealSpikeError("商品不是 T-103 已取证目标，已停止取证。") + link = parse_product_url(_TARGET_URL) + target = Path(output_directory) + _validate_new_target(target) + staging: Path | None = None + try: + staging = _prepare_staging(target) + deadline = self._clock() + self._timeout_seconds + inspection = self._adb_client.inspect(serial) + _require_expected_device(inspection) + adapter = _RevealEvidenceAdapter(self._connector(serial), self._timeout_seconds) + _require_expected_version(adapter.app_info(PDD_PACKAGE)) + if adapter.display_size() != EXPECTED_SCREEN_SIZE: + raise SkuRevealSpikeError("设备不是已取证的竖屏坐标空间，已停止取证。") + pre_intent = adapter.dump_window_hierarchy() + self._adb_client.start_pdd_view_intent(serial, link.goods_id) + + remaining = deadline - self._clock() + if remaining <= 0: + raise SkuRevealSpikeError("规格入口取证超时，未执行 reveal。") + flow = SkuSelectionFlow( + adapter, + entry_wait_timeout_seconds=remaining, + monotonic_clock=self._clock, + sleep_function=self._sleep, + ) + flow.open_sku_panel(link.canonical_url, pre_intent) + try: + flow.select_sku_options(resolve_task_selection("黑色CHA(纯棉)", "M(建议100-115)")) + except SkuSelectionError as error: + if error.args != (_REVEAL_NOT_PROVEN,): + raise + else: + raise SkuRevealSpikeError("规格流程未停在已取证的仅颜色状态。") + + before_hierarchy = adapter.dump_window_hierarchy() + before_nodes = _parse_nodes(before_hierarchy) + _require_color_only_panel(before_nodes) + _require_safe_reveal_path(before_nodes) + _capture_frame(adapter, staging / "before", before_hierarchy) + # 截图 RPC 期间页面也可能变化；真正发送手势前必须用新树再次证明同一前置与安全通道。 + before_hierarchy = adapter.dump_window_hierarchy() + before_nodes = _parse_nodes(before_hierarchy) + _require_color_only_panel(before_nodes) + _require_safe_reveal_path(before_nodes) + _require_expected_version(adapter.app_info(PDD_PACKAGE)) + current = adapter.app_current() + if not isinstance(current, dict) or current.get("package") != PDD_PACKAGE: + raise SkuRevealSpikeError("reveal 前拼多多不在前台，未执行手势。") + (staging / "before" / "hierarchy.xml").write_text(before_hierarchy, encoding="utf-8") + + rpc_outcome = "completed" + try: + adapter.reveal_size_options_once() + except SkuSelectionRunError: + rpc_outcome = "ambiguous_reconciled" + + projection, after_hierarchy = self._wait_for_candidate(adapter, deadline) + after_directory = staging / "after" + _capture_frame(adapter, after_directory, after_hierarchy) + reverified = adapter.dump_window_hierarchy() + if _candidate_projection(_parse_nodes(reverified)) != projection: + raise SkuRevealSpikeError("截图后候选状态漂移，未发布证据。") + (after_directory / "hierarchy.xml").write_text(reverified, encoding="utf-8") + + manifest = _manifest(inspection, serial, rpc_outcome, staging) + manifest_path = staging / "manifest.json" + manifest_path.write_text( + json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + os.rename(staging, target) + staging = None + except (DeviceConnectionError, SkuSelectionError, SkuSelectionRunError, SkuRevealSpikeError): + _clean_staging(staging) + raise + except Exception as error: + _clean_staging(staging) + raise SkuRevealSpikeError("规格 reveal 取证未完成，未发布本地证据目录。") from error + + return SkuRevealSpikeResult(target, target / "manifest.json") + + def _wait_for_candidate( + self, + adapter: _RevealEvidenceAdapter, + deadline: float, + ) -> tuple[tuple[tuple[str, ...], ...], str]: + stable: tuple[tuple[str, ...], ...] | None = None + while True: + _require_expected_version(adapter.app_info(PDD_PACKAGE)) + current = adapter.app_current() + if not isinstance(current, dict) or current.get("package") != PDD_PACKAGE: + raise SkuRevealSpikeError("reveal 后拼多多不在前台，未发布证据。") + hierarchy = adapter.dump_window_hierarchy() + try: + projection = _candidate_projection(_parse_nodes(hierarchy)) + except SkuSelectionError: + projection = None + if projection is not None and projection == stable: + return projection, hierarchy + stable = projection + remaining = deadline - self._clock() + if remaining <= 0: + raise SkuRevealSpikeError("reveal 后未形成稳定候选状态，未发布证据。") + self._sleep(min(0.2, remaining)) + + +def _require_safe_reveal_path(nodes: list[Any]) -> None: + # 当前真机证据证明 x=360 是两列规格卡之间的空隙；必须验证完整线段，离散采样会漏掉窄浮层。 + surface = _one( + [node for node in nodes if _exact_inert(node, "android.view.ViewGroup", _PANEL_SURFACE)], + "固定 reveal 通道无法绑定面板内容面。", + ) + _require_panel_chain(surface) + content = surface.parent + action_root = content.parent if content is not None else None + action_parent = action_root.parent if action_root is not None else None + if ( + content is None + or action_root is None + or action_parent is None + or not _exact_live_layout(action_root, "android.view.ViewGroup", "[0,366][1080,2328]") + or not _exact_live_layout(action_parent, "android.widget.LinearLayout", "[0,120][1080,2328]") + ): + raise SkuRevealSpikeError("固定 reveal 通道祖先身份漂移，未执行手势。") + allowed = {id(action_root.element), id(action_parent.element)} + occupants: set[int] = set() + for node in nodes: + if node.element.get("clickable") != "true": + continue + try: + left, top, right, bottom = _action_bounds(node.bounds) + except SkuSelectionError as error: + raise SkuRevealSpikeError("可点击节点坐标不可验证，未执行手势。") from error + if ( + left <= _REVEAL_START[0] < right + and top <= _REVEAL_START[1] + and bottom > _REVEAL_END[1] + ): + occupants.add(id(node.element)) + if occupants != allowed: + raise SkuRevealSpikeError("固定 reveal 通道被未取证可点击节点占用，未执行手势。") + if _REVEAL_START[1] >= 2079 or _REVEAL_END[1] >= 2079: + raise SkuRevealSpikeError("固定 reveal 通道越过规格内容区，未执行手势。") + + +def _candidate_projection(nodes: list[Any]) -> tuple[tuple[str, ...], ...]: + surface = _one( + [node for node in nodes if _exact_inert(node, "android.view.ViewGroup", _PANEL_SURFACE)], + "候选面板内容面不唯一。", + ) + _require_panel_chain(surface) + header = _one( + [node for node in nodes if node.parent is surface and _exact_inert(node, "android.widget.LinearLayout", "[0,366][1080,1000]")], + "候选面板头部不唯一。", + ) + outer = _one( + [node for node in nodes if node.parent is surface and _exact_recycler(node, "[0,1000][1080,2079]")], + "候选维度容器不唯一。", + ) + price_row = _one( + [node for node in nodes if _descendant(node, header) and _exact_inert(node, "android.widget.LinearLayout", "[396,498][895,570]")], + "候选价格行不唯一。", + ) + current = _one( + [node for node in nodes if node.parent is price_row and _exact_readonly_text(node, "快卖完 ¥12.88", "[396,503][712,570]")], + "候选当前价不唯一。", + ) + original = _one( + [node for node in nodes if node.parent is price_row and _exact_readonly_text(node, "¥29.88", "[730,503][895,570]")], + "候选原价不唯一。", + ) + if _clickable_before(current, surface): + raise SkuSelectionError("候选当前价位于可点击内容祖先下。") + summary = _one( + [node for node in nodes if _descendant(node, header) and _exact_readonly_text(node, _COLOR_ONLY_SUMMARY, "[396,654][1053,716]")], + "候选摘要不唯一。", + ) + color_region = _one( + [node for node in nodes if _descendant(node, outer) and _exact_recycler(node, "[36,1000][1080,1483]")], + "候选颜色容器不唯一。", + ) + color = _one( + [node for node in nodes if node.parent is color_region and _exact_color_action(node, "[372,1000][684,1024]", True)], + "候选目标颜色不唯一。", + ) + rolled_spec = _spec_for(_PanelProfile.TARGETS_SELECTED) + selected_color_nodes = _require_color_subtree(color, rolled_spec) + _require_color_action_chain(color, color_region, outer, surface, rolled_spec) + size_label = _one( + [node for node in nodes if _descendant(node, outer) and _exact_readonly_text(node, "尺码", "[36,1506][114,1552]")], + "候选尺码标题不唯一。", + ) + size_header = size_label.parent + if size_header is None or not _exact_live_layout(size_header, "android.widget.LinearLayout", "[36,1489][1044,1570]"): + raise SkuSelectionError("候选尺码标题父结构漂移。") + size_options = _one( + [node for node in nodes if _descendant(node, outer) and _exact_inert(node, "android.view.ViewGroup", "[36,1582][1044,1897]")], + "候选尺码 options 根不唯一。", + ) + actions = [node for node in nodes if _descendant(node, size_options) and _is_size_action_text(node)] + s_action = _one( + [node for node in actions if node.text == _S_SIZE_UI and node.bounds == "[36,1582][409,1667]"], + "候选 S action 不唯一。", + ) + m_action = _one( + [node for node in actions if node.text == _TARGET_SIZE_UI and node.bounds == "[439,1582][831,1667]"], + "候选 M action 不唯一。", + ) + _require_size_wrapper(s_action, size_options) + _require_size_wrapper(m_action, size_options) + _require_size_action_chain(s_action, size_options, outer, surface) + _require_size_action_chain(m_action, size_options, outer, surface) + if any(node.element.get("selected") == "true" for node in actions): + raise SkuSelectionError("reveal 候选态已有尺码被选中。") + _require_exact_selected_set(nodes, surface, selected_color_nodes) + dangerous = [ + node for node in nodes + if "提交订单" in node.text + and node.element.get("package") == PDD_PACKAGE + and node.element.get("class") == "android.widget.TextView" + ] + if len(dangerous) != 1 or _action_bounds(dangerous[0].bounds)[1] < 2079: + raise SkuSelectionError("提交硬拒绝区位置不唯一。") + return tuple( + ( + node.element.get("class", ""), + node.bounds, + node.text, + node.desc, + node.element.get("selected", ""), + node.element.get("clickable", ""), + ) + for node in (surface, header, outer, price_row, current, original, summary, color_region, color, size_label, size_options, s_action, m_action, dangerous[0]) + ) + + +def _capture_frame(adapter: _RevealEvidenceAdapter, directory: Path, hierarchy: str) -> None: + directory.mkdir() + hierarchy_path = directory / "hierarchy.xml" + hierarchy_path.write_text(hierarchy, encoding="utf-8") + screenshot_path = directory / "screenshot.png" + _save_base64_screenshot(adapter.capture_screenshot(), screenshot_path) + _require_screenshot_size(screenshot_path) + + +def _manifest( + inspection: DeviceInspection, + serial: str, + rpc_outcome: str, + staging: Path, +) -> dict[str, Any]: + artifacts = [] + for relative in ( + "before/screenshot.png", + "before/hierarchy.xml", + "after/screenshot.png", + "after/hierarchy.xml", + ): + path = staging / relative + artifacts.append({"path": relative, "sha256": _sha256_file(path)}) + return { + "schema_version": 1, + "captured_at": datetime.now(UTC).isoformat(), + "operation": "t103-sku-reveal-evidence", + "profile_id": "pdd-8.17.0-size-reveal-gap-v1", + "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": "8.17.0", + }, + "reveal_attempts": 1, + "rpc_outcome": rpc_outcome, + "candidate_status": "human_review_required", + "artifacts": artifacts, + } + + +def _validate_new_target(target: Path) -> None: + if target.exists() or not target.name: + raise SkuRevealSpikeError("输出目录必须是不存在的明确新目录。") + + +def _prepare_staging(target: Path) -> Path: + target.parent.mkdir(parents=True, exist_ok=True) + staging = target.parent / f".{target.name}.staging-{uuid4().hex}" + staging.mkdir() + return staging + + +def _clean_staging(staging: Path | None) -> None: + if staging is not None and staging.exists(): + shutil.rmtree(staging) + + +def _positive_finite(value: object) -> bool: + return ( + isinstance(value, (int, float)) + and not isinstance(value, bool) + and value > 0 + and isfinite(value) + ) diff --git a/client/src/cmbuyer_client/pdd/sku_selection.py b/client/src/cmbuyer_client/pdd/sku_selection.py index 21e3ee5..362e993 100644 --- a/client/src/cmbuyer_client/pdd/sku_selection.py +++ b/client/src/cmbuyer_client/pdd/sku_selection.py @@ -2,6 +2,7 @@ from __future__ import annotations from dataclasses import dataclass +from enum import Enum import re from time import monotonic, sleep from typing import Any, Callable, Protocol @@ -29,17 +30,16 @@ _FORBIDDEN_ENTRY_ACTION_DESC = ( ) _SIZE = "尺码" _W, _H = 1080, 2376 -_PRICE_PARENT = "[396,498][895,570]" -_CURRENT = "[396,503][712,570]" -_ORIGINAL = "[730,503][895,570]" -_SUMMARY = "[396,654][1053,716]" -_COLOR_REGION = "[36,1000][1080,1631]" -_SIZE_LABEL = "[36,1654][114,1700]" -_SIZE_HEADER = "[36,1637][1044,1718]" -_SIZE_OPTIONS = "[36,1730][1044,2045]" +_PANEL_SURFACE = "[0,366][1080,2079]" +_EMPTY_SUMMARY = "请选择： 颜色分类 尺码" +_COLOR_ONLY_SUMMARY = "请选择： 尺码" +_S_SIZE_UI = "S（建议80-100）" +_S_SUMMARY = f"已选： {_TARGET_COLOR_UI} {_S_SIZE_UI}" +_TARGET_SUMMARY = f"已选： {_TARGET_COLOR_UI} {_TARGET_SIZE_UI}" +_REVEAL_NOT_PROVEN = "尺码仍在已取证视口外；受控显示动作尚未取证，已停止后续点击。" _BOUNDS = re.compile(r"^\[(\d+),(\d+)\]\[(\d+),(\d+)\]$") -_PRICE = re.compile(r"^[^0-9¥￥]*[¥￥]([1-9][0-9]*\.[0-9]{2})$") -_ORIGINAL_PRICE = re.compile(r"^[¥￥][1-9][0-9]*\.[0-9]{2}$") +_ROLLED_CURRENT_PRICE = "快卖完 ¥12.88" +_ROLLED_ORIGINAL_PRICE = "¥29.88" _BAD_PRICE_ROLE = ("提交订单", "支付", "优惠", "券", "会员", "补贴", "区间", "实付", "到手", "原价", "划线价", "最低", "低至", "起价", "下单", "先用后付", "预估") @@ -118,6 +118,73 @@ class _Node: def bounds(self) -> str: return self.element.get("bounds", "") +class _PanelProfile(Enum): + PANEL_OPEN_EMPTY = "panel_open_empty" + COLOR_SELECTED_SIZE_HIDDEN = "color_selected_size_hidden" + SIZE_VISIBLE_NON_TARGET = "size_visible_non_target" + TARGETS_SELECTED = "targets_selected" + + +@dataclass(frozen=True) +class _PanelSpec: + profile: _PanelProfile + header_bounds: str + outer_bounds: str + price_row_bounds: str + current_text: str + current_bounds: str + original_text: str + original_bounds: str + summary_text: str + summary_bounds: str + color_label_bounds: str | None + color_region_bounds: str + color_bounds: str + color_selected: bool + size_label_bounds: str + selected_size: str | None + + +_PANEL_SPECS = ( + _PanelSpec( + _PanelProfile.PANEL_OPEN_EMPTY, + "[0,366][1080,1077]", "[0,1077][1080,2079]", + "[396,575][912,647]", "限1件 ¥12.88 ", "[396,580][675,647]", + "券前¥29.88", "[693,580][912,647]", + _EMPTY_SUMMARY, "[396,731][1053,793]", "[36,1106][192,1159]", + "[36,1188][1080,2046]", "[372,1188][684,1587]", False, + "[36,2069][114,2079]", None, + ), + _PanelSpec( + _PanelProfile.COLOR_SELECTED_SIZE_HIDDEN, + "[0,366][1080,1077]", "[0,1077][1080,2079]", + "[396,575][912,647]", "限1件 ¥12.88 ", "[396,580][675,647]", + "券前¥29.88", "[693,580][912,647]", + _COLOR_ONLY_SUMMARY, "[396,731][1053,793]", "[36,1106][192,1159]", + "[36,1188][1080,2046]", "[372,1188][684,1587]", True, + "[36,2069][114,2079]", None, + ), + _PanelSpec( + _PanelProfile.SIZE_VISIBLE_NON_TARGET, + "[0,366][1080,1000]", "[0,1000][1080,2079]", + "[396,498][895,570]", _ROLLED_CURRENT_PRICE, "[396,503][712,570]", + _ROLLED_ORIGINAL_PRICE, "[730,503][895,570]", + _S_SUMMARY, "[396,654][1053,716]", None, + "[36,1000][1080,1483]", "[372,1000][684,1024]", True, + "[36,1506][114,1552]", _S_SIZE_UI, + ), + _PanelSpec( + _PanelProfile.TARGETS_SELECTED, + "[0,366][1080,1000]", "[0,1000][1080,2079]", + "[396,498][895,570]", _ROLLED_CURRENT_PRICE, "[396,503][712,570]", + _ROLLED_ORIGINAL_PRICE, "[730,503][895,570]", + _TARGET_SUMMARY, "[396,654][1053,716]", None, + "[36,1000][1080,1483]", "[372,1000][684,1024]", True, + "[36,1506][114,1552]", _TARGET_SIZE_UI, + ), +) + + class SkuSelectionFlow: def __init__(self, device: SkuPanelDevice, entry_wait_timeout_seconds: float = 0.2, entry_poll_interval_seconds: float = 0.2, monotonic_clock: Callable[[], float] = monotonic, @@ -150,40 +217,63 @@ class SkuSelectionFlow: raise try: - self._pending = (before, _panel) + self._pending = (before, _require_empty_panel) self._device.tap_sku_entry(entry.bounds) except BaseException as error: _annotate_sku_entry_failure(error, "sku_entry_click") raise try: - self._wait_after_action(before, _panel) + self._wait_after_action(before, _require_empty_panel) except BaseException as error: _annotate_sku_entry_failure(error, "sku_entry_panel_verify") raise def select_sku_options(self, selection: SkuSelection) -> None: - if selection not in {SkuSelection(*item) for item in TASK_TO_UI_SELECTION.values()}: + if selection != SkuSelection(_TARGET_COLOR_UI, _TARGET_SIZE_UI): raise SkuSelectionError("规格 UI 文案不是获准目标，已停止操作。") - initial = self._verified_nodes() - _option(initial, "color", selection.color); _option(initial, "size", selection.size) - _selected_label(initial, "color"); _selected_label(initial, "size") - self._restore("color", selection.color) - self._restore("size", selection.size) + self._require_foreground() + before = self._read_hierarchy() + nodes = _parse_nodes(before) + profile = _classify_panel(nodes) + + if profile is _PanelProfile.PANEL_OPEN_EMPTY: + target = _target_color_action(nodes, selected=False) + _action_bounds(target.bounds) + _require_action_occupants(nodes, target) + self._pending = (before, _require_color_only_panel) + self._device.tap_sku_option(target.bounds) + self._wait_after_action(before, _require_color_only_panel) + # 当前证据只证明颜色选择；尺码仍在视口外。没有动作证据时必须在此停住， + # 不能把一次通用 swipe 或下一次规格点击伪装成已验证流程。 + raise SkuSelectionError(_REVEAL_NOT_PROVEN) + if profile is _PanelProfile.COLOR_SELECTED_SIZE_HIDDEN: + raise SkuSelectionError(_REVEAL_NOT_PROVEN) + if profile is _PanelProfile.SIZE_VISIBLE_NON_TARGET: + target = _target_size_action(nodes, selected=False) + _action_bounds(target.bounds) + _require_action_occupants(nodes, target) + self._pending = (before, _require_target_panel) + self._device.tap_sku_option(target.bounds) + self._wait_after_action(before, _require_target_panel) + return + if profile is _PanelProfile.TARGETS_SELECTED: + return + raise SkuSelectionError("规格面板状态不属于已取证 profile，已停止操作。") def read_sku_unit_price(self) -> str: return _unit_price(self._verified_nodes()) def verify_target_selection_and_read_price(self, selection: SkuSelection) -> str: + if selection != SkuSelection(_TARGET_COLOR_UI, _TARGET_SIZE_UI): + raise SkuSelectionError("规格 UI 文案不是获准目标，已停止读取。") nodes = self._verified_nodes() - _selected(nodes, "color", selection.color) - _selected(nodes, "size", selection.size) return _unit_price(nodes) def exit_sku_panel_safely(self) -> None: self._require_foreground() before = self._read_hierarchy() - _panel(_parse_nodes(before)) + _classify_panel(_parse_nodes(before)) self._device.leave_sku_panel() deadline = self._clock() + self._entry_timeout while True: @@ -191,7 +281,7 @@ class SkuSelectionFlow: raw = self._read_hierarchy() if raw != before: try: - _panel(_parse_nodes(raw)) + _classify_panel(_parse_nodes(raw)) except SkuSelectionError: return remaining = deadline - self._clock() @@ -206,26 +296,6 @@ class SkuSelectionFlow: before, condition = self._pending self._wait_after_action(before, condition) - def _restore(self, dimension: str, expected: str) -> None: - self._require_foreground() - before = self._read_hierarchy() - nodes = _panel(_parse_nodes(before)) - target = _option(nodes, dimension, expected) - if _selected_label(nodes, dimension) == expected: - return - _action_bounds(target.bounds) - condition: Callable[[list[_Node]], Any] - if dimension == "color": - condition = lambda refreshed: _post_color(refreshed, expected) - else: - condition = lambda refreshed: _post_all_targets(refreshed, expected) - self._pending = (before, condition) - self._device.tap_sku_option(target.bounds) - if dimension == "color": - self._wait_after_action(before, condition) - else: - self._wait_after_action(before, condition) - def _wait_for_entry(self, previous: str | None) -> tuple[_Node, str]: deadline, stable = self._clock() + self._entry_timeout, None while True: @@ -276,7 +346,9 @@ class SkuSelectionFlow: def _verified_nodes(self) -> list[_Node]: self._require_foreground() - return _panel(self._read_nodes()) + nodes = self._read_nodes() + _classify_panel(nodes) + return nodes def _require_version(self) -> None: info = self._device.app_info(PDD_PACKAGE) @@ -311,85 +383,415 @@ def _parse_nodes(raw: str) -> list[_Node]: return result -def _panel(nodes: list[_Node]) -> list[_Node]: - parent = _one([n for n in nodes if n.bounds == _PRICE_PARENT], "规格面板价格区域不唯一，已停止操作。") - _one([n for n in nodes if n.parent is parent and n.bounds == _ORIGINAL and _readonly(n) and _ORIGINAL_PRICE.fullmatch(n.text)], "规格面板原价槽位不唯一，已停止操作。") - _one([n for n in nodes if n.bounds == _SUMMARY and _readonly(n) and n.text.startswith("已选：")], "规格面板已选摘要不唯一，已停止操作。") - _color_container(nodes); _size_container(nodes) - return nodes +def _classify_panel(nodes: list[_Node]) -> _PanelProfile: + matches: list[_PanelProfile] = [] + for spec in _PANEL_SPECS: + try: + _match_panel_profile(nodes, spec) + except SkuSelectionError: + continue + matches.append(spec.profile) + if len(matches) != 1: + raise SkuSelectionError("规格面板不符合唯一完整取证 profile，已停止操作。") + return matches[0] + + +def _require_empty_panel(nodes: list[_Node]) -> None: + _require_profile(nodes, _PanelProfile.PANEL_OPEN_EMPTY) + + +def _require_color_only_panel(nodes: list[_Node]) -> None: + _require_profile(nodes, _PanelProfile.COLOR_SELECTED_SIZE_HIDDEN) + + +def _require_target_panel(nodes: list[_Node]) -> None: + _require_profile(nodes, _PanelProfile.TARGETS_SELECTED) + + +def _require_profile(nodes: list[_Node], expected: _PanelProfile) -> None: + if _classify_panel(nodes) is not expected: + raise SkuSelectionError("规格面板动作后状态与已取证 profile 不一致，已停止操作。") + + +def _match_panel_profile(nodes: list[_Node], spec: _PanelSpec) -> None: + surface = _one( + [node for node in nodes if _exact_inert(node, "android.view.ViewGroup", _PANEL_SURFACE)], + "规格面板内容面不唯一。", + ) + _require_panel_chain(surface) + header = _one( + [node for node in nodes if node.parent is surface and _exact_inert(node, "android.widget.LinearLayout", spec.header_bounds)], + "规格面板头部不唯一。", + ) + outer = _one( + [node for node in nodes if node.parent is surface and _exact_recycler(node, spec.outer_bounds)], + "规格面板维度容器不唯一。", + ) + price_row = _one( + [node for node in nodes if _descendant(node, header) and _exact_inert(node, "android.widget.LinearLayout", spec.price_row_bounds)], + "规格面板价格行不唯一。", + ) + if len([child for child in price_row.element if child.tag == "node"]) != 2: + raise SkuSelectionError("规格面板价格行子节点数量漂移。") + current = _one( + [node for node in nodes if node.parent is price_row and _exact_readonly_text(node, spec.current_text, spec.current_bounds)], + "规格面板当前价角色不唯一。", + ) + _one( + [node for node in nodes if node.parent is price_row and _exact_readonly_text(node, spec.original_text, spec.original_bounds)], + "规格面板原价角色不唯一。", + ) + if _clickable_before(current, surface): + raise SkuSelectionError("规格面板价格角色位于可点击内容祖先下。") + _one( + [node for node in nodes if _descendant(node, header) and _exact_readonly_text(node, spec.summary_text, spec.summary_bounds)], + "规格面板摘要不唯一。", + ) + + color_region = _one( + [node for node in nodes if _descendant(node, outer) and _exact_recycler(node, spec.color_region_bounds)], + "规格面板颜色容器不唯一。", + ) + if spec.color_label_bounds is None: + if any(node.text == "颜色分类" and _readonly(node) for node in nodes): + raise SkuSelectionError("滚动态出现未取证颜色标题。") + else: + _one( + [node for node in nodes if _descendant(node, outer) and not _descendant(node, color_region) and _exact_readonly_text(node, "颜色分类", spec.color_label_bounds)], + "规格面板颜色标题不唯一。", + ) + color = _one( + [node for node in nodes if node.parent is color_region and _exact_color_action(node, spec.color_bounds, spec.color_selected)], + "目标颜色 action 不唯一。", + ) + selected_nodes = _require_color_subtree(color, spec) + _require_color_action_chain(color, color_region, outer, surface, spec) + + size_label = _one( + [node for node in nodes if _descendant(node, outer) and not _descendant(node, color_region) and _exact_readonly_text(node, _SIZE, spec.size_label_bounds)], + "规格面板尺码标题不唯一。", + ) + if spec.selected_size is None: + if any( + node.text in {_S_SIZE_UI, _TARGET_SIZE_UI} + and _is_size_action_text(node) + for node in nodes + ): + raise SkuSelectionError("尺码隐藏 profile 出现可点击尺码。") + _require_exact_selected_set(nodes, surface, selected_nodes) + return + + size_header = size_label.parent + if size_header is None or not _exact_live_layout(size_header, "android.widget.LinearLayout", "[36,1489][1044,1570]"): + raise SkuSelectionError("规格面板尺码标题父结构漂移。") + size_options = _one( + [node for node in nodes if _descendant(node, outer) and _exact_inert(node, "android.view.ViewGroup", "[36,1582][1044,1897]")], + "规格面板尺码 options 根不唯一。", + ) + size_actions = [ + node for node in nodes + if _descendant(node, size_options) + and _is_size_action_text(node) + ] + s_action = _one([node for node in size_actions if node.text == _S_SIZE_UI and node.bounds == "[36,1582][409,1667]"], "S 尺码 action 不唯一。") + m_action = _one([node for node in size_actions if node.text == _TARGET_SIZE_UI and node.bounds == "[439,1582][831,1667]"], "M 尺码 action 不唯一。") + _require_size_wrapper(s_action, size_options) + _require_size_wrapper(m_action, size_options) + _require_size_action_chain(s_action, size_options, outer, surface) + _require_size_action_chain(m_action, size_options, outer, surface) + selected_sizes = [node for node in size_actions if node.element.get("selected") == "true"] + expected_action = s_action if spec.selected_size == _S_SIZE_UI else m_action + if selected_sizes != [expected_action]: + raise SkuSelectionError("尺码维度 selected 状态不唯一。") + _require_exact_selected_set(nodes, surface, [*selected_nodes, expected_action]) def _unit_price(nodes: list[_Node]) -> str: - parent = _one([n for n in nodes if n.bounds == _PRICE_PARENT], "规格面板价格区域不唯一，已停止读取。") - money = [n for n in nodes if n.parent is parent and _readonly(n) and any(mark in n.text for mark in "¥￥")] - if len(money) != 2: raise SkuSelectionError("规格面板金额槽位不唯一，已停止读取。") - current = _one([n for n in money if n.bounds == _CURRENT and not _clickable_ancestor(n) and not any(word in n.text for word in _BAD_PRICE_ROLE) and _PRICE.fullmatch(n.text)], "规格面板现价不唯一或不符合已取证槽位，已停止读取。") - if not any(n.bounds == _ORIGINAL and _ORIGINAL_PRICE.fullmatch(n.text) for n in money): - raise SkuSelectionError("规格面板原价槽位无效，已停止读取。") - match = _PRICE.fullmatch(current.text) - if match is None: raise SkuSelectionError("规格面板现价格式失效，已停止读取。") - return match.group(1) + _require_profile(nodes, _PanelProfile.TARGETS_SELECTED) + spec = _spec_for(_PanelProfile.TARGETS_SELECTED) + candidates = [ + node for node in nodes + if _exact_readonly_text(node, _ROLLED_CURRENT_PRICE, spec.current_bounds) + ] + current = _one(candidates, "规格面板现价不唯一，已停止读取。") + surface = _one([node for node in nodes if _exact_inert(node, "android.view.ViewGroup", _PANEL_SURFACE)], "规格面板内容面不唯一。") + if _clickable_before(current, surface) or any(word in current.text for word in _BAD_PRICE_ROLE): + raise SkuSelectionError("规格面板现价角色不可安全读取。") + return EXPECTED_UNIT_PRICE -def _option(nodes: list[_Node], dimension: str, expected: str) -> _Node: - _panel(nodes) - return _one([n for n in _options(nodes, dimension) if _label(n) == expected], "规格选项不唯一或不是精确匹配，已停止操作。") +def _target_color_action(nodes: list[_Node], *, selected: bool) -> _Node: + profile = _classify_panel(nodes) + expected_profile = _PanelProfile.COLOR_SELECTED_SIZE_HIDDEN if selected else _PanelProfile.PANEL_OPEN_EMPTY + if profile is not expected_profile: + raise SkuSelectionError("目标颜色 action 不属于预期 profile。") + spec = _spec_for(profile) + return _one([node for node in nodes if _exact_color_action(node, spec.color_bounds, selected)], "目标颜色 action 不唯一。") -def _selected(nodes: list[_Node], dimension: str, expected: str) -> None: - if _selected_label(nodes, dimension) != expected: - raise SkuSelectionError("规格选择后读回的 selected 文案不一致，已停止操作。") +def _target_size_action(nodes: list[_Node], *, selected: bool) -> _Node: + profile = _classify_panel(nodes) + expected_profile = _PanelProfile.TARGETS_SELECTED if selected else _PanelProfile.SIZE_VISIBLE_NON_TARGET + if profile is not expected_profile: + raise SkuSelectionError("目标尺码 action 不属于预期 profile。") + return _one( + [node for node in nodes if _is_size_action_text(node) and node.text == _TARGET_SIZE_UI and node.bounds == "[439,1582][831,1667]" and node.element.get("selected") == str(selected).lower()], + "目标尺码 action 不唯一。", + ) -def _selected_label(nodes: list[_Node], dimension: str) -> str: - selected = [n for n in _options(nodes, dimension) if n.element.get("selected") == "true"] - label = _label(_one(selected, "规格维度没有唯一 selected 状态，已停止操作。")) - if label is None: raise SkuSelectionError("规格维度 selected 文案无效，已停止操作。") - return label +def _spec_for(profile: _PanelProfile) -> _PanelSpec: + return next(spec for spec in _PANEL_SPECS if spec.profile is profile) -def _post_color(nodes: list[_Node], expected: str) -> None: - _panel(nodes) - _selected(nodes, "color", expected) - _selected_label(nodes, "size") +def _exact_inert(node: _Node, class_name: str, bounds: str) -> bool: + return _exact_common(node, class_name, bounds, clickable="false", selected="false", scrollable="false") and not node.text and not node.desc -def _post_all_targets(nodes: list[_Node], expected_size: str) -> None: - _panel(nodes) - _selected(nodes, "color", _TARGET_COLOR_UI) - _selected(nodes, "size", expected_size) +def _exact_recycler(node: _Node, bounds: str) -> bool: + return _exact_common(node, "androidx.recyclerview.widget.RecyclerView", bounds, clickable="false", selected="false", scrollable="true") and not node.text and not node.desc -def _options(nodes: list[_Node], dimension: str) -> list[_Node]: - container = _color_container(nodes) if dimension == "color" else _size_container(nodes) if dimension == "size" else None - if container is None: raise SkuSelectionError("未知规格维度，已停止操作。") - candidates = [n for n in nodes if _descendant(n, container) and _contained(n, container) and _choice(n) and _label(n) is not None] - return [n for n in candidates if not _labeled_ancestor(n, candidates)] +def _exact_readonly_text(node: _Node, text: str, bounds: str) -> bool: + return _exact_common(node, "android.widget.TextView", bounds, clickable="false", selected="false", scrollable="false") and node.text == text and not node.desc -def _color_container(nodes: list[_Node]) -> _Node: - return _one([n for n in nodes if n.element.get("package") == PDD_PACKAGE and n.element.get("class") == "androidx.recyclerview.widget.RecyclerView" and n.bounds == _COLOR_REGION], "规格面板颜色容器不唯一，已停止操作。") +def _exact_color_action(node: _Node, bounds: str, selected: bool) -> bool: + return ( + _exact_common(node, "android.view.ViewGroup", bounds, clickable="true", selected=str(selected).lower(), scrollable="false") + and not node.text + and node.desc == _TARGET_COLOR_UI + ) -def _size_container(nodes: list[_Node]) -> _Node: - label = _one([n for n in nodes if n.text == _SIZE and n.bounds == _SIZE_LABEL and _readonly(n)], "规格面板尺码标签不唯一，已停止操作。") - header = label.parent - if header is None or header.element.get("package") != PDD_PACKAGE or header.element.get("class") != "android.widget.LinearLayout" or header.bounds != _SIZE_HEADER or header.parent is None: - raise SkuSelectionError("规格面板尺码标题容器不符合已取证结构，已停止操作。") - return _one([n for n in nodes if n.parent is header.parent and n.element.get("package") == PDD_PACKAGE and n.element.get("class") == "android.widget.LinearLayout" and n.bounds == _SIZE_OPTIONS], "规格面板尺码选项容器不唯一，已停止操作。") +def _exact_live_layout(node: _Node, class_name: str, bounds: str) -> bool: + return _exact_common(node, class_name, bounds, clickable="true", selected="false", scrollable="false") and not node.text and not node.desc -def _label(node: _Node) -> str | None: - values = {value for value in (node.text, node.desc) if value} - return values.pop() if len(values) == 1 else None +def _exact_common(node: _Node, class_name: str, bounds: str, *, clickable: str, selected: str, scrollable: str) -> bool: + return ( + node.element.get("package") == PDD_PACKAGE + and node.element.get("class") == class_name + and node.bounds == bounds + and node.element.get("clickable") == clickable + and node.element.get("enabled") == "true" + and node.element.get("visible-to-user") == "true" + and node.element.get("selected") == selected + and node.element.get("scrollable") == scrollable + ) -def _labeled_ancestor(node: _Node, candidates: list[_Node]) -> bool: - ids, parent = {id(n.element) for n in candidates}, node.parent +def _is_size_action_text(node: _Node) -> bool: + return ( + node.element.get("package") == PDD_PACKAGE + and node.element.get("class") == "android.widget.TextView" + and node.element.get("clickable") == "true" + and node.element.get("enabled") == "true" + and node.element.get("visible-to-user") == "true" + and node.element.get("selected") in {"true", "false"} + and node.element.get("scrollable") == "false" + and not node.desc + ) + + +def _require_size_wrapper(action: _Node, size_options: _Node) -> None: + wrapper = action.parent + if ( + wrapper is None + or wrapper.parent is not size_options + or not _exact_live_layout(wrapper, "android.view.ViewGroup", action.bounds) + or wrapper.element[0] is not action.element + ): + raise SkuSelectionError("尺码 action 父结构漂移。") + children = [child for child in wrapper.element if child.tag == "node"] + if action.element.get("selected") == "true": + if len(children) != 2: + raise SkuSelectionError("已选尺码指示子树数量漂移。") + marker = _Node(children[1], wrapper) + if ( + not _exact_common( + marker, + "android.view.View", + action.bounds, + clickable="false", + selected="false", + scrollable="false", + ) + or marker.text + or marker.desc + ): + raise SkuSelectionError("已选尺码指示节点结构漂移。") + elif len(children) != 1: + raise SkuSelectionError("未选尺码 action 子树数量漂移。") + + +def _require_panel_chain(surface: _Node) -> None: + expected = ( + ("android.widget.LinearLayout", "[0,366][1080,2079]", "false"), + ("android.view.ViewGroup", "[0,366][1080,2328]", "true"), + ("android.widget.LinearLayout", "[0,120][1080,2328]", "true"), + ("android.widget.FrameLayout", "[0,120][1080,2328]", "false"), + ("android.widget.FrameLayout", "[0,120][1080,2328]", "false"), + ("android.widget.LinearLayout", "[0,0][1080,2328]", "false"), + ("android.widget.FrameLayout", "[0,0][1080,2376]", "false"), + ) + node = surface.parent + for class_name, bounds, clickable in expected: + if ( + node is None + or not _exact_common( + node, + class_name, + bounds, + clickable=clickable, + selected="false", + scrollable="false", + ) + or node.text + or node.desc + ): + raise SkuSelectionError("规格面板祖先链不符合已取证结构。") + node = node.parent + if node is None or node.element.tag != "hierarchy" or node.parent is not None: + raise SkuSelectionError("规格面板根节点结构漂移。") + + +def _require_color_subtree(color: _Node, spec: _PanelSpec) -> list[_Node]: + selected = str(spec.color_selected).lower() + children = [child for child in color.element if child.tag == "node"] + expected_selected: list[_Node] = [color] if spec.color_selected else [] + if spec.color_bounds == "[372,1188][684,1587]": + if len(children) != 4: + raise SkuSelectionError("目标颜色完整卡片子树数量漂移。") + child_nodes = [node for node in _walk_direct_children(color)] + view, image, zoom, label_layout = child_nodes + if not _exact_common(view, "android.view.View", spec.color_bounds, clickable="false", selected=selected, scrollable="false") or view.text or view.desc: + raise SkuSelectionError("目标颜色选中遮罩结构漂移。") + if not _exact_common(image, "android.widget.ImageView", "[372,1188][684,1500]", clickable="true", selected=selected, scrollable="false") or image.text or image.desc != _TARGET_COLOR_UI: + raise SkuSelectionError("目标颜色图片 action 结构漂移。") + if not _exact_common(zoom, "android.widget.ImageView", "[372,1188][483,1299]", clickable="true", selected=selected, scrollable="false") or zoom.text or zoom.desc != "打开大图": + raise SkuSelectionError("目标颜色大图 action 结构漂移。") + if not _exact_common(label_layout, "android.widget.LinearLayout", "[372,1479][684,1587]", clickable="false", selected=selected, scrollable="false") or label_layout.text or label_layout.desc or len(label_layout.element) != 1: + raise SkuSelectionError("目标颜色文字容器结构漂移。") + label = _Node(label_layout.element[0], label_layout) + if not _exact_common(label, "android.widget.TextView", "[372,1479][684,1587]", clickable="true", selected=selected, scrollable="false") or label.text != _TARGET_COLOR_UI or label.desc: + raise SkuSelectionError("目标颜色文字 action 结构漂移。") + if spec.color_selected: + expected_selected.extend((view, image, zoom, label_layout, label)) + return expected_selected + + if len(children) != 2: + raise SkuSelectionError("目标颜色滚动态子树数量漂移。") + view, label_layout = [node for node in _walk_direct_children(color)] + if not _exact_common(view, "android.view.View", spec.color_bounds, clickable="false", selected=selected, scrollable="false") or view.text or view.desc: + raise SkuSelectionError("目标颜色滚动态遮罩结构漂移。") + if not _exact_common(label_layout, "android.widget.LinearLayout", spec.color_bounds, clickable="false", selected=selected, scrollable="false") or label_layout.text or label_layout.desc or len(label_layout.element) != 1: + raise SkuSelectionError("目标颜色滚动态文字容器漂移。") + label = _Node(label_layout.element[0], label_layout) + if not _exact_common(label, "android.widget.TextView", spec.color_bounds, clickable="true", selected=selected, scrollable="false") or label.text != _TARGET_COLOR_UI or label.desc: + raise SkuSelectionError("目标颜色滚动态文字 action 漂移。") + if spec.color_selected: + expected_selected.extend((view, label_layout, label)) + return expected_selected + + +def _require_color_action_chain( + color: _Node, + color_region: _Node, + outer: _Node, + surface: _Node, + spec: _PanelSpec, +) -> None: + if color.parent is not color_region: + raise SkuSelectionError("目标颜色 action 不属于已取证颜色容器。") + if spec.color_bounds == "[372,1188][684,1587]": + expected = ( + ("android.widget.FrameLayout", "[0,1188][1080,2046]"), + ("android.widget.LinearLayout", "[0,1188][1080,2052]"), + ("android.widget.LinearLayout", "[0,1077][1080,2052]"), + ) + else: + expected = ( + ("android.widget.FrameLayout", "[0,1000][1080,1483]"), + ("android.widget.LinearLayout", "[0,1000][1080,1489]"), + ("android.widget.LinearLayout", "[0,1000][1080,1489]"), + ) + node = color_region.parent + for class_name, bounds in expected: + if node is None or not _exact_inert(node, class_name, bounds): + raise SkuSelectionError("目标颜色 action 父链不符合已取证结构。") + node = node.parent + if node is not outer or outer.parent is not surface: + raise SkuSelectionError("目标颜色 action 未沿已取证维度父链回到面板。") + + +def _require_size_action_chain( + action: _Node, + size_options: _Node, + outer: _Node, + surface: _Node, +) -> None: + wrapper = action.parent + if wrapper is None or wrapper.parent is not size_options: + raise SkuSelectionError("尺码 action 不属于已取证 options 容器。") + node = size_options.parent + for class_name, bounds in ( + ("android.widget.LinearLayout", "[36,1582][1044,1897]"), + ("android.widget.LinearLayout", "[0,1489][1080,1930]"), + ): + if node is None or not _exact_inert(node, class_name, bounds): + raise SkuSelectionError("尺码 action 父链不符合已取证结构。") + node = node.parent + if node is not outer or outer.parent is not surface: + raise SkuSelectionError("尺码 action 未沿已取证维度父链回到面板。") + + +def _walk_direct_children(parent: _Node) -> list[_Node]: + return [_Node(child, parent) for child in parent.element if child.tag == "node"] + + +def _require_exact_selected_set(nodes: list[_Node], surface: _Node, expected: list[_Node]) -> None: + actual = [ + node for node in nodes + if _descendant(node, surface) and node.element.get("selected") == "true" + ] + if {id(node.element) for node in actual} != {id(node.element) for node in expected}: + raise SkuSelectionError("规格面板 selected 节点集合与完整 profile 不一致。") + + +def _require_action_occupants(nodes: list[_Node], target: _Node) -> None: + left, top, right, bottom = _action_bounds(target.bounds) + point = (left + (right - left) // 2, top + (bottom - top) // 2) + allowed: set[int] = {id(target.element)} + parent = target.parent while parent is not None: - if id(parent.element) in ids and _label(parent) is not None: return True + if _is_live_clickable(parent): + if not ( + _exact_live_layout(parent, "android.view.ViewGroup", target.bounds) + or _exact_live_layout(parent, "android.view.ViewGroup", "[0,366][1080,2328]") + or _exact_live_layout(parent, "android.widget.LinearLayout", "[0,120][1080,2328]") + ): + raise SkuSelectionError("规格 action 存在未取证可点击祖先，已停止操作。") + allowed.add(id(parent.element)) parent = parent.parent - return False + for node in nodes: + if _descendant(node, target) and _is_live_clickable(node): + node_left, node_top, node_right, node_bottom = _action_bounds(node.bounds) + if node_left <= point[0] < node_right and node_top <= point[1] < node_bottom: + allowed.add(id(node.element)) + occupants = _live_clickables_covering(nodes, point) + if {id(node.element) for node in occupants} != allowed: + raise SkuSelectionError("规格 action 中心存在未取证可点击占用，已停止操作。") + + +def _clickable_before(node: _Node, stop: _Node) -> bool: + parent = node.parent + while parent is not None and parent.element is not stop.element: + if parent.element.get("clickable") == "true": + return True + parent = parent.parent + # 当前价必须能沿已取证内容祖先回到面板内容面；不接受另一个树枝上的同坐标文本。 + return parent is None def _descendant(node: _Node, ancestor: _Node) -> bool: @@ -400,20 +802,6 @@ def _descendant(node: _Node, ancestor: _Node) -> bool: return False -def _contained(node: _Node, container: _Node) -> bool: - left, top, right, bottom = _action_bounds(node.bounds) - outer_left, outer_top, outer_right, outer_bottom = _action_bounds(container.bounds) - return outer_left <= left < right <= outer_right and outer_top <= top < bottom <= outer_bottom - - -def _clickable_ancestor(node: _Node) -> bool: - parent = node.parent - while parent is not None: - if parent.element.get("clickable") == "true": return True - parent = parent.parent - return False - - def _readonly(node: _Node) -> bool: return node.element.get("package") == PDD_PACKAGE and node.element.get("class") == "android.widget.TextView" and node.element.get("clickable") == "false" and node.element.get("enabled") == "true" and node.element.get("visible-to-user") == "true" @@ -422,14 +810,15 @@ def _live(node: _Node) -> bool: return node.element.get("package") == PDD_PACKAGE and node.element.get("clickable") == "true" and node.element.get("enabled") == "true" and node.element.get("visible-to-user") == "true" and bool(node.bounds) -def _choice(node: _Node) -> bool: - return _live(node) and node.element.get("class") == "android.view.ViewGroup" and node.element.get("selected") in {"true", "false"} - - def _eligible_entries(nodes: list[_Node]) -> list[_Node]: # 入口文本本身不可点击：必须证明它仍是已取证底部父容器的直接子节点，但动作坐标继续 # 使用第一行文本的窄 bounds，避免把父容器中心或第二行“免拼购买”变成坐标兜底。 - if any(node.bounds == _PRICE_PARENT for node in nodes): + if any( + node.bounds == _PANEL_SURFACE + and node.element.get("package") == PDD_PACKAGE + and node.element.get("class") == "android.view.ViewGroup" + for node in nodes + ): return [] entries = _physical_entries(nodes) if len(entries) != 1: diff --git a/client/tests/pdd/fixtures/sku_panel_color_selected_size_hidden_8_17_0.xml b/client/tests/pdd/fixtures/sku_panel_color_selected_size_hidden_8_17_0.xml new file mode 100644 index 0000000..144948d --- /dev/null +++ b/client/tests/pdd/fixtures/sku_panel_color_selected_size_hidden_8_17_0.xml @@ -0,0 +1,52 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/client/tests/pdd/fixtures/sku_panel_empty_8_17_0.xml b/client/tests/pdd/fixtures/sku_panel_empty_8_17_0.xml new file mode 100644 index 0000000..529dab6 --- /dev/null +++ b/client/tests/pdd/fixtures/sku_panel_empty_8_17_0.xml @@ -0,0 +1,52 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/client/tests/pdd/fixtures/sku_panel_size_m_restored_8_17_0.xml b/client/tests/pdd/fixtures/sku_panel_size_m_restored_8_17_0.xml new file mode 100644 index 0000000..2333e62 --- /dev/null +++ b/client/tests/pdd/fixtures/sku_panel_size_m_restored_8_17_0.xml @@ -0,0 +1,58 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/client/tests/pdd/fixtures/sku_panel_size_s_selected_8_17_0.xml b/client/tests/pdd/fixtures/sku_panel_size_s_selected_8_17_0.xml new file mode 100644 index 0000000..0f7ec3c --- /dev/null +++ b/client/tests/pdd/fixtures/sku_panel_size_s_selected_8_17_0.xml @@ -0,0 +1,58 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/client/tests/pdd/test_sku_reveal_spike.py b/client/tests/pdd/test_sku_reveal_spike.py new file mode 100644 index 0000000..d1c88a8 --- /dev/null +++ b/client/tests/pdd/test_sku_reveal_spike.py @@ -0,0 +1,396 @@ +from __future__ import annotations + +import base64 +from contextlib import redirect_stderr +from io import BytesIO +import importlib.util +import io +import json +from pathlib import Path +from tempfile import TemporaryDirectory +import unittest +from unittest.mock import patch +from xml.etree import ElementTree + +from PIL import Image + +from cmbuyer_client.device.adb import AdbDevice, DeviceInspection +from cmbuyer_client.pdd.sku_reveal_spike import ( + SkuRevealSpikeCapturer, + SkuRevealSpikeError, + _require_safe_reveal_path, +) +from cmbuyer_client.pdd.sku_selection import SkuSelectionError, _parse_nodes + + +_FIXTURES = Path(__file__).with_name("fixtures") +_PRODUCT = (_FIXTURES / "product_entry_8_17_0.xml").read_text(encoding="utf-8") +_EMPTY = (_FIXTURES / "sku_panel_empty_8_17_0.xml").read_text(encoding="utf-8") +_COLOR_ONLY = (_FIXTURES / "sku_panel_color_selected_size_hidden_8_17_0.xml").read_text(encoding="utf-8") +_M_SELECTED = (_FIXTURES / "sku_panel_size_m_restored_8_17_0.xml").read_text(encoding="utf-8") + + +def _candidate_unselected() -> str: + root = ElementTree.fromstring(_M_SELECTED) + parents = {child: parent for parent in root.iter() for child in parent} + for node in root.iter("node"): + if node.get("text") == "已选： 黑色 CHA (纯棉） M（建议100-115）": + node.set("text", "请选择： 尺码") + if node.get("text") in {"S（建议80-100）", "M（建议100-115）"}: + node.set("selected", "false") + wrapper = parents[node] + for child in list(wrapper): + if child.get("class") == "android.view.View": + wrapper.remove(child) + action_root = next( + node + for node in root.iter("node") + if node.get("class") == "android.view.ViewGroup" + and node.get("clickable") == "true" + and node.get("bounds") == "[0,366][1080,2328]" + ) + button_frame = ElementTree.SubElement( + action_root, + "node", + { + "class": "android.widget.FrameLayout", + "package": "com.xunmeng.pinduoduo", + "text": "", + "content-desc": "", + "clickable": "true", + "enabled": "true", + "visible-to-user": "true", + "selected": "false", + "scrollable": "false", + "bounds": "[0,2181][1080,2328]", + }, + ) + button_content = ElementTree.SubElement( + button_frame, + "node", + { + "class": "android.widget.LinearLayout", + "package": "com.xunmeng.pinduoduo", + "text": "", + "content-desc": "", + "clickable": "false", + "enabled": "true", + "visible-to-user": "true", + "selected": "false", + "scrollable": "false", + "bounds": "[273,2181][807,2328]", + }, + ) + ElementTree.SubElement( + button_content, + "node", + { + "class": "android.widget.TextView", + "package": "com.xunmeng.pinduoduo", + "text": "选择尺码后，提交订单", + "content-desc": "", + "clickable": "false", + "enabled": "true", + "visible-to-user": "true", + "selected": "false", + "scrollable": "false", + "bounds": "[285,2225][795,2284]", + }, + ) + return ElementTree.tostring(root, encoding="unicode") + + +def _with_clickable_overlay(package: str, class_name: str, bounds: str) -> str: + root = ElementTree.fromstring(_COLOR_ONLY) + container = next(root.iter("node")) + ElementTree.SubElement( + container, + "node", + { + "package": package, + "class": class_name, + "text": "", + "content-desc": "", + "clickable": "true", + "enabled": "true", + "visible-to-user": "true", + "selected": "false", + "scrollable": "false", + "bounds": bounds, + }, + ) + return ElementTree.tostring(root, encoding="unicode") + + +def _png() -> str: + image = Image.new("RGB", (1080, 2376), "white") + raw = BytesIO() + image.save(raw, format="PNG") + return base64.b64encode(raw.getvalue()).decode("ascii") + + +class _FakeDevice: + def __init__(self) -> None: + self.hierarchy = "" + self.after_hierarchy = _candidate_unselected() + self.calls: list[tuple[object, ...]] = [] + self.swipe_error = False + + def app_info(self, package_name: str) -> dict[str, str]: + self.calls.append(("app_info", package_name)) + return {"versionName": "8.17.0"} + + def app_current(self) -> dict[str, str]: + self.calls.append(("app_current",)) + return {"package": "com.xunmeng.pinduoduo"} + + 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) -> str: + self.calls.append(("jsonrpc", method, params, timeout)) + if method == "dumpWindowHierarchy": + return self.hierarchy + if method == "takeScreenshot": + return _png() + if method == "click": + if params == [865, 2218]: + self.hierarchy = _EMPTY + return "" + if params == [528, 1387]: + self.hierarchy = _COLOR_ONLY + return "" + raise AssertionError(params) + if method == "swipe": + self.hierarchy = self.after_hierarchy + if self.swipe_error: + raise TimeoutError("unknown outcome") + return "" + raise AssertionError(method) + + +class _FakeAdb: + def __init__(self, device: _FakeDevice) -> None: + self.device = device + self.calls: list[tuple[object, ...]] = [] + + def inspect(self, serial: str) -> DeviceInspection: + self.calls.append(("inspect", serial)) + return DeviceInspection(AdbDevice(serial=serial, state="device"), "PKG110", "16") + + def start_pdd_view_intent(self, serial: str, goods_id: str) -> object: + self.calls.append(("intent", serial, goods_id)) + self.device.hierarchy = _PRODUCT + return object() + + +def _swipes(device: _FakeDevice) -> list[tuple[object, ...]]: + return [call for call in device.calls if call[:2] == ("jsonrpc", "swipe")] + + +class SkuRevealSpikeTests(unittest.TestCase): + def _capturer(self, device: _FakeDevice) -> SkuRevealSpikeCapturer: + now = [0.0] + return SkuRevealSpikeCapturer( + _FakeAdb(device), + lambda serial: device, + 30, + monotonic_clock=lambda: now[0], + sleep_function=lambda seconds: now.__setitem__(0, now[0] + seconds), + ) + + def test_success_publishes_before_after_and_exactly_one_reveal(self) -> None: + device = _FakeDevice() + with TemporaryDirectory() as directory: + target = Path(directory) / "evidence" + result = self._capturer(device).capture("192.168.0.173:5555", "937122477375", target) + + self.assertEqual(result.output_directory, target) + self.assertEqual(len(_swipes(device)), 1) + self.assertFalse(any(call[1] == "pressKey" for call in device.calls if call[0] == "jsonrpc")) + for relative in ( + "before/screenshot.png", + "before/hierarchy.xml", + "after/screenshot.png", + "after/hierarchy.xml", + "manifest.json", + ): + self.assertTrue((target / relative).is_file(), relative) + manifest_text = result.manifest_path.read_text(encoding="utf-8") + manifest = json.loads(manifest_text) + self.assertEqual(manifest["reveal_attempts"], 1) + self.assertEqual(manifest["rpc_outcome"], "completed") + self.assertNotIn("192.168.0.173:5555", manifest_text) + self.assertNotIn("gesture", manifest) + self.assertNotIn("coordinates", manifest) + self.assertNotIn("start", manifest) + self.assertNotIn("end", manifest) + + def test_ambiguous_rpc_is_read_only_reconciled_without_retry(self) -> None: + device = _FakeDevice() + device.swipe_error = True + with TemporaryDirectory() as directory: + target = Path(directory) / "evidence" + result = self._capturer(device).capture("192.168.0.173:5555", "937122477375", target) + self.assertEqual(len(_swipes(device)), 1) + manifest = json.loads(result.manifest_path.read_text(encoding="utf-8")) + self.assertEqual(manifest["rpc_outcome"], "ambiguous_reconciled") + + def test_selected_m_or_precondition_drift_never_publishes_or_retries(self) -> None: + for after in ( + _M_SELECTED, + _candidate_unselected().replace("请选择： 尺码", "已选： 尺码"), + ): + with self.subTest(): + device = _FakeDevice() + device.after_hierarchy = after + with TemporaryDirectory() as directory: + target = Path(directory) / "evidence" + with self.assertRaises(SkuRevealSpikeError): + self._capturer(device).capture("192.168.0.173:5555", "937122477375", target) + self.assertEqual(len(_swipes(device)), 1) + self.assertFalse(target.exists()) + self.assertEqual(list(Path(directory).glob(".*.staging-*")), []) + + device = _FakeDevice() + device.hierarchy = "" + invalid_color = _COLOR_ONLY.replace("请选择： 尺码", "请选择： 颜色分类 尺码") + original_call = device.jsonrpc_call + + def drift(method: str, params: object = None, timeout: float = 10) -> str: + value = original_call(method, params, timeout) + if method == "click" and params == [528, 1387]: + device.hierarchy = invalid_color + return value + + device.jsonrpc_call = drift # type: ignore[method-assign] + with TemporaryDirectory() as directory: + with self.assertRaises((SkuSelectionError, SkuRevealSpikeError)): + self._capturer(device).capture( + "192.168.0.173:5555", + "937122477375", + Path(directory) / "evidence", + ) + self.assertEqual(_swipes(device), []) + + def test_invalid_goods_and_existing_target_are_zero_action(self) -> None: + device = _FakeDevice() + with TemporaryDirectory() as directory: + target = Path(directory) / "existing" + target.mkdir() + with self.assertRaises(SkuRevealSpikeError): + self._capturer(device).capture("wifi", "1", Path(directory) / "new") + with self.assertRaises(SkuRevealSpikeError): + self._capturer(device).capture("wifi", "937122477375", target) + self.assertEqual(device.calls, []) + + def test_before_screenshot_drift_is_rechecked_before_zero_swipe(self) -> None: + device = _FakeDevice() + original_call = device.jsonrpc_call + + def drift(method: str, params: object = None, timeout: float = 10) -> str: + value = original_call(method, params, timeout) + if method == "takeScreenshot": + device.hierarchy = _COLOR_ONLY.replace("请选择： 尺码", "请选择： 颜色分类 尺码") + return value + + device.jsonrpc_call = drift # type: ignore[method-assign] + with TemporaryDirectory() as directory: + target = Path(directory) / "evidence" + with self.assertRaises(SkuSelectionError): + self._capturer(device).capture( + "192.168.0.173:5555", + "937122477375", + target, + ) + self.assertFalse(target.exists()) + self.assertEqual(_swipes(device), []) + + def test_complete_reveal_segment_rejects_narrow_impostor_and_invalid_bounds(self) -> None: + for hierarchy in ( + _with_clickable_overlay( + "com.xunmeng.pinduoduo", + "android.view.ViewGroup", + "[350,1450][370,1460]", + ), + _with_clickable_overlay( + "com.android.systemui", + "android.view.ViewGroup", + "[0,366][1080,2328]", + ), + _with_clickable_overlay( + "com.xunmeng.pinduoduo", + "android.view.ViewGroup", + "not-a-bound", + ), + ): + with self.subTest(), self.assertRaises(SkuRevealSpikeError): + _require_safe_reveal_path(_parse_nodes(hierarchy)) + + +class SkuRevealSpikeCliTests(unittest.TestCase): + def test_cli_has_no_gesture_or_task_specification_parameters(self) -> None: + script = _load_reveal_script() + arguments = script.parse_arguments( + [ + "--serial", "device-1", + "--goods-id", "937122477375", + "--output-dir", "evidence", + ] + ) + self.assertEqual( + set(vars(arguments)), + {"serial", "goods_id", "output_dir", "timeout", "adb"}, + ) + script.validate_arguments(arguments) + for field, value in ( + ("serial", ""), + ("goods_id", "1"), + ("timeout", 0), + ("timeout", float("inf")), + ): + with self.subTest(field=field), self.assertRaises(ValueError): + script.validate_arguments( + type("Arguments", (), vars(arguments) | {field: value})() + ) + + def test_cli_failure_is_redacted(self) -> None: + script = _load_reveal_script() + secret = "SERIAL=192.168.0.173:5555 private" + + class FailingCapturer: + def __init__(self, *args: object, **kwargs: object) -> None: + return None + + def capture(self, *args: object, **kwargs: object) -> object: + raise SkuRevealSpikeError(secret) + + stderr = io.StringIO() + with patch.object(script, "SkuRevealSpikeCapturer", FailingCapturer), redirect_stderr(stderr): + status = script.main( + [ + "--serial", "192.168.0.173:5555", + "--goods-id", "937122477375", + "--output-dir", "evidence", + ] + ) + output = stderr.getvalue() + self.assertEqual(status, 1) + self.assertNotIn("Traceback", output) + self.assertNotIn("192.168.0.173:5555", output) + self.assertNotIn("private", output) + + +def _load_reveal_script() -> object: + path = Path(__file__).resolve().parents[2] / "scripts" / "capture_sku_reveal_spike.py" + specification = importlib.util.spec_from_file_location("capture_sku_reveal_spike_test", path) + if specification is None or specification.loader is None: + raise RuntimeError("无法加载 T-103 reveal 取证脚本。") + module = importlib.util.module_from_spec(specification) + specification.loader.exec_module(module) + return module + + +if __name__ == "__main__": + unittest.main() diff --git a/client/tests/pdd/test_sku_selection.py b/client/tests/pdd/test_sku_selection.py index d741c07..2b2bbad 100644 --- a/client/tests/pdd/test_sku_selection.py +++ b/client/tests/pdd/test_sku_selection.py @@ -17,7 +17,13 @@ import cmbuyer_client.pdd as pdd import cmbuyer_client.pdd.sku_selection_runner as runner_module from cmbuyer_client.device.adb import AdbDevice, DeviceConnectionError, DeviceInspection from cmbuyer_client.pdd import SkuSelectionError, SkuSelectionFlow, SkuSelectionRunner -from cmbuyer_client.pdd.sku_selection import SkuPanelDevice, _action_bounds, resolve_task_selection +from cmbuyer_client.pdd.sku_selection import ( + SkuPanelDevice, + _action_bounds, + _classify_panel, + _parse_nodes, + resolve_task_selection, +) from cmbuyer_client.pdd.sku_selection_runner import ( SkuSelectionDeviceAdapterError, SkuSelectionRunError, @@ -27,7 +33,12 @@ from cmbuyer_client.pdd.sku_selection_runner import ( ) -_FIXTURE = Path(__file__).with_name("fixtures") / "sku_panel_8_17_0.xml" +_FIXTURES = Path(__file__).with_name("fixtures") +_EMPTY_FIXTURE = _FIXTURES / "sku_panel_empty_8_17_0.xml" +_COLOR_FIXTURE = _FIXTURES / "sku_panel_color_selected_size_hidden_8_17_0.xml" +_S_FIXTURE = _FIXTURES / "sku_panel_size_s_selected_8_17_0.xml" +_M_FIXTURE = _FIXTURES / "sku_panel_size_m_restored_8_17_0.xml" +_FIXTURE = _M_FIXTURE _ENTRY_FIXTURE = Path(__file__).with_name("fixtures") / "product_entry_8_17_0.xml" _TARGET_URL = "https://mobile.yangkeduo.com/goods.html?goods_id=937122477375" _TASK_COLOR = "黑色CHA(纯棉)" @@ -45,7 +56,7 @@ def _png_base64() -> str: class _RawDevice: def __init__(self, hierarchy: str = _PRODUCT_PAGE, screenshot: str | None = None) -> None: self.hierarchy = hierarchy - self.panel_hierarchy = _FIXTURE.read_text(encoding="utf-8") + self.panel_hierarchy = _EMPTY_FIXTURE.read_text(encoding="utf-8") self.version = "8.17.0" self.package = "com.xunmeng.pinduoduo" self.screenshot = _png_base64() if screenshot is None else screenshot @@ -84,30 +95,24 @@ class _RawDevice: ): self.hierarchy = self.panel_hierarchy return - root = ElementTree.fromstring(self.hierarchy) - target = next(node for node in root.iter("node") if _center(node.get("bounds", "")) == (x, y)) - color = target.get("bounds", "").endswith("][438,1172]") - for node in root.iter("node"): - if node.get("selected") is not None and ((color and ",1000]" in node.get("bounds", "")) or (not color and ",1730]" in node.get("bounds", ""))): - node.set("selected", "false") - if color and self.fail_color_readback: - next(node for node in root.iter("node") if node.get("content-desc") == "粉红").set("selected", "true") - else: - target.set("selected", "true") - self.hierarchy = ElementTree.tostring(root, encoding="unicode") + if (x, y) == (528, 1387): + self.hierarchy = ( + _EMPTY_FIXTURE.read_text(encoding="utf-8") + if self.fail_color_readback + else _COLOR_FIXTURE.read_text(encoding="utf-8") + ) + return + if (x, y) == (635, 1624): + self.hierarchy = _M_FIXTURE.read_text(encoding="utf-8") + return + raise AssertionError((x, y)) def window_size(self) -> tuple[int, int]: self.calls.append(("window_size",)) return 1080, 2376 def select_alternates(self) -> None: - root = ElementTree.fromstring(self.panel_hierarchy) - for node in root.iter("node"): - if node.get("selected") is not None: - node.set("selected", "false") - next(node for node in root.iter("node") if node.get("content-desc") == "粉红").set("selected", "true") - next(node for node in root.iter("node") if node.get("text") == "L（建议115-130）").set("selected", "true") - self.panel_hierarchy = ElementTree.tostring(root, encoding="unicode") + self.panel_hierarchy = _S_FIXTURE.read_text(encoding="utf-8") if self.hierarchy != _PRODUCT_PAGE: self.hierarchy = self.panel_hierarchy @@ -119,6 +124,34 @@ def _center(bounds: str) -> tuple[int, int]: return left + (right - left) // 2, top + (bottom - top) // 2 +def _mutate_unique_node( + hierarchy: str, + *, + attribute: str, + value: str | None, + text: str | None = None, + desc: str | None = None, + class_name: str | None = None, + bounds: str | None = None, +) -> str: + root = ElementTree.fromstring(hierarchy) + matches = [ + node + for node in root.iter("node") + if (text is None or node.get("text") == text) + and (desc is None or node.get("content-desc") == desc) + and (class_name is None or node.get("class") == class_name) + and (bounds is None or node.get("bounds") == bounds) + ] + if len(matches) != 1: + raise AssertionError(f"fixture node match count: {len(matches)}") + if value is None: + matches[0].attrib.pop(attribute, None) + else: + matches[0].set(attribute, value) + return ElementTree.tostring(root, encoding="unicode") + + def _entry_chain(root: ElementTree.Element) -> list[ElementTree.Element]: parents = {child: parent for parent in root.iter() for child in parent} child = next(node for node in root.iter("node") if node.get("text") == "快要抢光 ¥ 12.88") @@ -286,18 +319,26 @@ class SkuSelectionFlowTests(unittest.TestCase): flow.open_sku_panel(_TARGET_URL) self.assertEqual(_actions(device, "click"), []) - def test_target_mapping_is_exact_and_success_path_restores_target(self) -> None: + def test_target_mapping_stops_at_unproven_reveal_then_s_to_m_is_exact(self) -> None: device = _RawDevice() adapter = UiautomatorSkuPanelAdapter(device, 10) flow = SkuSelectionFlow(adapter) flow.open_sku_panel(_TARGET_URL) - flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE)) - self.assertEqual(flow.read_sku_unit_price(), "12.88") - flow.exit_sku_panel_safely() + with self.assertRaisesRegex(SkuSelectionError, "受控显示动作尚未取证"): + flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE)) - self.assertEqual(_tap_centers(device), [(865, 2218)]) - self.assertEqual(_actions(device, "pressKey"), [("jsonrpc", "pressKey", ["back"], 10)]) + self.assertEqual(_tap_centers(device), [(865, 2218), (528, 1387)]) + self.assertEqual(_actions(device, "pressKey"), []) + + restored = _RawDevice(_S_FIXTURE.read_text(encoding="utf-8")) + restored_flow = SkuSelectionFlow(UiautomatorSkuPanelAdapter(restored, 10)) + restored_flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE)) + self.assertEqual(restored_flow.read_sku_unit_price(), "12.88") + restored_flow.exit_sku_panel_safely() + + self.assertEqual(_tap_centers(restored), [(635, 1624)]) + self.assertEqual(_actions(restored, "pressKey"), [("jsonrpc", "pressKey", ["back"], 10)]) def test_full_verified_entry_structure_taps_exact_text_child_once(self) -> None: device = _RawDevice() @@ -498,15 +539,45 @@ class SkuSelectionFlowTests(unittest.TestCase): def test_option_selected_and_container_drift_fail_closed_before_click(self) -> None: base = _FIXTURE.read_text(encoding="utf-8") cases = ( - base.replace('selected="true" clickable="true" enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]"', 'clickable="true" enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]"'), - base.replace('selected="true" clickable="true" enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]"', 'selected="maybe" clickable="true" enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]"'), - base.replace('bounds="[126,1000][438,1172]"', 'bounds="[1,1][20,20]"'), - base.replace('enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]"', 'enabled="false" visible-to-user="true" bounds="[126,1000][438,1172]"'), + _mutate_unique_node(base, desc="黑色 CHA (纯棉）", class_name="android.view.ViewGroup", attribute="selected", value=None), + _mutate_unique_node(base, desc="黑色 CHA (纯棉）", class_name="android.view.ViewGroup", attribute="selected", value="maybe"), + _mutate_unique_node(base, desc="黑色 CHA (纯棉）", class_name="android.view.ViewGroup", attribute="bounds", value="[1,1][20,20]"), + _mutate_unique_node(base, desc="黑色 CHA (纯棉）", class_name="android.view.ViewGroup", attribute="enabled", value="false"), ) for hierarchy in cases: with self.subTest(), self.assertRaises(SkuSelectionError): SkuSelectionFlow(UiautomatorSkuPanelAdapter(_RawDevice(hierarchy), 10)).select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE)) + def test_unknown_clickable_action_ancestor_is_rejected_by_profile(self) -> None: + root = ElementTree.fromstring(_EMPTY_FIXTURE.read_text(encoding="utf-8")) + parents = {child: parent for parent in root.iter() for child in parent} + target = next( + node + for node in root.iter("node") + if node.get("content-desc") == "黑色 CHA (纯棉）" + and node.get("class") == "android.view.ViewGroup" + ) + region = parents[target] + region.remove(target) + unknown = ElementTree.SubElement( + region, + "node", + { + "package": "com.xunmeng.pinduoduo", + "class": "android.view.ViewGroup", + "bounds": "[0,1188][1080,2046]", + "clickable": "true", + "enabled": "true", + "visible-to-user": "true", + "selected": "false", + "scrollable": "false", + }, + ) + unknown.append(target) + + with self.assertRaises(SkuSelectionError): + _classify_panel(_parse_nodes(ElementTree.tostring(root, encoding="unicode"))) + def test_invalid_bounds_stop_before_action(self) -> None: for bounds in ("", "[1,2][1,3]", "[1,2][3,2]", "[0,0][1081,1]", "[0,0][1,2377]", "[a,0][1,1]"): with self.subTest(bounds=bounds), self.assertRaises(SkuSelectionError): @@ -522,10 +593,9 @@ class SkuSelectionFlowTests(unittest.TestCase): device.fail_color_readback = True flow = SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)) flow.open_sku_panel(_TARGET_URL) - device.select_alternates() with self.assertRaises(SkuSelectionError): flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE)) - self.assertEqual(_tap_centers(device), [(865, 2218), (282, 1086)]) + self.assertEqual(_tap_centers(device), [(865, 2218), (528, 1387)]) def test_non_target_selection_restores_each_dimension_once(self) -> None: device = _RawDevice() @@ -535,7 +605,7 @@ class SkuSelectionFlowTests(unittest.TestCase): flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE)) self.assertEqual( _tap_centers(device), - [(865, 2218), (282, 1086), (635, 1772)], + [(865, 2218), (635, 1624)], ) def test_price_rejects_coupon_prefix_extra_amount_and_bottom_action(self) -> None: @@ -547,9 +617,12 @@ class SkuSelectionFlowTests(unittest.TestCase): device = _RawDevice(_FIXTURE.read_text(encoding="utf-8").replace("快卖完 ¥12.88", "提交订单 ¥12.88")) with self.assertRaises(SkuSelectionError): SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).read_sku_unit_price() - clickable_parent = _FIXTURE.read_text(encoding="utf-8").replace( - '', - '', + clickable_parent = _mutate_unique_node( + _FIXTURE.read_text(encoding="utf-8"), + class_name="android.widget.LinearLayout", + bounds="[396,498][895,570]", + attribute="clickable", + value="true", ) with self.assertRaises(SkuSelectionError): SkuSelectionFlow(UiautomatorSkuPanelAdapter(_RawDevice(clickable_parent), 10)).read_sku_unit_price() @@ -566,6 +639,8 @@ class SkuSelectionFlowTests(unittest.TestCase): root / "src" / "cmbuyer_client" / "pdd" / "sku_selection.py", root / "src" / "cmbuyer_client" / "pdd" / "sku_selection_runner.py", root / "scripts" / "run_t103_sku_selection.py", + root / "src" / "cmbuyer_client" / "pdd" / "sku_reveal_spike.py", + root / "scripts" / "capture_sku_reveal_spike.py", ) forbidden = ("quantity", "confirm", "authorization", "fence", "submit_order", "payment") for path in files: @@ -606,31 +681,56 @@ class SkuSelectionFlowTests(unittest.TestCase): self.assertEqual(_actions(device, "click"), []) def test_fixture_contains_no_address_phone_or_payment_credentials(self) -> None: - for fixture in (_FIXTURE, _ENTRY_FIXTURE): + for fixture in ( + _EMPTY_FIXTURE, + _COLOR_FIXTURE, + _S_FIXTURE, + _M_FIXTURE, + _ENTRY_FIXTURE, + ): content = fixture.read_text(encoding="utf-8") with self.subTest(fixture=fixture.name): self.assertNotRegex(content, r"1[3-9]\d{9}") for forbidden in ("地址", "收货", "支付", "银行卡", "身份证"): self.assertNotIn(forbidden, content) - content = _FIXTURE.read_text(encoding="utf-8") - root = ElementTree.fromstring(content) - leaf = next(node for node in root.iter("node") if node.get("text") == "提交订单 ¥12.88") - self.assertEqual(leaf.get("clickable"), "false") - self.assertEqual(leaf.get("bounds"), "[369,2225][710,2284]") + self.assertNotIn("提交订单", content) + + +class _CompletedFlow: + """仅隔离 runner 文件发布测试；生产 Flow 在 reveal 取证前仍必须停止。""" + + def __init__(self, device: object, *args: object, **kwargs: object) -> None: + self.device = device + + def open_sku_panel(self, product_url: str, pre_intent_hierarchy: str | None = None) -> None: + return None + + def select_sku_options(self, selection: object) -> None: + return None + + def verify_target_selection_and_read_price(self, selection: object) -> str: + return "12.88" + + def exit_sku_panel_safely(self) -> None: + return None + + def reconcile_pending_action(self) -> None: + return None class SkuSelectionRunnerTests(unittest.TestCase): def _runner(self, adb: _FakeAdb, device: _RawDevice) -> SkuSelectionRunner: device.hierarchy = "" adb.on_intent = lambda: setattr(device, "hierarchy", _PRODUCT_PAGE.replace("", '')) - return SkuSelectionRunner(adb, lambda serial: device, 10) + return SkuSelectionRunner(adb, lambda serial: device, 0.03) def test_runner_atomically_publishes_screenshot_and_redacted_manifest(self) -> None: adb = _FakeAdb() device = _RawDevice() with TemporaryDirectory() as temporary: target = Path(temporary) / "result" - result = self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target) + with patch.object(runner_module, "SkuSelectionFlow", _CompletedFlow): + result = self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target) self.assertEqual(result.unit_price, "12.88") manifest = result.manifest_path.read_text(encoding="utf-8") @@ -643,7 +743,7 @@ class SkuSelectionRunnerTests(unittest.TestCase): self.assertIn('"panel_status": "verified"', manifest) self.assertIn('"safe_exit": "completed"', manifest) self.assertFalse((target / "hierarchy.xml").exists()) - self.assertEqual(_actions(device, "pressKey"), [("jsonrpc", "pressKey", ["back"], 10)]) + self.assertEqual(_actions(device, "pressKey"), []) def test_failure_stage_is_fixed_control_flow_metadata_without_error_text(self) -> None: class NoPanelAfterEntry(_RawDevice): @@ -729,7 +829,11 @@ class SkuSelectionRunnerTests(unittest.TestCase): (Path(destination) / "sentinel").write_text("keep", encoding="utf-8") original_rename(source, destination) - with patch.object(runner_module.os, "rename", side_effect=create_target_then_rename), self.assertRaises(SkuSelectionRunError): + with ( + patch.object(runner_module, "SkuSelectionFlow", _CompletedFlow), + patch.object(runner_module.os, "rename", side_effect=create_target_then_rename), + self.assertRaises(SkuSelectionRunError), + ): self._runner(_FakeAdb(), _RawDevice()).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target) self.assertEqual((target / "sentinel").read_text(encoding="utf-8"), "keep") self.assertEqual(list(Path(temporary).glob(".result.staging-*")), []) @@ -737,14 +841,20 @@ class SkuSelectionRunnerTests(unittest.TestCase): def test_bad_screenshot_or_existing_target_never_publishes_manifest(self) -> None: with TemporaryDirectory() as temporary: target = Path(temporary) / "result" - with self.assertRaises(SkuSelectionScreenshotError) as screenshot_failure: + with ( + patch.object(runner_module, "SkuSelectionFlow", _CompletedFlow), + self.assertRaises(SkuSelectionScreenshotError) as screenshot_failure, + ): self._runner(_FakeAdb(), _RawDevice(screenshot="not-image")).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target) self.assertEqual(safe_failure_stage(screenshot_failure.exception), "screenshot_capture") self.assertFalse(target.exists()) self.assertEqual(list(Path(temporary).glob(".result.staging-*")), []) target = Path(temporary) / "write-failure" - with patch.object(runner_module, "_save_base64_screenshot", side_effect=OSError("private path")): + with ( + patch.object(runner_module, "SkuSelectionFlow", _CompletedFlow), + patch.object(runner_module, "_save_base64_screenshot", side_effect=OSError("private path")), + ): with self.assertRaises(SkuSelectionScreenshotError): self._runner(_FakeAdb(), _RawDevice()).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target) self.assertFalse(target.exists()) @@ -792,7 +902,11 @@ class SkuSelectionRunnerTests(unittest.TestCase): def test_small_but_valid_png_is_not_accepted(self) -> None: image = Image.new("RGB", (1, 1), "white") raw = BytesIO(); image.save(raw, format="PNG") - with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionScreenshotError): + with ( + TemporaryDirectory() as temporary, + patch.object(runner_module, "SkuSelectionFlow", _CompletedFlow), + self.assertRaises(SkuSelectionScreenshotError), + ): self._runner(_FakeAdb(), _RawDevice(screenshot=base64.b64encode(raw.getvalue()).decode("ascii"))).run( "device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result" ) @@ -907,14 +1021,8 @@ class SkuSelectionRunnerTests(unittest.TestCase): with self.assertRaises(SkuSelectionError): SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).exit_sku_panel_safely() self.assertEqual(_actions(device, "pressKey"), []) - def test_screenshot_then_foreground_drift_publishes_nothing_and_never_back(self) -> None: - class ForegroundDriftDevice(_RawDevice): - def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str: - value = super().jsonrpc_call(method, params, timeout) - if method == "takeScreenshot": self.package = "other" - return value - - device = ForegroundDriftDevice() + def test_unproven_reveal_publishes_nothing_and_safely_exits_once(self) -> None: + device = _RawDevice() with TemporaryDirectory() as temporary: target = Path(temporary) / "out" with self.assertRaises(SkuSelectionError): @@ -922,7 +1030,7 @@ class SkuSelectionRunnerTests(unittest.TestCase): self.assertFalse(target.exists()) self.assertFalse((target / "manifest.json").exists()) self.assertEqual(list(Path(temporary).glob(".out.staging-*")), []) - self.assertEqual(_actions(device, "pressKey"), []) + self.assertEqual(len(_actions(device, "pressKey")), 1) def test_option_timeout_reconciliation_controls_back_once(self) -> None: class OptionTimeoutDevice(_RawDevice): @@ -939,7 +1047,7 @@ class SkuSelectionRunnerTests(unittest.TestCase): for delivered, expected_back in ((False, 0), (True, 1)): with self.subTest(delivered=delivered), TemporaryDirectory() as temporary: - device = OptionTimeoutDevice(delivered); device.select_alternates() + device = OptionTimeoutDevice(delivered) adb = _FakeAdb(); device.hierarchy = "" adb.on_intent = lambda: setattr(device, "hierarchy", _PRODUCT_PAGE.replace("", '')) runner = SkuSelectionRunner(adb, lambda serial: device, .03) @@ -956,9 +1064,9 @@ class SkuSelectionRunnerTests(unittest.TestCase): raise TimeoutError("back uncertain") return super().jsonrpc_call(method, params, timeout) - device = BackTimeoutDevice() - with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionRunError): - self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "out") + device = BackTimeoutDevice(_M_FIXTURE.read_text(encoding="utf-8")) + with self.assertRaises(SkuSelectionRunError): + SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 0.03)).exit_sku_panel_safely() self.assertEqual(len(_actions(device, "pressKey")), 1) def test_entry_click_timeout_reconciles_only_through_verified_flow_exit(self) -> None: diff --git a/docs/tasks/T-103.md b/docs/tasks/T-103.md index 8e39ffb..e24528b 100644 --- a/docs/tasks/T-103.md +++ b/docs/tasks/T-103.md @@ -25,7 +25,7 @@ write_paths: - docs/current-state.md --- - + ## 问题 / 背景 T-102 已证明 canonical 链接可进入目标商品。T-103 在 PKG110 / Android 16 / 拼多多 8.17.0、goods_id `937122477375` 上确认：详情页底部购买区第一行“快要抢光 + 金额”是唯一获准的受控规格入口；商品内容区同名小字和第二行“免拼购买”都不得点击。 @@ -270,6 +270,10 @@ T-103 sanitizer v2 坐标修正与主审：提交 44c027a 将 screenshot space ### 2026-08-05T03:44:34Z · ila 2026-08-05 reveal 取证方案收口：不改造只读 capture_sku_panel_spike.py，也不把未经验证的滑动放进生产 SkuSelectionFlow。新增独立 capture_sku_reveal_spike.py：无颜色/尺码/坐标/手势参数，只在精确 COLOR_SELECTED_SIZE_HIDDEN 前置成立后沿已取证列间安全通道发出一次固定手势；发出前封存，RPC 正常/超时/异常后均只读调和且绝不重试。后置必须颜色仍选中、摘要仍请选择尺码、S/M 可见且全部 selected=false、未触发提交区；成功后不 Back，由人核对 before/after 后才允许固化生产 reveal 判据。T-103 继续 Doing。 + +### 2026-08-05T04:33:56Z · ila + +2026-08-05 T-103 reveal 离线实现与终审通过：新增独立 `capture_sku_reveal_spike.py`，只在拼多多 8.17.0 / PKG110 / Android 16 / 1080×2376、goods_id `937122477375`、完整 `COLOR_SELECTED_SIZE_HIDDEN` 前置两次重证后，封存并发出唯一一次固定 reveal 手势；RPC 正常或结果不明均只读调和且绝不重试。后置要求目标颜色保持精确 selected、S/M 均可见且 selected=false、摘要仍为“请选择：尺码”、提交硬拒绝区仍在面板下方；成功不 Back，等待人工核对。四态 classifier 已绑定 A/B/S/M 真机证据的完整父链与 exact selected set；未知 clickable 祖先、窄浮层、SystemUI 同 bounds、坏 bounds 均失败关闭。独立终审 PASS；聚焦 55/55、client 全量 255/255、compileall、agent-context、diff-check 全部 PASS。未引用数量、确认页、围栏、提交或支付能力。`needs_device=true`，T-103 保持 Doing，等待项目所有者从拼多多首页、仅 Wi-Fi ADB、使用全新 output-dir 执行唯一一次真机 reveal 并人工确认 before/after。 ## 边界