feat(client): project sanitized SKU price candidates

client/src/cmbuyer_client/device/sku_evidence_sanitizer.py

@@ -1,6 +1,7 @@
 """T-103 原始规格面板证据的本机确定性隐私脱敏。
 
-此模块只处理人工采集的本地文件：不连接设备、不理解拼多多页面，也不识别规格或价格。
+此模块只处理人工采集的本地文件：不连接设备、不识别规格；仅可按已取证的固定
+几何和严格格式，将跨隐私边界的价格叶节点投影到派生 XML。
 """
 
 from __future__ import annotations
@@ -21,7 +22,7 @@ from ..pdd.product_url import ProductUrl, ProductUrlError, parse_product_url
 from ..pdd.sku_panel_state import HUMAN_DECLARED_STATES
 
 
-SANITIZER_VERSION = "t103-privacy-v3"
+SANITIZER_VERSION = "t103-privacy-v4"
 EXPECTED_GOODS_ID = "937122477375"
 EXPECTED_PDD_VERSION = "8.17.0"
 EXPECTED_DEVICE_MODEL = "PKG110"
@@ -37,6 +38,21 @@ _FULL_PHONE_RE = re.compile(r"(?:\+?86)?1[3-9]\d{9}")
 _MASKED_PHONE_RE = re.compile(r"1[3-9]\d\*{4}\d{4}")
 _MASK_TRANSLATION = str.maketrans({"＊": "*", "•": "*", "·": "*", "×": "*", "x": "*", "X": "*"})
 _SEPARATOR_RE = re.compile(r"[\s\-‐‑‒–—―()（）]+")
+# 这两个槽位来自 T-103 当前第一态、1080×2376 XML 坐标的人工审查。它们不是通用
+# 页面判据；坐标、文本或结构任何变化都停止发布，交由人重新取证。
+_CROSSING_PRICE_BOUNDS = frozenset({(396, 503, 712, 570), (730, 503, 895, 570)})
+_PRICE_PROJECTION_ATTRIBUTES = (
+    "bounds",
+    "text",
+    "package",
+    "class",
+    "clickable",
+    "enabled",
+    "visible-to-user",
+)
+# 仅接受普通 ASCII 空格，且每个可分隔位置最多一个；禁止换行、折扣、支付/提交文案和
+# 任何其它字符。前缀捕获组用于区分当前价与至多一个划线/原价候选。
+_CROSSING_PRICE_TEXT_RE = re.compile(r" {0,1}(?:(快卖光) {0,1})?[¥￥] {0,1}[1-9]\d*\.\d{2} {0,1}\Z")
 
 
 class SkuEvidenceSanitizationError(RuntimeError):
@@ -49,9 +65,12 @@ class _CleanupStats:
 
     removed_nodes: int = 0
     cleared_crossing_nodes: int = 0
+    preserved_crossing_price_nodes: int = 0
     retained_below_nodes: int = 0
     max_right: int = 0
     max_bottom: int = 0
+    current_price_candidates: int = 0
+    original_price_candidates: int = 0
 
 
 @dataclass(frozen=True)
@@ -271,6 +290,7 @@ def _sanitize_hierarchy(source: Path, target: Path) -> _CleanupStats:
     _require_expected_xml_coordinate_space(stats)
     if stats.removed_nodes < 1 or stats.retained_below_nodes < 1:
         raise SkuEvidenceSanitizationError("原始节点树未满足隐私几何结构。")
+    _require_safe_crossing_price_projection(stats)
     if _contains_phone(root):
         raise SkuEvidenceSanitizationError("派生节点树仍包含手机号，拒绝发布。")
     ElementTree.ElementTree(root).write(target, encoding="utf-8", xml_declaration=True)
@@ -294,6 +314,9 @@ def _sanitize_node(parent: ElementTree.Element, node: ElementTree.Element, stats
         parent.remove(node)
         return
     if position == "crossing":
+        if bounds in _CROSSING_PRICE_BOUNDS and node.get("text"):
+            _project_crossing_price_node(node, stats)
+        else:
             # 全屏/跨界容器可保留其下方子节点，但自身所有属性和文本都可能含地址或手机号。
             _clear_node_text(node)
             stats.cleared_crossing_nodes += 1
@@ -340,6 +363,47 @@ def _vertical_position(bounds: tuple[int, int, int, int]) -> str:
     return "crossing"
 
 
+def _project_crossing_price_node(node: ElementTree.Element, stats: _CleanupStats) -> None:
+    """投影唯一允许的跨界价格叶节点；任何结构漂移一律拒绝发布。"""
+
+    if (
+        len(node) != 0
+        or node.get("package") != "com.xunmeng.pinduoduo"
+        or node.get("class") != "android.widget.TextView"
+        or node.get("clickable") != "false"
+        or node.get("enabled") != "true"
+        or node.get("visible-to-user") != "true"
+    ):
+        raise SkuEvidenceSanitizationError("跨界价格节点结构不匹配，拒绝发布。")
+    text = node.get("text")
+    if text is None:
+        raise SkuEvidenceSanitizationError("跨界价格节点文本不匹配，拒绝发布。")
+    match = _CROSSING_PRICE_TEXT_RE.fullmatch(text)
+    if match is None:
+        raise SkuEvidenceSanitizationError("跨界价格节点文本不匹配，拒绝发布。")
+
+    # 只有这七项经上述检查后可进入派生 XML；尤其不复制 content-desc、resource-id 等原始属性。
+    node.attrib = {attribute: node.attrib[attribute] for attribute in _PRICE_PROJECTION_ATTRIBUTES}
+    node.text = None
+    node.tail = None
+    stats.preserved_crossing_price_nodes += 1
+    if match.group(1) is not None:
+        stats.current_price_candidates += 1
+    else:
+        stats.original_price_candidates += 1
+
+
+def _require_safe_crossing_price_projection(stats: _CleanupStats) -> None:
+    """当前价必须唯一；原价仅可选且唯一，避免把任意金额释放为价格证据。"""
+
+    if (
+        stats.current_price_candidates != 1
+        or stats.original_price_candidates > 1
+        or stats.preserved_crossing_price_nodes != stats.current_price_candidates + stats.original_price_candidates
+    ):
+        raise SkuEvidenceSanitizationError("跨界价格候选不唯一或缺失，拒绝发布。")
+
+
 def _clear_node_text(node: ElementTree.Element) -> None:
     node.attrib = {"bounds": node.attrib["bounds"]} if "bounds" in node.attrib else {}
     node.text = None
@@ -415,6 +479,7 @@ def _derived_manifest(
         "privacy_cleanup": {
             "removed_nodes": cleanup_stats.removed_nodes,
             "cleared_crossing_nodes": cleanup_stats.cleared_crossing_nodes,
+            "preserved_crossing_price_nodes": cleanup_stats.preserved_crossing_price_nodes,
             "retained_below_nodes": cleanup_stats.retained_below_nodes,
             "max_right": cleanup_stats.max_right,
             "max_bottom": cleanup_stats.max_bottom,

client/tests/device/test_sku_evidence_sanitizer.py

@@ -34,6 +34,10 @@ TEST_ADDRESS = "SYNTHETIC_ADDRESS_NEVER_PUBLISH"
 FULL_PHONE = "13800138000"
 MASKED_PHONE = "138****0000"
 SAFE_TEXT = "synthetic-safe-lower-content"
+CURRENT_PRICE = "快卖光 ¥12.88"
+ORIGINAL_PRICE = "¥29.00"
+PRICE_CURRENT_BOUNDS = "[396,503][712,570]"
+PRICE_ORIGINAL_BOUNDS = "[730,503][895,570]"
 
 
 def _hash(path: Path) -> str:
@@ -48,11 +52,56 @@ def _default_xml() -> str:
     return (
         "<hierarchy rotation='0'>"
         f"<node bounds='[0,0][1080,540]' text='{TEST_ADDRESS}' content-desc='{MASKED_PHONE} {FULL_PHONE}' />"
+        f"{_price_node(CURRENT_PRICE, PRICE_CURRENT_BOUNDS)}"
+        f"{_price_node(ORIGINAL_PRICE, PRICE_ORIGINAL_BOUNDS)}"
         f"<node bounds='[0,540][1080,2376]' text='{SAFE_TEXT}' />"
         "</hierarchy>"
     )
 
 
+def _price_node(
+    text: str,
+    bounds: str,
+    *,
+    package: str = "com.xunmeng.pinduoduo",
+    node_class: str = "android.widget.TextView",
+    clickable: str = "false",
+    enabled: str = "true",
+    visible: str = "true",
+    extra_attributes: str = "",
+    children: str = "",
+) -> str:
+    attributes = (
+        f"bounds='{bounds}' text='{text}' package='{package}' class='{node_class}' "
+        f"clickable='{clickable}' enabled='{enabled}' visible-to-user='{visible}'{extra_attributes}"
+    )
+    return f"<node {attributes}>{children}</node>"
+
+
+def _xml_with_prices(
+    current: str = CURRENT_PRICE,
+    original: str = ORIGINAL_PRICE,
+    *,
+    current_node: str | None = None,
+    original_node: str | None = None,
+    include_original: bool = True,
+    extra_nodes: str = "",
+) -> str:
+    current_markup = current_node if current_node is not None else _price_node(current, PRICE_CURRENT_BOUNDS)
+    original_markup = (
+        original_node if original_node is not None else _price_node(original, PRICE_ORIGINAL_BOUNDS)
+    ) if include_original else ""
+    return (
+        "<hierarchy>"
+        f"<node bounds='[0,0][1080,540]' text='{TEST_ADDRESS}' />"
+        f"{current_markup}"
+        f"{original_markup}"
+        f"{extra_nodes}"
+        f"<node bounds='[0,570][1080,2376]' text='{SAFE_TEXT}' />"
+        "</hierarchy>"
+    )
+
+
 def _write_raw(
     root: Path,
     *,
@@ -117,7 +166,7 @@ class SkuEvidenceSanitizerTests(unittest.TestCase):
                 self.assertNotIn(MASKED_PHONE, derived_xml)
                 self.assertIn(f'"human_declared_state": "{state}"', manifest)
                 self.assertIn('"privacy_tier": "SANITIZED"', manifest)
-                self.assertIn('"sanitizer_version": "t103-privacy-v3"', manifest)
+                self.assertIn('"sanitizer_version": "t103-privacy-v4"', manifest)
                 self.assertIn('"screenshot_space": {', manifest)
                 self.assertIn('"xml_coordinate_space": {', manifest)
                 self.assertIn('"height": 2376', manifest)
@@ -125,6 +174,7 @@ class SkuEvidenceSanitizerTests(unittest.TestCase):
                 self.assertIn('"privacy_mask_rectangle": [', manifest)
                 self.assertIn('"removed_nodes": 1', manifest)
                 self.assertIn('"cleared_crossing_nodes": 0', manifest)
+                self.assertIn('"preserved_crossing_price_nodes": 2', manifest)
                 self.assertIn('"retained_below_nodes": 1', manifest)
                 self.assertIn('"max_right": 1080', manifest)
                 self.assertIn('"max_bottom": 2376', manifest)
@@ -140,6 +190,8 @@ class SkuEvidenceSanitizerTests(unittest.TestCase):
             "<hierarchy>"
             f"<node bounds='[0,0][1080,2376]' text='{TEST_ADDRESS}' content-desc='{MASKED_PHONE}'>"
             f"<node bounds='[0,0][1080,540]' text='{FULL_PHONE}' />"
+            f"{_price_node(CURRENT_PRICE, PRICE_CURRENT_BOUNDS)}"
+            f"{_price_node(ORIGINAL_PRICE, PRICE_ORIGINAL_BOUNDS)}"
             f"<node bounds='[0,540][1080,2376]' text='{SAFE_TEXT}' />"
             "</node></hierarchy>"
         )
@@ -152,20 +204,165 @@ class SkuEvidenceSanitizerTests(unittest.TestCase):
             self.assertIsNotNone(crossing)
             assert crossing is not None
             self.assertEqual(crossing.attrib, {"bounds": "[0,0][1080,2376]"})
-            self.assertEqual(len(list(crossing)), 1)
+            self.assertEqual(len(list(crossing)), 3)
-            self.assertEqual(list(crossing)[0].get("text"), SAFE_TEXT)
+            self.assertEqual(list(crossing)[-1].get("text"), SAFE_TEXT)
             manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
             self.assertEqual(
                 manifest["privacy_cleanup"],
                 {
                     "removed_nodes": 1,
                     "cleared_crossing_nodes": 1,
+                    "preserved_crossing_price_nodes": 2,
                     "retained_below_nodes": 1,
                     "max_right": 1080,
                     "max_bottom": 2376,
                 },
             )
 
+    def test_only_strict_crossing_price_leaves_are_projected_with_whitelisted_attributes(self) -> None:
+        xml = _xml_with_prices(
+            current_node=_price_node(
+                CURRENT_PRICE,
+                PRICE_CURRENT_BOUNDS,
+                extra_attributes=" content-desc='discard' resource-id='discard' focused='true'",
+            ),
+            original_node=_price_node(
+                ORIGINAL_PRICE,
+                PRICE_ORIGINAL_BOUNDS,
+                extra_attributes=" content-desc='discard-too' resource-id='discard-too'",
+            ),
+        )
+        with TemporaryDirectory() as temporary:
+            raw = _write_raw(Path(temporary), xml=xml)
+            result = sanitize_sku_panel_evidence(raw, raw.parent / "derived")
+            root = ElementTree.parse(result.hierarchy_path).getroot()
+            prices = [node for node in root.findall("node") if node.get("text") in {CURRENT_PRICE, ORIGINAL_PRICE}]
+
+            self.assertEqual(len(prices), 2)
+            for node in prices:
+                self.assertEqual(
+                    set(node.attrib),
+                    {"bounds", "text", "package", "class", "clickable", "enabled", "visible-to-user"},
+                )
+                self.assertEqual(node.get("package"), "com.xunmeng.pinduoduo")
+                self.assertEqual(node.get("class"), "android.widget.TextView")
+                self.assertEqual(node.get("clickable"), "false")
+                self.assertEqual(node.get("enabled"), "true")
+                self.assertEqual(node.get("visible-to-user"), "true")
+
+            manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
+            self.assertEqual(manifest["privacy_cleanup"]["preserved_crossing_price_nodes"], 2)
+            self.assertNotIn("discard", result.hierarchy_path.read_text(encoding="utf-8"))
+
+    def test_crossing_price_window_rejects_text_and_structure_drift(self) -> None:
+        bad_texts = (
+            f"快卖光 ¥12.88 {TEST_ADDRESS}",
+            f"快卖光 ¥12.88 {FULL_PHONE}",
+            "快卖光 ¥12.88 使用微信支付",
+            "快卖光 ¥12.88 提交订单",
+            "快卖光 ¥12.88 优惠-11元",
+            "快要抢光 ¥12.88",
+            "快卖光 ¥0.00",
+            "快卖光 ¥12.8",
+            "快卖光 ¥12.880",
+            "快卖光　¥12.88",
+        )
+        for text in bad_texts:
+            with self.subTest(text=text), TemporaryDirectory() as temporary:
+                raw = _write_raw(Path(temporary), xml=_xml_with_prices(current=text))
+                with self.assertRaises(SkuEvidenceSanitizationError):
+                    sanitize_sku_panel_evidence(raw, raw.parent / "derived")
+                self.assertFalse((raw.parent / "derived").exists())
+
+    def test_crossing_price_projection_allows_only_limited_ascii_spaces_and_yen_variants(self) -> None:
+        for current in ("快卖光 ¥12.88", " 快卖光 ¥ 12.88 ", "快卖光 ￥12.88"):
+            with self.subTest(current=current), TemporaryDirectory() as temporary:
+                raw = _write_raw(Path(temporary), xml=_xml_with_prices(current=current))
+                result = sanitize_sku_panel_evidence(raw, raw.parent / "derived")
+                hierarchy = result.hierarchy_path.read_text(encoding="utf-8")
+                self.assertIn(current, hierarchy)
+
+    def test_unique_current_price_without_original_price_is_published(self) -> None:
+        with TemporaryDirectory() as temporary:
+            raw = _write_raw(Path(temporary), xml=_xml_with_prices(include_original=False))
+            result = sanitize_sku_panel_evidence(raw, raw.parent / "derived")
+            hierarchy = result.hierarchy_path.read_text(encoding="utf-8")
+            manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
+
+            self.assertIn(CURRENT_PRICE, hierarchy)
+            self.assertNotIn(ORIGINAL_PRICE, hierarchy)
+            self.assertEqual(manifest["privacy_cleanup"]["preserved_crossing_price_nodes"], 1)
+
+    def test_crossing_price_projection_rejects_newline_and_structure_drift(self) -> None:
+        encoded_newline = _price_node(CURRENT_PRICE, PRICE_CURRENT_BOUNDS).replace(
+            "快卖光 ¥12.88", "快卖光&#10;¥12.88"
+        )
+        with TemporaryDirectory() as temporary:
+            raw = _write_raw(Path(temporary), xml=_xml_with_prices(current_node=encoded_newline))
+            with self.assertRaises(SkuEvidenceSanitizationError):
+                sanitize_sku_panel_evidence(raw, raw.parent / "derived")
+            self.assertFalse((raw.parent / "derived").exists())
+
+        bad_structure = (
+            ("package", {"package": "com.android.systemui"}),
+            ("class", {"node_class": "android.view.View"}),
+            ("clickable", {"clickable": "true"}),
+            ("disabled", {"enabled": "false"}),
+            ("hidden", {"visible": "false"}),
+            ("children", {"children": "<node bounds='[400,510][500,520]' />"}),
+        )
+        for name, kwargs in bad_structure:
+            with self.subTest(structure=name), TemporaryDirectory() as temporary:
+                raw = _write_raw(
+                    Path(temporary),
+                    xml=_xml_with_prices(current_node=_price_node(CURRENT_PRICE, PRICE_CURRENT_BOUNDS, **kwargs)),
+                )
+                with self.assertRaises(SkuEvidenceSanitizationError):
+                    sanitize_sku_panel_evidence(raw, raw.parent / "derived")
+                self.assertFalse((raw.parent / "derived").exists())
+
+    def test_crossing_price_candidates_require_unique_current_and_at_most_one_original(self) -> None:
+        scenarios = (
+            (
+                "missing-current",
+                _xml_with_prices(current="¥12.88", original=ORIGINAL_PRICE),
+            ),
+            (
+                "duplicate-current",
+                _xml_with_prices(current=CURRENT_PRICE, original=f"快卖光 {ORIGINAL_PRICE}"),
+            ),
+            (
+                "multiple-original",
+                _xml_with_prices(
+                    extra_nodes=_price_node("¥39.88", PRICE_ORIGINAL_BOUNDS),
+                ),
+            ),
+        )
+        for name, xml in scenarios:
+            with self.subTest(name=name), TemporaryDirectory() as temporary:
+                raw = _write_raw(Path(temporary), xml=xml)
+                with self.assertRaises(SkuEvidenceSanitizationError):
+                    sanitize_sku_panel_evidence(raw, raw.parent / "derived")
+                self.assertFalse((raw.parent / "derived").exists())
+
+    def test_crossing_price_outside_fixed_windows_is_cleared_and_submit_price_is_not_candidate(self) -> None:
+        outside_crossing = _price_node("快卖光 ¥99.99", "[396,498][712,570]")
+        submit = (
+            "<node bounds='[369,2225][710,2284]' text='提交订单 ¥12.88' "
+            "package='com.xunmeng.pinduoduo' class='android.widget.TextView' clickable='false' "
+            "enabled='true' visible-to-user='true' resource-id='submit-button' />"
+        )
+        with TemporaryDirectory() as temporary:
+            raw = _write_raw(Path(temporary), xml=_xml_with_prices(extra_nodes=outside_crossing + submit))
+            result = sanitize_sku_panel_evidence(raw, raw.parent / "derived")
+            hierarchy = result.hierarchy_path.read_text(encoding="utf-8")
+            manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
+
+            self.assertNotIn("快卖光 ¥99.99", hierarchy)
+            self.assertIn("提交订单 ¥12.88", hierarchy)
+            self.assertIn("submit-button", hierarchy)
+            self.assertEqual(manifest["privacy_cleanup"]["preserved_crossing_price_nodes"], 2)
+
     def test_same_raw_and_config_produce_identical_derived_files(self) -> None:
         with TemporaryDirectory() as temporary:
             root = Path(temporary)