feat(client): capture one-shot sku reveal evidence

2026-08-05 12:34:59 +08:00
parent 429c2d33db
commit e8ca738b33
10 changed files with 1822 additions and 180 deletions
@@ -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(
-            '<node package="" class="android.view.ViewGroup" bounds="[396,498][895,570]">',
-            '<node package="" class="android.view.ViewGroup" clickable="true" bounds="[396,498][895,570]">',
+        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 = "<hierarchy />"
        adb.on_intent = lambda: setattr(device, "hierarchy", _PRODUCT_PAGE.replace("<hierarchy>", '<hierarchy post-intent="1">'))
-        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 = "<hierarchy />"
                adb.on_intent = lambda: setattr(device, "hierarchy", _PRODUCT_PAGE.replace("<hierarchy>", '<hierarchy post-intent="1">'))
                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: