feat(client): implement verified SKU selection flow

client/src/cmbuyer_client/pdd/sku_selection.py

@@ -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]