cmbuyer/client/scripts/capture_order_confirm_spike.py

"""采集 T-106 人工准备的提交前来源态证据；不执行页面操作。"""

from __future__ import annotations

import argparse
from math import isfinite
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.adb import AdbClient, DeviceConnectionError, SubprocessAdbRunner
from cmbuyer_client.device.baseline import NoReconnectUiautomatorConnector
from cmbuyer_client.pdd.order_confirm_spike import (
    Android16ForegroundReader,
    DECLARED_STATES,
    EXPECTED_GOODS_ID,
    OrderConfirmEvidenceCapturer,
    OrderConfirmEvidenceError,
)


def parse_arguments(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="采集 T-106 人工准备的提交前来源态本机证据。")
    parser.add_argument("--serial", required=True, help="ADB device serial；禁止自动选择。")
    parser.add_argument("--goods-id", required=True, help="T-106 已批准的目标商品标识。")
    parser.add_argument(
        "--state",
        required=True,
        choices=DECLARED_STATES,
        help="仅允许已取证的 Gate2 后提交前来源态；未证实的后续状态全部禁用。",
    )
    parser.add_argument("--output-dir", required=True, type=Path, help="全新本机证据目录；不得覆盖。")
    parser.add_argument("--timeout", type=float, default=10.0, help="ADB 与只读设备 RPC 超时（秒）。")
    parser.add_argument("--adb", default="adb", help="adb 可执行文件路径。")
    return parser.parse_args(argv)


def validate_arguments(arguments: argparse.Namespace) -> None:
    if (
        type(arguments.serial) is not str
        or not arguments.serial.strip()
        or arguments.serial != arguments.serial.strip()
    ):
        raise ValueError("必须显式提供非空 --serial。")
    if type(arguments.goods_id) is not str or arguments.goods_id != EXPECTED_GOODS_ID:
        raise ValueError("--goods-id 不是 T-106 已批准目标。")
    if type(arguments.state) is not str or arguments.state not in DECLARED_STATES:
        raise ValueError("--state 必须是批准的人工声明状态。")
    if not isinstance(arguments.output_dir, Path) or not arguments.output_dir.name:
        raise ValueError("--output-dir 必须是明确的全新目录。")
    if (
        not isinstance(arguments.timeout, (int, float))
        or isinstance(arguments.timeout, bool)
        or arguments.timeout <= 0
        or not isfinite(arguments.timeout)
    ):
        raise ValueError("--timeout 必须是大于 0 的有限数值。")


def main(argv: list[str] | None = None) -> int:
    arguments = parse_arguments(argv)
    try:
        validate_arguments(arguments)
    except ValueError as error:
        print(f"失败：{error}", file=sys.stderr)
        return 2

    try:
        import adbutils
        import uiautomator2 as u2
    except ImportError:
        print("失败：缺少 uiautomator2；请在采购工具虚拟环境中运行。", file=sys.stderr)
        return 2

    adb_runner = SubprocessAdbRunner(arguments.adb)
    capturer = OrderConfirmEvidenceCapturer(
        AdbClient(adb_runner, timeout_seconds=arguments.timeout),
        NoReconnectUiautomatorConnector(
            adbutils.AdbClient(socket_timeout=arguments.timeout).device_list,
            u2.connect,
        ),
        Android16ForegroundReader(adb_runner, arguments.timeout),
        timeout_seconds=arguments.timeout,
    )
    try:
        capturer.capture(
            arguments.serial,
            arguments.goods_id,
            arguments.state,
            arguments.output_dir,
        )
    except (DeviceConnectionError, OrderConfirmEvidenceError):
        # 第三方异常可能带设备、页面正文或本机目录，命令行只输出固定摘要。
        print("T-106 只读取证失败：已停止，未发布本机证据目录。", file=sys.stderr)
        return 1
    except OSError:
        print("T-106 只读取证失败：无法发布本机证据目录。", file=sys.stderr)
        return 1

    print("T-106 提交前来源态只读取证完成。")
    print("人工复核：请在指定目录检查截图、XML、应用摘要和 manifest。")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())