421 lines
20 KiB
Python
421 lines
20 KiB
Python
"""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 = "快要抢光"
|
||
_ENTRY_TEXT_BOUNDS = "[900,1312][1056,1355]"
|
||
_ENTRY_INNER_BOUNDS = "[712,1312][1056,1355]"
|
||
_ENTRY_ACTION_BOUNDS = "[0,1256][1080,1355]"
|
||
_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]:
|
||
# 入口文本本身不可点击：必须逐层证明它仍位于已取证的唯一可点击祖先中，但动作坐标继续
|
||
# 使用文本子节点的窄 bounds，避免把同一祖先内未知区域变成坐标兜底。“免拼购买”等底部
|
||
# 容器既不属于这条祖先链，也绝不能作为替代入口。
|
||
if any(node.bounds == _PRICE_PARENT for node in nodes): return []
|
||
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"
|
||
]
|
||
if len(entry_labels) != 1:
|
||
return entry_labels
|
||
action_ancestors = [
|
||
node for node in nodes
|
||
if _exact_entry_node(node, "android.view.ViewGroup", _ENTRY_ACTION_BOUNDS, "true")
|
||
]
|
||
if len(action_ancestors) != 1:
|
||
return []
|
||
action_ancestor = action_ancestors[0]
|
||
entries: list[_Node] = []
|
||
for node in entry_labels:
|
||
if not _exact_entry_node(node, "android.widget.TextView", _ENTRY_TEXT_BOUNDS, "false"):
|
||
return []
|
||
inner = node.parent
|
||
switcher = inner.parent if inner is not None else None
|
||
frame = switcher.parent if switcher is not None else None
|
||
ancestor = frame.parent if frame is not None else None
|
||
if (
|
||
inner is not None
|
||
and _exact_entry_node(inner, "android.view.ViewGroup", _ENTRY_INNER_BOUNDS, "false")
|
||
and switcher is not None
|
||
and _exact_entry_node(switcher, "android.widget.ViewSwitcher", _ENTRY_INNER_BOUNDS, "false")
|
||
and frame is not None
|
||
and _exact_entry_node(frame, "android.widget.FrameLayout", _ENTRY_INNER_BOUNDS, "false")
|
||
and ancestor is action_ancestor
|
||
):
|
||
entries.append(node)
|
||
return entries
|
||
|
||
|
||
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]
|