"""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 = "快要抢光 ¥ 12.88" _ENTRY_PROMOTION_LABEL = "快要抢光" _ENTRY_TEXT_BOUNDS = "[688,2184][1042,2253]" _ENTRY_ACTION_DESC = "快要抢光¥12.88" _ENTRY_ACTION_BOUNDS = "[446,2166][1080,2328]" _ENTRY_SIBLING = "免拼购买" _ENTRY_SIBLING_BOUNDS = "[688,2256][856,2305]" _FORBIDDEN_ENTRY_ACTION_DESC = ( "购买", "下单", "付款", "订单", "单独购买", "直接拼成", "提交订单", "支付", "先用后付", "0元下单", "0 元下单", ) _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): """已取证判据不成立时的脱敏停止。""" _SKU_ENTRY_FAILURE_STAGES = frozenset( ( "sku_entry_pre_intent", "sku_entry_discovery", "sku_entry_click", "sku_entry_panel_verify", ) ) _SKU_ENTRY_FAILURE_MARKER = object() def _annotate_sku_entry_failure(error: BaseException, stage: str) -> None: """把 Flow 实际经过的固定入口子阶段附到原异常,不改变异常类型。""" if type(stage) is not str or stage not in _SKU_ENTRY_FAILURE_STAGES: return try: # marker 最后写入:若第三方异常拒绝任一属性写入,就不能形成可信诊断。 setattr(error, "_cmbuyer_failure_stage", stage) setattr(error, "_cmbuyer_sku_entry_failure_marker", _SKU_ENTRY_FAILURE_MARKER) except BaseException: pass def _safe_sku_entry_failure_stage(error: BaseException) -> str | None: """只读取由本模块写入的入口子阶段;任意异常自报的值不可信。""" try: marker = getattr(error, "_cmbuyer_sku_entry_failure_marker", None) stage = getattr(error, "_cmbuyer_failure_stage", None) if marker is not _SKU_ENTRY_FAILURE_MARKER or type(stage) is not str: return None return stage if stage in _SKU_ENTRY_FAILURE_STAGES else None except BaseException: return None 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: try: 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) # intent 前只判断旧页是否已经存在完整入口链;浮层或额外动作节点不能把旧商品 # 伪装成“不安全所以不存在”,否则 intent 后可能误把旧页当成新目标页。 if _physical_entries(previous_nodes): raise SkuSelectionError("intent 前页面已出现规格入口,已拒绝旧商品误点。") except BaseException as error: _annotate_sku_entry_failure(error, "sku_entry_pre_intent") raise try: entry, before = self._wait_for_entry(pre_intent_hierarchy) _action_bounds(entry.bounds) except BaseException as error: _annotate_sku_entry_failure(error, "sku_entry_discovery") raise try: self._pending = (before, _panel) self._device.tap_sku_entry(entry.bounds) except BaseException as error: _annotate_sku_entry_failure(error, "sku_entry_click") raise try: self._wait_after_action(before, _panel) except BaseException as error: _annotate_sku_entry_failure(error, "sku_entry_panel_verify") raise 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() nodes = _parse_nodes(raw) entries = _eligible_entries(nodes) if len(entries) > 1: raise SkuSelectionError("商品页规格入口不唯一,已停止操作。") if len(entries) == 1 and raw != previous: # 商品详情正文包含倒计时等动态节点,全文 XML 稳定不是已取证入口的安全属性。 # 连续两帧只比较证据绑定的底部入口、直接父容器和不可点击兄弟节点;商品正文及 # 父容器内其他非危险动态节点不是入口身份,不能迫使实现退回坐标兜底。 projection = _entry_projection(entries[0], nodes) if projection is None: raise SkuSelectionError("商品页规格入口结构失效,已停止操作。") if stable == projection: return entries[0], raw stable = projection 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]: # 入口文本本身不可点击:必须证明它仍是已取证底部父容器的直接子节点,但动作坐标继续 # 使用第一行文本的窄 bounds,避免把父容器中心或第二行“免拼购买”变成坐标兜底。 if any(node.bounds == _PRICE_PARENT for node in nodes): return [] entries = _physical_entries(nodes) if len(entries) != 1: return entries entry = entries[0] chain = _physical_entry_chain(entry, nodes) if chain is None or _entry_projection(entry, nodes) is None: return [] ancestor = chain[1] left, top, right, bottom = _action_bounds(entry.bounds) center = (left + (right - left) // 2, top + (bottom - top) // 2) occupants = _live_clickables_covering(nodes, center) # RPC 点的是文本中心而不是祖先对象。只有完整入口链自己的动作祖先占用该坐标时才可点击; # SystemUI 浮层、额外按钮或任意部分覆盖矩形都可能截获触摸,必须零点击失败关闭。 return entries if len(occupants) == 1 and occupants[0].element is ancestor.element else [] def _physical_entries(nodes: list[_Node]) -> list[_Node]: entry_labels = [ node for node in nodes if node.text == _ENTRY and node.element.get("package") == PDD_PACKAGE and node.element.get("class") == "android.widget.TextView" ] # 这里故意只证明物理结构。pre-intent 必须识别旧商品,不能让父容器描述或浮层等 # post-intent 安全条件把已经存在的旧入口伪装成“不存在”。 return [node for node in entry_labels if _physical_entry_chain(node, nodes) is not None] def _physical_entry_chain(node: _Node, nodes: list[_Node]) -> tuple[_Node, ...] | None: if node.text != _ENTRY or not _exact_entry_node( node, "android.widget.TextView", _ENTRY_TEXT_BOUNDS, "false" ) or len(node.element) != 0: return None ancestor = node.parent if ancestor is None or not _exact_entry_node( ancestor, "android.view.ViewGroup", _ENTRY_ACTION_BOUNDS, "true" ): return None siblings = [ candidate for candidate in nodes if candidate.parent is ancestor and candidate.text == _ENTRY_SIBLING and _exact_entry_node( candidate, "android.widget.TextView", _ENTRY_SIBLING_BOUNDS, "false" ) and not candidate.desc and len(candidate.element) == 0 ] if len(siblings) != 1: return None return node, ancestor, siblings[0] def _entry_projection(node: _Node, nodes: list[_Node]) -> tuple[object, ...] | None: chain = _physical_entry_chain(node, nodes) if chain is None: return None _, ancestor, sibling = chain if node.desc: return None # 金额只作为这个已取证入口的不可变身份。这里既不解析也不返回它,价格闸门仍只能读取规格面板。 if ancestor.text or ancestor.desc != _ENTRY_ACTION_DESC: return None sibling_mentions = [ candidate for candidate in nodes if _ENTRY_SIBLING in candidate.text or _ENTRY_SIBLING in candidate.desc ] if len(sibling_mentions) != 1 or sibling_mentions[0].element is not sibling.element: return None subtree = [ candidate for candidate in nodes if candidate.element is ancestor.element or _descendant(candidate, ancestor) ] if any( forbidden in value for candidate in subtree for value in (candidate.text, candidate.desc) if candidate.element is not sibling.element for forbidden in _FORBIDDEN_ENTRY_ACTION_DESC ): return None if any( value == _ENTRY_PROMOTION_LABEL for candidate in subtree for value in (candidate.text, candidate.desc) ): return None entry_text_nodes = [candidate for candidate in subtree if candidate.text == _ENTRY] if len(entry_text_nodes) != 1 or entry_text_nodes[0].element is not node.element: return None return tuple( _entry_node_projection(candidate) for candidate in (node, ancestor, sibling) ) def _entry_node_projection(node: _Node) -> tuple[str, ...]: return ( node.element.tag, node.element.get("package", ""), node.element.get("class", ""), node.bounds, node.element.get("clickable", ""), node.element.get("enabled", ""), node.element.get("visible-to-user", ""), node.text, node.desc, ) def _is_live_clickable(node: _Node) -> bool: return ( node.element.get("clickable") == "true" and node.element.get("enabled") == "true" and node.element.get("visible-to-user") == "true" ) def _live_clickables_covering(nodes: list[_Node], point: tuple[int, int]) -> list[_Node]: occupants: list[_Node] = [] x, y = point for node in nodes: if not _is_live_clickable(node): continue # 活跃可点击节点的 bounds 无法验证时,无法证明它不会截获入口坐标,因此整体失败关闭。 left, top, right, bottom = _action_bounds(node.bounds) if left <= x < right and top <= y < bottom: occupants.append(node) return occupants def _exact_entry_node(node: _Node, class_name: str, bounds: str, clickable: str) -> bool: return ( node.element.get("package") == PDD_PACKAGE and node.element.get("class") == class_name and node.bounds == bounds and node.element.get("clickable") == clickable and node.element.get("enabled") == "true" and node.element.get("visible-to-user") == "true" ) 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]