feat(client): implement verified SKU selection flow
This commit is contained in:
QiuSW
@@ -0,0 +1,84 @@
|
||||
"""恢复 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
|
||||
|
||||
|
||||
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("规格恢复失败：已停止，未发布本地证据目录。", 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())
|
||||
@@ -1,15 +1,23 @@
|
||||
"""拼多多链接的受限打开与只读取证。
|
||||
"""拼多多链接的受限打开、只读取证与经取证的规格面板选择。
|
||||
|
||||
此包不提供页面选择器、输入、滑动、下单或支付能力。
|
||||
此包不提供通用页面选择器、输入、滑动或任何订单动作。
|
||||
"""
|
||||
|
||||
from .product_open import ProductOpenCapturer, ProductOpenResult
|
||||
from .product_url import ProductUrl, ProductUrlError, parse_product_url
|
||||
from .sku_selection import SkuSelection, SkuSelectionError, SkuSelectionFlow
|
||||
from .sku_selection_runner import SkuSelectionRunError, SkuSelectionRunResult, SkuSelectionRunner
|
||||
|
||||
__all__ = [
|
||||
"ProductOpenCapturer",
|
||||
"ProductOpenResult",
|
||||
"ProductUrl",
|
||||
"ProductUrlError",
|
||||
"SkuSelection",
|
||||
"SkuSelectionError",
|
||||
"SkuSelectionFlow",
|
||||
"SkuSelectionRunError",
|
||||
"SkuSelectionRunResult",
|
||||
"SkuSelectionRunner",
|
||||
"parse_product_url",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,371 @@
|
||||
"""T-103：仅限已取证 PDD 8.17.0 的规格面板恢复。"""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
import re
|
||||
from time import monotonic, sleep
|
||||
from typing import Any, Callable, Protocol
|
||||
from xml.etree import ElementTree
|
||||
|
||||
from ..device.baseline import PDD_PACKAGE
|
||||
from .product_open import EXPECTED_PDD_VERSION
|
||||
from .product_url import parse_product_url
|
||||
|
||||
EXPECTED_GOODS_ID = "937122477375"
|
||||
EXPECTED_UNIT_PRICE = "12.88"
|
||||
# 任务值不是页面判据；右侧是 v5 取证的唯一 accessibility 文案（空格/全角括号均有意义）。
|
||||
TASK_TO_UI_SELECTION = {("黑色CHA(纯棉)", "M(建议100-115)"): ("黑色 CHA (纯棉）", "M（建议100-115）")}
|
||||
_TARGET_COLOR_UI, _TARGET_SIZE_UI = next(iter(TASK_TO_UI_SELECTION.values()))
|
||||
_ENTRY = "快要抢光"
|
||||
_SIZE = "尺码"
|
||||
_W, _H = 1080, 2376
|
||||
_PRICE_PARENT = "[396,498][895,570]"
|
||||
_CURRENT = "[396,503][712,570]"
|
||||
_ORIGINAL = "[730,503][895,570]"
|
||||
_SUMMARY = "[396,654][1053,716]"
|
||||
_COLOR_REGION = "[36,1000][1080,1631]"
|
||||
_SIZE_LABEL = "[36,1654][114,1700]"
|
||||
_SIZE_HEADER = "[36,1637][1044,1718]"
|
||||
_SIZE_OPTIONS = "[36,1730][1044,2045]"
|
||||
_BOUNDS = re.compile(r"^\[(\d+),(\d+)\]\[(\d+),(\d+)\]$")
|
||||
_PRICE = re.compile(r"^[^0-9¥￥]*[¥￥]([1-9][0-9]*\.[0-9]{2})$")
|
||||
_ORIGINAL_PRICE = re.compile(r"^[¥￥][1-9][0-9]*\.[0-9]{2}$")
|
||||
_BAD_PRICE_ROLE = ("提交订单", "支付", "优惠", "券", "会员", "补贴", "区间", "实付", "到手", "原价", "划线价", "最低", "低至", "起价", "下单", "先用后付", "预估")
|
||||
|
||||
|
||||
class SkuSelectionError(RuntimeError):
|
||||
"""已取证判据不成立时的脱敏停止。"""
|
||||
|
||||
|
||||
class SkuPanelDevice(Protocol):
|
||||
def app_info(self, package_name: str) -> dict[str, Any]: ...
|
||||
def app_current(self) -> dict[str, Any]: ...
|
||||
def dump_window_hierarchy(self) -> str: ...
|
||||
def tap_sku_entry(self, bounds: str) -> None: ...
|
||||
def tap_sku_option(self, bounds: str) -> None: ...
|
||||
def leave_sku_panel(self) -> None: ...
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SkuSelection:
|
||||
color: str
|
||||
size: str
|
||||
|
||||
|
||||
def resolve_task_selection(color: str, size: str) -> SkuSelection:
|
||||
mapped = TASK_TO_UI_SELECTION.get((color, size))
|
||||
if mapped is None:
|
||||
raise SkuSelectionError("规格任务值不是已取证的唯一目标，已停止操作。")
|
||||
return SkuSelection(*mapped)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _Node:
|
||||
element: ElementTree.Element
|
||||
parent: "_Node | None"
|
||||
@property
|
||||
def text(self) -> str: return self.element.get("text", "")
|
||||
@property
|
||||
def desc(self) -> str: return self.element.get("content-desc", "")
|
||||
@property
|
||||
def bounds(self) -> str: return self.element.get("bounds", "")
|
||||
|
||||
|
||||
class SkuSelectionFlow:
|
||||
def __init__(self, device: SkuPanelDevice, entry_wait_timeout_seconds: float = 0.2,
|
||||
entry_poll_interval_seconds: float = 0.2, monotonic_clock: Callable[[], float] = monotonic,
|
||||
sleep_function: Callable[[float], None] = sleep) -> None:
|
||||
if entry_wait_timeout_seconds < 0 or entry_poll_interval_seconds <= 0:
|
||||
raise ValueError("入口等待参数无效。")
|
||||
self._device, self._entry_timeout, self._poll = device, entry_wait_timeout_seconds, entry_poll_interval_seconds
|
||||
self._clock, self._sleep = monotonic_clock, sleep_function
|
||||
self._pending: tuple[str, Callable[[list[_Node]], Any]] | None = None
|
||||
|
||||
def open_sku_panel(self, product_url: str, pre_intent_hierarchy: str | None = None) -> None:
|
||||
if parse_product_url(product_url).goods_id != EXPECTED_GOODS_ID:
|
||||
raise SkuSelectionError("商品不是已取证目标，已停止操作。")
|
||||
if pre_intent_hierarchy is not None:
|
||||
previous_nodes = _parse_nodes(pre_intent_hierarchy)
|
||||
if _eligible_entries(previous_nodes):
|
||||
raise SkuSelectionError("intent 前页面已出现规格入口，已拒绝旧商品误点。")
|
||||
entry, before = self._wait_for_entry(pre_intent_hierarchy)
|
||||
_action_bounds(entry.bounds)
|
||||
self._pending = (before, _panel)
|
||||
self._device.tap_sku_entry(entry.bounds)
|
||||
self._wait_after_action(before, _panel)
|
||||
|
||||
def select_sku_options(self, selection: SkuSelection) -> None:
|
||||
if selection not in {SkuSelection(*item) for item in TASK_TO_UI_SELECTION.values()}:
|
||||
raise SkuSelectionError("规格 UI 文案不是获准目标，已停止操作。")
|
||||
initial = self._verified_nodes()
|
||||
_option(initial, "color", selection.color); _option(initial, "size", selection.size)
|
||||
_selected_label(initial, "color"); _selected_label(initial, "size")
|
||||
self._restore("color", selection.color)
|
||||
self._restore("size", selection.size)
|
||||
|
||||
def read_sku_unit_price(self) -> str:
|
||||
return _unit_price(self._verified_nodes())
|
||||
|
||||
def verify_target_selection_and_read_price(self, selection: SkuSelection) -> str:
|
||||
nodes = self._verified_nodes()
|
||||
_selected(nodes, "color", selection.color)
|
||||
_selected(nodes, "size", selection.size)
|
||||
return _unit_price(nodes)
|
||||
|
||||
def exit_sku_panel_safely(self) -> None:
|
||||
self._require_foreground()
|
||||
before = self._read_hierarchy()
|
||||
_panel(_parse_nodes(before))
|
||||
self._device.leave_sku_panel()
|
||||
deadline = self._clock() + self._entry_timeout
|
||||
while True:
|
||||
self._require_foreground()
|
||||
raw = self._read_hierarchy()
|
||||
if raw != before:
|
||||
try:
|
||||
_panel(_parse_nodes(raw))
|
||||
except SkuSelectionError:
|
||||
return
|
||||
remaining = deadline - self._clock()
|
||||
if remaining <= 0:
|
||||
raise SkuSelectionError("安全退出后未确认离开规格面板，未重试返回。")
|
||||
self._sleep(min(self._poll, remaining))
|
||||
|
||||
def reconcile_pending_action(self) -> None:
|
||||
"""仅只读调和一次已发出但尚未得到后置条件确认的动作。"""
|
||||
if self._pending is None:
|
||||
return
|
||||
before, condition = self._pending
|
||||
self._wait_after_action(before, condition)
|
||||
|
||||
def _restore(self, dimension: str, expected: str) -> None:
|
||||
self._require_foreground()
|
||||
before = self._read_hierarchy()
|
||||
nodes = _panel(_parse_nodes(before))
|
||||
target = _option(nodes, dimension, expected)
|
||||
if _selected_label(nodes, dimension) == expected:
|
||||
return
|
||||
_action_bounds(target.bounds)
|
||||
condition: Callable[[list[_Node]], Any]
|
||||
if dimension == "color":
|
||||
condition = lambda refreshed: _post_color(refreshed, expected)
|
||||
else:
|
||||
condition = lambda refreshed: _post_all_targets(refreshed, expected)
|
||||
self._pending = (before, condition)
|
||||
self._device.tap_sku_option(target.bounds)
|
||||
if dimension == "color":
|
||||
self._wait_after_action(before, condition)
|
||||
else:
|
||||
self._wait_after_action(before, condition)
|
||||
|
||||
def _wait_for_entry(self, previous: str | None) -> tuple[_Node, str]:
|
||||
deadline, stable = self._clock() + self._entry_timeout, None
|
||||
while True:
|
||||
self._require_version()
|
||||
current = self._device.app_current()
|
||||
if isinstance(current, dict) and current.get("package") == PDD_PACKAGE:
|
||||
raw = self._read_hierarchy()
|
||||
entries = _eligible_entries(_parse_nodes(raw))
|
||||
if len(entries) > 1:
|
||||
raise SkuSelectionError("商品页规格入口不唯一，已停止操作。")
|
||||
if len(entries) == 1 and raw != previous:
|
||||
if stable == raw:
|
||||
return entries[0], raw
|
||||
stable = raw
|
||||
else:
|
||||
stable = None
|
||||
else:
|
||||
stable = None
|
||||
remaining = deadline - self._clock()
|
||||
if remaining <= 0:
|
||||
raise SkuSelectionError("等待已取证规格入口超时，未执行点击。")
|
||||
self._sleep(min(self._poll, remaining))
|
||||
|
||||
def _wait_after_action(self, previous: str, condition: Callable[[list[_Node]], Any]) -> list[_Node]:
|
||||
deadline = self._clock() + self._entry_timeout
|
||||
while True:
|
||||
self._require_foreground()
|
||||
raw = self._read_hierarchy()
|
||||
if raw != previous:
|
||||
nodes = _parse_nodes(raw)
|
||||
try:
|
||||
condition(nodes)
|
||||
self._pending = None
|
||||
return nodes
|
||||
except SkuSelectionError:
|
||||
pass
|
||||
remaining = deadline - self._clock()
|
||||
if remaining <= 0:
|
||||
raise SkuSelectionError("动作后页面未在限定时间内满足已取证后置条件，未重试动作。")
|
||||
self._sleep(min(self._poll, remaining))
|
||||
|
||||
def _verified_nodes(self) -> list[_Node]:
|
||||
self._require_foreground()
|
||||
return _panel(self._read_nodes())
|
||||
|
||||
def _require_version(self) -> None:
|
||||
info = self._device.app_info(PDD_PACKAGE)
|
||||
version = (info.get("versionName") or info.get("version_name")) if isinstance(info, dict) else None
|
||||
if version != EXPECTED_PDD_VERSION:
|
||||
raise SkuSelectionError("拼多多版本与已取证版本不一致，已停止操作。")
|
||||
|
||||
def _require_foreground(self) -> None:
|
||||
self._require_version()
|
||||
current = self._device.app_current()
|
||||
if not isinstance(current, dict) or current.get("package") != PDD_PACKAGE:
|
||||
raise SkuSelectionError("拼多多不在前台，已停止操作。")
|
||||
|
||||
def _read_hierarchy(self) -> str:
|
||||
try: raw = self._device.dump_window_hierarchy()
|
||||
except Exception as error: raise SkuSelectionError("节点树读取失败，已停止操作。") from error
|
||||
if not isinstance(raw, str) or not raw: raise SkuSelectionError("节点树不可用，已停止操作。")
|
||||
return raw
|
||||
|
||||
def _read_nodes(self) -> list[_Node]: return _parse_nodes(self._read_hierarchy())
|
||||
|
||||
|
||||
def _parse_nodes(raw: str) -> list[_Node]:
|
||||
try: root = ElementTree.fromstring(raw)
|
||||
except ElementTree.ParseError as error: raise SkuSelectionError("节点树格式无效，已停止操作。") from error
|
||||
if root.tag != "hierarchy": raise SkuSelectionError("节点树根节点无效，已停止操作。")
|
||||
result: list[_Node] = []
|
||||
def visit(element: ElementTree.Element, parent: _Node | None) -> None:
|
||||
node = _Node(element, parent); result.append(node)
|
||||
for child in element: visit(child, node)
|
||||
visit(root, None)
|
||||
return result
|
||||
|
||||
|
||||
def _panel(nodes: list[_Node]) -> list[_Node]:
|
||||
parent = _one([n for n in nodes if n.bounds == _PRICE_PARENT], "规格面板价格区域不唯一，已停止操作。")
|
||||
_one([n for n in nodes if n.parent is parent and n.bounds == _ORIGINAL and _readonly(n) and _ORIGINAL_PRICE.fullmatch(n.text)], "规格面板原价槽位不唯一，已停止操作。")
|
||||
_one([n for n in nodes if n.bounds == _SUMMARY and _readonly(n) and n.text.startswith("已选：")], "规格面板已选摘要不唯一，已停止操作。")
|
||||
_color_container(nodes); _size_container(nodes)
|
||||
return nodes
|
||||
|
||||
|
||||
def _unit_price(nodes: list[_Node]) -> str:
|
||||
parent = _one([n for n in nodes if n.bounds == _PRICE_PARENT], "规格面板价格区域不唯一，已停止读取。")
|
||||
money = [n for n in nodes if n.parent is parent and _readonly(n) and any(mark in n.text for mark in "¥￥")]
|
||||
if len(money) != 2: raise SkuSelectionError("规格面板金额槽位不唯一，已停止读取。")
|
||||
current = _one([n for n in money if n.bounds == _CURRENT and not _clickable_ancestor(n) and not any(word in n.text for word in _BAD_PRICE_ROLE) and _PRICE.fullmatch(n.text)], "规格面板现价不唯一或不符合已取证槽位，已停止读取。")
|
||||
if not any(n.bounds == _ORIGINAL and _ORIGINAL_PRICE.fullmatch(n.text) for n in money):
|
||||
raise SkuSelectionError("规格面板原价槽位无效，已停止读取。")
|
||||
match = _PRICE.fullmatch(current.text)
|
||||
if match is None: raise SkuSelectionError("规格面板现价格式失效，已停止读取。")
|
||||
return match.group(1)
|
||||
|
||||
|
||||
def _option(nodes: list[_Node], dimension: str, expected: str) -> _Node:
|
||||
_panel(nodes)
|
||||
return _one([n for n in _options(nodes, dimension) if _label(n) == expected], "规格选项不唯一或不是精确匹配，已停止操作。")
|
||||
|
||||
|
||||
def _selected(nodes: list[_Node], dimension: str, expected: str) -> None:
|
||||
if _selected_label(nodes, dimension) != expected:
|
||||
raise SkuSelectionError("规格选择后读回的 selected 文案不一致，已停止操作。")
|
||||
|
||||
|
||||
def _selected_label(nodes: list[_Node], dimension: str) -> str:
|
||||
selected = [n for n in _options(nodes, dimension) if n.element.get("selected") == "true"]
|
||||
label = _label(_one(selected, "规格维度没有唯一 selected 状态，已停止操作。"))
|
||||
if label is None: raise SkuSelectionError("规格维度 selected 文案无效，已停止操作。")
|
||||
return label
|
||||
|
||||
|
||||
def _post_color(nodes: list[_Node], expected: str) -> None:
|
||||
_panel(nodes)
|
||||
_selected(nodes, "color", expected)
|
||||
_selected_label(nodes, "size")
|
||||
|
||||
|
||||
def _post_all_targets(nodes: list[_Node], expected_size: str) -> None:
|
||||
_panel(nodes)
|
||||
_selected(nodes, "color", _TARGET_COLOR_UI)
|
||||
_selected(nodes, "size", expected_size)
|
||||
|
||||
|
||||
def _options(nodes: list[_Node], dimension: str) -> list[_Node]:
|
||||
container = _color_container(nodes) if dimension == "color" else _size_container(nodes) if dimension == "size" else None
|
||||
if container is None: raise SkuSelectionError("未知规格维度，已停止操作。")
|
||||
candidates = [n for n in nodes if _descendant(n, container) and _contained(n, container) and _choice(n) and _label(n) is not None]
|
||||
return [n for n in candidates if not _labeled_ancestor(n, candidates)]
|
||||
|
||||
|
||||
def _color_container(nodes: list[_Node]) -> _Node:
|
||||
return _one([n for n in nodes if n.element.get("package") == PDD_PACKAGE and n.element.get("class") == "androidx.recyclerview.widget.RecyclerView" and n.bounds == _COLOR_REGION], "规格面板颜色容器不唯一，已停止操作。")
|
||||
|
||||
|
||||
def _size_container(nodes: list[_Node]) -> _Node:
|
||||
label = _one([n for n in nodes if n.text == _SIZE and n.bounds == _SIZE_LABEL and _readonly(n)], "规格面板尺码标签不唯一，已停止操作。")
|
||||
header = label.parent
|
||||
if header is None or header.element.get("package") != PDD_PACKAGE or header.element.get("class") != "android.widget.LinearLayout" or header.bounds != _SIZE_HEADER or header.parent is None:
|
||||
raise SkuSelectionError("规格面板尺码标题容器不符合已取证结构，已停止操作。")
|
||||
return _one([n for n in nodes if n.parent is header.parent and n.element.get("package") == PDD_PACKAGE and n.element.get("class") == "android.widget.LinearLayout" and n.bounds == _SIZE_OPTIONS], "规格面板尺码选项容器不唯一，已停止操作。")
|
||||
|
||||
|
||||
def _label(node: _Node) -> str | None:
|
||||
values = {value for value in (node.text, node.desc) if value}
|
||||
return values.pop() if len(values) == 1 else None
|
||||
|
||||
|
||||
def _labeled_ancestor(node: _Node, candidates: list[_Node]) -> bool:
|
||||
ids, parent = {id(n.element) for n in candidates}, node.parent
|
||||
while parent is not None:
|
||||
if id(parent.element) in ids and _label(parent) is not None: return True
|
||||
parent = parent.parent
|
||||
return False
|
||||
|
||||
|
||||
def _descendant(node: _Node, ancestor: _Node) -> bool:
|
||||
parent = node.parent
|
||||
while parent is not None:
|
||||
if parent.element is ancestor.element: return True
|
||||
parent = parent.parent
|
||||
return False
|
||||
|
||||
|
||||
def _contained(node: _Node, container: _Node) -> bool:
|
||||
left, top, right, bottom = _action_bounds(node.bounds)
|
||||
outer_left, outer_top, outer_right, outer_bottom = _action_bounds(container.bounds)
|
||||
return outer_left <= left < right <= outer_right and outer_top <= top < bottom <= outer_bottom
|
||||
|
||||
|
||||
def _clickable_ancestor(node: _Node) -> bool:
|
||||
parent = node.parent
|
||||
while parent is not None:
|
||||
if parent.element.get("clickable") == "true": return True
|
||||
parent = parent.parent
|
||||
return False
|
||||
|
||||
|
||||
def _readonly(node: _Node) -> bool:
|
||||
return node.element.get("package") == PDD_PACKAGE and node.element.get("class") == "android.widget.TextView" and node.element.get("clickable") == "false" and node.element.get("enabled") == "true" and node.element.get("visible-to-user") == "true"
|
||||
|
||||
|
||||
def _live(node: _Node) -> bool:
|
||||
return node.element.get("package") == PDD_PACKAGE and node.element.get("clickable") == "true" and node.element.get("enabled") == "true" and node.element.get("visible-to-user") == "true" and bool(node.bounds)
|
||||
|
||||
|
||||
def _choice(node: _Node) -> bool:
|
||||
return _live(node) and node.element.get("class") == "android.view.ViewGroup" and node.element.get("selected") in {"true", "false"}
|
||||
|
||||
|
||||
def _eligible_entries(nodes: list[_Node]) -> list[_Node]:
|
||||
# 面板与底部硬拒绝区绝不能被误作商品页入口；入口仍只按经取证精确文案、包、状态判定。
|
||||
if any(node.bounds == _PRICE_PARENT for node in nodes): return []
|
||||
return [node for node in nodes if _live(node) and node.element.get("class") == "android.widget.TextView" and node.text == _ENTRY]
|
||||
|
||||
|
||||
def _action_bounds(bounds: str) -> tuple[int, int, int, int]:
|
||||
match = _BOUNDS.fullmatch(bounds)
|
||||
if match is None: raise SkuSelectionError("规格节点坐标格式无效，已停止操作。")
|
||||
left, top, right, bottom = (int(item) for item in match.groups())
|
||||
if not (0 <= left < right <= _W and 0 <= top < bottom <= _H):
|
||||
raise SkuSelectionError("规格节点坐标不在已取证屏幕范围内，已停止操作。")
|
||||
return left, top, right, bottom
|
||||
|
||||
|
||||
def _one(nodes: list[_Node], message: str) -> _Node:
|
||||
if len(nodes) != 1: raise SkuSelectionError(message)
|
||||
return nodes[0]
|
||||
@@ -0,0 +1,348 @@
|
||||
"""T-103 真机运行边界：窄适配器、原始截图和无页面正文的摘要。"""
|
||||
|
||||
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
|
||||
from typing import Any
|
||||
from uuid import uuid4
|
||||
|
||||
from PIL import Image, UnidentifiedImageError
|
||||
|
||||
from adbutils.errors import AdbTimeout
|
||||
from uiautomator2.exceptions import HTTPTimeoutError
|
||||
|
||||
from ..device.adb import AdbClient, DeviceConnectionError, DeviceInspection
|
||||
from ..device.baseline import PDD_PACKAGE, SCREENSHOT_PARAMS, _save_base64_screenshot, _sha256_file
|
||||
from .product_open import EXPECTED_PDD_VERSION
|
||||
from .product_url import ProductUrl, parse_product_url
|
||||
from .sku_selection import (
|
||||
EXPECTED_GOODS_ID,
|
||||
EXPECTED_UNIT_PRICE,
|
||||
SkuPanelDevice,
|
||||
SkuSelectionError,
|
||||
SkuSelectionFlow,
|
||||
_action_bounds,
|
||||
resolve_task_selection,
|
||||
)
|
||||
|
||||
|
||||
EXPECTED_DEVICE_MODEL = "PKG110"
|
||||
EXPECTED_ANDROID_VERSION = "16"
|
||||
EXPECTED_SCREEN_SIZE = (1080, 2376)
|
||||
|
||||
|
||||
class SkuSelectionRunError(RuntimeError):
|
||||
"""T-103 运行未完整完成；错误文本不携带设备或页面原文。"""
|
||||
|
||||
|
||||
class SkuSelectionRunTimeoutError(SkuSelectionRunError):
|
||||
"""设备 RPC 或操作超时。"""
|
||||
|
||||
|
||||
class SkuSelectionScreenshotError(SkuSelectionRunError):
|
||||
"""原始截图无法作为完整 PNG 原子发布。"""
|
||||
|
||||
|
||||
class SkuSelectionUnexpectedPriceError(SkuSelectionRunError):
|
||||
"""取证面板现价不是本任务已确认值。"""
|
||||
|
||||
|
||||
class SkuSelectionDeviceAdapterError(SkuSelectionRunError):
|
||||
"""第三方设备接口失败的脱敏映射。"""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SkuSelectionRunResult:
|
||||
"""已发布的截图和无页面正文 manifest 摘要。"""
|
||||
|
||||
output_directory: Path
|
||||
screenshot_path: Path
|
||||
manifest_path: Path
|
||||
unit_price: str
|
||||
|
||||
|
||||
class UiautomatorSkuPanelAdapter(SkuPanelDevice):
|
||||
"""把 uiautomator2 缩为 T-103 所需的读取与三种命名操作。
|
||||
|
||||
``tap_sku_entry``、``tap_sku_option`` 和 ``leave_sku_panel`` 是仅有的状态改变方法；
|
||||
坐标由 Flow 和本类双重检查后才计算中心点，每次调用只执行一次底层动作。
|
||||
"""
|
||||
|
||||
def __init__(self, device: Any, timeout_seconds: float) -> None:
|
||||
if not _is_positive_finite(timeout_seconds):
|
||||
raise ValueError("timeout_seconds 必须是大于 0 的有限数值")
|
||||
self._device = device
|
||||
self._timeout_seconds = timeout_seconds
|
||||
self._entry_was_tapped = False
|
||||
self._left_panel = False
|
||||
|
||||
@property
|
||||
def entry_was_tapped(self) -> bool:
|
||||
"""仅供运行器决定故障后的单次尽力返回，不是页面操作。"""
|
||||
|
||||
return self._entry_was_tapped
|
||||
|
||||
@property
|
||||
def left_panel(self) -> bool:
|
||||
return self._left_panel
|
||||
|
||||
def app_info(self, package_name: str) -> dict[str, Any]:
|
||||
value = self._call("app_info", package_name)
|
||||
if not isinstance(value, dict):
|
||||
raise SkuSelectionDeviceAdapterError("无法读取应用版本，已停止操作。")
|
||||
return value
|
||||
|
||||
def app_current(self) -> dict[str, Any]:
|
||||
value = self._call("app_current")
|
||||
if not isinstance(value, dict):
|
||||
raise SkuSelectionDeviceAdapterError("无法读取前台应用，已停止操作。")
|
||||
return value
|
||||
|
||||
def dump_window_hierarchy(self) -> str:
|
||||
value = self._call("jsonrpc_call", "dumpWindowHierarchy", [False, 50], timeout=self._timeout_seconds)
|
||||
if not isinstance(value, str):
|
||||
raise SkuSelectionDeviceAdapterError("节点树读取失败，已停止操作。")
|
||||
return value
|
||||
|
||||
def tap_sku_entry(self, bounds: str) -> None:
|
||||
# 超时也可能表示底层事件已经送达；必须先封存 attempt，后续绝不重试该入口。
|
||||
self._entry_was_tapped = True
|
||||
self._tap_bounds_once(bounds)
|
||||
|
||||
def tap_sku_option(self, bounds: str) -> None:
|
||||
self._tap_bounds_once(bounds)
|
||||
|
||||
def leave_sku_panel(self) -> None:
|
||||
if self._left_panel:
|
||||
raise SkuSelectionDeviceAdapterError("规格面板已经执行过返回，已停止操作。")
|
||||
# 底层调用即使报错也可能已把返回事件送达；先封存本次机会，finally 不得再次返回。
|
||||
self._left_panel = True
|
||||
self._call("jsonrpc_call", "pressKey", ["back"], timeout=self._timeout_seconds)
|
||||
|
||||
def capture_screenshot(self) -> str:
|
||||
value = self._call("jsonrpc_call", "takeScreenshot", SCREENSHOT_PARAMS, timeout=self._timeout_seconds)
|
||||
if not isinstance(value, str):
|
||||
raise SkuSelectionScreenshotError("规格面板原始截图读取失败，未发布任何证据产物。")
|
||||
return value
|
||||
|
||||
def display_size(self) -> tuple[int, int]:
|
||||
value = self._call("window_size")
|
||||
if not isinstance(value, tuple) or len(value) != 2 or any(not isinstance(item, int) for item in value):
|
||||
raise SkuSelectionDeviceAdapterError("无法读取屏幕坐标空间，已停止操作。")
|
||||
return value
|
||||
|
||||
def _tap_bounds_once(self, bounds: str) -> None:
|
||||
left, top, right, bottom = _action_bounds(bounds)
|
||||
center_x = left + (right - left) // 2
|
||||
center_y = top + (bottom - top) // 2
|
||||
self._call("jsonrpc_call", "click", [center_x, center_y], timeout=self._timeout_seconds)
|
||||
|
||||
def _call(self, method: str, *args: Any, **kwargs: Any) -> Any:
|
||||
try:
|
||||
operation = getattr(self._device, method)
|
||||
return operation(*args, **kwargs)
|
||||
except (AdbTimeout, HTTPTimeoutError, TimeoutError) as error:
|
||||
raise SkuSelectionRunTimeoutError("规格面板设备操作超时，已停止操作。") from error
|
||||
except SkuSelectionRunError:
|
||||
raise
|
||||
except Exception as error:
|
||||
raise SkuSelectionDeviceAdapterError("规格面板设备操作失败，已停止操作。") from error
|
||||
|
||||
|
||||
class SkuSelectionRunner:
|
||||
"""只运行 T-103 目标规格恢复、价格确认、原始截图和一次安全退出。"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
adb_client: AdbClient,
|
||||
connector: Callable[[str], Any],
|
||||
timeout_seconds: float,
|
||||
monotonic_clock: Callable[[], float] = monotonic,
|
||||
) -> None:
|
||||
if not _is_positive_finite(timeout_seconds):
|
||||
raise ValueError("timeout_seconds 必须是大于 0 的有限数值")
|
||||
self._adb_client = adb_client
|
||||
self._connector = connector
|
||||
self._timeout_seconds = timeout_seconds
|
||||
self._monotonic_clock = monotonic_clock
|
||||
|
||||
def run(
|
||||
self,
|
||||
serial: str,
|
||||
product_url: str,
|
||||
task_color: str,
|
||||
task_size: str,
|
||||
output_directory: Path,
|
||||
) -> SkuSelectionRunResult:
|
||||
link = parse_product_url(product_url)
|
||||
if link.goods_id != EXPECTED_GOODS_ID:
|
||||
raise SkuSelectionRunError("商品不是 T-103 已取证目标，已停止操作。")
|
||||
selection = resolve_task_selection(task_color, task_size)
|
||||
target = Path(output_directory)
|
||||
_validate_new_target(target)
|
||||
|
||||
adapter: UiautomatorSkuPanelAdapter | None = None
|
||||
flow: SkuSelectionFlow | None = None
|
||||
staging = _prepare_staging(target)
|
||||
deadline = self._monotonic_clock() + self._timeout_seconds
|
||||
try:
|
||||
inspection = self._adb_client.inspect(serial)
|
||||
_require_expected_device(inspection)
|
||||
adapter = UiautomatorSkuPanelAdapter(self._connector(serial), self._timeout_seconds)
|
||||
_require_expected_version(adapter.app_info(PDD_PACKAGE))
|
||||
if adapter.display_size() != EXPECTED_SCREEN_SIZE:
|
||||
raise SkuSelectionRunError("设备不是已取证的竖屏坐标空间，已停止操作。")
|
||||
pre_intent_hierarchy = adapter.dump_window_hierarchy()
|
||||
# 固定 ACTION_VIEW、固定 PDD package 和 canonical goods_id；不接受任意 URL 或 shell。
|
||||
self._adb_client.start_pdd_view_intent(serial, link.goods_id)
|
||||
|
||||
remaining = deadline - self._monotonic_clock()
|
||||
if remaining <= 0:
|
||||
raise SkuSelectionRunTimeoutError("等待规格入口超时，未执行点击。")
|
||||
flow = SkuSelectionFlow(adapter, entry_wait_timeout_seconds=remaining)
|
||||
flow.open_sku_panel(link.canonical_url, pre_intent_hierarchy)
|
||||
flow.select_sku_options(selection)
|
||||
unit_price = flow.verify_target_selection_and_read_price(selection)
|
||||
if unit_price != EXPECTED_UNIT_PRICE:
|
||||
raise SkuSelectionUnexpectedPriceError("规格面板现价不是本任务已确认值，已停止操作。")
|
||||
|
||||
screenshot_path = staging / "screenshot.png"
|
||||
try:
|
||||
_save_base64_screenshot(adapter.capture_screenshot(), screenshot_path)
|
||||
_require_screenshot_size(screenshot_path)
|
||||
except SkuSelectionRunError:
|
||||
raise
|
||||
except Exception as error:
|
||||
raise SkuSelectionScreenshotError("规格面板原始截图保存失败，未发布任何证据产物。") from error
|
||||
|
||||
manifest_path = staging / "manifest.json"
|
||||
# 截图可能落在动态页面切换边界；发布前必须用一棵更新节点树同时重证两维和现价。
|
||||
final_price = flow.verify_target_selection_and_read_price(selection)
|
||||
if final_price != EXPECTED_UNIT_PRICE:
|
||||
raise SkuSelectionUnexpectedPriceError("截图后规格面板现价不是本任务已确认值，已停止操作。")
|
||||
# 正常路径仍经 Flow 做最后一次前台和面板判定；返回操作只发生一次。
|
||||
flow.exit_sku_panel_safely()
|
||||
manifest_path.write_text(
|
||||
json.dumps(_manifest(inspection, serial, link, screenshot_path, task_color, task_size), ensure_ascii=False, indent=2, sort_keys=True) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
# Windows 的 rename 不替换既有目标；并发创建 target 时保留其内容并把本次运行判失败。
|
||||
os.rename(staging, target)
|
||||
staging = None
|
||||
except (DeviceConnectionError, SkuSelectionRunError, SkuSelectionError):
|
||||
_clean_staging(staging)
|
||||
raise
|
||||
except (AdbTimeout, HTTPTimeoutError, TimeoutError) as error:
|
||||
_clean_staging(staging)
|
||||
raise SkuSelectionRunTimeoutError("规格面板运行超时，未发布任何证据产物。") from error
|
||||
except OSError as error:
|
||||
_clean_staging(staging)
|
||||
raise SkuSelectionRunError("规格面板证据目录无法创建或发布，未发布任何证据产物。") from error
|
||||
except Exception as error:
|
||||
_clean_staging(staging)
|
||||
raise SkuSelectionRunError("规格面板运行未完成，未发布任何证据产物。") from error
|
||||
finally:
|
||||
# 失败路径只能复用 Flow 的版本、前台和面板证明；证明不了便停止，绝不盲目返回。
|
||||
if flow is not None and adapter is not None and adapter.entry_was_tapped and not adapter.left_panel:
|
||||
try:
|
||||
flow.reconcile_pending_action()
|
||||
flow.exit_sku_panel_safely()
|
||||
except (SkuSelectionRunError, SkuSelectionError):
|
||||
pass
|
||||
|
||||
return SkuSelectionRunResult(
|
||||
output_directory=target,
|
||||
screenshot_path=target / "screenshot.png",
|
||||
manifest_path=target / "manifest.json",
|
||||
unit_price=EXPECTED_UNIT_PRICE,
|
||||
)
|
||||
|
||||
|
||||
def _is_positive_finite(value: object) -> bool:
|
||||
return isinstance(value, (int, float)) and not isinstance(value, bool) and value > 0 and isfinite(value)
|
||||
|
||||
|
||||
def _validate_new_target(target: Path) -> None:
|
||||
if target.exists():
|
||||
raise SkuSelectionRunError("输出目录已存在；为防止覆盖旧证据，已停止操作。")
|
||||
if not target.name:
|
||||
raise SkuSelectionRunError("输出目录必须是明确的新目录。")
|
||||
|
||||
|
||||
def _prepare_staging(target: Path) -> Path:
|
||||
staging: Path | None = None
|
||||
try:
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
staging = target.parent / f".{target.name}.staging-{uuid4().hex}"
|
||||
staging.mkdir()
|
||||
probe = staging / ".write-probe"
|
||||
probe.write_bytes(b"ok")
|
||||
probe.unlink()
|
||||
return staging
|
||||
except OSError as error:
|
||||
_clean_staging(staging)
|
||||
raise SkuSelectionRunError("输出目录不可写，已停止操作。") from error
|
||||
|
||||
|
||||
def _clean_staging(staging: Path | None) -> None:
|
||||
if staging is not None and staging.exists():
|
||||
shutil.rmtree(staging)
|
||||
|
||||
|
||||
def _require_expected_version(app_info: object) -> str:
|
||||
version = (app_info.get("versionName") or app_info.get("version_name")) if isinstance(app_info, dict) else None
|
||||
if version != EXPECTED_PDD_VERSION:
|
||||
raise SkuSelectionRunError("拼多多版本与已取证版本不一致，已停止操作。")
|
||||
return version
|
||||
|
||||
|
||||
def _require_expected_device(inspection: DeviceInspection) -> None:
|
||||
if inspection.model != EXPECTED_DEVICE_MODEL or inspection.android_version != EXPECTED_ANDROID_VERSION:
|
||||
raise SkuSelectionRunError("设备型号或 Android 版本不是已取证组合，已停止操作。")
|
||||
|
||||
|
||||
def _require_screenshot_size(screenshot_path: Path) -> None:
|
||||
try:
|
||||
with Image.open(screenshot_path) as image:
|
||||
image.load()
|
||||
if image.size != EXPECTED_SCREEN_SIZE:
|
||||
raise SkuSelectionScreenshotError("原始截图坐标空间不是已取证尺寸，未发布任何证据产物。")
|
||||
except SkuSelectionRunError:
|
||||
raise
|
||||
except (UnidentifiedImageError, OSError) as error:
|
||||
raise SkuSelectionScreenshotError("原始截图无效，未发布任何证据产物。") from error
|
||||
|
||||
|
||||
def _manifest(inspection: DeviceInspection, serial: str, link: ProductUrl, screenshot_path: Path, task_color: str, task_size: str) -> dict[str, Any]:
|
||||
"""仅写可审计摘要；原始 serial、节点树、页面文案和实际截图内容均不写入 manifest。"""
|
||||
|
||||
return {
|
||||
"schema_version": 1,
|
||||
"captured_at": datetime.now(UTC).isoformat(),
|
||||
"operation": "t103-sku-selection",
|
||||
"product": {"goods_id": link.goods_id, "canonical_url": link.canonical_url},
|
||||
"target_selection": {"color": task_color, "size": task_size},
|
||||
"unit_price": EXPECTED_UNIT_PRICE,
|
||||
"selection_status": "restored",
|
||||
"panel_status": "verified",
|
||||
"safe_exit": "completed",
|
||||
"page_identity": "human_review_required",
|
||||
"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": EXPECTED_PDD_VERSION,
|
||||
},
|
||||
"artifacts": [{"path": screenshot_path.name, "sha256": _sha256_file(screenshot_path)}],
|
||||
}
|
||||
@@ -0,0 +1,26 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<hierarchy>
|
||||
<node package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" bounds="[0,474][1080,2328]">
|
||||
<node package="" class="android.view.ViewGroup" bounds="[396,498][895,570]">
|
||||
<node text="快卖完 ¥12.88" package="com.xunmeng.pinduoduo" class="android.widget.TextView" clickable="false" enabled="true" visible-to-user="true" bounds="[396,503][712,570]" />
|
||||
<node text="¥29.88" package="com.xunmeng.pinduoduo" class="android.widget.TextView" clickable="false" enabled="true" visible-to-user="true" bounds="[730,503][895,570]" />
|
||||
</node>
|
||||
<node text="已选： 黑色 CHA (纯棉） M（建议100-115）" package="com.xunmeng.pinduoduo" class="android.widget.TextView" clickable="false" enabled="true" visible-to-user="true" bounds="[396,654][1053,716]" />
|
||||
<node package="com.xunmeng.pinduoduo" class="androidx.recyclerview.widget.RecyclerView" bounds="[36,1000][1080,1631]">
|
||||
<node content-desc="黑色 CHA (纯棉）" package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" selected="true" clickable="true" enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]" />
|
||||
<node content-desc="粉红" package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" selected="false" clickable="true" enabled="true" visible-to-user="true" bounds="[456,1000][690,1172]" />
|
||||
</node>
|
||||
<node package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" bounds="[36,1637][1044,2045]">
|
||||
<node package="com.xunmeng.pinduoduo" class="android.widget.LinearLayout" bounds="[36,1637][1044,1718]">
|
||||
<node text="尺码" package="com.xunmeng.pinduoduo" class="android.widget.TextView" clickable="false" enabled="true" visible-to-user="true" bounds="[36,1654][114,1700]" />
|
||||
</node>
|
||||
<node package="com.xunmeng.pinduoduo" class="android.widget.LinearLayout" bounds="[36,1730][1044,2045]">
|
||||
<node text="M（建议100-115）" package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" selected="true" clickable="true" enabled="true" visible-to-user="true" bounds="[439,1730][831,1815]" />
|
||||
<node text="L（建议115-130）" package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" selected="false" clickable="true" enabled="true" visible-to-user="true" bounds="[840,1730][1044,1815]" />
|
||||
</node>
|
||||
</node>
|
||||
<node package="com.xunmeng.pinduoduo" class="android.widget.LinearLayout" clickable="true" enabled="true" visible-to-user="true" bounds="[357,2181][722,2328]">
|
||||
<node text="提交订单 ¥12.88" package="com.xunmeng.pinduoduo" class="android.widget.TextView" clickable="false" enabled="true" visible-to-user="true" bounds="[369,2225][710,2284]" />
|
||||
</node>
|
||||
</node>
|
||||
</hierarchy>
|
||||
@@ -0,0 +1,619 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
import base64
|
||||
from contextlib import redirect_stderr
|
||||
from io import BytesIO
|
||||
import importlib.util
|
||||
from pathlib import Path
|
||||
from tempfile import TemporaryDirectory
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
from xml.etree import ElementTree
|
||||
|
||||
from PIL import Image
|
||||
|
||||
import cmbuyer_client.pdd as pdd
|
||||
import cmbuyer_client.pdd.sku_selection_runner as runner_module
|
||||
from cmbuyer_client.device.adb import AdbDevice, DeviceInspection
|
||||
from cmbuyer_client.pdd import SkuSelectionError, SkuSelectionFlow, SkuSelectionRunner
|
||||
from cmbuyer_client.pdd.sku_selection import SkuPanelDevice, _action_bounds, resolve_task_selection
|
||||
from cmbuyer_client.pdd.sku_selection_runner import (
|
||||
SkuSelectionDeviceAdapterError,
|
||||
SkuSelectionRunError,
|
||||
SkuSelectionScreenshotError,
|
||||
UiautomatorSkuPanelAdapter,
|
||||
)
|
||||
|
||||
|
||||
_FIXTURE = Path(__file__).with_name("fixtures") / "sku_panel_8_17_0.xml"
|
||||
_TARGET_URL = "https://mobile.yangkeduo.com/goods.html?goods_id=937122477375"
|
||||
_TASK_COLOR = "黑色CHA(纯棉)"
|
||||
_TASK_SIZE = "M(建议100-115)"
|
||||
_PRODUCT_PAGE = """<hierarchy><node text="快要抢光" package="com.xunmeng.pinduoduo"
|
||||
class="android.widget.TextView" clickable="true" enabled="true" visible-to-user="true"
|
||||
bounds="[20,800][320,900]" /></hierarchy>"""
|
||||
|
||||
|
||||
def _png_base64() -> str:
|
||||
image = Image.new("RGB", (1080, 2376), "white")
|
||||
raw = BytesIO()
|
||||
image.save(raw, format="PNG")
|
||||
return base64.b64encode(raw.getvalue()).decode("ascii")
|
||||
|
||||
|
||||
class _RawDevice:
|
||||
def __init__(self, hierarchy: str = _PRODUCT_PAGE, screenshot: str | None = None) -> None:
|
||||
self.hierarchy = hierarchy
|
||||
self.panel_hierarchy = _FIXTURE.read_text(encoding="utf-8")
|
||||
self.version = "8.17.0"
|
||||
self.package = "com.xunmeng.pinduoduo"
|
||||
self.screenshot = _png_base64() if screenshot is None else screenshot
|
||||
self.calls: list[tuple[object, ...]] = []
|
||||
self.fail_color_readback = False
|
||||
|
||||
def app_info(self, package_name: str) -> dict[str, str]:
|
||||
self.calls.append(("app_info", package_name))
|
||||
return {"versionName": self.version}
|
||||
|
||||
def app_current(self) -> dict[str, str]:
|
||||
self.calls.append(("app_current",))
|
||||
return {"package": self.package}
|
||||
|
||||
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
|
||||
self.calls.append(("jsonrpc", method, params, timeout))
|
||||
if method == "dumpWindowHierarchy":
|
||||
return self.hierarchy
|
||||
if method == "takeScreenshot":
|
||||
return self.screenshot
|
||||
if method == "pressKey":
|
||||
self.hierarchy = "<hierarchy />"
|
||||
return ""
|
||||
if method == "click":
|
||||
if not isinstance(params, list) or len(params) != 2:
|
||||
raise AssertionError(params)
|
||||
self._apply_tap(int(params[0]), int(params[1]))
|
||||
return ""
|
||||
raise AssertionError(method)
|
||||
|
||||
def _apply_tap(self, x: int, y: int) -> None:
|
||||
if "快要抢光" in self.hierarchy and "[396,498][895,570]" not in self.hierarchy:
|
||||
self.hierarchy = self.panel_hierarchy
|
||||
return
|
||||
root = ElementTree.fromstring(self.hierarchy)
|
||||
target = next(node for node in root.iter("node") if _center(node.get("bounds", "")) == (x, y))
|
||||
color = target.get("bounds", "").endswith("][438,1172]")
|
||||
for node in root.iter("node"):
|
||||
if node.get("selected") is not None and ((color and ",1000]" in node.get("bounds", "")) or (not color and ",1730]" in node.get("bounds", ""))):
|
||||
node.set("selected", "false")
|
||||
if color and self.fail_color_readback:
|
||||
next(node for node in root.iter("node") if node.get("content-desc") == "粉红").set("selected", "true")
|
||||
else:
|
||||
target.set("selected", "true")
|
||||
self.hierarchy = ElementTree.tostring(root, encoding="unicode")
|
||||
|
||||
def window_size(self) -> tuple[int, int]:
|
||||
self.calls.append(("window_size",))
|
||||
return 1080, 2376
|
||||
|
||||
def select_alternates(self) -> None:
|
||||
root = ElementTree.fromstring(self.panel_hierarchy)
|
||||
for node in root.iter("node"):
|
||||
if node.get("selected") is not None:
|
||||
node.set("selected", "false")
|
||||
next(node for node in root.iter("node") if node.get("content-desc") == "粉红").set("selected", "true")
|
||||
next(node for node in root.iter("node") if node.get("text") == "L（建议115-130）").set("selected", "true")
|
||||
self.panel_hierarchy = ElementTree.tostring(root, encoding="unicode")
|
||||
if self.hierarchy != _PRODUCT_PAGE:
|
||||
self.hierarchy = self.panel_hierarchy
|
||||
|
||||
|
||||
def _center(bounds: str) -> tuple[int, int]:
|
||||
left_top, right_bottom = bounds.split("][")
|
||||
left, top = (int(value) for value in left_top.removeprefix("[").split(","))
|
||||
right, bottom = (int(value) for value in right_bottom.removesuffix("]").split(","))
|
||||
return left + (right - left) // 2, top + (bottom - top) // 2
|
||||
|
||||
|
||||
def _actions(device: _RawDevice, method: str) -> list[tuple[object, ...]]:
|
||||
return [call for call in device.calls if call[0] == "jsonrpc" and call[1] == method]
|
||||
|
||||
|
||||
def _tap_centers(device: _RawDevice) -> list[tuple[int, int]]:
|
||||
return [tuple(call[2]) for call in _actions(device, "click")] # type: ignore[misc]
|
||||
|
||||
|
||||
class _FakeAdb:
|
||||
def __init__(self) -> None:
|
||||
self.calls: list[tuple[object, ...]] = []
|
||||
self.inspection = DeviceInspection(AdbDevice(serial="device-1", state="device"), "PKG110", "16")
|
||||
self.on_intent: callable | None = None
|
||||
|
||||
def inspect(self, serial: str) -> DeviceInspection:
|
||||
self.calls.append(("inspect", serial))
|
||||
return self.inspection
|
||||
|
||||
def start_pdd_view_intent(self, serial: str, goods_id: str) -> object:
|
||||
self.calls.append(("intent", serial, goods_id))
|
||||
if self.on_intent is not None:
|
||||
self.on_intent()
|
||||
return object()
|
||||
|
||||
|
||||
class SkuSelectionFlowTests(unittest.TestCase):
|
||||
def test_target_mapping_is_exact_and_success_path_restores_target(self) -> None:
|
||||
device = _RawDevice()
|
||||
adapter = UiautomatorSkuPanelAdapter(device, 10)
|
||||
flow = SkuSelectionFlow(adapter)
|
||||
|
||||
flow.open_sku_panel(_TARGET_URL)
|
||||
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
|
||||
self.assertEqual(flow.read_sku_unit_price(), "12.88")
|
||||
flow.exit_sku_panel_safely()
|
||||
|
||||
self.assertEqual(_tap_centers(device), [(170, 850)])
|
||||
self.assertEqual(_actions(device, "pressKey"), [("jsonrpc", "pressKey", ["back"], 10)])
|
||||
|
||||
def test_unknown_task_or_ui_variants_are_rejected_without_action(self) -> None:
|
||||
for color, size in (("黑色 CHA (纯棉）", _TASK_SIZE), (_TASK_COLOR, "M（建议100-115）"), ("黑色CHA（纯棉）", _TASK_SIZE)):
|
||||
with self.subTest(color=color, size=size), self.assertRaises(SkuSelectionError):
|
||||
resolve_task_selection(color, size)
|
||||
|
||||
device = _RawDevice(_FIXTURE.read_text(encoding="utf-8"))
|
||||
with self.assertRaises(SkuSelectionError):
|
||||
SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).select_sku_options(
|
||||
resolve_task_selection(_TASK_COLOR, _TASK_SIZE).__class__("粉红", "L（建议115-130）")
|
||||
)
|
||||
self.assertEqual(_actions(device, "click"), [])
|
||||
|
||||
def test_option_selected_and_container_drift_fail_closed_before_click(self) -> None:
|
||||
base = _FIXTURE.read_text(encoding="utf-8")
|
||||
cases = (
|
||||
base.replace('selected="true" clickable="true" enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]"', 'clickable="true" enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]"'),
|
||||
base.replace('selected="true" clickable="true" enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]"', 'selected="maybe" clickable="true" enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]"'),
|
||||
base.replace('bounds="[126,1000][438,1172]"', 'bounds="[1,1][20,20]"'),
|
||||
base.replace('enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]"', 'enabled="false" visible-to-user="true" bounds="[126,1000][438,1172]"'),
|
||||
)
|
||||
for hierarchy in cases:
|
||||
with self.subTest(), self.assertRaises(SkuSelectionError):
|
||||
SkuSelectionFlow(UiautomatorSkuPanelAdapter(_RawDevice(hierarchy), 10)).select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
|
||||
|
||||
def test_invalid_bounds_stop_before_action(self) -> None:
|
||||
for bounds in ("", "[1,2][1,3]", "[1,2][3,2]", "[0,0][1081,1]", "[0,0][1,2377]", "[a,0][1,1]"):
|
||||
with self.subTest(bounds=bounds), self.assertRaises(SkuSelectionError):
|
||||
_action_bounds(bounds)
|
||||
|
||||
device = _RawDevice(_PRODUCT_PAGE.replace("[20,800][320,900]", "[0,0][1081,1]"))
|
||||
with self.assertRaises(SkuSelectionError):
|
||||
SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).open_sku_panel(_TARGET_URL)
|
||||
self.assertEqual(_actions(device, "click"), [])
|
||||
|
||||
def test_color_readback_failure_never_attempts_second_option(self) -> None:
|
||||
device = _RawDevice()
|
||||
device.fail_color_readback = True
|
||||
flow = SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10))
|
||||
flow.open_sku_panel(_TARGET_URL)
|
||||
device.select_alternates()
|
||||
with self.assertRaises(SkuSelectionError):
|
||||
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
|
||||
self.assertEqual(_tap_centers(device), [(170, 850), (282, 1086)])
|
||||
|
||||
def test_non_target_selection_restores_each_dimension_once(self) -> None:
|
||||
device = _RawDevice()
|
||||
flow = SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10))
|
||||
flow.open_sku_panel(_TARGET_URL)
|
||||
device.select_alternates()
|
||||
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
|
||||
self.assertEqual(
|
||||
_tap_centers(device),
|
||||
[(170, 850), (282, 1086), (635, 1772)],
|
||||
)
|
||||
|
||||
def test_price_rejects_coupon_prefix_extra_amount_and_bottom_action(self) -> None:
|
||||
for replacement in ("券后 ¥12.88", "会员补贴 ¥12.88", "到手 ¥12.88", "实付 ¥12.88", "区间 ¥12.88", "原价 ¥12.88", "划线价 ¥12.88", "最低 ¥12.88", "低至 ¥12.88", "起价 ¥12.88", "快卖完 1 ¥12.88", "快卖完 ¥12.88 ¥11.88"):
|
||||
with self.subTest(replacement=replacement):
|
||||
device = _RawDevice(_FIXTURE.read_text(encoding="utf-8").replace("快卖完 ¥12.88", replacement))
|
||||
with self.assertRaises(SkuSelectionError):
|
||||
SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).read_sku_unit_price()
|
||||
device = _RawDevice(_FIXTURE.read_text(encoding="utf-8").replace("快卖完 ¥12.88", "提交订单 ¥12.88"))
|
||||
with self.assertRaises(SkuSelectionError):
|
||||
SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).read_sku_unit_price()
|
||||
clickable_parent = _FIXTURE.read_text(encoding="utf-8").replace(
|
||||
'<node package="" class="android.view.ViewGroup" bounds="[396,498][895,570]">',
|
||||
'<node package="" class="android.view.ViewGroup" clickable="true" bounds="[396,498][895,570]">',
|
||||
)
|
||||
with self.assertRaises(SkuSelectionError):
|
||||
SkuSelectionFlow(UiautomatorSkuPanelAdapter(_RawDevice(clickable_parent), 10)).read_sku_unit_price()
|
||||
|
||||
def test_public_api_and_protocol_have_no_broad_or_order_operations(self) -> None:
|
||||
forbidden = {"quantity", "confirm", "authorization", "fence", "submit", "payment", "click"}
|
||||
self.assertTrue(forbidden.isdisjoint(SkuSelectionFlow.__dict__))
|
||||
self.assertTrue(forbidden.isdisjoint(SkuPanelDevice.__dict__))
|
||||
self.assertTrue(forbidden.isdisjoint(pdd.__all__))
|
||||
|
||||
def test_static_ast_boundary_limits_flow_runner_adapter_and_cli(self) -> None:
|
||||
root = Path(__file__).resolve().parents[2]
|
||||
files = (
|
||||
root / "src" / "cmbuyer_client" / "pdd" / "sku_selection.py",
|
||||
root / "src" / "cmbuyer_client" / "pdd" / "sku_selection_runner.py",
|
||||
root / "scripts" / "run_t103_sku_selection.py",
|
||||
)
|
||||
forbidden = ("quantity", "confirm", "authorization", "fence", "submit_order", "payment")
|
||||
for path in files:
|
||||
source = path.read_text(encoding="utf-8")
|
||||
with self.subTest(path=path.name):
|
||||
self.assertTrue(all(token not in source.lower() for token in forbidden))
|
||||
tree = ast.parse(source)
|
||||
self.assertFalse(any(isinstance(node, ast.ImportFrom) and node.module in {"selenium", "requests"} for node in ast.walk(tree)))
|
||||
runner_tree = ast.parse(files[1].read_text(encoding="utf-8"))
|
||||
click_calls = [node for node in ast.walk(runner_tree) if isinstance(node, ast.Constant) and node.value == "click"]
|
||||
self.assertEqual(len(click_calls), 1)
|
||||
|
||||
def test_entry_wait_rejects_unchanged_or_duplicate_page_without_click(self) -> None:
|
||||
now = [0.0]
|
||||
device = _RawDevice()
|
||||
flow = SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10), 0.01, 0.01, lambda: now[0], lambda seconds: now.__setitem__(0, now[0] + seconds))
|
||||
with self.assertRaises(SkuSelectionError):
|
||||
flow.open_sku_panel(_TARGET_URL, _PRODUCT_PAGE)
|
||||
self.assertEqual(_actions(device, "click"), [])
|
||||
|
||||
def test_action_postcondition_wait_never_repeats_entry_click(self) -> None:
|
||||
class NoPanelAfterEntry(_RawDevice):
|
||||
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
|
||||
if method == "click":
|
||||
self.calls.append(("jsonrpc", method, params, timeout))
|
||||
return ""
|
||||
return super().jsonrpc_call(method, params, timeout)
|
||||
|
||||
device = NoPanelAfterEntry()
|
||||
with self.assertRaises(SkuSelectionError):
|
||||
SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).open_sku_panel(_TARGET_URL)
|
||||
self.assertEqual(_tap_centers(device), [(170, 850)])
|
||||
|
||||
duplicate = _PRODUCT_PAGE.replace("</hierarchy>", _PRODUCT_PAGE.removeprefix("<hierarchy>"))
|
||||
device = _RawDevice(duplicate)
|
||||
with self.assertRaises(SkuSelectionError):
|
||||
SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).open_sku_panel(_TARGET_URL)
|
||||
self.assertEqual(_actions(device, "click"), [])
|
||||
|
||||
def test_fixture_contains_no_address_phone_or_payment_credentials(self) -> None:
|
||||
content = _FIXTURE.read_text(encoding="utf-8")
|
||||
self.assertNotRegex(content, r"1[3-9]\d{9}")
|
||||
for forbidden in ("地址", "收货", "支付", "银行卡", "身份证"):
|
||||
self.assertNotIn(forbidden, content)
|
||||
root = ElementTree.fromstring(content)
|
||||
leaf = next(node for node in root.iter("node") if node.get("text") == "提交订单 ¥12.88")
|
||||
self.assertEqual(leaf.get("clickable"), "false")
|
||||
self.assertEqual(leaf.get("bounds"), "[369,2225][710,2284]")
|
||||
|
||||
|
||||
class SkuSelectionRunnerTests(unittest.TestCase):
|
||||
def _runner(self, adb: _FakeAdb, device: _RawDevice) -> SkuSelectionRunner:
|
||||
device.hierarchy = "<hierarchy />"
|
||||
adb.on_intent = lambda: setattr(device, "hierarchy", _PRODUCT_PAGE.replace("<hierarchy>", '<hierarchy post-intent="1">'))
|
||||
return SkuSelectionRunner(adb, lambda serial: device, 10)
|
||||
|
||||
def test_runner_atomically_publishes_screenshot_and_redacted_manifest(self) -> None:
|
||||
adb = _FakeAdb()
|
||||
device = _RawDevice()
|
||||
with TemporaryDirectory() as temporary:
|
||||
target = Path(temporary) / "result"
|
||||
result = self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
|
||||
|
||||
self.assertEqual(result.unit_price, "12.88")
|
||||
manifest = result.manifest_path.read_text(encoding="utf-8")
|
||||
self.assertTrue(result.screenshot_path.is_file())
|
||||
self.assertNotIn("device-1", manifest)
|
||||
self.assertNotIn("hierarchy", manifest)
|
||||
self.assertNotIn("已选", manifest)
|
||||
self.assertIn('"unit_price": "12.88"', manifest)
|
||||
self.assertIn('"selection_status": "restored"', manifest)
|
||||
self.assertIn('"panel_status": "verified"', manifest)
|
||||
self.assertIn('"safe_exit": "completed"', manifest)
|
||||
self.assertFalse((target / "hierarchy.xml").exists())
|
||||
self.assertEqual(_actions(device, "pressKey"), [("jsonrpc", "pressKey", ["back"], 10)])
|
||||
|
||||
def test_target_created_during_publish_is_preserved_without_staging_residue(self) -> None:
|
||||
with TemporaryDirectory() as temporary:
|
||||
target = Path(temporary) / "result"
|
||||
original_rename = runner_module.os.rename
|
||||
|
||||
def create_target_then_rename(source: str | Path, destination: str | Path) -> None:
|
||||
Path(destination).mkdir()
|
||||
(Path(destination) / "sentinel").write_text("keep", encoding="utf-8")
|
||||
original_rename(source, destination)
|
||||
|
||||
with patch.object(runner_module.os, "rename", side_effect=create_target_then_rename), self.assertRaises(SkuSelectionRunError):
|
||||
self._runner(_FakeAdb(), _RawDevice()).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
|
||||
self.assertEqual((target / "sentinel").read_text(encoding="utf-8"), "keep")
|
||||
self.assertEqual(list(Path(temporary).glob(".result.staging-*")), [])
|
||||
|
||||
def test_bad_screenshot_or_existing_target_never_publishes_manifest(self) -> None:
|
||||
with TemporaryDirectory() as temporary:
|
||||
target = Path(temporary) / "result"
|
||||
with self.assertRaises(SkuSelectionScreenshotError):
|
||||
self._runner(_FakeAdb(), _RawDevice(screenshot="not-image")).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
|
||||
self.assertFalse(target.exists())
|
||||
self.assertEqual(list(Path(temporary).glob(".result.staging-*")), [])
|
||||
|
||||
target = Path(temporary) / "write-failure"
|
||||
with patch.object(runner_module, "_save_base64_screenshot", side_effect=OSError("private path")):
|
||||
with self.assertRaises(SkuSelectionScreenshotError):
|
||||
self._runner(_FakeAdb(), _RawDevice()).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
|
||||
self.assertFalse(target.exists())
|
||||
self.assertEqual(list(Path(temporary).glob(".write-failure.staging-*")), [])
|
||||
|
||||
adb = _FakeAdb()
|
||||
device = _RawDevice()
|
||||
target.mkdir()
|
||||
sentinel = target / "keep"
|
||||
sentinel.write_text("keep", encoding="utf-8")
|
||||
with self.assertRaises(SkuSelectionRunError):
|
||||
self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
|
||||
self.assertEqual(adb.calls, [])
|
||||
self.assertEqual(device.calls, [])
|
||||
self.assertEqual(sentinel.read_text(encoding="utf-8"), "keep")
|
||||
|
||||
def test_device_screen_and_output_preflight_fail_before_any_click(self) -> None:
|
||||
with TemporaryDirectory() as temporary:
|
||||
adb = _FakeAdb()
|
||||
adb.inspection = DeviceInspection(AdbDevice(serial="device-1", state="device"), "wrong", "16")
|
||||
device = _RawDevice()
|
||||
with self.assertRaises(SkuSelectionRunError):
|
||||
self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result")
|
||||
self.assertEqual(device.calls, [])
|
||||
|
||||
class WrongScreenDevice(_RawDevice):
|
||||
def window_size(self) -> tuple[int, int]:
|
||||
return 1080, 1920
|
||||
|
||||
device = WrongScreenDevice()
|
||||
with self.assertRaises(SkuSelectionRunError):
|
||||
self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "screen")
|
||||
self.assertEqual(_actions(device, "click"), [])
|
||||
|
||||
parent_file = Path(temporary) / "not-a-directory"
|
||||
parent_file.write_text("x", encoding="utf-8")
|
||||
adb = _FakeAdb()
|
||||
device = _RawDevice()
|
||||
with self.assertRaises(SkuSelectionRunError):
|
||||
self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, parent_file / "result")
|
||||
self.assertEqual(adb.calls, [])
|
||||
self.assertEqual(device.calls, [])
|
||||
|
||||
def test_small_but_valid_png_is_not_accepted(self) -> None:
|
||||
image = Image.new("RGB", (1, 1), "white")
|
||||
raw = BytesIO(); image.save(raw, format="PNG")
|
||||
with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionScreenshotError):
|
||||
self._runner(_FakeAdb(), _RawDevice(screenshot=base64.b64encode(raw.getvalue()).decode("ascii"))).run(
|
||||
"device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result"
|
||||
)
|
||||
|
||||
def test_failure_after_entry_attempts_one_safe_exit_and_hides_device_detail(self) -> None:
|
||||
adb = _FakeAdb()
|
||||
device = _RawDevice()
|
||||
device.fail_color_readback = True
|
||||
device.select_alternates()
|
||||
with TemporaryDirectory() as temporary:
|
||||
with self.assertRaises(SkuSelectionError):
|
||||
self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result")
|
||||
self.assertEqual(_actions(device, "pressKey"), [])
|
||||
|
||||
class FailingRawDevice(_RawDevice):
|
||||
def app_info(self, package_name: str) -> dict[str, str]:
|
||||
raise RuntimeError("device-1 <xml>private</xml>")
|
||||
|
||||
with self.assertRaises(SkuSelectionDeviceAdapterError) as raised:
|
||||
UiautomatorSkuPanelAdapter(FailingRawDevice(), 10).app_info("com.xunmeng.pinduoduo")
|
||||
self.assertNotIn("device-1", str(raised.exception))
|
||||
self.assertNotIn("private", str(raised.exception))
|
||||
|
||||
def test_unverified_failure_never_sends_blind_back(self) -> None:
|
||||
class InvalidAfterOptionDevice(_RawDevice):
|
||||
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
|
||||
value = super().jsonrpc_call(method, params, timeout)
|
||||
if method == "click" and "[396,498][895,570]" in self.hierarchy:
|
||||
self.hierarchy = "<hierarchy />"
|
||||
return value
|
||||
|
||||
device = InvalidAfterOptionDevice()
|
||||
device.select_alternates()
|
||||
with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionError):
|
||||
self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result")
|
||||
self.assertEqual(_actions(device, "pressKey"), [])
|
||||
|
||||
def test_adapter_timeout_is_mapped_without_third_party_detail(self) -> None:
|
||||
class TimeoutRawDevice(_RawDevice):
|
||||
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
|
||||
raise TimeoutError("device-1 <hierarchy>private</hierarchy>")
|
||||
|
||||
with self.assertRaises(SkuSelectionRunError) as raised:
|
||||
UiautomatorSkuPanelAdapter(TimeoutRawDevice(), 10).dump_window_hierarchy()
|
||||
self.assertNotIn("device-1", str(raised.exception))
|
||||
self.assertNotIn("private", str(raised.exception))
|
||||
|
||||
def test_entry_attempt_is_recorded_before_unconfirmed_click_and_not_retried(self) -> None:
|
||||
class TimeoutTapDevice(_RawDevice):
|
||||
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
|
||||
self.calls.append(("jsonrpc", method, params, timeout))
|
||||
if method == "click":
|
||||
raise TimeoutError("device detail")
|
||||
return super().jsonrpc_call(method, params, timeout)
|
||||
|
||||
adapter = UiautomatorSkuPanelAdapter(TimeoutTapDevice(), 10)
|
||||
with self.assertRaises(SkuSelectionRunError):
|
||||
adapter.tap_sku_entry("[20,800][320,900]")
|
||||
self.assertTrue(adapter.entry_was_tapped)
|
||||
self.assertEqual(_actions(adapter._device, "click"), [("jsonrpc", "click", [170, 850], 10)])
|
||||
|
||||
def test_entry_stability_interruptions_never_click(self) -> None:
|
||||
now = [0.0]
|
||||
class SequenceDevice(_RawDevice):
|
||||
def __init__(self) -> None:
|
||||
super().__init__(); self.frames = [_PRODUCT_PAGE, "<hierarchy />", _PRODUCT_PAGE]
|
||||
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
|
||||
if method == "dumpWindowHierarchy" and self.frames:
|
||||
self.hierarchy = self.frames.pop(0)
|
||||
return super().jsonrpc_call(method, params, timeout)
|
||||
device = SequenceDevice()
|
||||
flow = SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10), .02, .01, lambda: now[0], lambda x: now.__setitem__(0, now[0] + x))
|
||||
with self.assertRaises(SkuSelectionError): flow.open_sku_panel(_TARGET_URL, "<hierarchy />")
|
||||
self.assertEqual(_actions(device, "click"), [])
|
||||
|
||||
def test_screenshot_drift_and_foreground_drift_publish_nothing_and_never_back(self) -> None:
|
||||
for drift in ("color", "size", "price"):
|
||||
class DriftDevice(_RawDevice):
|
||||
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
|
||||
value = super().jsonrpc_call(method, params, timeout)
|
||||
if method == "takeScreenshot":
|
||||
if drift == "price":
|
||||
self.hierarchy = self.hierarchy.replace("快卖完 ¥12.88", "快卖完 ¥13.88")
|
||||
else:
|
||||
root = ElementTree.fromstring(self.hierarchy)
|
||||
if drift == "color":
|
||||
for node in root.iter("node"):
|
||||
if node.get("selected") is not None and ",1000]" in node.get("bounds", ""):
|
||||
node.set("selected", "false")
|
||||
next(node for node in root.iter("node") if node.get("content-desc") == "粉红").set("selected", "true")
|
||||
else:
|
||||
for node in root.iter("node"):
|
||||
if node.get("selected") is not None and ",1730]" in node.get("bounds", ""):
|
||||
node.set("selected", "false")
|
||||
next(node for node in root.iter("node") if node.get("text") == "L（建议115-130）").set("selected", "true")
|
||||
self.hierarchy = ElementTree.tostring(root, encoding="unicode")
|
||||
return value
|
||||
with self.subTest(drift=drift), TemporaryDirectory() as temporary:
|
||||
target = Path(temporary) / "out"
|
||||
with self.assertRaises((SkuSelectionError, SkuSelectionRunError)):
|
||||
self._runner(_FakeAdb(), DriftDevice()).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
|
||||
self.assertFalse(target.exists())
|
||||
self.assertFalse((target / "manifest.json").exists())
|
||||
self.assertEqual(list(Path(temporary).glob(".out.staging-*")), [])
|
||||
|
||||
device = _RawDevice(); device.select_alternates()
|
||||
device.package = "other"
|
||||
with self.assertRaises(SkuSelectionError): SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).exit_sku_panel_safely()
|
||||
self.assertEqual(_actions(device, "pressKey"), [])
|
||||
|
||||
def test_screenshot_then_foreground_drift_publishes_nothing_and_never_back(self) -> None:
|
||||
class ForegroundDriftDevice(_RawDevice):
|
||||
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
|
||||
value = super().jsonrpc_call(method, params, timeout)
|
||||
if method == "takeScreenshot": self.package = "other"
|
||||
return value
|
||||
|
||||
device = ForegroundDriftDevice()
|
||||
with TemporaryDirectory() as temporary:
|
||||
target = Path(temporary) / "out"
|
||||
with self.assertRaises(SkuSelectionError):
|
||||
self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
|
||||
self.assertFalse(target.exists())
|
||||
self.assertFalse((target / "manifest.json").exists())
|
||||
self.assertEqual(list(Path(temporary).glob(".out.staging-*")), [])
|
||||
self.assertEqual(_actions(device, "pressKey"), [])
|
||||
|
||||
def test_option_timeout_reconciliation_controls_back_once(self) -> None:
|
||||
class OptionTimeoutDevice(_RawDevice):
|
||||
def __init__(self, delivered: bool) -> None:
|
||||
super().__init__(); self.delivered = delivered; self.clicks = 0
|
||||
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
|
||||
if method == "click":
|
||||
self.clicks += 1
|
||||
if self.clicks == 2:
|
||||
if self.delivered: super().jsonrpc_call(method, params, timeout)
|
||||
else: self.calls.append(("jsonrpc", method, params, timeout))
|
||||
raise TimeoutError("uncertain option")
|
||||
return super().jsonrpc_call(method, params, timeout)
|
||||
|
||||
for delivered, expected_back in ((False, 0), (True, 1)):
|
||||
with self.subTest(delivered=delivered), TemporaryDirectory() as temporary:
|
||||
device = OptionTimeoutDevice(delivered); device.select_alternates()
|
||||
adb = _FakeAdb(); device.hierarchy = "<hierarchy />"
|
||||
adb.on_intent = lambda: setattr(device, "hierarchy", _PRODUCT_PAGE.replace("<hierarchy>", '<hierarchy post-intent="1">'))
|
||||
runner = SkuSelectionRunner(adb, lambda serial: device, .03)
|
||||
with self.assertRaises(SkuSelectionRunError):
|
||||
runner.run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "out")
|
||||
self.assertEqual(len(_actions(device, "click")), 2)
|
||||
self.assertEqual(len(_actions(device, "pressKey")), expected_back)
|
||||
|
||||
def test_back_timeout_is_never_retried(self) -> None:
|
||||
class BackTimeoutDevice(_RawDevice):
|
||||
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
|
||||
if method == "pressKey":
|
||||
super().jsonrpc_call(method, params, timeout)
|
||||
raise TimeoutError("back uncertain")
|
||||
return super().jsonrpc_call(method, params, timeout)
|
||||
|
||||
device = BackTimeoutDevice()
|
||||
with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionRunError):
|
||||
self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "out")
|
||||
self.assertEqual(len(_actions(device, "pressKey")), 1)
|
||||
|
||||
def test_entry_click_timeout_reconciles_only_through_verified_flow_exit(self) -> None:
|
||||
class DeliveredThenTimeout(_RawDevice):
|
||||
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
|
||||
if method == "click" and self.hierarchy != _FIXTURE.read_text(encoding="utf-8"):
|
||||
super().jsonrpc_call(method, params, timeout)
|
||||
raise TimeoutError("delivery uncertain")
|
||||
return super().jsonrpc_call(method, params, timeout)
|
||||
|
||||
device = DeliveredThenTimeout()
|
||||
with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionRunError):
|
||||
self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result")
|
||||
self.assertEqual(len(_actions(device, "click")), 1)
|
||||
self.assertEqual(len(_actions(device, "pressKey")), 1)
|
||||
|
||||
|
||||
class SkuSelectionCliTests(unittest.TestCase):
|
||||
def test_cli_accepts_only_target_url_and_task_values(self) -> None:
|
||||
script = _load_runner_script()
|
||||
valid = {
|
||||
"serial": "device-1",
|
||||
"url": _TARGET_URL,
|
||||
"color": _TASK_COLOR,
|
||||
"size": _TASK_SIZE,
|
||||
"output_dir": Path("evidence"),
|
||||
"timeout": 10.0,
|
||||
"adb": "adb",
|
||||
}
|
||||
script.validate_arguments(type("Arguments", (), valid)())
|
||||
for field, value in (("serial", ""), ("url", "https://mobile.yangkeduo.com/goods.html?goods_id=1"), ("color", "黑色 CHA (纯棉）"), ("size", "M（建议100-115）"), ("timeout", 0), ("timeout", float("inf"))):
|
||||
with self.subTest(field=field, value=value), self.assertRaises((ValueError, SkuSelectionError)):
|
||||
script.validate_arguments(type("Arguments", (), valid | {field: value})())
|
||||
|
||||
def test_cli_main_catches_flow_error_without_traceback_or_page_body(self) -> None:
|
||||
script = _load_runner_script()
|
||||
|
||||
class FlowFailingRunner:
|
||||
def __init__(self, *args: object, **kwargs: object) -> None: pass
|
||||
def run(self, *args: object, **kwargs: object) -> object:
|
||||
raise SkuSelectionError("<hierarchy>page-body</hierarchy>")
|
||||
|
||||
stderr = BytesIO()
|
||||
# TextIOWrapper keeps the assertion independent from host console encoding.
|
||||
import io
|
||||
text_stderr = io.TextIOWrapper(stderr, encoding="utf-8")
|
||||
with patch.object(script, "SkuSelectionRunner", FlowFailingRunner), redirect_stderr(text_stderr):
|
||||
status = script.main([
|
||||
"--serial", "device-1", "--url", _TARGET_URL, "--color", _TASK_COLOR,
|
||||
"--size", _TASK_SIZE, "--output-dir", "evidence",
|
||||
])
|
||||
text_stderr.flush()
|
||||
output = stderr.getvalue().decode("utf-8")
|
||||
self.assertEqual(status, 1)
|
||||
self.assertNotIn("Traceback", output)
|
||||
self.assertNotIn("page-body", output)
|
||||
|
||||
|
||||
def _load_runner_script() -> object:
|
||||
path = Path(__file__).resolve().parents[2] / "scripts" / "run_t103_sku_selection.py"
|
||||
specification = importlib.util.spec_from_file_location("run_t103_sku_selection_test", path)
|
||||
if specification is None or specification.loader is None:
|
||||
raise RuntimeError("无法加载 T-103 运行脚本。")
|
||||
module = importlib.util.module_from_spec(specification)
|
||||
specification.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -17,6 +17,7 @@ write_paths:
|
||||
- client/tests/pdd/**
|
||||
- client/tests/device/**
|
||||
- client/scripts/capture_sku_panel_spike.py
|
||||
- client/scripts/run_t103_sku_selection.py
|
||||
- client/scripts/sanitize_sku_panel_evidence.py
|
||||
- docs/02-requirements.md
|
||||
- docs/03-tech-stack.md
|
||||
|
||||
Reference in New Issue
Block a user