diff --git a/client/src/cmbuyer_client/pdd/product_open.py b/client/src/cmbuyer_client/pdd/product_open.py index db12e45..a5feac2 100644 --- a/client/src/cmbuyer_client/pdd/product_open.py +++ b/client/src/cmbuyer_client/pdd/product_open.py @@ -7,9 +7,11 @@ 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, Protocol from uuid import uuid4 @@ -40,7 +42,7 @@ class ProductVersionMismatchError(ProductOpenError): class ProductPackageMismatchError(ProductOpenError): - """Intent 后当前前台包不是拼多多。""" + """Intent 后在有限时间内未观察到拼多多前台包。""" class ProductOpenTimeoutError(ProductOpenError): @@ -90,12 +92,20 @@ class ProductOpenCapturer: adb_client: AdbClient, connector: Callable[[str], ProductOpenUiDevice], timeout_seconds: float, + foreground_poll_interval_seconds: float = 0.2, + monotonic_clock: Callable[[], float] = monotonic, + sleep_function: Callable[[float], None] = sleep, ) -> None: - if timeout_seconds <= 0: - raise ValueError("timeout_seconds 必须大于 0") + if not _is_positive_finite(timeout_seconds): + raise ValueError("timeout_seconds 必须是大于 0 的有限数值") + if not _is_positive_finite(foreground_poll_interval_seconds): + raise ValueError("foreground_poll_interval_seconds 必须是大于 0 的有限数值") self._adb_client = adb_client self._connector = connector self._timeout_seconds = timeout_seconds + self._foreground_poll_interval_seconds = foreground_poll_interval_seconds + self._monotonic_clock = monotonic_clock + self._sleep_function = sleep_function def open_and_capture(self, serial: str, product_url: str, output_directory: Path) -> ProductOpenResult: """完成唯一允许的 Intent 打开及其后的只读取证。""" @@ -114,7 +124,7 @@ class ProductOpenCapturer: # 版本精确匹配是 Intent 的前置条件,失败时绝不调用 start_pdd_view_intent。 intent = self._adb_client.start_pdd_view_intent(serial, link.goods_id) - _require_pdd_foreground(device.app_current()) + self._wait_for_pdd_foreground(device) target.parent.mkdir(parents=True, exist_ok=True) staging = target.parent / f".{target.name}.staging-{uuid4().hex}" @@ -174,6 +184,21 @@ class ProductOpenCapturer: hierarchy_path=target / "hierarchy.xml", ) + def _wait_for_pdd_foreground(self, device: ProductOpenUiDevice) -> None: + """只轮询当前 package,直到 deadline;Activity 和节点树均不参与本判据。""" + + deadline = self._monotonic_clock() + self._timeout_seconds + while True: + if _is_pdd_foreground(device.app_current()): + return + remaining = deadline - self._monotonic_clock() + if remaining <= 0: + raise ProductPackageMismatchError( + "商品链接打开后未在限定时间内进入拼多多,已停止后续取证。" + ) + # 每个失败观察后都等待正的、受 deadline 约束的时长,避免 busy-loop。 + self._sleep_function(min(self._foreground_poll_interval_seconds, remaining)) + def _validate_new_target(target: Path) -> None: if target.exists(): @@ -197,9 +222,12 @@ def _require_expected_version(app_info: dict[str, Any]) -> str: return version -def _require_pdd_foreground(current: dict[str, Any]) -> None: - if not isinstance(current, dict) or current.get("package") != PDD_PACKAGE: - raise ProductPackageMismatchError("商品链接打开后前台应用不是拼多多,已停止后续取证。") +def _is_positive_finite(value: object) -> bool: + return isinstance(value, (int, float)) and not isinstance(value, bool) and value > 0 and isfinite(value) + + +def _is_pdd_foreground(current: object) -> bool: + return isinstance(current, dict) and current.get("package") == PDD_PACKAGE def _manifest( diff --git a/client/tests/pdd/test_product_open.py b/client/tests/pdd/test_product_open.py index 29c12f7..68ad09b 100644 --- a/client/tests/pdd/test_product_open.py +++ b/client/tests/pdd/test_product_open.py @@ -63,11 +63,13 @@ class FakeUiDevice: *, version: str = "8.17.0", current_package: str = "com.xunmeng.pinduoduo", + current_packages: list[str] | None = None, hierarchy: str = HIERARCHY, timeout_on_screenshot: bool = False, ) -> None: self.version = version self.current_package = current_package + self.current_packages = list(current_packages) if current_packages is not None else None self.hierarchy = hierarchy self.timeout_on_screenshot = timeout_on_screenshot self.calls: list[str] = [] @@ -78,6 +80,9 @@ class FakeUiDevice: def app_current(self) -> dict[str, str]: self.calls.append("app_current") + if self.current_packages: + package = self.current_packages.pop(0) + self.current_package = package return {"package": self.current_package, "activity": "sensitive.activity.name"} def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str: @@ -92,8 +97,8 @@ class FakeUiDevice: class ProductOpenTests(unittest.TestCase): - def _capturer(self, adb: FakeAdbClient, device: FakeUiDevice) -> ProductOpenCapturer: - return ProductOpenCapturer(adb, lambda serial: device, timeout_seconds=2) + def _capturer(self, adb: FakeAdbClient, device: FakeUiDevice, **kwargs: object) -> ProductOpenCapturer: + return ProductOpenCapturer(adb, lambda serial: device, timeout_seconds=2, **kwargs) def test_success_uses_canonical_url_and_redacted_atomic_manifest(self) -> None: adb = FakeAdbClient() @@ -139,15 +144,77 @@ class ProductOpenTests(unittest.TestCase): def test_foreground_package_mismatch_halts_before_capture(self) -> None: adb = FakeAdbClient() device = FakeUiDevice(current_package="com.example.other") + clock = FakeClock() with TemporaryDirectory() as temporary: target = Path(temporary) / "evidence" with self.assertRaises(ProductPackageMismatchError): - self._capturer(adb, device).open_and_capture(SERIAL, URL, target) + self._capturer( + adb, + device, + foreground_poll_interval_seconds=0.5, + monotonic_clock=clock.monotonic, + sleep_function=clock.sleep, + ).open_and_capture(SERIAL, URL, target) self.assertEqual(adb.calls, [("inspect", SERIAL), ("intent", "123")]) - self.assertEqual(device.calls, ["app_info", "app_current"]) + self.assertEqual(device.calls, ["app_info", "app_current", "app_current", "app_current", "app_current", "app_current"]) + self.assertEqual(clock.sleeps, [0.5, 0.5, 0.5, 0.5]) self.assertFalse(target.exists()) + def test_foreground_package_poll_waits_for_pdd_before_reading_evidence(self) -> None: + adb = FakeAdbClient() + device = FakeUiDevice(current_packages=["com.example.other", "com.xunmeng.pinduoduo"]) + clock = FakeClock() + with TemporaryDirectory() as temporary: + target = Path(temporary) / "evidence" + result = self._capturer( + adb, + device, + foreground_poll_interval_seconds=0.25, + monotonic_clock=clock.monotonic, + sleep_function=clock.sleep, + ).open_and_capture(SERIAL, URL, target) + + self.assertTrue(result.manifest_path.exists()) + self.assertEqual(clock.sleeps, [0.25]) + self.assertEqual( + device.calls, + ["app_info", "app_current", "app_current", "takeScreenshot", "dumpWindowHierarchy"], + ) + + def test_foreground_package_poll_stops_at_deadline_without_evidence(self) -> None: + adb = FakeAdbClient() + device = FakeUiDevice(current_packages=["com.example.other", "", "com.example.other"]) + clock = FakeClock() + with TemporaryDirectory() as temporary: + target = Path(temporary) / "evidence" + with self.assertRaises(ProductPackageMismatchError): + self._capturer( + adb, + device, + foreground_poll_interval_seconds=0.8, + monotonic_clock=clock.monotonic, + sleep_function=clock.sleep, + ).open_and_capture(SERIAL, URL, target) + + self.assertEqual(len(clock.sleeps), 3) + for actual, expected in zip(clock.sleeps, (0.8, 0.8, 0.4), strict=True): + self.assertAlmostEqual(actual, expected) + self.assertEqual(device.calls, ["app_info", "app_current", "app_current", "app_current", "app_current"]) + self.assertFalse(target.exists()) + self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), []) + + def test_foreground_poll_interval_must_be_positive_and_finite(self) -> None: + for interval in (0, -0.1, float("inf"), float("nan"), True): + with self.subTest(interval=interval): + with self.assertRaises(ValueError): + ProductOpenCapturer( + FakeAdbClient(), + lambda serial: FakeUiDevice(), + timeout_seconds=2, + foreground_poll_interval_seconds=interval, # type: ignore[arg-type] + ) + def test_timeout_and_invalid_hierarchy_leave_no_partial_evidence(self) -> None: scenarios = ( (FakeUiDevice(timeout_on_screenshot=True), ProductOpenTimeoutError), @@ -184,3 +251,16 @@ class ProductOpenTests(unittest.TestCase): self.assertTrue(forbidden.isdisjoint(ProductOpenUiDevice.__dict__)) self.assertEqual(base64.b64decode(_png_base64())[:8], b"\x89PNG\r\n\x1a\n") + + +class FakeClock: + def __init__(self) -> None: + self.value = 0.0 + self.sleeps: list[float] = [] + + def monotonic(self) -> float: + return self.value + + def sleep(self, seconds: float) -> None: + self.sleeps.append(seconds) + self.value += seconds diff --git a/docs/03-tech-stack.md b/docs/03-tech-stack.md index 7612884..19315b4 100644 --- a/docs/03-tech-stack.md +++ b/docs/03-tech-stack.md @@ -117,8 +117,9 @@ D:\Portable\adb\adb.exe devices -l .\client\.venv\Scripts\python.exe client\scripts\capture_product_open.py --serial --url "https://mobile.yangkeduo.com/goods.html?goods_id=" --output-dir "$env:LOCALAPPDATA\cmbuyer\artifacts\T-102\product-open-" --timeout 10 --adb D:\Portable\adb\adb.exe ``` -脚本只在 intent 成功且前台 package 为拼多多后采集截图与完整 XML,不根据 Activity、节点文本或 -旧项目常量声称已到详情页。成功目录以原子方式发布,manifest 仅记录 `goods_id`、canonical URL、 +脚本在 intent 成功后,以 `--timeout` 为明确上限只读轮询前台 package;只有观察到拼多多才采集截图与 +完整 XML,超时仍 fail closed。这一等待只解决 App 异步切换,不根据 Activity、节点文本或旧项目常量 +声称已到详情页。成功目录以原子方式发布,manifest 仅记录 `goods_id`、canonical URL、 设备/App 非敏感元数据、受限命令摘要、文件路径和 SHA-256,不含原始 serial、Activity 或页面正文。 必须由人本地确认截图对应目标商品并检查截图/XML 无地址、手机号、支付信息或其他无关隐私;原始 证据不得提交 Git,人工确认前 T-102 保持 `DOING`。 diff --git a/docs/current-state.md b/docs/current-state.md index a3baba5..424af1e 100644 --- a/docs/current-state.md +++ b/docs/current-state.md @@ -20,7 +20,7 @@ - 生产代码:`admin/` 已有最小 Go 服务、健康检查、核心领域模型、SQLite 迁移与任务状态机; `client/` 已有 Python 包、PySide6 最小入口、运行目录与日志脱敏策略,以及显式 serial 的 ADB 连接边界、本地基线取证 CLI 和受限商品链接打开取证 CLI;尚无规格选择、价格读取或下单流程 -- 测试:采购服务已覆盖健康检查、核心模型、迁移与状态机等离线包级测试;采购工具 43 项离线单元测试 +- 测试:采购服务已覆盖健康检查、核心模型、迁移与状态机等离线包级测试;采购工具 46 项离线单元测试 (全部 mock,不连接真机) - 数据:SQLite 核心表与迁移已落成;无业务实例数据 - 标准启动路径:Windows PowerShell 运行 `./init.ps1`,Unix shell 运行 `./init.sh`。Windows 入口 @@ -126,8 +126,9 @@ D:\Portable\adb\adb.exe devices -l .\client\.venv\Scripts\python.exe client\scripts\capture_product_open.py --serial --url "https://mobile.yangkeduo.com/goods.html?goods_id=" --output-dir "$env:LOCALAPPDATA\cmbuyer\artifacts\T-102\product-open-" --timeout 10 --adb D:\Portable\adb\adb.exe ``` -脚本只允许 Android `VIEW` intent,并把 package 固定为 `com.xunmeng.pinduoduo`;它不点击、滑动、 -输入或判断商品页节点,也不打开规格、读取价格、进入下单或支付。运行拼多多版本必须精确为 +脚本只允许 Android `VIEW` intent,并把 package 固定为 `com.xunmeng.pinduoduo`;intent 后会在 +`--timeout` 的有限窗口内只读轮询前台 package,解决 App 异步切换造成的一次性误判,超时仍会停止。 +它不点击、滑动、输入或判断商品页节点,也不打开规格、读取价格、进入下单或支付。运行拼多多版本必须精确为 `8.17.0`,否则在 intent 前停止。成功后由人本地查看截图/XML,确认页面确为该 `goods_id` 对应商品并 检查无地址、手机号、支付信息或其他无关隐私;只回报 manifest 路径及截图/XML SHA-256,原始证据 不得提交 Git。人工确认前 T-102 必须保持 `DOING`。