cmbuyer/client/src/cmbuyer_client/pdd/product_open.py

"""安全打开 canonical 商品链接后的只读取证。"""

from __future__ import annotations

from collections.abc import Callable
from dataclasses import dataclass
from datetime import UTC, datetime
from hashlib import sha256
import json
from math import isfinite
import os
from pathlib import Path
import shutil
from time import monotonic, sleep
from typing import Any, Protocol
from uuid import uuid4

from adbutils.errors import AdbTimeout
from uiautomator2.exceptions import HTTPTimeoutError

from ..device.adb import AdbClient, DeviceConnectionError, DeviceInspection, IntentLaunchSummary
from ..device.baseline import (
    HIERARCHY_PARAMS,
    PDD_PACKAGE,
    SCREENSHOT_PARAMS,
    _save_base64_screenshot,
    _sha256_file,
    _validate_hierarchy,
)
from .product_url import ProductUrl, parse_product_url


EXPECTED_PDD_VERSION = "8.17.0"


class ProductOpenError(RuntimeError):
    """商品打开或证据发布未完整完成。"""


class ProductVersionMismatchError(ProductOpenError):
    """运行时拼多多版本不是经取证允许的版本。"""


class ProductPackageMismatchError(ProductOpenError):
    """Intent 后在有限时间内未观察到拼多多前台包。"""


class ProductOpenTimeoutError(ProductOpenError):
    """商品打开后的只读取证超时。"""


class ProductScreenshotCaptureError(ProductOpenError):
    """Intent 后截图不能作为完整 PNG 证据保存。"""


class ProductHierarchyCaptureError(ProductOpenError):
    """Intent 后完整节点树不能作为有效 XML 证据保存。"""


class ProductOpenUiDevice(Protocol):
    """本任务所需的只读 uiautomator2 接口；故意没有任何 UI 操作方法。"""

    def app_info(self, package_name: str) -> dict[str, Any]:
        """读取应用元数据。"""

    def app_current(self) -> dict[str, Any]:
        """读取当前前台应用元数据。"""

    def jsonrpc_call(self, method: str, params: Any = None, timeout: float = 10) -> Any:
        """调用只读取证所需的公开 JSON-RPC 方法。"""


@dataclass(frozen=True)
class ProductOpenResult:
    """已原子发布的商品打开证据位置。"""

    output_directory: Path
    manifest_path: Path
    screenshot_path: Path
    hierarchy_path: Path


class ProductOpenCapturer:
    """以 fail-closed 顺序打开已重建链接，并在打开后只读留证。

    本类不判断商品页、Activity、文案或控件；打开后只确认当前 package，随后采集截图与
    完整节点树。任何失败都不会发布半成品证据目录。
    """

    def __init__(
        self,
        adb_client: AdbClient,
        connector: Callable[[str], ProductOpenUiDevice],
        timeout_seconds: float,
        foreground_poll_interval_seconds: float = 0.2,
        monotonic_clock: Callable[[], float] = monotonic,
        sleep_function: Callable[[float], None] = sleep,
    ) -> None:
        if not _is_positive_finite(timeout_seconds):
            raise ValueError("timeout_seconds 必须是大于 0 的有限数值")
        if not _is_positive_finite(foreground_poll_interval_seconds):
            raise ValueError("foreground_poll_interval_seconds 必须是大于 0 的有限数值")
        self._adb_client = adb_client
        self._connector = connector
        self._timeout_seconds = timeout_seconds
        self._foreground_poll_interval_seconds = foreground_poll_interval_seconds
        self._monotonic_clock = monotonic_clock
        self._sleep_function = sleep_function

    def open_and_capture(self, serial: str, product_url: str, output_directory: Path) -> ProductOpenResult:
        """完成唯一允许的 Intent 打开及其后的只读取证。"""

        # 公共入口只接收原始字符串并每次重新解析，不能由调用方构造不一致的值对象伪造 manifest。
        link = parse_product_url(product_url)
        target = Path(output_directory)
        _validate_new_target(target)

        staging: Path | None = None
        try:
            # inspect 必须先于连接和 Intent，复用 T-101 的显式 serial、重复物理设备拒绝逻辑。
            inspection = self._adb_client.inspect(serial)
            device = self._connector(serial)
            pdd_version = _require_expected_version(device.app_info(PDD_PACKAGE))

            # 版本精确匹配是 Intent 的前置条件，失败时绝不调用 start_pdd_view_intent。
            intent = self._adb_client.start_pdd_view_intent(serial, link.goods_id)
            self._wait_for_pdd_foreground(device)

            target.parent.mkdir(parents=True, exist_ok=True)
            staging = target.parent / f".{target.name}.staging-{uuid4().hex}"
            staging.mkdir()
            screenshot_path = staging / "screenshot.png"
            try:
                _save_base64_screenshot(
                    device.jsonrpc_call("takeScreenshot", SCREENSHOT_PARAMS, timeout=self._timeout_seconds),
                    screenshot_path,
                )
            except (AdbTimeout, HTTPTimeoutError, TimeoutError):
                raise
            except Exception as error:
                raise ProductScreenshotCaptureError("商品打开后截图取证失败，未发布任何证据产物。") from error

            try:
                hierarchy = device.jsonrpc_call(
                    "dumpWindowHierarchy",
                    HIERARCHY_PARAMS,
                    timeout=self._timeout_seconds,
                )
                _validate_hierarchy(hierarchy)
            except (AdbTimeout, HTTPTimeoutError, TimeoutError):
                raise
            except Exception as error:
                raise ProductHierarchyCaptureError("商品打开后节点树取证失败，未发布任何证据产物。") from error
            hierarchy_path = staging / "hierarchy.xml"
            hierarchy_path.write_text(hierarchy, encoding="utf-8")

            manifest_path = staging / "manifest.json"
            manifest_path.write_text(
                json.dumps(
                    _manifest(inspection, serial, link, pdd_version, intent, screenshot_path, hierarchy_path),
                    ensure_ascii=False,
                    indent=2,
                    sort_keys=True,
                )
                + "\n",
                encoding="utf-8",
            )
            os.replace(staging, target)
        except (ProductOpenError, DeviceConnectionError):
            _clean_staging(staging)
            raise
        except (AdbTimeout, HTTPTimeoutError, TimeoutError) as error:
            _clean_staging(staging)
            raise ProductOpenTimeoutError("商品打开后的只读取证超时，未发布任何证据产物。") from error
        except Exception as error:
            _clean_staging(staging)
            # 底层异常可能含 serial、路径或远端页面内容，不能直接向 CLI 或日志传播。
            raise ProductOpenError("商品打开或只读取证未完成，未发布任何证据产物。") from error

        return ProductOpenResult(
            output_directory=target,
            manifest_path=target / "manifest.json",
            screenshot_path=target / "screenshot.png",
            hierarchy_path=target / "hierarchy.xml",
        )

    def _wait_for_pdd_foreground(self, device: ProductOpenUiDevice) -> None:
        """只轮询当前 package，直到 deadline；Activity 和节点树均不参与本判据。"""

        deadline = self._monotonic_clock() + self._timeout_seconds
        while True:
            if _is_pdd_foreground(device.app_current()):
                return
            remaining = deadline - self._monotonic_clock()
            if remaining <= 0:
                raise ProductPackageMismatchError(
                    "商品链接打开后未在限定时间内进入拼多多，已停止后续取证。"
                )
            # 每个失败观察后都等待正的、受 deadline 约束的时长，避免 busy-loop。
            self._sleep_function(min(self._foreground_poll_interval_seconds, remaining))


def _validate_new_target(target: Path) -> None:
    if target.exists():
        raise ProductOpenError("输出目录已存在；为防止混入旧证据，拒绝覆盖。")
    if not target.name:
        raise ProductOpenError("输出目录必须是明确的新目录。")


def _clean_staging(staging: Path | None) -> None:
    if staging is not None and staging.exists():
        # staging 仅在本次调用中创建，删除前不解析或扩展任何调用方提供的路径。
        shutil.rmtree(staging)


def _require_expected_version(app_info: dict[str, Any]) -> str:
    if not isinstance(app_info, dict):
        raise ProductVersionMismatchError("拼多多版本与已取证版本不一致，已停止打开商品链接。")
    version = app_info.get("versionName") or app_info.get("version_name")
    if not isinstance(version, str) or version != EXPECTED_PDD_VERSION:
        raise ProductVersionMismatchError("拼多多版本与已取证版本不一致，已停止打开商品链接。")
    return version


def _is_positive_finite(value: object) -> bool:
    return isinstance(value, (int, float)) and not isinstance(value, bool) and value > 0 and isfinite(value)


def _is_pdd_foreground(current: object) -> bool:
    return isinstance(current, dict) and current.get("package") == PDD_PACKAGE


def _manifest(
    inspection: DeviceInspection,
    serial: str,
    link: ProductUrl,
    pdd_version: str,
    intent: IntentLaunchSummary,
    screenshot_path: Path,
    hierarchy_path: Path,
) -> dict[str, Any]:
    """只写审计摘要；原始 serial、Activity、ADB 输出和页面正文均不进入 manifest。"""

    return {
        "schema_version": 1,
        "captured_at": datetime.now(UTC).isoformat(),
        "product": {"goods_id": link.goods_id, "canonical_url": link.canonical_url},
        "channel": "wifi" if ":" in serial else "usb",
        "serial_sha256": sha256(serial.encode("utf-8")).hexdigest(),
        "device": {
            "model": inspection.model,
            "android_version": inspection.android_version,
            "pdd_package": PDD_PACKAGE,
            "pdd_version": pdd_version,
        },
        "intent": {"status": intent.status, "returncode": intent.returncode},
        "current_package": PDD_PACKAGE,
        "artifacts": [
            {"path": screenshot_path.name, "sha256": _sha256_file(screenshot_path)},
            {"path": hierarchy_path.name, "sha256": _sha256_file(hierarchy_path)},
        ],
    }