diff --git a/client/src/cmbuyer_client/device/sku_evidence_sanitizer.py b/client/src/cmbuyer_client/device/sku_evidence_sanitizer.py
index ca9df74..b71e73c 100644
--- a/client/src/cmbuyer_client/device/sku_evidence_sanitizer.py
+++ b/client/src/cmbuyer_client/device/sku_evidence_sanitizer.py
@@ -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,11 +281,15 @@ 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")):
- raise SkuEvidenceSanitizationError("原始节点树 bounds 结构不匹配。")
+ 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)
return
@@ -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,
diff --git a/client/tests/device/test_sku_evidence_sanitizer.py b/client/tests/device/test_sku_evidence_sanitizer.py
index c85095e..90fa7f4 100644
--- a/client/tests/device/test_sku_evidence_sanitizer.py
+++ b/client/tests/device/test_sku_evidence_sanitizer.py
@@ -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",
+ ""
+ "",
+ "1079x2400",
+ ),
+ (
+ "short-height",
+ ""
+ "",
+ "1080x2376",
+ ),
+ (
+ "wide-width",
+ ""
+ "",
+ "1081x2400",
+ ),
+ (
+ "tall-height",
+ ""
+ "",
+ "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":