chengma/cmbuyer
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(client): project sanitized SKU price candidates

Browse Source
This commit is contained in:
QiuSW
2026-08-04 14:42:49 +08:00
parent a45a1b5afc
commit 1e69d274b8
2 changed files with 270 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/cmbuyer_client/device/sku_evidence_sanitizer.py
+70 -5
View File
@@ -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,
client/tests/device/test_sku_evidence_sanitizer.py
+200 -3
View File
@@ -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(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", "快卖光&#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)