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/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)