diff --git a/client/src/cmbuyer_client/device/sku_evidence_sanitizer.py b/client/src/cmbuyer_client/device/sku_evidence_sanitizer.py index 0eef776..f7bc41c 100644 --- a/client/src/cmbuyer_client/device/sku_evidence_sanitizer.py +++ b/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,9 +314,12 @@ def _sanitize_node(parent: ElementTree.Element, node: ElementTree.Element, stats parent.remove(node) return if position == "crossing": - # 全屏/跨界容器可保留其下方子节点，但自身所有属性和文本都可能含地址或手机号。 - _clear_node_text(node) - stats.cleared_crossing_nodes += 1 + 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 else: stats.retained_below_nodes += 1 for child in list(node): @@ -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, diff --git a/client/tests/device/test_sku_evidence_sanitizer.py b/client/tests/device/test_sku_evidence_sanitizer.py index d729928..fa9f830 100644 --- a/client/tests/device/test_sku_evidence_sanitizer.py +++ b/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 ( "" f"" + f"{_price_node(CURRENT_PRICE, PRICE_CURRENT_BOUNDS)}" + f"{_price_node(ORIGINAL_PRICE, PRICE_ORIGINAL_BOUNDS)}" f"" "" ) +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"{children}" + + +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 ( + "" + f"" + f"{current_markup}" + f"{original_markup}" + f"{extra_nodes}" + f"" + "" + ) + + 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): "" f"" f"" + f"{_price_node(CURRENT_PRICE, PRICE_CURRENT_BOUNDS)}" + f"{_price_node(ORIGINAL_PRICE, PRICE_ORIGINAL_BOUNDS)}" f"" "" ) @@ -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(list(crossing)[0].get("text"), SAFE_TEXT) + self.assertEqual(len(list(crossing)), 3) + 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", "快卖光 ¥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": ""}), + ) + 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 = ( + "" + ) + 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)