From 9b2eb744787b7db4f3815de834c6477c6e8e9200 Mon Sep 17 00:00:00 2001
From: QiuSW <105186638@qq.com>
Date: Tue, 4 Aug 2026 11:16:11 +0800
Subject: [PATCH] feat(client): sanitize T-103 device evidence

---
 client/scripts/sanitize_sku_panel_evidence.py |  41 ++
 .../device/sku_evidence_sanitizer.py          | 426 ++++++++++++++++++
 .../device/test_sku_evidence_sanitizer.py     | 341 ++++++++++++++
 3 files changed, 808 insertions(+)
 create mode 100644 client/scripts/sanitize_sku_panel_evidence.py
 create mode 100644 client/src/cmbuyer_client/device/sku_evidence_sanitizer.py
 create mode 100644 client/tests/device/test_sku_evidence_sanitizer.py

diff --git a/client/scripts/sanitize_sku_panel_evidence.py b/client/scripts/sanitize_sku_panel_evidence.py
new file mode 100644
index 0000000..c87f61f
--- /dev/null
+++ b/client/scripts/sanitize_sku_panel_evidence.py
@@ -0,0 +1,41 @@
+"""本机脱敏 T-103 raw 证据到同级 derived；不连接设备或解析页面语义。"""
+
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+import sys
+
+
+CLIENT_ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(CLIENT_ROOT / "src"))
+
+from cmbuyer_client.device.sku_evidence_sanitizer import (
+    SkuEvidenceSanitizationError,
+    sanitize_sku_panel_evidence,
+)
+
+
+def parse_arguments(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="将本机 raw 规格面板证据确定性脱敏到同级 derived。")
+    parser.add_argument("--raw-dir", required=True, type=Path, help="仅允许名为 raw 的本机原始证据目录。")
+    parser.add_argument("--output-dir", required=True, type=Path, help="仅允许 raw 同级且名为 derived 的新目录。")
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    arguments = parse_arguments(argv)
+    try:
+        result = sanitize_sku_panel_evidence(arguments.raw_dir, arguments.output_dir)
+    except SkuEvidenceSanitizationError as error:
+        # 错误不回显 raw 路径、manifest/XML、地址、手机号或 serial。
+        print(f"证据脱敏失败：{error}", file=sys.stderr)
+        return 1
+
+    print(f"派生证据脱敏完成：{result.output_directory}")
+    print(f"manifest：{result.manifest_path}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/client/src/cmbuyer_client/device/sku_evidence_sanitizer.py b/client/src/cmbuyer_client/device/sku_evidence_sanitizer.py
new file mode 100644
index 0000000..5ee06df
--- /dev/null
+++ b/client/src/cmbuyer_client/device/sku_evidence_sanitizer.py
@@ -0,0 +1,426 @@
+"""T-103 原始规格面板证据的本机确定性隐私脱敏。
+
+此模块只处理人工采集的本地文件：不连接设备、不理解拼多多页面，也不识别规格或价格。
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from hashlib import sha256
+import json
+from pathlib import Path
+import re
+import shutil
+from typing import Any
+from uuid import uuid4
+from xml.etree import ElementTree
+
+from PIL import Image, ImageDraw, UnidentifiedImageError
+
+from ..pdd.product_url import ProductUrl, ProductUrlError, parse_product_url
+
+
+SANITIZER_VERSION = "t103-privacy-v1"
+EXPECTED_GOODS_ID = "937122477375"
+EXPECTED_PDD_VERSION = "8.17.0"
+EXPECTED_DEVICE_MODEL = "PKG110"
+EXPECTED_ANDROID_VERSION = "16"
+EXPECTED_WIDTH = 1080
+EXPECTED_HEIGHT = 2400
+HUMAN_DECLARED_STATES = frozenset(
+    {"initial", "one-dimension-selected", "all-dimensions-selected"}
+)
+_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")
+_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\-‐‑‒–—―()（）]+")
+
+
+class SkuEvidenceSanitizationError(RuntimeError):
+    """原始证据不能被安全地发布为派生证据。"""
+
+
+@dataclass
+class _CleanupStats:
+    """仅记录节点数量，供派生 manifest 审计；不记录任何页面文本。"""
+
+    removed_nodes: int = 0
+    cleared_crossing_nodes: int = 0
+    retained_below_nodes: int = 0
+
+
+@dataclass(frozen=True)
+class PrivacyMaskConfig:
+    """仅描述已人工确认的隐私几何区域，绝不承担页面或规格判据。"""
+
+    version: str
+    width: int
+    height: int
+    privacy_top: int
+
+
+PRIVACY_MASK_CONFIG = PrivacyMaskConfig(
+    version=SANITIZER_VERSION,
+    width=EXPECTED_WIDTH,
+    height=EXPECTED_HEIGHT,
+    # 主审在原始截图确认 y < 540 为收货/手机号区域；整宽遮罩优先保护隐私而非保留版面。
+    privacy_top=540,
+)
+
+
+@dataclass(frozen=True)
+class SkuEvidenceSanitizationResult:
+    """已经原子发布的派生证据位置。"""
+
+    output_directory: Path
+    manifest_path: Path
+    screenshot_path: Path
+    hierarchy_path: Path
+
+
+def sanitize_sku_panel_evidence(raw_directory: Path, output_directory: Path) -> SkuEvidenceSanitizationResult:
+    """校验 raw 三文件，并发布同级 ``derived`` 的脱敏副本。
+
+    目标已存在时在读取任何输入前拒绝，避免混入旧派生物或覆盖人工保留文件。
+    """
+
+    raw = Path(raw_directory)
+    target = Path(output_directory)
+    _validate_directories(raw, target)
+    if target.exists():
+        raise SkuEvidenceSanitizationError("派生证据目录已存在，拒绝覆盖。")
+
+    staging: Path | None = None
+    try:
+        source_manifest_path = _required_file(raw, "manifest.json")
+        source_screenshot_path = _required_file(raw, "screenshot.png")
+        source_hierarchy_path = _required_file(raw, "hierarchy.xml")
+        manifest = _read_source_manifest(source_manifest_path)
+        link, state, source_hashes = _validate_source_manifest(manifest)
+        _verify_source_hashes(source_screenshot_path, source_hierarchy_path, source_hashes)
+
+        staging = raw.parent / f".derived.staging-{uuid4().hex}"
+        staging.mkdir()
+        derived_screenshot_path = staging / "screenshot.png"
+        _sanitize_screenshot(source_screenshot_path, derived_screenshot_path)
+        derived_hierarchy_path = staging / "hierarchy.xml"
+        cleanup_stats = _sanitize_hierarchy(source_hierarchy_path, derived_hierarchy_path)
+
+        derived_manifest_path = staging / "manifest.json"
+        derived_manifest_path.write_text(
+            json.dumps(
+                _derived_manifest(
+                    manifest,
+                    link,
+                    state,
+                    source_manifest_path,
+                    source_screenshot_path,
+                    source_hierarchy_path,
+                    derived_screenshot_path,
+                    derived_hierarchy_path,
+                    cleanup_stats,
+                ),
+                ensure_ascii=False,
+                indent=2,
+                sort_keys=True,
+            )
+            + "\n",
+            encoding="utf-8",
+        )
+        _publish_staging(staging, target)
+    except SkuEvidenceSanitizationError:
+        _clean_staging(staging)
+        raise
+    except (OSError, ValueError, ElementTree.ParseError, UnidentifiedImageError) as error:
+        _clean_staging(staging)
+        # 原始异常可能含文件路径、JSON/XML 文本或其他敏感内容，不能向 CLI/日志传播。
+        raise SkuEvidenceSanitizationError("原始证据无法安全脱敏，未发布任何派生产物。") from error
+    except Exception as error:
+        _clean_staging(staging)
+        raise SkuEvidenceSanitizationError("原始证据脱敏未完成，未发布任何派生产物。") from error
+
+    return SkuEvidenceSanitizationResult(
+        output_directory=target,
+        manifest_path=target / "manifest.json",
+        screenshot_path=target / "screenshot.png",
+        hierarchy_path=target / "hierarchy.xml",
+    )
+
+
+def _validate_directories(raw: Path, target: Path) -> None:
+    if raw.name != "raw" or not raw.is_dir():
+        raise SkuEvidenceSanitizationError("原始证据目录必须是存在的 raw 目录。")
+    if target.name != "derived" or target.parent != raw.parent:
+        raise SkuEvidenceSanitizationError("派生证据目录必须是 raw 同级的 derived 目录。")
+
+
+def _required_file(raw: Path, filename: str) -> Path:
+    candidate = raw / filename
+    if not candidate.is_file():
+        raise SkuEvidenceSanitizationError("原始证据文件集合不完整。")
+    return candidate
+
+
+def _read_source_manifest(path: Path) -> dict[str, Any]:
+    try:
+        value = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as error:
+        raise SkuEvidenceSanitizationError("原始证据 manifest 无效。") from error
+    if not isinstance(value, dict):
+        raise SkuEvidenceSanitizationError("原始证据 manifest 结构无效。")
+    return value
+
+
+def _validate_source_manifest(manifest: dict[str, Any]) -> tuple[ProductUrl, str, dict[str, str]]:
+    product = manifest.get("product")
+    device = manifest.get("device")
+    state = manifest.get("human_declared_state")
+    if manifest.get("schema_version") != 1 or not isinstance(product, dict) or not isinstance(device, dict):
+        raise SkuEvidenceSanitizationError("原始证据 manifest 缺少必要元数据。")
+    canonical_url = product.get("canonical_url")
+    goods_id = product.get("goods_id")
+    try:
+        link = parse_product_url(canonical_url)
+    except ProductUrlError as error:
+        raise SkuEvidenceSanitizationError("原始证据商品元数据不匹配。") from error
+    if (
+        link.goods_id != EXPECTED_GOODS_ID
+        or goods_id != EXPECTED_GOODS_ID
+        or device.get("model") != EXPECTED_DEVICE_MODEL
+        or device.get("android_version") != EXPECTED_ANDROID_VERSION
+        or device.get("pdd_version") != EXPECTED_PDD_VERSION
+        or device.get("pdd_package") != "com.xunmeng.pinduoduo"
+        or not isinstance(state, str)
+        or state not in HUMAN_DECLARED_STATES
+    ):
+        raise SkuEvidenceSanitizationError("原始证据元数据与脱敏配置不匹配。")
+    return link, state, _artifact_hashes(manifest)
+
+
+def _artifact_hashes(manifest: dict[str, Any]) -> dict[str, str]:
+    artifacts = manifest.get("artifacts")
+    if not isinstance(artifacts, list):
+        raise SkuEvidenceSanitizationError("原始证据 manifest 缺少文件校验信息。")
+    hashes: dict[str, str] = {}
+    for artifact in artifacts:
+        if not isinstance(artifact, dict):
+            raise SkuEvidenceSanitizationError("原始证据 manifest 文件校验信息无效。")
+        path = artifact.get("path")
+        digest = artifact.get("sha256")
+        if path not in _ARTIFACT_FILES or path in hashes or not isinstance(digest, str) or not _SHA256_RE.fullmatch(digest):
+            raise SkuEvidenceSanitizationError("原始证据 manifest 文件校验信息无效。")
+        hashes[path] = digest
+    if set(hashes) != set(_ARTIFACT_FILES):
+        raise SkuEvidenceSanitizationError("原始证据 manifest 文件校验信息不完整。")
+    return hashes
+
+
+def _verify_source_hashes(screenshot_path: Path, hierarchy_path: Path, expected: dict[str, str]) -> None:
+    if (
+        _sha256_file(screenshot_path) != expected["screenshot.png"]
+        or _sha256_file(hierarchy_path) != expected["hierarchy.xml"]
+    ):
+        raise SkuEvidenceSanitizationError("原始证据文件校验失败。")
+
+
+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):
+                raise SkuEvidenceSanitizationError("原始截图分辨率或格式与脱敏配置不匹配。")
+            sanitized = image.convert("RGBA")
+    except SkuEvidenceSanitizationError:
+        raise
+    except (OSError, UnidentifiedImageError) as error:
+        raise SkuEvidenceSanitizationError("原始截图无效。") from error
+
+    # 用不透明黑色覆盖 y < 540，保证截图与 XML 使用相同的隐私几何边界。
+    ImageDraw.Draw(sanitized).rectangle(
+        (0, 0, PRIVACY_MASK_CONFIG.width - 1, PRIVACY_MASK_CONFIG.privacy_top - 1),
+        fill=(0, 0, 0, 255),
+    )
+    sanitized.save(target, format="PNG", optimize=False, compress_level=9)
+
+
+def _sanitize_hierarchy(source: Path, target: Path) -> _CleanupStats:
+    try:
+        root = ElementTree.parse(source).getroot()
+    except (OSError, ElementTree.ParseError) as error:
+        raise SkuEvidenceSanitizationError("原始节点树无效。") from error
+    if root.tag != "hierarchy":
+        raise SkuEvidenceSanitizationError("原始节点树结构不匹配。")
+    if not list(root):
+        raise SkuEvidenceSanitizationError("原始节点树结构不匹配。")
+    stats = _CleanupStats()
+    _clear_node_text(root)
+    for child in list(root):
+        _sanitize_node(root, child, stats)
+    if stats.removed_nodes < 1 or stats.retained_below_nodes < 1:
+        raise SkuEvidenceSanitizationError("原始节点树未满足隐私几何结构。")
+    if _contains_phone(root):
+        raise SkuEvidenceSanitizationError("派生节点树仍包含手机号，拒绝发布。")
+    ElementTree.ElementTree(root).write(target, encoding="utf-8", xml_declaration=True)
+    return stats
+
+
+def _sanitize_node(parent: ElementTree.Element, node: ElementTree.Element, stats: _CleanupStats) -> None:
+    if node.tag != "node":
+        raise SkuEvidenceSanitizationError("原始节点树结构不匹配。")
+    bounds = _parse_bounds(node.get("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 结构不匹配。")
+        stats.removed_nodes += sum(1 for _ in node.iter("node"))
+        parent.remove(node)
+        return
+    if position == "crossing":
+        # 全屏/跨界容器可保留其下方子节点，但自身所有属性和文本都可能含地址或手机号。
+        _clear_node_text(node)
+        stats.cleared_crossing_nodes += 1
+    else:
+        stats.retained_below_nodes += 1
+    for child in list(node):
+        _sanitize_node(node, child, stats)
+
+
+def _parse_bounds(value: object) -> tuple[int, int, int, int]:
+    if not isinstance(value, str):
+        raise SkuEvidenceSanitizationError("原始节点树 bounds 缺失或无效。")
+    match = _BOUNDS_RE.fullmatch(value)
+    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 超出脱敏配置。")
+    return left, top, right, bottom
+
+
+def _vertical_position(bounds: tuple[int, int, int, int]) -> str:
+    _, top, _, bottom = bounds
+    if bottom <= PRIVACY_MASK_CONFIG.privacy_top:
+        return "private"
+    if top >= PRIVACY_MASK_CONFIG.privacy_top:
+        return "below"
+    return "crossing"
+
+
+def _clear_node_text(node: ElementTree.Element) -> None:
+    node.attrib = {"bounds": node.attrib["bounds"]} if "bounds" in node.attrib else {}
+    node.text = None
+    node.tail = None
+
+
+def _contains_phone(root: ElementTree.Element) -> bool:
+    """逐项与跨节点复检电话，避免分隔符、遮罩字符或节点切分绕过。"""
+
+    all_values: list[str] = []
+    content_values: list[str] = []
+    for element in root.iter():
+        if element.text:
+            all_values.append(element.text)
+            content_values.append(element.text)
+        for attribute, value in element.attrib.items():
+            all_values.append(value)
+            if attribute != "bounds":
+                content_values.append(value)
+        if element.tail:
+            all_values.append(element.tail)
+            content_values.append(element.tail)
+    normalized_values = [_normalize_phone_value(value) for value in all_values]
+    normalized_all_document = "".join(normalized_values)
+    normalized_document = "".join(_normalize_phone_value(value) for value in content_values)
+    return (
+        any(_matches_phone(value) for value in normalized_values)
+        or _matches_phone(normalized_all_document)
+        or _matches_phone(normalized_document)
+    )
+
+
+def _normalize_phone_value(value: str) -> str:
+    return _SEPARATOR_RE.sub("", value.translate(_MASK_TRANSLATION))
+
+
+def _matches_phone(value: str) -> bool:
+    return _FULL_PHONE_RE.search(value) is not None or _MASKED_PHONE_RE.search(value) is not None
+
+
+def _derived_manifest(
+    source_manifest: dict[str, Any],
+    link: ProductUrl,
+    state: str,
+    source_manifest_path: Path,
+    source_screenshot_path: Path,
+    source_hierarchy_path: Path,
+    derived_screenshot_path: Path,
+    derived_hierarchy_path: Path,
+    cleanup_stats: _CleanupStats,
+) -> dict[str, Any]:
+    device = source_manifest["device"]
+    return {
+        "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],
+        },
+        "privacy_cleanup": {
+            "removed_nodes": cleanup_stats.removed_nodes,
+            "cleared_crossing_nodes": cleanup_stats.cleared_crossing_nodes,
+            "retained_below_nodes": cleanup_stats.retained_below_nodes,
+        },
+        "product": {"goods_id": link.goods_id},
+        "human_declared_state": state,
+        "device": {
+            "model": device["model"],
+            "android_version": device.get("android_version"),
+            "pdd_package": device["pdd_package"],
+            "pdd_version": device["pdd_version"],
+        },
+        # source hashes stay only in the local derived manifest; no raw path, serial or body is retained.
+        "source": {
+            "manifest_sha256": _sha256_file(source_manifest_path),
+            "artifacts": [
+                {"path": "screenshot.png", "sha256": _sha256_file(source_screenshot_path)},
+                {"path": "hierarchy.xml", "sha256": _sha256_file(source_hierarchy_path)},
+            ],
+        },
+        "derived": {
+            "artifacts": [
+                {"path": "screenshot.png", "sha256": _sha256_file(derived_screenshot_path)},
+                {"path": "hierarchy.xml", "sha256": _sha256_file(derived_hierarchy_path)},
+            ]
+        },
+    }
+
+
+def _sha256_file(path: Path) -> str:
+    digest = sha256()
+    with path.open("rb") as source:
+        for chunk in iter(lambda: source.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _clean_staging(staging: Path | None) -> None:
+    if staging is not None and staging.exists():
+        shutil.rmtree(staging)
+
+
+def _publish_staging(staging: Path, target: Path) -> None:
+    """发布前二次检查，并使用目录 rename 而不是会覆盖目标的 replace。"""
+
+    if target.exists():
+        raise SkuEvidenceSanitizationError("派生证据目录已存在，拒绝覆盖。")
+    try:
+        staging.rename(target)
+    except OSError as error:
+        # 竞态中新目标出现或文件系统拒绝 rename 时一律不尝试覆盖或重试。
+        raise SkuEvidenceSanitizationError("派生证据目录发布失败，未覆盖已有目录。") from error
diff --git a/client/tests/device/test_sku_evidence_sanitizer.py b/client/tests/device/test_sku_evidence_sanitizer.py
new file mode 100644
index 0000000..b38f0fb
--- /dev/null
+++ b/client/tests/device/test_sku_evidence_sanitizer.py
@@ -0,0 +1,341 @@
+"""T-103 脱敏器测试：全部证据为合成数据，绝不读取真实 raw 目录。"""
+
+from __future__ import annotations
+
+from hashlib import sha256
+import json
+from pathlib import Path
+import sys
+from tempfile import TemporaryDirectory
+import unittest
+from unittest.mock import patch
+from xml.etree import ElementTree
+
+from PIL import Image
+
+
+CLIENT_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(CLIENT_ROOT / "src"))
+
+from cmbuyer_client.device.sku_evidence_sanitizer import (
+    EXPECTED_GOODS_ID,
+    EXPECTED_HEIGHT,
+    EXPECTED_WIDTH,
+    HUMAN_DECLARED_STATES,
+    SkuEvidenceSanitizationError,
+    sanitize_sku_panel_evidence,
+)
+
+
+TEST_SERIAL = "synthetic-serial-never-publish"
+TEST_ADDRESS = "SYNTHETIC_ADDRESS_NEVER_PUBLISH"
+FULL_PHONE = "13800138000"
+MASKED_PHONE = "138****0000"
+SAFE_TEXT = "synthetic-safe-lower-content"
+
+
+def _hash(path: Path) -> str:
+    digest = sha256()
+    with path.open("rb") as source:
+        for chunk in iter(lambda: source.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+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"<node bounds='[0,540][1080,2400]' text='{SAFE_TEXT}' />"
+        "</hierarchy>"
+    )
+
+
+def _write_raw(
+    root: Path,
+    *,
+    state: str = "initial",
+    goods_id: str = EXPECTED_GOODS_ID,
+    model: str = "PKG110",
+    android_version: str = "16",
+    pdd_version: str = "8.17.0",
+    size: tuple[int, int] = (EXPECTED_WIDTH, EXPECTED_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):
+        for y in range(540):
+            for x in range(8):
+                image.putpixel((x, y), (255, 0, 0))
+    image.save(screenshot, format="PNG")
+    hierarchy = raw / "hierarchy.xml"
+    hierarchy.write_text(_default_xml() if xml is None else xml, encoding="utf-8")
+    manifest = {
+        "schema_version": 1,
+        "product": {
+            "goods_id": goods_id,
+            "canonical_url": f"https://mobile.yangkeduo.com/goods.html?goods_id={goods_id}",
+        },
+        "human_declared_state": state,
+        "serial": TEST_SERIAL,
+        "channel": "wifi",
+        "device": {
+            "model": model,
+            "android_version": android_version,
+            "pdd_package": "com.xunmeng.pinduoduo",
+            "pdd_version": pdd_version,
+        },
+        "artifacts": [
+            {"path": "screenshot.png", "sha256": _hash(screenshot)},
+            {"path": "hierarchy.xml", "sha256": _hash(hierarchy)},
+        ],
+    }
+    (raw / "manifest.json").write_text(json.dumps(manifest, sort_keys=True), encoding="utf-8")
+    return raw
+
+
+class SkuEvidenceSanitizerTests(unittest.TestCase):
+    def test_all_declared_states_mask_screenshot_and_xml_without_raw_metadata(self) -> None:
+        for state in sorted(HUMAN_DECLARED_STATES):
+            with self.subTest(state=state), TemporaryDirectory() as temporary:
+                raw = _write_raw(Path(temporary), state=state)
+                result = sanitize_sku_panel_evidence(raw, raw.parent / "derived")
+                with Image.open(result.screenshot_path) as image:
+                    self.assertEqual(image.getpixel((0, 0)), (0, 0, 0, 255))
+                    self.assertEqual(image.getpixel((100, 600)), (0, 180, 0, 255))
+                derived_xml = result.hierarchy_path.read_text(encoding="utf-8")
+                manifest = result.manifest_path.read_text(encoding="utf-8")
+
+                self.assertIn(SAFE_TEXT, derived_xml)
+                self.assertNotIn(TEST_ADDRESS, derived_xml)
+                self.assertNotIn(FULL_PHONE, derived_xml)
+                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('"removed_nodes": 1', manifest)
+                self.assertIn('"cleared_crossing_nodes": 0', manifest)
+                self.assertIn('"retained_below_nodes": 1', manifest)
+                self.assertNotIn("canonical_url", manifest)
+                self.assertNotIn(TEST_SERIAL, manifest)
+                self.assertNotIn("serial", manifest)
+                self.assertNotIn("channel", manifest)
+                self.assertNotIn(TEST_ADDRESS, manifest)
+                self.assertNotIn(FULL_PHONE, manifest)
+
+    def test_crossing_container_keeps_lower_children_but_clears_its_text(self) -> None:
+        xml = (
+            "<hierarchy>"
+            f"<node bounds='[0,0][1080,2400]' text='{TEST_ADDRESS}' content-desc='{MASKED_PHONE}'>"
+            f"<node bounds='[0,0][1080,540]' text='{FULL_PHONE}' />"
+            f"<node bounds='[0,540][1080,2400]' text='{SAFE_TEXT}' />"
+            "</node></hierarchy>"
+        )
+        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()
+            crossing = root.find("node")
+
+            self.assertIsNotNone(crossing)
+            assert crossing is not None
+            self.assertEqual(crossing.attrib, {"bounds": "[0,0][1080,2400]"})
+            self.assertEqual(len(list(crossing)), 1)
+            self.assertEqual(list(crossing)[0].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, "retained_below_nodes": 1},
+            )
+
+    def test_same_raw_and_config_produce_identical_derived_files(self) -> None:
+        with TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            left_raw = _write_raw(root / "left")
+            right_raw = _write_raw(root / "right")
+            left = sanitize_sku_panel_evidence(left_raw, left_raw.parent / "derived")
+            right = sanitize_sku_panel_evidence(right_raw, right_raw.parent / "derived")
+
+            for left_path, right_path in (
+                (left.screenshot_path, right.screenshot_path),
+                (left.hierarchy_path, right.hierarchy_path),
+                (left.manifest_path, right.manifest_path),
+            ):
+                self.assertEqual(left_path.read_bytes(), right_path.read_bytes())
+
+    def test_hash_and_metadata_mismatch_fail_closed_without_leak(self) -> None:
+        scenarios = (
+            ("hash", {}, "screenshot"),
+            ("model", {"model": "other"}, None),
+            ("android", {"android_version": "15"}, None),
+            ("version", {"pdd_version": "8.17.1"}, None),
+            ("goods", {"goods_id": "123"}, None),
+            ("state", {"state": "guessed"}, None),
+            ("resolution", {"size": (100, 100)}, None),
+        )
+        for name, kwargs, corrupt_file in scenarios:
+            with self.subTest(name=name), TemporaryDirectory() as temporary:
+                raw = _write_raw(Path(temporary), **kwargs)
+                if corrupt_file is not None:
+                    (raw / f"{corrupt_file}.png").write_bytes(b"changed")
+                with self.assertRaises(SkuEvidenceSanitizationError) as raised:
+                    sanitize_sku_panel_evidence(raw, raw.parent / "derived")
+
+                message = str(raised.exception)
+                self.assertNotIn(TEST_SERIAL, message)
+                self.assertNotIn(TEST_ADDRESS, message)
+                self.assertNotIn(FULL_PHONE, message)
+                self.assertFalse((raw.parent / "derived").exists())
+                self.assertEqual(list(raw.parent.glob(".derived.staging-*")), [])
+
+    def test_malformed_inputs_and_bounds_or_phone_residue_fail_closed(self) -> None:
+        malformed = (
+            ("manifest", None),
+            ("png", None),
+            ("xml", None),
+            ("bounds", "<hierarchy><node text='missing bounds' /></hierarchy>"),
+            (
+                "private-parent-with-lower-child",
+                "<hierarchy><node bounds='[0,0][1080,540]'><node bounds='[0,540][1080,2400]' text='x' /></node></hierarchy>",
+            ),
+            ("full-phone-below", f"<hierarchy><node bounds='[0,540][1080,2400]' text='{FULL_PHONE}' /></hierarchy>"),
+            ("masked-phone-below", f"<hierarchy><node bounds='[0,540][1080,2400]' text='{MASKED_PHONE}' /></hierarchy>"),
+        )
+        for kind, xml in malformed:
+            with self.subTest(kind=kind), TemporaryDirectory() as temporary:
+                raw = _write_raw(Path(temporary), xml=xml)
+                if kind == "manifest":
+                    (raw / "manifest.json").write_text("{invalid", encoding="utf-8")
+                elif kind == "png":
+                    (raw / "screenshot.png").write_bytes(b"not a png")
+                    manifest = json.loads((raw / "manifest.json").read_text(encoding="utf-8"))
+                    manifest["artifacts"][0]["sha256"] = _hash(raw / "screenshot.png")
+                    (raw / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
+                elif kind == "xml":
+                    (raw / "hierarchy.xml").write_text("<hierarchy>", encoding="utf-8")
+                    manifest = json.loads((raw / "manifest.json").read_text(encoding="utf-8"))
+                    manifest["artifacts"][1]["sha256"] = _hash(raw / "hierarchy.xml")
+                    (raw / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
+                with self.assertRaises(SkuEvidenceSanitizationError):
+                    sanitize_sku_panel_evidence(raw, raw.parent / "derived")
+
+                self.assertFalse((raw.parent / "derived").exists())
+                self.assertEqual(list(raw.parent.glob(".derived.staging-*")), [])
+
+    def test_phone_recheck_rejects_separator_mask_and_cross_node_bypasses(self) -> None:
+        variants = (
+            "138 0013-8000",
+            "138＊＊＊＊0000",
+            "138••••0000",
+            "138xxxx0000",
+            "138XXXX0000",
+        )
+        for value in variants:
+            xml = (
+                "<hierarchy>"
+                "<node bounds='[0,0][1080,540]' text='private' />"
+                f"<node bounds='[0,540][1080,2400]' text='{value}' />"
+                "</hierarchy>"
+            )
+            with self.subTest(value=value), 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())
+
+        split_xml = (
+            "<hierarchy>"
+            "<node bounds='[0,0][1080,540]' text='private' />"
+            "<node bounds='[0,540][1080,1000]' text='138' content-desc='0013' />"
+            "<node bounds='[0,1000][1080,2400]' text='8000' />"
+            "</hierarchy>"
+        )
+        with TemporaryDirectory() as temporary:
+            raw = _write_raw(Path(temporary), xml=split_xml)
+            with self.assertRaises(SkuEvidenceSanitizationError):
+                sanitize_sku_panel_evidence(raw, raw.parent / "derived")
+            self.assertFalse((raw.parent / "derived").exists())
+
+    def test_privacy_geometry_requires_removed_and_retained_nodes(self) -> None:
+        scenarios = (
+            ("no-private", f"<hierarchy><node bounds='[0,540][1080,2400]' text='{SAFE_TEXT}' /></hierarchy>"),
+            ("no-below", "<hierarchy><node bounds='[0,0][1080,540]' text='private' /></hierarchy>"),
+        )
+        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())
+                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":
+                manifest["schema_version"] = 2
+            elif kind == "package":
+                manifest["device"]["pdd_package"] = "com.example.other"  # type: ignore[index]
+            elif kind == "missing":
+                manifest.pop("artifacts")
+            elif kind == "duplicate":
+                manifest["artifacts"].append(manifest["artifacts"][0])  # type: ignore[index]
+            elif kind == "bad-hash":
+                manifest["artifacts"][0]["sha256"] = "g" * 64  # type: ignore[index]
+
+        for kind in ("schema", "package", "missing", "duplicate", "bad-hash"):
+            with self.subTest(kind=kind), TemporaryDirectory() as temporary:
+                raw = _write_raw(Path(temporary))
+                manifest_path = raw / "manifest.json"
+                manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+                mutate(manifest, kind)
+                manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
+                with self.assertRaises(SkuEvidenceSanitizationError) as raised:
+                    sanitize_sku_panel_evidence(raw, raw.parent / "derived")
+
+                self.assertNotIn(TEST_SERIAL, str(raised.exception))
+                self.assertNotIn(TEST_ADDRESS, str(raised.exception))
+                self.assertFalse((raw.parent / "derived").exists())
+
+    def test_target_created_during_publish_is_preserved_and_staging_is_removed(self) -> None:
+        with TemporaryDirectory() as temporary:
+            raw = _write_raw(Path(temporary))
+            target = raw.parent / "derived"
+
+            def race_rename(destination: Path) -> None:
+                destination.mkdir()
+                (destination / "sentinel.txt").write_text("keep", encoding="utf-8")
+                raise FileExistsError("simulated publish race")
+
+            with patch("cmbuyer_client.device.sku_evidence_sanitizer.Path.rename", side_effect=race_rename):
+                with self.assertRaises(SkuEvidenceSanitizationError):
+                    sanitize_sku_panel_evidence(raw, target)
+
+            self.assertEqual((target / "sentinel.txt").read_text(encoding="utf-8"), "keep")
+            self.assertEqual(list(raw.parent.glob(".derived.staging-*")), [])
+
+    def test_existing_derived_is_preserved_without_reading_or_writing_raw(self) -> None:
+        with TemporaryDirectory() as temporary:
+            raw = _write_raw(Path(temporary))
+            target = raw.parent / "derived"
+            target.mkdir()
+            sentinel = target / "sentinel.txt"
+            sentinel.write_text("keep", encoding="utf-8")
+
+            with self.assertRaises(SkuEvidenceSanitizationError):
+                sanitize_sku_panel_evidence(raw, target)
+
+            self.assertEqual(sentinel.read_text(encoding="utf-8"), "keep")
+            self.assertEqual(list(raw.parent.glob(".derived.staging-*")), [])
+
+    def test_directory_contract_rejects_non_sibling_paths(self) -> None:
+        with TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            raw = _write_raw(root)
+            with self.assertRaises(SkuEvidenceSanitizationError):
+                sanitize_sku_panel_evidence(raw, root / "not-derived")
+            with self.assertRaises(SkuEvidenceSanitizationError):
+                sanitize_sku_panel_evidence(root / "not-raw", root / "derived")