88 lines
4.1 KiB
Python
88 lines
4.1 KiB
Python
"""恢复 T-103 已取证目标规格、验证现价并保存本地原始截图。"""
|
||
|
||
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.product_url import ProductUrlError, parse_product_url
|
||
from cmbuyer_client.pdd.sku_selection import EXPECTED_GOODS_ID, SkuSelectionError, TASK_TO_UI_SELECTION
|
||
from cmbuyer_client.pdd.sku_selection_runner import SkuSelectionRunError, SkuSelectionRunner, safe_failure_stage
|
||
|
||
|
||
def parse_arguments(argv: list[str] | None = None) -> argparse.Namespace:
|
||
parser = argparse.ArgumentParser(description="恢复 T-103 已取证规格并保存本地原始截图。")
|
||
parser.add_argument("--serial", required=True, help="ADB device serial；禁止自动选择。")
|
||
parser.add_argument("--url", required=True, help="唯一 canonical goods.html?goods_id= 直链。")
|
||
parser.add_argument("--color", required=True, help="T-103 任务颜色值。")
|
||
parser.add_argument("--size", required=True, help="T-103 任务尺码值。")
|
||
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 not isinstance(arguments.serial, str) or not arguments.serial.strip():
|
||
raise ValueError("必须显式提供非空 --serial。")
|
||
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 的有限数值。")
|
||
link = parse_product_url(arguments.url)
|
||
if link.goods_id != EXPECTED_GOODS_ID:
|
||
raise ValueError("--url 不是 T-103 已取证商品。")
|
||
if (arguments.color, arguments.size) not in TASK_TO_UI_SELECTION:
|
||
raise ValueError("--color 与 --size 必须是 T-103 已取证任务值。")
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
arguments = parse_arguments(argv)
|
||
try:
|
||
validate_arguments(arguments)
|
||
except (ValueError, ProductUrlError) 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
|
||
|
||
runner = SkuSelectionRunner(
|
||
AdbClient(SubprocessAdbRunner(arguments.adb), timeout_seconds=arguments.timeout),
|
||
NoReconnectUiautomatorConnector(adbutils.AdbClient(socket_timeout=arguments.timeout).device_list, u2.connect),
|
||
timeout_seconds=arguments.timeout,
|
||
)
|
||
try:
|
||
result = runner.run(arguments.serial, arguments.url, arguments.color, arguments.size, arguments.output_dir)
|
||
except (DeviceConnectionError, SkuSelectionRunError, SkuSelectionError) as error:
|
||
# Flow 可能来自测试替身或未来实现；CLI 不回显任何异常正文，避免泄露节点树或页面文本。
|
||
print(
|
||
f"规格恢复失败：stage={safe_failure_stage(error)}；已停止，未发布本地证据目录。",
|
||
file=sys.stderr,
|
||
)
|
||
return 1
|
||
except OSError:
|
||
print("规格恢复失败：无法创建或发布本地证据目录。", file=sys.stderr)
|
||
return 1
|
||
|
||
print(f"规格恢复完成：{result.output_directory}")
|
||
print(f"manifest：{result.manifest_path}")
|
||
print(f"目标规格：{arguments.color} / {arguments.size}")
|
||
print(f"确认单价：{result.unit_price}")
|
||
print("页面对应性：请人工核对本地原始截图。")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|