121 lines
4.7 KiB
Python
121 lines
4.7 KiB
Python
"""采集 T-105 人工准备的数量 1/2 两态证据；不执行页面操作。"""
|
||
|
||
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.quantity_gate2_spike import (
|
||
Android16TopResumedForegroundReader,
|
||
DECLARED_QUANTITIES,
|
||
EXPECTED_GOODS_ID,
|
||
QuantityGate2EvidenceCapturer,
|
||
QuantityGate2EvidenceError,
|
||
)
|
||
|
||
|
||
def parse_arguments(argv: list[str] | None = None) -> argparse.Namespace:
|
||
parser = argparse.ArgumentParser(description="采集 T-105 人工准备的数量两态本机证据。")
|
||
parser.add_argument("--serial", required=True, help="ADB device serial；禁止自动选择。")
|
||
parser.add_argument("--goods-id", required=True, help="T-105 已批准的目标商品标识。")
|
||
parser.add_argument(
|
||
"--state",
|
||
required=True,
|
||
choices=sorted(DECLARED_QUANTITIES),
|
||
help="人工声明状态：initial=数量1，target=数量2。",
|
||
)
|
||
parser.add_argument(
|
||
"--declared-quantity",
|
||
required=True,
|
||
type=int,
|
||
help="人工看到的数量；必须与状态一致。",
|
||
)
|
||
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-105 已批准目标。")
|
||
if type(arguments.state) is not str or arguments.state not in DECLARED_QUANTITIES:
|
||
raise ValueError("--state 必须是批准的人工声明状态。")
|
||
if type(arguments.declared_quantity) is not int or (
|
||
arguments.declared_quantity != DECLARED_QUANTITIES[arguments.state]
|
||
):
|
||
raise ValueError("--declared-quantity 必须与人工声明状态一致。")
|
||
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 = QuantityGate2EvidenceCapturer(
|
||
AdbClient(adb_runner, timeout_seconds=arguments.timeout),
|
||
NoReconnectUiautomatorConnector(
|
||
adbutils.AdbClient(socket_timeout=arguments.timeout).device_list,
|
||
u2.connect,
|
||
),
|
||
Android16TopResumedForegroundReader(adb_runner, arguments.timeout),
|
||
timeout_seconds=arguments.timeout,
|
||
)
|
||
try:
|
||
capturer.capture(
|
||
arguments.serial,
|
||
arguments.goods_id,
|
||
arguments.state,
|
||
arguments.declared_quantity,
|
||
arguments.output_dir,
|
||
)
|
||
except (DeviceConnectionError, QuantityGate2EvidenceError):
|
||
# 第三方异常可能含 serial、Activity、本机路径或页面正文，CLI 只输出固定摘要。
|
||
print("数量两态只读取证失败：已停止，未发布本机证据目录。", file=sys.stderr)
|
||
return 1
|
||
except OSError:
|
||
print("数量两态只读取证失败：无法发布本机证据目录。", file=sys.stderr)
|
||
return 1
|
||
|
||
print("数量两态只读取证完成。")
|
||
print("人工复核：请在指定目录检查截图、XML、应用摘要和 manifest。")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|