From b0b71d5a29b4f5c2e7abdc518fc849d2c71123ba Mon Sep 17 00:00:00 2001 From: QiuSW <105186638@qq.com> Date: Thu, 6 Aug 2026 10:21:11 +0800 Subject: [PATCH] feat(client): add T-106 confirmation evidence capture --- client/scripts/capture_order_confirm_spike.py | 112 ++++ .../cmbuyer_client/pdd/order_confirm_spike.py | 345 ++++++++++ client/tests/pdd/test_order_confirm_spike.py | 597 ++++++++++++++++++ docs/current-state.md | 28 +- docs/tasks/T-106.md | 7 +- 5 files changed, 1085 insertions(+), 4 deletions(-) create mode 100644 client/scripts/capture_order_confirm_spike.py create mode 100644 client/src/cmbuyer_client/pdd/order_confirm_spike.py create mode 100644 client/tests/pdd/test_order_confirm_spike.py diff --git a/client/scripts/capture_order_confirm_spike.py b/client/scripts/capture_order_confirm_spike.py new file mode 100644 index 0000000..bd98b4b --- /dev/null +++ b/client/scripts/capture_order_confirm_spike.py @@ -0,0 +1,112 @@ +"""采集 T-106 人工准备的确认页四态证据；不执行页面操作。""" + +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.order_confirm_spike import ( + Android16ForegroundReader, + DECLARED_STATES, + EXPECTED_GOODS_ID, + OrderConfirmEvidenceCapturer, + OrderConfirmEvidenceError, +) + + +def parse_arguments(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description="采集 T-106 人工准备的确认页四态本机证据。") + parser.add_argument("--serial", required=True, help="ADB device serial；禁止自动选择。") + parser.add_argument("--goods-id", required=True, help="T-106 已批准的目标商品标识。") + parser.add_argument( + "--state", + required=True, + choices=DECLARED_STATES, + help=( + "人工声明状态：Gate2 后导航来源、确认页 Gate3、最终控件可见、" + "人工一次返回后的安全页。" + ), + ) + parser.add_argument("--output-dir", required=True, type=Path, help="全新本机证据目录；不得覆盖。") + parser.add_argument("--timeout", type=float, default=10.0, help="ADB 与只读设备 RPC 超时（秒）。") + 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() + or arguments.serial != 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-106 已批准目标。") + if type(arguments.state) is not str or arguments.state not in DECLARED_STATES: + raise ValueError("--state 必须是批准的人工声明状态。") + if not isinstance(arguments.output_dir, Path) or not arguments.output_dir.name: + 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 的有限数值。") + + +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 + + adb_runner = SubprocessAdbRunner(arguments.adb) + capturer = OrderConfirmEvidenceCapturer( + AdbClient(adb_runner, timeout_seconds=arguments.timeout), + NoReconnectUiautomatorConnector( + adbutils.AdbClient(socket_timeout=arguments.timeout).device_list, + u2.connect, + ), + Android16ForegroundReader(adb_runner, arguments.timeout), + timeout_seconds=arguments.timeout, + ) + try: + capturer.capture( + arguments.serial, + arguments.goods_id, + arguments.state, + arguments.output_dir, + ) + except (DeviceConnectionError, OrderConfirmEvidenceError): + # 第三方异常可能带设备、页面正文或本机目录，命令行只输出固定摘要。 + print("确认页四态只读取证失败：已停止，未发布本机证据目录。", file=sys.stderr) + return 1 + except OSError: + print("确认页四态只读取证失败：无法发布本机证据目录。", file=sys.stderr) + return 1 + + print("确认页四态只读取证完成。") + print("人工复核：请在指定目录检查截图、XML、应用摘要和 manifest。") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/client/src/cmbuyer_client/pdd/order_confirm_spike.py b/client/src/cmbuyer_client/pdd/order_confirm_spike.py new file mode 100644 index 0000000..0ec1b49 --- /dev/null +++ b/client/src/cmbuyer_client/pdd/order_confirm_spike.py @@ -0,0 +1,345 @@ +"""T-106 确认页四态的纯只读取证；页面对应性只由人工确认。""" + +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 re +import shutil +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, CommandRunner, DeviceConnectionError, DeviceInspection +from ..device.baseline import ( + HIERARCHY_PARAMS, + PDD_PACKAGE, + SCREENSHOT_PARAMS, + _save_base64_screenshot, + _sha256_file, + _validate_hierarchy, +) +from .product_open import EXPECTED_PDD_VERSION +from .product_url import parse_product_url + + +EXPECTED_GOODS_ID = "937122477375" +EXPECTED_DEVICE_MODEL = "PKG110" +EXPECTED_ANDROID_VERSION = "16" +EXPECTED_SCREEN_SIZE = (1080, 2376) +DECLARED_STATES = ( + "gate2-navigation-source", + "confirm-gate3", + "submit-control-visible", + "returned-safe-page", +) + + +class OrderConfirmEvidenceError(RuntimeError): + """T-106 四态证据未形成完整原子产物。""" + + +class OrderConfirmEvidenceTimeoutError(OrderConfirmEvidenceError): + """只读取证设备调用超时。""" + + +class OrderConfirmReadDevice(Protocol): + """T-106 唯一设备边界；故意只有读取能力。""" + + def app_info(self, package_name: str) -> dict[str, Any]: ... + + def window_size(self) -> tuple[int, int]: ... + + def jsonrpc_call(self, method: str, params: Any = None, timeout: float = 10) -> Any: ... + + +class OrderConfirmForegroundReader(Protocol): + """读取 Android 前台摘要，不暴露通用命令执行。""" + + def read(self, serial: str) -> dict[str, str]: ... + + +class Android16ForegroundReader: + """读取 Android 16 的唯一 top-resumed Activity。""" + + _TOP_RESUMED_PATTERN = re.compile( + r"(?m)^\s*topResumedActivity=ActivityRecord\{[^\r\n}]*?\s+u\d+\s+" + r"(?P[^/\s]+)/(?P[^\s}]+)\s+t\d+\}\s*$" + ) + + def __init__(self, runner: CommandRunner, timeout_seconds: float) -> None: + if not _is_positive_finite(timeout_seconds): + raise ValueError("timeout_seconds 必须是大于 0 的有限数值") + self._runner = runner + self._timeout_seconds = timeout_seconds + + def read(self, serial: str) -> dict[str, str]: + if type(serial) is not str or not serial.strip() or serial != serial.strip(): + raise OrderConfirmEvidenceError("必须显式提供非空设备通道。") + result = self._runner.run( + ("-s", serial, "shell", "dumpsys", "activity", "activities"), + self._timeout_seconds, + ) + if result.returncode != 0: + raise OrderConfirmEvidenceError("Android 前台摘要读取失败，未发布证据。") + matches = list(self._TOP_RESUMED_PATTERN.finditer(result.stdout)) + if len(matches) != 1: + raise OrderConfirmEvidenceError("Android 前台摘要不唯一，未发布证据。") + match = matches[0] + return {"package": match.group("package"), "activity": match.group("activity")} + + +@dataclass(frozen=True) +class OrderConfirmEvidenceResult: + output_directory: Path + manifest_path: Path + screenshot_path: Path + hierarchy_path: Path + app_path: Path + + +class OrderConfirmEvidenceCapturer: + """采集人工准备的单个稳定状态，不解析页面业务字段。""" + + def __init__( + self, + adb_client: AdbClient, + connector: Callable[[str], OrderConfirmReadDevice], + foreground_reader: OrderConfirmForegroundReader, + timeout_seconds: float, + ) -> None: + if not _is_positive_finite(timeout_seconds): + raise ValueError("timeout_seconds 必须是大于 0 的有限数值") + self._adb_client = adb_client + self._connector = connector + self._foreground_reader = foreground_reader + self._timeout_seconds = timeout_seconds + self._started = False + + def capture( + self, + serial: str, + goods_id: str, + human_declared_state: str, + output_directory: Path, + ) -> OrderConfirmEvidenceResult: + if self._started: + raise OrderConfirmEvidenceError("同一取证器不可重复调用。") + self._started = True + _validate_inputs(serial, goods_id, human_declared_state) + target = Path(output_directory) + _validate_new_target(target) + + staging: Path | None = None + try: + inspection = self._adb_client.inspect(serial) + _require_expected_device(inspection) + device = self._connector(serial) + initial_app = _require_read_precondition( + device, + self._foreground_reader.read(serial), + ) + + target.parent.mkdir(parents=True, exist_ok=True) + staging = target.parent / f".{target.name}.staging-{uuid4().hex}" + staging.mkdir() + + screenshot_path = staging / "screenshot.png" + screenshot_payload = _read_rpc( + device, + "takeScreenshot", + SCREENSHOT_PARAMS, + self._timeout_seconds, + ) + if not isinstance(screenshot_payload, str): + raise OrderConfirmEvidenceError("页面截图无效，未发布证据。") + _save_base64_screenshot(screenshot_payload, screenshot_path) + _require_screenshot(screenshot_path) + + hierarchy = _read_rpc( + device, + "dumpWindowHierarchy", + HIERARCHY_PARAMS, + self._timeout_seconds, + ) + _validate_hierarchy(hierarchy) + hierarchy_path = staging / "hierarchy.xml" + hierarchy_path.write_text(hierarchy, encoding="utf-8") + + final_app = _require_read_precondition( + device, + self._foreground_reader.read(serial), + ) + if final_app != initial_app: + raise OrderConfirmEvidenceError("取证期间前台页面漂移，未发布证据。") + + app_path = staging / "app.json" + app_path.write_text( + json.dumps(final_app, ensure_ascii=False, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + manifest_path = staging / "manifest.json" + manifest_path.write_text( + json.dumps( + _manifest( + inspection, + serial, + goods_id, + human_declared_state, + screenshot_path, + hierarchy_path, + app_path, + ), + ensure_ascii=False, + indent=2, + sort_keys=True, + ) + + "\n", + encoding="utf-8", + ) + os.rename(staging, target) + staging = None + except (DeviceConnectionError, OrderConfirmEvidenceError): + _clean_staging(staging) + raise + except (AdbTimeout, HTTPTimeoutError, TimeoutError) as error: + _clean_staging(staging) + raise OrderConfirmEvidenceTimeoutError("确认页四态只读取证超时，未发布证据。") from error + except (OSError, UnidentifiedImageError, ValueError) as error: + _clean_staging(staging) + raise OrderConfirmEvidenceError("确认页四态证据无法原子发布，未发布证据。") from error + except Exception as error: + _clean_staging(staging) + raise OrderConfirmEvidenceError("确认页四态只读取证未完成，未发布证据。") from error + + return OrderConfirmEvidenceResult( + output_directory=target, + manifest_path=target / "manifest.json", + screenshot_path=target / "screenshot.png", + hierarchy_path=target / "hierarchy.xml", + app_path=target / "app.json", + ) + + +def _is_positive_finite(value: object) -> bool: + return ( + isinstance(value, (int, float)) + and not isinstance(value, bool) + and value > 0 + and isfinite(value) + ) + + +def _validate_inputs(serial: object, goods_id: object, state: object) -> None: + if type(serial) is not str or not serial.strip() or serial != serial.strip(): + raise OrderConfirmEvidenceError("必须显式提供非空设备通道。") + if type(goods_id) is not str or goods_id != EXPECTED_GOODS_ID: + raise OrderConfirmEvidenceError("商品不是 T-106 已批准取证目标。") + parse_product_url(f"https://mobile.yangkeduo.com/goods.html?goods_id={goods_id}") + if type(state) is not str or state not in DECLARED_STATES: + raise OrderConfirmEvidenceError("人工声明状态无效。") + + +def _validate_new_target(target: Path) -> None: + if target.exists() or not target.name: + raise OrderConfirmEvidenceError("输出目录必须是不存在的明确新目录。") + + +def _require_expected_device(inspection: DeviceInspection) -> None: + if inspection.model != EXPECTED_DEVICE_MODEL or inspection.android_version != EXPECTED_ANDROID_VERSION: + raise OrderConfirmEvidenceError("设备不是已批准取证组合。") + + +def _require_read_precondition( + device: OrderConfirmReadDevice, + current: object, +) -> dict[str, str]: + info = 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 OrderConfirmEvidenceError("拼多多版本不是已批准取证版本。") + if not isinstance(current, dict) or current.get("package") != PDD_PACKAGE: + raise OrderConfirmEvidenceError("拼多多不在前台。") + activity = current.get("activity") + if not isinstance(activity, str) or not activity.strip(): + raise OrderConfirmEvidenceError("前台应用摘要不完整。") + if device.window_size() != EXPECTED_SCREEN_SIZE: + raise OrderConfirmEvidenceError("屏幕坐标空间不是已批准尺寸。") + return {"package": PDD_PACKAGE, "activity": activity, "pdd_version": EXPECTED_PDD_VERSION} + + +def _read_rpc( + device: OrderConfirmReadDevice, + method: str, + params: object, + timeout_seconds: float, +) -> object: + return device.jsonrpc_call(method, params, timeout=timeout_seconds) + + +def _require_screenshot(path: Path) -> None: + with Image.open(path) as image: + image.load() + if image.size != EXPECTED_SCREEN_SIZE or image.format != "PNG": + raise OrderConfirmEvidenceError("页面截图格式或尺寸无效。") + + +def _clean_staging(staging: Path | None) -> None: + if staging is not None and staging.exists(): + shutil.rmtree(staging) + + +def _manifest( + inspection: DeviceInspection, + serial: str, + goods_id: str, + state: str, + screenshot_path: Path, + hierarchy_path: Path, + app_path: Path, +) -> dict[str, Any]: + return { + "schema_version": 1, + "operation": "t106-order-confirm-readonly-evidence", + "captured_at": datetime.now(UTC).isoformat(), + "product": { + "goods_id": goods_id, + "canonical_url": f"https://mobile.yangkeduo.com/goods.html?goods_id={goods_id}", + }, + "human_declared_state": state, + "review_status": "human_review_required", + "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, + }, + "artifacts": [ + { + "path": screenshot_path.name, + "role": "human_prepared_state_raw_screenshot", + "sha256": _sha256_file(screenshot_path), + }, + { + "path": hierarchy_path.name, + "role": "human_prepared_state_raw_hierarchy_local_only", + "sha256": _sha256_file(hierarchy_path), + }, + { + "path": app_path.name, + "role": "human_prepared_state_app_identity", + "sha256": _sha256_file(app_path), + }, + ], + } diff --git a/client/tests/pdd/test_order_confirm_spike.py b/client/tests/pdd/test_order_confirm_spike.py new file mode 100644 index 0000000..fc2b76f --- /dev/null +++ b/client/tests/pdd/test_order_confirm_spike.py @@ -0,0 +1,597 @@ +"""T-106 确认页四态纯只读取证的离线测试。""" + +from __future__ import annotations + +import argparse +import ast +import base64 +from contextlib import redirect_stderr, redirect_stdout +from functools import lru_cache +from importlib.util import module_from_spec, spec_from_file_location +from io import BytesIO, StringIO +import json +from pathlib import Path +import sys +from tempfile import TemporaryDirectory +import unittest +from unittest.mock import patch + +from PIL import Image + + +CLIENT_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(CLIENT_ROOT / "src")) + +from cmbuyer_client.device.adb import AdbDevice, CommandResult, DeviceInspection +from cmbuyer_client.pdd.order_confirm_spike import ( + Android16ForegroundReader, + DECLARED_STATES, + EXPECTED_GOODS_ID, + OrderConfirmEvidenceCapturer, + OrderConfirmEvidenceError, + OrderConfirmEvidenceTimeoutError, + OrderConfirmForegroundReader, + OrderConfirmReadDevice, +) + + +SERIAL = "192.168.0.173:5555" +HIERARCHY = ( + "" + "" +) +TOP_RESUMED = ( + " topResumedActivity=ActivityRecord{101034589 u0 " + "com.xunmeng.pinduoduo/.activity.NewPageActivity t1816}\n" +) + + +@lru_cache(maxsize=1) +def _png_base64() -> str: + raw = BytesIO() + Image.new("RGB", (1080, 2376), color="white").save(raw, format="PNG") + return base64.b64encode(raw.getvalue()).decode("ascii") + + +class FakeAdbClient: + def __init__(self, *, model: str = "PKG110", android_version: str = "16") -> None: + self.calls: list[str] = [] + self.inspection = DeviceInspection( + device=AdbDevice(serial=SERIAL, state="device", model=model), + model=model, + android_version=android_version, + ) + + def inspect(self, serial: str) -> DeviceInspection: + self.calls.append(serial) + return self.inspection + + +class FakeReadDevice: + def __init__( + self, + *, + version: str = "8.17.0", + screen_size: tuple[int, int] = (1080, 2376), + screenshot: object | None = None, + hierarchy: object = HIERARCHY, + ) -> None: + self.version = version + self.screen_size = screen_size + self.screenshot = _png_base64() if screenshot is None else screenshot + self.hierarchy = hierarchy + self.calls: list[tuple[object, ...]] = [] + self.app_info_reads = 0 + self.window_reads = 0 + self.post_version: str | None = None + self.post_screen_size: tuple[int, int] | None = None + + def app_info(self, package_name: str) -> dict[str, str]: + self.calls.append(("app_info", package_name)) + self.app_info_reads += 1 + version = self.post_version if self.app_info_reads > 1 and self.post_version else self.version + return {"versionName": version} + + def window_size(self) -> tuple[int, int]: + self.calls.append(("window_size",)) + self.window_reads += 1 + if self.window_reads > 1 and self.post_screen_size is not None: + return self.post_screen_size + return self.screen_size + + def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> object: + self.calls.append(("jsonrpc", method, params, timeout)) + if method == "takeScreenshot": + return self.screenshot + if method == "dumpWindowHierarchy": + return self.hierarchy + raise AssertionError("unexpected read RPC") + + +class FakeForegroundReader: + def __init__( + self, + *, + package: str = "com.xunmeng.pinduoduo", + activity: str = ".activity.NewPageActivity", + ) -> None: + self.package = package + self.activity = activity + self.post_package: str | None = None + self.post_activity: str | None = None + self.calls: list[str] = [] + + def read(self, serial: str) -> dict[str, str]: + self.calls.append(serial) + package = self.post_package if len(self.calls) > 1 and self.post_package else self.package + activity = self.post_activity if len(self.calls) > 1 and self.post_activity else self.activity + return {"package": package, "activity": activity} + + +class FakeCommandRunner: + def __init__(self, result: CommandResult) -> None: + self.result = result + self.calls: list[tuple[tuple[str, ...], float]] = [] + + def run(self, arguments: tuple[str, ...], timeout_seconds: float) -> CommandResult: + self.calls.append((arguments, timeout_seconds)) + return self.result + + +def _load_script() -> object: + path = CLIENT_ROOT / "scripts" / "capture_order_confirm_spike.py" + specification = spec_from_file_location("capture_order_confirm_spike_for_test", path) + assert specification is not None and specification.loader is not None + module = module_from_spec(specification) + specification.loader.exec_module(module) + return module + + +def _namespace(**changes: object) -> argparse.Namespace: + values: dict[str, object] = { + "serial": SERIAL, + "goods_id": EXPECTED_GOODS_ID, + "state": DECLARED_STATES[0], + "output_dir": Path("evidence"), + "timeout": 10.0, + "adb": "adb", + } + values.update(changes) + return argparse.Namespace(**values) + + +class Android16ForegroundReaderTests(unittest.TestCase): + def test_reads_exact_unique_top_resumed_activity(self) -> None: + runner = FakeCommandRunner(CommandResult(stdout=TOP_RESUMED)) + reader = Android16ForegroundReader(runner, timeout_seconds=7) + + self.assertEqual( + reader.read(SERIAL), + { + "package": "com.xunmeng.pinduoduo", + "activity": ".activity.NewPageActivity", + }, + ) + self.assertEqual( + runner.calls, + [ + ( + ("-s", SERIAL, "shell", "dumpsys", "activity", "activities"), + 7, + ) + ], + ) + + def test_missing_legacy_duplicate_or_failed_summary_is_rejected(self) -> None: + outputs = ( + "", + "mResumedActivity: ActivityRecord{1 u0 com.xunmeng.pinduoduo/.Main t1}\n", + TOP_RESUMED + TOP_RESUMED, + "topResumedActivity=null\n", + ) + for output in outputs: + with self.subTest(output=output): + reader = Android16ForegroundReader( + FakeCommandRunner(CommandResult(stdout=output)), + timeout_seconds=7, + ) + with self.assertRaises(OrderConfirmEvidenceError): + reader.read(SERIAL) + + reader = Android16ForegroundReader( + FakeCommandRunner(CommandResult(stdout=TOP_RESUMED, returncode=1)), + timeout_seconds=7, + ) + with self.assertRaises(OrderConfirmEvidenceError): + reader.read(SERIAL) + with self.assertRaises(OrderConfirmEvidenceError): + reader.read(f" {SERIAL}") + + +class OrderConfirmEvidenceTests(unittest.TestCase): + def _capturer( + self, + adb: FakeAdbClient, + device: FakeReadDevice, + *, + foreground: FakeForegroundReader | None = None, + connector_calls: list[str] | None = None, + ) -> OrderConfirmEvidenceCapturer: + def connect(serial: str) -> FakeReadDevice: + if connector_calls is not None: + connector_calls.append(serial) + return device + + return OrderConfirmEvidenceCapturer( + adb, # type: ignore[arg-type] + connect, + foreground or FakeForegroundReader(), + timeout_seconds=2, + ) + + def test_all_four_human_states_publish_only_raw_read_evidence(self) -> None: + self.assertEqual(len(DECLARED_STATES), 4) + for state in DECLARED_STATES: + with self.subTest(state=state), TemporaryDirectory() as temporary: + adb = FakeAdbClient() + device = FakeReadDevice() + foreground = FakeForegroundReader() + connector_calls: list[str] = [] + target = Path(temporary) / state + + result = self._capturer( + adb, + device, + foreground=foreground, + connector_calls=connector_calls, + ).capture(SERIAL, EXPECTED_GOODS_ID, state, target) + + manifest_text = result.manifest_path.read_text(encoding="utf-8") + manifest = json.loads(manifest_text) + app = json.loads(result.app_path.read_text(encoding="utf-8")) + self.assertEqual(adb.calls, [SERIAL]) + self.assertEqual(connector_calls, [SERIAL]) + self.assertEqual(foreground.calls, [SERIAL, SERIAL]) + self.assertEqual( + [call[1] for call in device.calls if call[0] == "jsonrpc"], + ["takeScreenshot", "dumpWindowHierarchy"], + ) + self.assertEqual(manifest["operation"], "t106-order-confirm-readonly-evidence") + self.assertEqual(manifest["human_declared_state"], state) + self.assertEqual(manifest["review_status"], "human_review_required") + self.assertEqual(manifest["product"]["goods_id"], EXPECTED_GOODS_ID) + self.assertEqual(app["package"], "com.xunmeng.pinduoduo") + self.assertEqual( + {path.name for path in target.iterdir()}, + {"screenshot.png", "hierarchy.xml", "app.json", "manifest.json"}, + ) + self.assertNotIn(SERIAL, manifest_text) + self.assertNotIn("NewPageActivity", manifest_text) + self.assertNotIn("local raw page", manifest_text) + for artifact in manifest["artifacts"]: + self.assertEqual(len(artifact["sha256"]), 64) + + def test_invalid_inputs_and_existing_target_stop_before_device_access(self) -> None: + scenarios = ( + ("", EXPECTED_GOODS_ID, DECLARED_STATES[0]), + (f" {SERIAL}", EXPECTED_GOODS_ID, DECLARED_STATES[0]), + (SERIAL, "958756616606", DECLARED_STATES[0]), + (SERIAL, EXPECTED_GOODS_ID, "unknown"), + ) + for serial, goods_id, state in scenarios: + with self.subTest(state=state), TemporaryDirectory() as temporary: + adb = FakeAdbClient() + connector_calls: list[str] = [] + with self.assertRaises(OrderConfirmEvidenceError): + self._capturer( + adb, + FakeReadDevice(), + connector_calls=connector_calls, + ).capture(serial, goods_id, state, Path(temporary) / "evidence") + self.assertEqual(adb.calls, []) + self.assertEqual(connector_calls, []) + + with TemporaryDirectory() as temporary: + target = Path(temporary) / "evidence" + target.mkdir() + sentinel = target / "sentinel.txt" + sentinel.write_text("keep", encoding="utf-8") + adb = FakeAdbClient() + with self.assertRaises(OrderConfirmEvidenceError): + self._capturer(adb, FakeReadDevice()).capture( + SERIAL, EXPECTED_GOODS_ID, DECLARED_STATES[0], target + ) + self.assertEqual(adb.calls, []) + self.assertEqual(sentinel.read_text(encoding="utf-8"), "keep") + + def test_device_version_package_activity_and_screen_mismatch_fail_closed(self) -> None: + scenarios = ( + (FakeAdbClient(model="OTHER"), FakeReadDevice(), FakeForegroundReader()), + (FakeAdbClient(android_version="15"), FakeReadDevice(), FakeForegroundReader()), + (FakeAdbClient(), FakeReadDevice(version="8.17.1"), FakeForegroundReader()), + (FakeAdbClient(), FakeReadDevice(), FakeForegroundReader(package="com.example.other")), + (FakeAdbClient(), FakeReadDevice(), FakeForegroundReader(activity="")), + (FakeAdbClient(), FakeReadDevice(screen_size=(1080, 2400)), FakeForegroundReader()), + ) + for adb, device, foreground in scenarios: + with self.subTest(device=device.__dict__), TemporaryDirectory() as temporary: + target = Path(temporary) / "evidence" + with self.assertRaises(OrderConfirmEvidenceError): + self._capturer(adb, device, foreground=foreground).capture( + SERIAL, EXPECTED_GOODS_ID, DECLARED_STATES[0], target + ) + self.assertFalse(target.exists()) + self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), []) + self.assertEqual([call for call in device.calls if call[0] == "jsonrpc"], []) + + def test_invalid_screenshot_or_hierarchy_never_publishes(self) -> None: + devices = ( + FakeReadDevice(screenshot="not base64"), + FakeReadDevice(screenshot=123), + FakeReadDevice(hierarchy=""), + FakeReadDevice(hierarchy=123), + ) + for device in devices: + with self.subTest(value=device.screenshot), TemporaryDirectory() as temporary: + target = Path(temporary) / "evidence" + with self.assertRaises(OrderConfirmEvidenceError): + self._capturer(FakeAdbClient(), device).capture( + SERIAL, EXPECTED_GOODS_ID, DECLARED_STATES[0], target + ) + self.assertFalse(target.exists()) + self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), []) + + def test_timeout_and_unexpected_failure_are_redacted(self) -> None: + class BrokenDevice(FakeReadDevice): + def __init__(self, failure: BaseException) -> None: + super().__init__() + self.failure = failure + + def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> object: + raise self.failure + + failures = ( + (TimeoutError(f"secret {SERIAL} C:\\private\\raw.xml"), OrderConfirmEvidenceTimeoutError), + (RuntimeError(f"secret {SERIAL} C:\\private\\raw.xml"), OrderConfirmEvidenceError), + ) + for failure, expected in failures: + with self.subTest(expected=expected), TemporaryDirectory() as temporary: + target = Path(temporary) / "evidence" + with self.assertRaises(expected) as raised: + self._capturer(FakeAdbClient(), BrokenDevice(failure)).capture( + SERIAL, EXPECTED_GOODS_ID, DECLARED_STATES[0], target + ) + self.assertNotIn(SERIAL, str(raised.exception)) + self.assertNotIn("private", str(raised.exception)) + self.assertFalse(target.exists()) + self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), []) + + def test_post_capture_identity_drift_never_publishes(self) -> None: + mutators = ( + lambda device, foreground: setattr(device, "post_version", "8.17.1"), + lambda device, foreground: setattr(foreground, "post_package", "com.example.other"), + lambda device, foreground: setattr(foreground, "post_activity", "OtherActivity"), + lambda device, foreground: setattr(device, "post_screen_size", (1080, 2400)), + ) + for mutate in mutators: + with TemporaryDirectory() as temporary: + device = FakeReadDevice() + foreground = FakeForegroundReader() + mutate(device, foreground) + target = Path(temporary) / "evidence" + with self.assertRaises(OrderConfirmEvidenceError): + self._capturer( + FakeAdbClient(), device, foreground=foreground + ).capture(SERIAL, EXPECTED_GOODS_ID, DECLARED_STATES[0], target) + self.assertFalse(target.exists()) + self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), []) + + def test_atomic_publish_failure_cleans_staging(self) -> None: + with TemporaryDirectory() as temporary: + target = Path(temporary) / "evidence" + with ( + patch( + "cmbuyer_client.pdd.order_confirm_spike.os.rename", + side_effect=OSError(f"private {SERIAL}"), + ), + self.assertRaises(OrderConfirmEvidenceError) as raised, + ): + self._capturer(FakeAdbClient(), FakeReadDevice()).capture( + SERIAL, EXPECTED_GOODS_ID, DECLARED_STATES[0], target + ) + self.assertNotIn(SERIAL, str(raised.exception)) + self.assertFalse(target.exists()) + self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), []) + + def test_capturer_is_one_shot_even_after_rejection(self) -> None: + capturer = self._capturer(FakeAdbClient(), FakeReadDevice()) + with TemporaryDirectory() as temporary: + with self.assertRaises(OrderConfirmEvidenceError): + capturer.capture("", EXPECTED_GOODS_ID, DECLARED_STATES[0], Path(temporary) / "bad") + with self.assertRaises(OrderConfirmEvidenceError): + capturer.capture( + SERIAL, EXPECTED_GOODS_ID, DECLARED_STATES[0], Path(temporary) / "good" + ) + + def test_public_boundaries_expose_reading_only(self) -> None: + forbidden = { + "click", + "swipe", + "press", + "pressKey", + "send_keys", + "set_text", + "start_pdd_view_intent", + "open_product", + "open_sku_panel", + "set_quantity", + "go_to_order_confirm", + "submit_order_once", + "pay", + } + self.assertTrue(forbidden.isdisjoint(OrderConfirmReadDevice.__dict__)) + self.assertTrue(forbidden.isdisjoint(OrderConfirmForegroundReader.__dict__)) + self.assertEqual( + {name for name in Android16ForegroundReader.__dict__ if not name.startswith("_")}, + {"read"}, + ) + self.assertEqual( + {name for name in OrderConfirmEvidenceCapturer.__dict__ if not name.startswith("_")}, + {"capture"}, + ) + + +class OrderConfirmCliAndStaticBoundaryTests(unittest.TestCase): + def test_cli_parses_each_approved_state(self) -> None: + script = _load_script() + for state in DECLARED_STATES: + arguments = script.parse_arguments( # type: ignore[attr-defined] + [ + "--serial", + SERIAL, + "--goods-id", + EXPECTED_GOODS_ID, + "--state", + state, + "--output-dir", + f"evidence-{state}", + ] + ) + script.validate_arguments(arguments) # type: ignore[attr-defined] + + def test_cli_validation_rejects_unapproved_values(self) -> None: + script = _load_script() + invalid = ( + _namespace(serial=""), + _namespace(serial=f" {SERIAL}"), + _namespace(goods_id="958756616606"), + _namespace(state="unknown"), + _namespace(output_dir=Path("")), + _namespace(timeout=0), + _namespace(timeout=float("inf")), + ) + for arguments in invalid: + with self.subTest(arguments=arguments), self.assertRaises(ValueError): + script.validate_arguments(arguments) # type: ignore[attr-defined] + + def test_cli_runtime_failure_never_echoes_sensitive_values(self) -> None: + script = _load_script() + fake_capturer = unittest.mock.Mock() + fake_capturer.capture.side_effect = OrderConfirmEvidenceError( + f"secret {SERIAL} C:\\private\\raw.xml body" + ) + stderr = StringIO() + with ( + patch.object(script, "OrderConfirmEvidenceCapturer", return_value=fake_capturer), + 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, + "--state", + DECLARED_STATES[0], + "--output-dir", + "C:\\private\\evidence", + ] + ) + self.assertEqual(result, 1) + self.assertNotIn(SERIAL, stderr.getvalue()) + self.assertNotIn("private", stderr.getvalue()) + self.assertNotIn("body", stderr.getvalue()) + + def test_cli_success_does_not_echo_local_path_or_device(self) -> None: + script = _load_script() + fake_capturer = unittest.mock.Mock() + stdout = StringIO() + with ( + patch.object(script, "OrderConfirmEvidenceCapturer", return_value=fake_capturer), + patch.dict( + sys.modules, + {"adbutils": unittest.mock.Mock(), "uiautomator2": unittest.mock.Mock()}, + ), + redirect_stdout(stdout), + ): + result = script.main( # type: ignore[attr-defined] + [ + "--serial", + SERIAL, + "--goods-id", + EXPECTED_GOODS_ID, + "--state", + DECLARED_STATES[0], + "--output-dir", + "C:\\private\\evidence", + ] + ) + self.assertEqual(result, 0) + self.assertNotIn(SERIAL, stdout.getvalue()) + self.assertNotIn("private", stdout.getvalue()) + fake_capturer.capture.assert_called_once() + + def test_sources_have_only_approved_read_rpc_literals_and_no_ui_mutators(self) -> None: + expected_by_source = { + Path("src/cmbuyer_client/pdd/order_confirm_spike.py"): { + "takeScreenshot", + "dumpWindowHierarchy", + }, + Path("scripts/capture_order_confirm_spike.py"): set(), + } + forbidden_attributes = { + "click", + "swipe", + "press", + "pressKey", + "send_keys", + "set_text", + "start_pdd_view_intent", + "open_product", + "open_sku_panel", + "set_quantity_and_readback", + "go_to_order_confirm", + "submit_order_once", + } + forbidden_import_fragments = { + "quantity_gate2_runner", + "sku_selection_runner", + "submission", + "payment", + } + for relative, expected_rpc in expected_by_source.items(): + source = (CLIENT_ROOT / relative).read_text(encoding="utf-8") + tree = ast.parse(source) + rpc_literals = { + node.args[1].value + for node in ast.walk(tree) + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Name) + and node.func.id == "_read_rpc" + and len(node.args) > 1 + and isinstance(node.args[1], ast.Constant) + and isinstance(node.args[1].value, str) + } + self.assertEqual(rpc_literals, expected_rpc) + observed_attributes = { + node.attr + for node in ast.walk(tree) + if isinstance(node, ast.Attribute) and node.attr in forbidden_attributes + } + self.assertEqual(observed_attributes, set()) + imported_modules = { + alias.name + for node in ast.walk(tree) + if isinstance(node, (ast.Import, ast.ImportFrom)) + for alias in node.names + } + for fragment in forbidden_import_fragments: + self.assertTrue(all(fragment not in module for module in imported_modules)) + + +if __name__ == "__main__": + unittest.main() diff --git a/docs/current-state.md b/docs/current-state.md index 265d2bd..524aa79 100644 --- a/docs/current-state.md +++ b/docs/current-state.md @@ -30,7 +30,7 @@ 已取证的固定尺码 reveal、精确 M 选择、规格面板单价 `12.88` 与一次 Back 已进入生产 Flow 并完成 真机人工验收；T-104 已把同商品详情页安全退出收紧为版本绑定、连续两帧稳定判据，数量/确认页/提交仍未开放。 - 测试：采购服务已覆盖登录、建单、授权、详情/证据、设备身份、迁移、原子 claim/renew 与竞态； - 采购工具 330 项离线单元测试（全部 mock，不连接真机）。 + 采购工具 346 项离线单元测试（全部 mock，不连接真机）。 - 数据：受保护迁移已落到 `00005_task_claims.sql`。旧两趟模型已移除；开始采购会锁定任务快照并签发 一次性授权，领取会创建唯一 attempt/claim、HMAC claim token 与有界租约。T-211 已统一 title、SKU、 goods_id 与金额的双端 wire 上限，并证明最坏合法 claim 响应小于 32 KiB。仓库不含业务实例数据。 @@ -62,11 +62,11 @@ | 路径 | 状态 | 说明 | | --- | --- | --- | | `docs/` | 已有 | 项目规范化文档，本次已完整生成 | -| `docs/tasks/` | 已有（含 T-001～T-111、T-201～T-211、T-301～T-308、T-400～T-405） | T-103～T-105、T-303、T-211 已完成；T-304 代码完成后保持 DOING 等人工 UI 复核；T-106 确认页只读取证依赖已解除；双端围栏、真实提交、调和、打包与只读验收均已落成任务 | +| `docs/tasks/` | 已有（含 T-001～T-111、T-201～T-211、T-301～T-308、T-400～T-405） | T-103～T-105、T-303、T-211 已完成；T-304 代码完成后保持 DOING 等人工 UI 复核；T-106 已领取并完成只读取证代码，等待四态真机人工验收；双端围栏、真实提交、调和、打包与只读验收均已落成任务 | | `docs/design/` | 已有（6 个原型） | web 登录 / 建单 / 工作台 / 详情，desk 采购执行 / 配置；均已人工确认 | | `scripts/` | 已有 | 上下文门禁、Vikunja 单向导出与 MCP 启动包装 | | `admin/` | 已初始化 | Go 1.23+ / gin / SQLite，含建单/授权/详情/证据/设备身份与原子 claim/renew；不执行真机动作 | -| `client/` | 已初始化 | Python 3.11+、PySide6/uiautomator2、固定双 Tab 主界面、安全轮询、严格 HTTP/DPAPI/SQLite 恢复底座及受控规格选择/读价/安全退出；数量、确认页和提交未开放 | +| `client/` | 已初始化 | Python 3.11+、PySide6/uiautomator2、固定双 Tab 主界面、安全轮询、严格 HTTP/DPAPI/SQLite 恢复底座及受控规格选择/读价/数量 Gate2/安全退出；T-106 确认页四态仅开放纯只读取证，确认页生产导航和提交仍未开放 | | `init.ps1` / `init.sh` | 已完成 | 统一安装与离线验证入口；PowerShell 优先复用合规 venv，缺失时自动选择最高的 Python 3.11+，Unix 缺工具链明确失败 | ## 任务状态 @@ -225,6 +225,28 @@ T-105 阶段二生产验收已在人工确认的同一手机、拼多多 `8.17.0 项目所有者已确认数量精确为 2、颜色尺码未变、Gate1/Gate2 分别为 `12.88/32.76`、Gate2 截图正确、 一次 Back 回到同一商品详情，且未进入确认/提交/支付页、未创建订单。该命令只作已验运行记录，不再重跑。 +T-106 已提供四态纯只读取证命令。每次运行都会重新核对设备、拼多多 `8.17.0`、前台包和 +1080×2376 屏幕，只调用截图与 `compressed=false` XML；不会打开页面、点击、返回、滑动、输入或识别 +确认页业务字段。四个输出目录必须均不存在，完整 XML 只留本机： + +```powershell +# 1. 人工准备：Gate2 已通过、尚未进入确认页；运行后再由人记录实际进入确认页的精确控件文本/容器/匹配数 +.\client\.venv\Scripts\python.exe client\scripts\capture_order_confirm_spike.py --serial 192.168.0.173:5555 --goods-id 937122477375 --state gate2-navigation-source --output-dir "$env:LOCALAPPDATA\cmbuyer\artifacts\T-106\gate2-navigation-source-937122477375" --timeout 10 --adb D:\Portable\adb\adb.exe + +# 2. 人工只点击已核对的确认页入口并停在 Gate3 页面；绝不点击最终“提交订单” +.\client\.venv\Scripts\python.exe client\scripts\capture_order_confirm_spike.py --serial 192.168.0.173:5555 --goods-id 937122477375 --state confirm-gate3 --output-dir "$env:LOCALAPPDATA\cmbuyer\artifacts\T-106\confirm-gate3-937122477375" --timeout 10 --adb D:\Portable\adb\adb.exe + +# 3. 人工把最终“提交订单”控件准备为可见但不点击；记录精确文案、匹配数和启用态 +.\client\.venv\Scripts\python.exe client\scripts\capture_order_confirm_spike.py --serial 192.168.0.173:5555 --goods-id 937122477375 --state submit-control-visible --output-dir "$env:LOCALAPPDATA\cmbuyer\artifacts\T-106\submit-control-visible-937122477375" --timeout 10 --adb D:\Portable\adb\adb.exe + +# 4. 人工只返回一次并停手；结果不明时不再返回，直接停止 +.\client\.venv\Scripts\python.exe client\scripts\capture_order_confirm_spike.py --serial 192.168.0.173:5555 --goods-id 937122477375 --state returned-safe-page --output-dir "$env:LOCALAPPDATA\cmbuyer\artifacts\T-106\returned-safe-page-937122477375" --timeout 10 --adb D:\Portable\adb\adb.exe +``` + +四态都必须由项目所有者本地核对截图/XML：规格、数量、Gate3 应付总额、最终控件精确文案/匹配数/ +启用态、一次返回后的页面身份；同时确认未点击最终提交、未创建订单、未进入支付或安全验证页面。 +满足全部人工验收前 T-106 保持 `DOING`，这些人工观察也不能在 T-106 中直接变成生产 selector。 + ## 关键背景 本项目是 `cmroubao`（Go 后端 + Android AccessibilityService）与 `cmpdd` diff --git a/docs/tasks/T-106.md b/docs/tasks/T-106.md index 96db02e..3354294 100644 --- a/docs/tasks/T-106.md +++ b/docs/tasks/T-106.md @@ -18,7 +18,7 @@ write_paths: - docs/current-state.md --- - + ## 问题 / 背景 T-105 验证规格面板数量与闸门二后，下一步会进入更高风险的订单确认页。T-106 只由人手工准备确认页、最终提交控件可见态和一次返回后的页面，再用完全只读脚本采集真机事实；本任务不自动导航、不写确认页判据。 @@ -63,6 +63,11 @@ T-105 验证规格面板数量与闸门二后，下一步会进入更高风险 2026-08-06T02:12:16.724Z · Codex 已领取 T-106。基线 init.ps1 通过（client 330 tests）；按任务边界只实现四态纯只读取证，禁止 click/press/back/swipe/input/intent，代码完成后等待人工真机准备与验收。 + +### 2026-08-06T02:20:23Z · ila + +2026-08-06T02:20:23.748Z · Codex +T-106 四态纯只读取证代码已完成：新增 order_confirm_spike、capture_order_confirm_spike.py 与 16 项专项测试。静态 AST 证明生产模块/CLI 只允许 takeScreenshot、dumpWindowHierarchy，且不可达页面修改、确认页导航、提交、围栏或付款能力。client 全量 346 tests、compileall、validate_agent_context 与完整 init.ps1 均通过。任务保持 DOING，等待项目所有者依次人工准备并确认 Gate2 导航前、Gate3、最终控件可见、一次返回后四态；绝不点击最终提交。 ## 边界