cmbuyer/client/src/cmbuyer_client/device/sku_evidence_sanitizer.py

"""T-103 原始规格面板证据的本机确定性隐私脱敏。

此模块只处理人工采集的本地文件：不连接设备、不识别规格；仅可按已取证的固定
几何和严格格式，将跨隐私边界的价格叶节点投影到派生 XML。
"""

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
from ..pdd.sku_panel_state import HUMAN_DECLARED_STATES


SANITIZER_VERSION = "t103-privacy-v4"
EXPECTED_GOODS_ID = "937122477375"
EXPECTED_PDD_VERSION = "8.17.0"
EXPECTED_DEVICE_MODEL = "PKG110"
EXPECTED_ANDROID_VERSION = "16"
EXPECTED_SCREENSHOT_WIDTH = 1080
EXPECTED_SCREENSHOT_HEIGHT = 2376
EXPECTED_XML_WIDTH = 1080
EXPECTED_XML_HEIGHT = 2376
_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\-‐‑‒–—―()（）]+")
# 这两个槽位来自 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):
    """原始证据不能被安全地发布为派生证据。"""


@dataclass
class _CleanupStats:
    """仅记录节点数量，供派生 manifest 审计；不记录任何页面文本。"""

    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)
class PrivacyMaskConfig:
    """仅描述已人工确认的隐私几何区域，绝不承担页面或规格判据。"""

    version: str
    screenshot_width: int
    screenshot_height: int
    xml_width: int
    xml_height: int
    privacy_top: int


PRIVACY_MASK_CONFIG = PrivacyMaskConfig(
    version=SANITIZER_VERSION,
    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,
)


@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.screenshot_width,
                PRIVACY_MASK_CONFIG.screenshot_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.screenshot_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)
    _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)
    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"))
    _observe_bounds(stats, bounds)
    position = _vertical_position(bounds)
    if position == "private":
        # 私有带内的父节点不可以悄然包含下方子节点，否则会把仍需审计的下方内容一起丢失。
        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
    if position == "crossing":
        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):
        _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 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:
        return "private"
    if top >= PRIVACY_MASK_CONFIG.privacy_top:
        return "below"
    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
    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,
        "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,
            "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,
        },
        "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