139 lines
5.9 KiB
Python
139 lines
5.9 KiB
Python
"""运行 T-107 最终面板 Gate3 只读观察与一次安全返回。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from datetime import datetime
|
|
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.final_submit_panel import FinalSubmitPanelError
|
|
from cmbuyer_client.pdd.final_submit_panel_runner import FinalSubmitPanelRunner
|
|
from cmbuyer_client.pdd.quantity_gate2 import (
|
|
EXPECTED_GATE1_UNIT_PRICE,
|
|
EXPECTED_GOODS_ID,
|
|
TASK_COLOR,
|
|
TASK_SIZE,
|
|
Gate2Observation,
|
|
)
|
|
from cmbuyer_client.pdd.quantity_gate2_spike import Android16TopResumedForegroundReader
|
|
|
|
|
|
def parse_arguments(argv: list[str] | None = None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="运行 T-107 围栏前最终面板只读 dry-run。")
|
|
parser.add_argument("--serial", required=True, help="显式 ADB serial；禁止自动选择。")
|
|
parser.add_argument("--goods-id", required=True)
|
|
parser.add_argument("--color", required=True)
|
|
parser.add_argument("--size", required=True)
|
|
parser.add_argument("--quantity", required=True, type=int)
|
|
parser.add_argument("--gate1-unit-price", required=True)
|
|
parser.add_argument("--gate2-panel-total-price", required=True)
|
|
parser.add_argument("--gate2-screenshot", required=True, type=Path)
|
|
parser.add_argument("--gate2-captured-at", required=True)
|
|
parser.add_argument("--max-total-price", required=True)
|
|
parser.add_argument("--output-dir", required=True, type=Path)
|
|
parser.add_argument("--timeout", type=float, default=10.0)
|
|
parser.add_argument("--adb", default="adb")
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def validate_arguments(arguments: argparse.Namespace) -> datetime:
|
|
if type(arguments.serial) is not str or not arguments.serial.strip() or arguments.serial != arguments.serial.strip():
|
|
raise ValueError("必须显式提供非空 --serial。")
|
|
if arguments.goods_id != EXPECTED_GOODS_ID:
|
|
raise ValueError("--goods-id 不是 T-107 已批准目标。")
|
|
if arguments.color != TASK_COLOR or arguments.size != TASK_SIZE:
|
|
raise ValueError("颜色或尺码不是 T-107 已批准目标。")
|
|
if type(arguments.quantity) is not int or arguments.quantity != 2:
|
|
raise ValueError("本次真机验收只批准 --quantity 2。")
|
|
if arguments.gate1_unit_price != EXPECTED_GATE1_UNIT_PRICE:
|
|
raise ValueError("--gate1-unit-price 与已确认 Gate1 不一致。")
|
|
if arguments.gate2_panel_total_price != "32.76":
|
|
raise ValueError("--gate2-panel-total-price 与 T-106 已确认值不一致。")
|
|
if not isinstance(arguments.gate2_screenshot, Path) or not arguments.gate2_screenshot.is_file():
|
|
raise ValueError("--gate2-screenshot 必须是现有显式文件。")
|
|
try:
|
|
captured_at = datetime.fromisoformat(arguments.gate2_captured_at)
|
|
except (TypeError, ValueError) as error:
|
|
raise ValueError("--gate2-captured-at 必须是带时区 ISO 时间。") from error
|
|
if captured_at.utcoffset() is None:
|
|
raise ValueError("--gate2-captured-at 必须带时区。")
|
|
if arguments.max_total_price != "40.00":
|
|
raise ValueError("本次真机验收固定 --max-total-price 40.00。")
|
|
if not isinstance(arguments.output_dir, Path) or not arguments.output_dir.name or arguments.output_dir.exists():
|
|
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 的有限数值。")
|
|
return captured_at
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
arguments = parse_arguments(argv)
|
|
try:
|
|
captured_at = 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("失败：缺少采购工具真机依赖。", file=sys.stderr)
|
|
return 2
|
|
|
|
adb_runner = SubprocessAdbRunner(arguments.adb)
|
|
runner = FinalSubmitPanelRunner(
|
|
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,
|
|
)
|
|
gate2 = Gate2Observation(
|
|
requested_color=arguments.color,
|
|
requested_size=arguments.size,
|
|
actual_color=arguments.color,
|
|
actual_size=arguments.size,
|
|
requested_quantity=arguments.quantity,
|
|
quantity_read=arguments.quantity,
|
|
gate1_unit_price=arguments.gate1_unit_price,
|
|
gate2_panel_total_price=arguments.gate2_panel_total_price,
|
|
max_total_price=arguments.max_total_price,
|
|
screenshot_path=arguments.gate2_screenshot,
|
|
captured_at=captured_at,
|
|
)
|
|
try:
|
|
runner.run(
|
|
arguments.serial,
|
|
arguments.goods_id,
|
|
gate2,
|
|
arguments.output_dir,
|
|
)
|
|
except (DeviceConnectionError, FinalSubmitPanelError, OSError):
|
|
# 不回显第三方异常、serial、本机路径或页面正文。
|
|
print("T-107 最终面板 dry-run 失败：已停止，未发布证据目录。", file=sys.stderr)
|
|
return 1
|
|
|
|
print("T-107 最终面板 dry-run 完成。")
|
|
print("人工复核：Gate2/Gate3、最终控件唯一、一次安全返回，且未创建订单或进入付款。")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|