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

fix(client): separate screenshot and XML coordinates

Browse Source
This commit is contained in:
QiuSW
2026-08-04 11:57:45 +08:00
parent cf3692112c
commit 44c027a18f
2 changed files with 143 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/cmbuyer_client/device/sku_evidence_sanitizer.py
+61 -16
View File
@@ -21,13 +21,15 @@ from ..pdd.product_url import ProductUrl, ProductUrlError, parse_product_url
from ..pdd.sku_panel_state import HUMAN_DECLARED_STATES
SANITIZER_VERSION = "t103-privacy-v1"
SANITIZER_VERSION = "t103-privacy-v2"
EXPECTED_GOODS_ID = "937122477375"
EXPECTED_PDD_VERSION = "8.17.0"
EXPECTED_DEVICE_MODEL = "PKG110"
EXPECTED_ANDROID_VERSION = "16"
EXPECTED_WIDTH = 1080
EXPECTED_HEIGHT = 2400
EXPECTED_SCREENSHOT_WIDTH = 1080
EXPECTED_SCREENSHOT_HEIGHT = 2376
EXPECTED_XML_WIDTH = 1080
EXPECTED_XML_HEIGHT = 2400
_ARTIFACT_FILES = ("screenshot.png", "hierarchy.xml")
_SHA256_RE = re.compile(r"[0-9a-f]{64}\Z")
_BOUNDS_RE = re.compile(r"\[(-?\d+),(-?\d+)\]\[(-?\d+),(-?\d+)\]\Z")
@@ -48,6 +50,8 @@ class _CleanupStats:
removed_nodes: int = 0
cleared_crossing_nodes: int = 0
retained_below_nodes: int = 0
max_right: int = 0
max_bottom: int = 0
@dataclass(frozen=True)
@@ -55,15 +59,19 @@ class PrivacyMaskConfig:
"""仅描述已人工确认的隐私几何区域，绝不承担页面或规格判据。"""
version: str
width: int
height: int
screenshot_width: int
screenshot_height: int
xml_width: int
xml_height: int
privacy_top: int
PRIVACY_MASK_CONFIG = PrivacyMaskConfig(
version=SANITIZER_VERSION,
width=EXPECTED_WIDTH,
height=EXPECTED_HEIGHT,
screenshot_width=EXPECTED_SCREENSHOT_WIDTH,
screenshot_height=EXPECTED_SCREENSHOT_HEIGHT,
xml_width=EXPECTED_XML_WIDTH,
xml_height=EXPECTED_XML_HEIGHT,
# 主审在原始截图确认 y < 540 为收货/手机号区域；整宽遮罩优先保护隐私而非保留版面。
privacy_top=540,
)
@@ -228,7 +236,10 @@ def _sanitize_screenshot(source: Path, target: Path) -> None:
try:
with Image.open(source) as image:
image.load()
if image.format != "PNG" or image.size != (PRIVACY_MASK_CONFIG.width, PRIVACY_MASK_CONFIG.height):
if image.format != "PNG" or image.size != (
PRIVACY_MASK_CONFIG.screenshot_width,
PRIVACY_MASK_CONFIG.screenshot_height,
):
raise SkuEvidenceSanitizationError("原始截图分辨率或格式与脱敏配置不匹配。")
sanitized = image.convert("RGBA")
except SkuEvidenceSanitizationError:
@@ -238,7 +249,7 @@ def _sanitize_screenshot(source: Path, target: Path) -> None:
# 用不透明黑色覆盖 y < 540，保证截图与 XML 使用相同的隐私几何边界。
ImageDraw.Draw(sanitized).rectangle(
(0, 0, PRIVACY_MASK_CONFIG.width - 1, PRIVACY_MASK_CONFIG.privacy_top - 1),
(0, 0, PRIVACY_MASK_CONFIG.screenshot_width - 1, PRIVACY_MASK_CONFIG.privacy_top - 1),
fill=(0, 0, 0, 255),
)
sanitized.save(target, format="PNG", optimize=False, compress_level=9)
@@ -257,6 +268,7 @@ def _sanitize_hierarchy(source: Path, target: Path) -> _CleanupStats:
_clear_node_text(root)
for child in list(root):
_sanitize_node(root, child, stats)
_require_expected_xml_coordinate_space(stats)
if stats.removed_nodes < 1 or stats.retained_below_nodes < 1:
raise SkuEvidenceSanitizationError("原始节点树未满足隐私几何结构。")
if _contains_phone(root):
@@ -269,10 +281,14 @@ def _sanitize_node(parent: ElementTree.Element, node: ElementTree.Element, stats
if node.tag != "node":
raise SkuEvidenceSanitizationError("原始节点树结构不匹配。")
bounds = _parse_bounds(node.get("bounds"))
_observe_bounds(stats, bounds)
position = _vertical_position(bounds)
if position == "private":
# 私有带内的父节点不可以悄然包含下方子节点，否则会把仍需审计的下方内容一起丢失。
if any(_vertical_position(_parse_bounds(descendant.get("bounds"))) != "private" for descendant in node.iter("node")):
for descendant in node.iter("node"):
descendant_bounds = _parse_bounds(descendant.get("bounds"))
_observe_bounds(stats, descendant_bounds)
if _vertical_position(descendant_bounds) != "private":
raise SkuEvidenceSanitizationError("原始节点树 bounds 结构不匹配。")
stats.removed_nodes += sum(1 for _ in node.iter("node"))
parent.remove(node)
@@ -294,11 +310,27 @@ def _parse_bounds(value: object) -> tuple[int, int, int, int]:
if match is None:
raise SkuEvidenceSanitizationError("原始节点树 bounds 缺失或无效。")
left, top, right, bottom = (int(group) for group in match.groups())
if not (0 <= left < right <= PRIVACY_MASK_CONFIG.width and 0 <= top < bottom <= PRIVACY_MASK_CONFIG.height):
raise SkuEvidenceSanitizationError("原始节点树 bounds 超出脱敏配置。")
if not (0 <= left < right and 0 <= top < bottom):
raise SkuEvidenceSanitizationError("原始节点树 bounds 缺失或无效。")
return left, top, right, bottom
def _observe_bounds(stats: _CleanupStats, bounds: tuple[int, int, int, int]) -> None:
_, _, right, bottom = bounds
stats.max_right = max(stats.max_right, right)
stats.max_bottom = max(stats.max_bottom, bottom)
def _require_expected_xml_coordinate_space(stats: _CleanupStats) -> None:
if (
stats.max_right != PRIVACY_MASK_CONFIG.xml_width
or stats.max_bottom != PRIVACY_MASK_CONFIG.xml_height
):
raise SkuEvidenceSanitizationError(
f"原始节点树坐标范围不匹配（observed {stats.max_right}x{stats.max_bottom}）。"
)
def _vertical_position(bounds: tuple[int, int, int, int]) -> str:
_, top, _, bottom = bounds
if bottom <= PRIVACY_MASK_CONFIG.privacy_top:
@@ -364,15 +396,28 @@ def _derived_manifest(
"schema_version": 1,
"privacy_tier": "SANITIZED",
"sanitizer_version": PRIVACY_MASK_CONFIG.version,
"privacy_mask": {
"width": PRIVACY_MASK_CONFIG.width,
"height": PRIVACY_MASK_CONFIG.height,
"rectangle": [0, 0, PRIVACY_MASK_CONFIG.width, PRIVACY_MASK_CONFIG.privacy_top],
"screenshot_space": {
"width": PRIVACY_MASK_CONFIG.screenshot_width,
"height": PRIVACY_MASK_CONFIG.screenshot_height,
"privacy_mask_rectangle": [
0,
0,
PRIVACY_MASK_CONFIG.screenshot_width,
PRIVACY_MASK_CONFIG.privacy_top,
],
},
"xml_coordinate_space": {
"width": PRIVACY_MASK_CONFIG.xml_width,
"height": PRIVACY_MASK_CONFIG.xml_height,
"privacy_mask_rectangle": [0, 0, PRIVACY_MASK_CONFIG.xml_width, PRIVACY_MASK_CONFIG.privacy_top],
"observed_max": {"right": cleanup_stats.max_right, "bottom": cleanup_stats.max_bottom},
},
"privacy_cleanup": {
"removed_nodes": cleanup_stats.removed_nodes,
"cleared_crossing_nodes": cleanup_stats.cleared_crossing_nodes,
"retained_below_nodes": cleanup_stats.retained_below_nodes,
"max_right": cleanup_stats.max_right,
"max_bottom": cleanup_stats.max_bottom,
},
"product": {"goods_id": link.goods_id},
"human_declared_state": state,
client/tests/device/test_sku_evidence_sanitizer.py
+81 -8
View File
@@ -19,8 +19,10 @@ sys.path.insert(0, str(CLIENT_ROOT / "src"))
from cmbuyer_client.device.sku_evidence_sanitizer import (
EXPECTED_GOODS_ID,
EXPECTED_HEIGHT,
EXPECTED_WIDTH,
EXPECTED_SCREENSHOT_HEIGHT,
EXPECTED_SCREENSHOT_WIDTH,
EXPECTED_XML_HEIGHT,
EXPECTED_XML_WIDTH,
HUMAN_DECLARED_STATES,
SkuEvidenceSanitizationError,
sanitize_sku_panel_evidence,
@@ -59,14 +61,14 @@ def _write_raw(
model: str = "PKG110",
android_version: str = "16",
pdd_version: str = "8.17.0",
size: tuple[int, int] = (EXPECTED_WIDTH, EXPECTED_HEIGHT),
size: tuple[int, int] = (EXPECTED_SCREENSHOT_WIDTH, EXPECTED_SCREENSHOT_HEIGHT),
xml: str | None = None,
) -> Path:
raw = root / "raw"
raw.mkdir(parents=True)
screenshot = raw / "screenshot.png"
image = Image.new("RGB", size, color=(0, 180, 0))
if size == (EXPECTED_WIDTH, EXPECTED_HEIGHT):
if size == (EXPECTED_SCREENSHOT_WIDTH, EXPECTED_SCREENSHOT_HEIGHT):
for y in range(540):
for x in range(8):
image.putpixel((x, y), (255, 0, 0))
@@ -115,11 +117,17 @@ 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-v1"', manifest)
self.assertIn('"rectangle": [', manifest)
self.assertIn('"sanitizer_version": "t103-privacy-v2"', manifest)
self.assertIn('"screenshot_space": {', manifest)
self.assertIn('"xml_coordinate_space": {', manifest)
self.assertIn('"height": 2376', manifest)
self.assertIn('"height": 2400', manifest)
self.assertIn('"privacy_mask_rectangle": [', manifest)
self.assertIn('"removed_nodes": 1', manifest)
self.assertIn('"cleared_crossing_nodes": 0', manifest)
self.assertIn('"retained_below_nodes": 1', manifest)
self.assertIn('"max_right": 1080', manifest)
self.assertIn('"max_bottom": 2400', manifest)
self.assertNotIn("canonical_url", manifest)
self.assertNotIn(TEST_SERIAL, manifest)
self.assertNotIn("serial", manifest)
@@ -149,7 +157,13 @@ class SkuEvidenceSanitizerTests(unittest.TestCase):
manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
self.assertEqual(
manifest["privacy_cleanup"],
{"removed_nodes": 1, "cleared_crossing_nodes": 1, "retained_below_nodes": 1},
{
"removed_nodes": 1,
"cleared_crossing_nodes": 1,
"retained_below_nodes": 1,
"max_right": 1080,
"max_bottom": 2400,
},
)
def test_same_raw_and_config_produce_identical_derived_files(self) -> None:
@@ -167,6 +181,26 @@ class SkuEvidenceSanitizerTests(unittest.TestCase):
):
self.assertEqual(left_path.read_bytes(), right_path.read_bytes())
def test_manifest_records_distinct_screenshot_and_xml_coordinate_spaces(self) -> None:
with TemporaryDirectory() as temporary:
raw = _write_raw(Path(temporary))
result = sanitize_sku_panel_evidence(raw, raw.parent / "derived")
manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
self.assertEqual(
manifest["screenshot_space"],
{"width": 1080, "height": 2376, "privacy_mask_rectangle": [0, 0, 1080, 540]},
)
self.assertEqual(
manifest["xml_coordinate_space"],
{
"width": 1080,
"height": 2400,
"privacy_mask_rectangle": [0, 0, 1080, 540],
"observed_max": {"right": 1080, "bottom": 2400},
},
)
def test_hash_and_metadata_mismatch_fail_closed_without_leak(self) -> None:
scenarios = (
("hash", {}, "screenshot"),
@@ -175,7 +209,8 @@ class SkuEvidenceSanitizerTests(unittest.TestCase):
("version", {"pdd_version": "8.17.1"}, None),
("goods", {"goods_id": "123"}, None),
("state", {"state": "guessed"}, None),
("resolution", {"size": (100, 100)}, None),
("old-screenshot-space", {"size": (1080, 2400)}, None),
("other-screenshot-space", {"size": (100, 100)}, None),
)
for name, kwargs, corrupt_file in scenarios:
with self.subTest(name=name), TemporaryDirectory() as temporary:
@@ -281,6 +316,44 @@ class SkuEvidenceSanitizerTests(unittest.TestCase):
self.assertFalse((raw.parent / "derived").exists())
self.assertEqual(list(raw.parent.glob(".derived.staging-*")), [])
def test_xml_coordinate_space_must_have_exact_configured_maximums(self) -> None:
scenarios = (
(
"short-width",
"<hierarchy><node bounds='[0,0][1079,540]' text='private' />"
"<node bounds='[0,540][1079,2400]' text='safe' /></hierarchy>",
"1079x2400",
),
(
"short-height",
"<hierarchy><node bounds='[0,0][1080,540]' text='private' />"
"<node bounds='[0,540][1080,2376]' text='safe' /></hierarchy>",
"1080x2376",
),
(
"wide-width",
"<hierarchy><node bounds='[0,0][1081,540]' text='private' />"
"<node bounds='[0,540][1081,2400]' text='safe' /></hierarchy>",
"1081x2400",
),
(
"tall-height",
"<hierarchy><node bounds='[0,0][1080,540]' text='private' />"
"<node bounds='[0,540][1080,2401]' text='safe' /></hierarchy>",
"1080x2401",
),
)
for name, xml, observed in scenarios:
with self.subTest(name=name), TemporaryDirectory() as temporary:
raw = _write_raw(Path(temporary), xml=xml)
with self.assertRaises(SkuEvidenceSanitizationError) as raised:
sanitize_sku_panel_evidence(raw, raw.parent / "derived")
self.assertIn(observed, str(raised.exception))
self.assertNotIn(TEST_SERIAL, str(raised.exception))
self.assertFalse((raw.parent / "derived").exists())
self.assertEqual(list(raw.parent.glob(".derived.staging-*")), [])
def test_manifest_schema_package_and_artifact_structure_are_required(self) -> None:
def mutate(manifest: dict[str, object], kind: str) -> None:
if kind == "schema":