diff --git a/client/src/cmbuyer_client/pdd/sku_selection.py b/client/src/cmbuyer_client/pdd/sku_selection.py
index 5efac1c..e69253b 100644
--- a/client/src/cmbuyer_client/pdd/sku_selection.py
+++ b/client/src/cmbuyer_client/pdd/sku_selection.py
@@ -20,6 +20,10 @@ _ENTRY = "快要抢光"
_ENTRY_TEXT_BOUNDS = "[900,1312][1056,1355]"
_ENTRY_INNER_BOUNDS = "[712,1312][1056,1355]"
_ENTRY_ACTION_BOUNDS = "[0,1256][1080,1355]"
+_FORBIDDEN_ENTRY_ACTION_DESC = (
+ "购买", "下单", "付款", "订单", "免拼购买", "单独购买", "直接拼成", "提交订单", "支付",
+ "先用后付", "0元下单", "0 元下单",
+)
_SIZE = "尺码"
_W, _H = 1080, 2376
_PRICE_PARENT = "[396,498][895,570]"
@@ -89,7 +93,9 @@ class SkuSelectionFlow:
raise SkuSelectionError("商品不是已取证目标，已停止操作。")
if pre_intent_hierarchy is not None:
previous_nodes = _parse_nodes(pre_intent_hierarchy)
- if _eligible_entries(previous_nodes):
+ # intent 前只判断旧页是否已经存在完整入口链；浮层或额外动作节点不能把旧商品
+ # 伪装成“不安全所以不存在”，否则 intent 后可能误把旧页当成新目标页。
+ if _physical_entries(previous_nodes):
raise SkuSelectionError("intent 前页面已出现规格入口，已拒绝旧商品误点。")
entry, before = self._wait_for_entry(pre_intent_hierarchy)
_action_bounds(entry.bounds)
@@ -168,13 +174,19 @@ class SkuSelectionFlow:
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))
+ nodes = _parse_nodes(raw)
+ entries = _eligible_entries(nodes)
if len(entries) > 1:
raise SkuSelectionError("商品页规格入口不唯一，已停止操作。")
if len(entries) == 1 and raw != previous:
- if stable == raw:
+ # 商品详情正文包含倒计时等动态节点，全文 XML 稳定不是已取证入口的安全属性。
+ # 连续两帧只比较完整五层入口判据投影；任一层失配仍会在 eligible 阶段清空稳定态。
+ projection = _entry_projection(entries[0], nodes)
+ if projection is None:
+ raise SkuSelectionError("商品页规格入口结构失效，已停止操作。")
+ if stable == projection:
return entries[0], raw
- stable = raw
+ stable = projection
else:
stable = None
else:
@@ -358,41 +370,128 @@ def _eligible_entries(nodes: list[_Node]) -> list[_Node]:
# 入口文本本身不可点击：必须逐层证明它仍位于已取证的唯一可点击祖先中，但动作坐标继续
# 使用文本子节点的窄 bounds，避免把同一祖先内未知区域变成坐标兜底。“免拼购买”等底部
# 容器既不属于这条祖先链，也绝不能作为替代入口。
- if any(node.bounds == _PRICE_PARENT for node in nodes): return []
+ 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)
+ 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"
]
- 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")
+ # 这里故意只证明物理五层链。pre-intent 必须识别旧商品，不能让禁词、浮层或其他
+ # post-intent 安全条件把已经存在的旧入口伪装成“不存在”。
+ return [node for node in entry_labels if _physical_entry_chain(node) is not None]
+
+
+def _physical_entry_chain(node: _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
+ 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
+ chain = (
+ (node, "android.widget.TextView", _ENTRY_TEXT_BOUNDS, "false"),
+ (inner, "android.view.ViewGroup", _ENTRY_INNER_BOUNDS, "false"),
+ (switcher, "android.widget.ViewSwitcher", _ENTRY_INNER_BOUNDS, "false"),
+ (frame, "android.widget.FrameLayout", _ENTRY_INNER_BOUNDS, "false"),
+ (ancestor, "android.view.ViewGroup", _ENTRY_ACTION_BOUNDS, "true"),
+ )
+ if any(
+ candidate is None or not _exact_entry_node(candidate, class_name, bounds, clickable)
+ for candidate, class_name, bounds, clickable in chain
+ ):
+ return None
+ return tuple(candidate for candidate, _, _, _ in chain if candidate is not None)
+
+
+def _entry_projection(node: _Node, nodes: list[_Node]) -> tuple[object, ...] | None:
+ chain = _physical_entry_chain(node)
+ if chain is None:
+ return None
+ _, inner, switcher, frame, ancestor = chain
+ if node.desc:
+ return None
+ if any(candidate.text or candidate.desc for candidate in (inner, switcher, frame)):
+ return None
+ if (
+ ancestor.text
+ or (ancestor.desc and not ancestor.desc.endswith(_ENTRY))
+ or any(forbidden in ancestor.desc for forbidden in _FORBIDDEN_ENTRY_ACTION_DESC)
+ ):
+ return None
+ subtree = [
+ candidate for candidate in nodes
+ if candidate.element is ancestor.element or _descendant(candidate, ancestor)
]
- 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
+ if any(
+ forbidden in value
+ for candidate in subtree
+ for value in (candidate.text, candidate.desc)
+ for forbidden in _FORBIDDEN_ENTRY_ACTION_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 _subtree_projection(ancestor.element)
+
+
+def _subtree_projection(element: ElementTree.Element) -> tuple[object, ...]:
+ return (
+ element.tag,
+ element.get("package", ""),
+ element.get("class", ""),
+ element.get("bounds", ""),
+ element.get("clickable", ""),
+ element.get("enabled", ""),
+ element.get("visible-to-user", ""),
+ element.get("text", ""),
+ element.get("content-desc", ""),
+ tuple(_subtree_projection(child) for child in element),
+ )
+
+
+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:
diff --git a/client/tests/pdd/test_sku_selection.py b/client/tests/pdd/test_sku_selection.py
index 6b28bbb..b780077 100644
--- a/client/tests/pdd/test_sku_selection.py
+++ b/client/tests/pdd/test_sku_selection.py
@@ -143,6 +143,86 @@ def _duplicate_entry() -> str:
return ElementTree.tostring(root, encoding="unicode")
+def _extra_entry_action_ancestor() -> str:
+ root = ElementTree.fromstring(_PRODUCT_PAGE)
+ action = _entry_chain(root)[4]
+ root.append(ElementTree.Element("node", dict(action.attrib)))
+ return ElementTree.tostring(root, encoding="unicode")
+
+
+def _entry_with_panel_price_marker() -> str:
+ root = ElementTree.fromstring(_PRODUCT_PAGE)
+ root.append(ElementTree.Element("node", {"bounds": "[396,498][895,570]"}))
+ return ElementTree.tostring(root, encoding="unicode")
+
+
+def _dynamic_product_page(value: str, action_desc: str = "") -> str:
+ root = ElementTree.fromstring(_PRODUCT_PAGE)
+ root.set("dynamic-page-value", value)
+ _entry_chain(root)[4].set("content-desc", action_desc)
+ return ElementTree.tostring(root, encoding="unicode")
+
+
+def _with_overlapping_clickable(package: str, class_name: str, bounds: str) -> str:
+ root = ElementTree.fromstring(_PRODUCT_PAGE)
+ root.append(
+ ElementTree.Element(
+ "node",
+ {
+ "package": package,
+ "class": class_name,
+ "bounds": bounds,
+ "clickable": "true",
+ "enabled": "true",
+ "visible-to-user": "true",
+ },
+ )
+ )
+ return ElementTree.tostring(root, encoding="unicode")
+
+
+def _with_action_subtree_child(
+ text: str,
+ *,
+ bounds: str = "[10,1260][100,1300]",
+) -> str:
+ root = ElementTree.fromstring(_PRODUCT_PAGE)
+ action = _entry_chain(root)[4]
+ ElementTree.SubElement(
+ action,
+ "node",
+ {
+ "text": text,
+ "package": "com.xunmeng.pinduoduo",
+ "class": "android.widget.TextView",
+ "bounds": bounds,
+ "clickable": "false",
+ "enabled": "true",
+ "visible-to-user": "true",
+ },
+ )
+ return ElementTree.tostring(root, encoding="unicode")
+
+
+def _home_with_unverified_entry_labels() -> str:
+ root = ElementTree.Element("hierarchy")
+ for index in range(2):
+ ElementTree.SubElement(
+ root,
+ "node",
+ {
+ "text": "快要抢光",
+ "package": "com.xunmeng.pinduoduo",
+ "class": "android.widget.TextView",
+ "clickable": "false",
+ "enabled": "true",
+ "visible-to-user": "true",
+ "bounds": f"[{index},{index}][{index + 1},{index + 1}]",
+ },
+ )
+ return ElementTree.tostring(root, encoding="unicode")
+
+
def _actions(device: _RawDevice, method: str) -> list[tuple[object, ...]]:
return [call for call in device.calls if call[0] == "jsonrpc" and call[1] == method]
@@ -202,6 +282,73 @@ class SkuSelectionFlowTests(unittest.TestCase):
self.assertEqual(_tap_centers(device), [(978, 1333)])
+ def test_entry_stability_uses_verified_chain_projection_not_whole_xml(self) -> None:
+ class DynamicProductDevice(_RawDevice):
+ def __init__(self) -> None:
+ super().__init__()
+ self.frame = 0
+
+ def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+ if method == "dumpWindowHierarchy" and "快要抢光" in self.hierarchy:
+ self.frame += 1
+ self.hierarchy = _dynamic_product_page(str(self.frame), "活动剩余快要抢光")
+ return super().jsonrpc_call(method, params, timeout)
+
+ now = [0.0]
+ device = DynamicProductDevice()
+ flow = SkuSelectionFlow(
+ UiautomatorSkuPanelAdapter(device, 10),
+ 0.03,
+ 0.01,
+ lambda: now[0],
+ lambda seconds: now.__setitem__(0, now[0] + seconds),
+ )
+
+ flow.open_sku_panel(_TARGET_URL, "")
+
+ self.assertEqual(device.frame, 2)
+ self.assertEqual(_tap_centers(device), [(978, 1333)])
+
+ def test_entry_action_description_must_be_stable_across_frames(self) -> None:
+ class ChangingActionDescriptionDevice(_RawDevice):
+ def __init__(self) -> None:
+ super().__init__()
+ self.frame = 0
+
+ def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+ if method == "dumpWindowHierarchy":
+ self.frame += 1
+ self.hierarchy = _dynamic_product_page(
+ str(self.frame),
+ f"活动{self.frame}快要抢光",
+ )
+ return super().jsonrpc_call(method, params, timeout)
+
+ now = [0.0]
+ device = ChangingActionDescriptionDevice()
+ flow = SkuSelectionFlow(
+ UiautomatorSkuPanelAdapter(device, 10),
+ 0.02,
+ 0.01,
+ lambda: now[0],
+ lambda seconds: now.__setitem__(0, now[0] + seconds),
+ )
+
+ with self.assertRaises(SkuSelectionError):
+ flow.open_sku_panel(_TARGET_URL, "")
+
+ self.assertEqual(_actions(device, "click"), [])
+
+ def test_unverified_home_entry_labels_do_not_trigger_old_product_rejection(self) -> None:
+ device = _RawDevice()
+
+ SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).open_sku_panel(
+ _TARGET_URL,
+ _home_with_unverified_entry_labels(),
+ )
+
+ self.assertEqual(_tap_centers(device), [(978, 1333)])
+
def test_entry_child_and_every_ancestor_attribute_drift_never_clicks(self) -> None:
expected_clickable = ("false", "false", "false", "false", "true")
for depth in range(5):
@@ -217,8 +364,74 @@ class SkuSelectionFlowTests(unittest.TestCase):
with self.subTest(depth=depth, attribute=attribute):
self._assert_entry_rejected_without_click(_mutate_entry(depth, attribute, value))
+ def test_entry_chain_text_and_description_drift_never_clicks(self) -> None:
+ cases = [
+ _mutate_entry(depth, "text", "祖先文字漂移") for depth in range(1, 5)
+ ] + [
+ _mutate_entry(depth, "content-desc", "祖先描述漂移") for depth in range(0, 4)
+ ] + [
+ _mutate_entry(4, "content-desc", "祖先描述漂移")
+ ] + [
+ _mutate_entry(4, "content-desc", forbidden + "快要抢光")
+ for forbidden in (
+ "免拼购买", "单独购买", "直接拼成", "提交订单", "支付", "先用后付", "0元下单",
+ "立即购买", "确认下单", "立即付款", "订单详情",
+ )
+ ]
+ for hierarchy in cases:
+ with self.subTest():
+ self._assert_entry_rejected_without_click(hierarchy)
+
+ def test_any_live_clickable_covering_entry_center_blocks_click(self) -> None:
+ cases = (
+ _with_overlapping_clickable(
+ "com.xunmeng.pinduoduo",
+ "android.widget.Button",
+ "[950,1300][1000,1340]",
+ ),
+ _with_overlapping_clickable(
+ "com.android.systemui",
+ "android.view.ViewGroup",
+ "[936,1208][1080,1352]",
+ ),
+ _with_overlapping_clickable(
+ "com.android.systemui",
+ "android.view.ViewGroup",
+ "not-a-bound",
+ ),
+ )
+ for hierarchy in cases:
+ with self.subTest():
+ self._assert_entry_rejected_without_click(hierarchy)
+
+ def test_pre_intent_old_entry_is_rejected_even_with_extra_action_node(self) -> None:
+ for old_page in (
+ _extra_entry_action_ancestor(),
+ _entry_with_panel_price_marker(),
+ _mutate_entry(4, "content-desc", "立即购买快要抢光"),
+ ):
+ with self.subTest():
+ device = _RawDevice()
+ flow = SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10))
+
+ with self.assertRaises(SkuSelectionError):
+ flow.open_sku_panel(_TARGET_URL, old_page)
+
+ self.assertEqual(_actions(device, "click"), [])
+
+ def test_action_subtree_dangerous_or_ambiguous_children_never_click(self) -> None:
+ cases = (
+ _with_action_subtree_child("提交订单"),
+ _with_action_subtree_child("免拼购买"),
+ _with_action_subtree_child("快要抢光"),
+ )
+ for hierarchy in cases:
+ with self.subTest():
+ self._assert_entry_rejected_without_click(hierarchy)
+
def test_duplicate_entry_and_forbidden_sibling_entry_never_click(self) -> None:
self._assert_entry_rejected_without_click(_duplicate_entry())
+ self._assert_entry_rejected_without_click(_extra_entry_action_ancestor())
self._assert_entry_rejected_without_click(_without_entry())
self._assert_entry_rejected_without_click(_mutate_entry(0, "clickable", "true"))
@@ -555,7 +768,11 @@ class SkuSelectionRunnerTests(unittest.TestCase):
now = [0.0]
class SequenceDevice(_RawDevice):
def __init__(self) -> None:
- super().__init__(); self.frames = [_PRODUCT_PAGE, "", _PRODUCT_PAGE]
+ super().__init__(); self.frames = [
+ _dynamic_product_page("first"),
+ "",
+ _dynamic_product_page("second"),
+ ]
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)