fix: 兼容采购规格面板确定按钮 (#157)

This commit is contained in:
chengma
2026-08-11 15:29:41 +08:00
parent 753f816f32
commit f7f688833e
2 changed files with 263 additions and 3 deletions
+139 -1
View File
@@ -34,8 +34,10 @@ from .pdd_page_classifier import (
PAGE_HOME, PAGE_HOME,
PAGE_LOGIN_REQUIRED, PAGE_LOGIN_REQUIRED,
PAGE_NETWORK_ERROR, PAGE_NETWORK_ERROR,
PAGE_ORDER_CONFIRMATION,
PAGE_PAYMENT, PAGE_PAYMENT,
PAGE_RISK_CONTROL, PAGE_RISK_CONTROL,
PAGE_UNKNOWN,
GoodsOpenTracker, GoodsOpenTracker,
PddPageObservation, PddPageObservation,
classify_pdd_page, classify_pdd_page,
@@ -80,6 +82,11 @@ _DIAGNOSTIC_MARKERS = (
"手机号登录", "手机号登录",
"操作频繁", "操作频繁",
"网络不给力", "网络不给力",
"确认款式",
"确认规格",
"已选",
"请选择",
"确定",
) )
@@ -167,7 +174,11 @@ def _goods_id_from_url(goods_url: str) -> str:
def _page_kind(root: ET.Element, current_package: str) -> str: def _page_kind(root: ET.Element, current_package: str) -> str:
return classify_pdd_page(root, current_package).kind kind = classify_pdd_page(root, current_package).kind
contextual_targets = _contextual_confirm_targets(root)
if kind in {PAGE_UNKNOWN, PAGE_GOODS} and len(contextual_targets) == 1:
return PAGE_ORDER_CONFIRMATION
return kind
def _selected(root: ET.Element, target: str) -> bool: def _selected(root: ET.Element, target: str) -> bool:
@@ -324,9 +335,136 @@ def _final_submit_targets(root: ET.Element) -> list[Bounds]:
if (bounds[1] + bounds[3]) // 2 < screen_bottom * 0.6: if (bounds[1] + bounds[3]) // 2 < screen_bottom * 0.6:
continue continue
targets.add(bounds) targets.add(bounds)
targets.update(_contextual_confirm_targets(root))
return sorted(targets) return sorted(targets)
def _contextual_confirm_targets(root: ET.Element) -> list[Bounds]:
"""只在强证据规格面板内接受底部“确定”作为最终提交按钮。"""
nodes = list(root.iter("node"))
if sum(
node.get("package") == PDD_PACKAGE_NAME for node in nodes
) < 3:
return []
parsed_bounds = [
bounds
for node in nodes
if (bounds := _parse_bounds(node.get("bounds", ""))) is not None
]
if not parsed_bounds:
return []
screen_right = max(bounds[2] for bounds in parsed_bounds)
screen_bottom = max(bounds[3] for bounds in parsed_bounds)
screen_area = screen_right * screen_bottom
parents = {
child: parent
for parent in root.iter()
for child in parent
if child.tag == "node"
}
targets: set[Bounds] = set()
for confirm in nodes:
if _label(confirm).strip() not in {"确定", "確定"}:
continue
bounds = _parse_bounds(confirm.get("bounds", ""))
if bounds is None or confirm.get("clickable") != "true":
continue
if (
confirm.get("visible-to-user") != "true"
or confirm.get("enabled") != "true"
):
continue
if (bounds[1] + bounds[3]) // 2 < screen_bottom * 0.6:
continue
panel = parents.get(confirm)
while panel is not None:
panel_bounds = _parse_bounds(panel.get("bounds", ""))
if panel_bounds is not None:
area = (panel_bounds[2] - panel_bounds[0]) * (
panel_bounds[3] - panel_bounds[1]
)
if (not screen_area or area < screen_area * 0.95) and (
_has_confirmation_panel_evidence(panel)
):
targets.add(bounds)
break
panel = parents.get(panel)
return sorted(targets)
def _has_confirmation_panel_evidence(panel: ET.Element) -> bool:
"""确认标题、已选摘要、规格标题和数量控件位于同一面板。"""
nodes = list(panel.iter("node"))
labels = [_label(node).replace(" ", "") for node in nodes]
has_title = any(
label in {"确认款式", "確認款式", "确认规格", "確認規格"}
for label in labels
)
has_summary = any(
label.startswith(("已选", "已選", "已选择", "已選擇", "请选择", "請選擇"))
for label in labels
)
has_dimension = any(
re.sub(r"[((]\d+[))]$", "", label)
in {
"颜色分类",
"顏色分類",
"颜色",
"顏色",
"尺码",
"尺碼",
"规格",
"規格",
}
for label in labels
)
editors = [
node
for node in nodes
if node.get("class") == "android.widget.EditText"
and node.get("text", "").strip().isdigit()
and int(node.get("text", "0")) > 0
]
has_decrease = len(_quantity_targets_in_nodes(nodes, "减少数量")) == 1
has_increase = len(_quantity_targets_in_nodes(nodes, "增加数量")) == 1
return (
has_title
and has_summary
and has_dimension
and len(editors) == 1
and has_decrease
and has_increase
)
def _quantity_targets_in_nodes(
nodes: list[ET.Element], description: str
) -> set[Bounds]:
"""读取指定面板内唯一、可用的数量按钮。"""
targets: set[Bounds] = set()
for node in nodes:
if description not in {
node.get("text", "").strip(),
node.get("content-desc", "").strip(),
}:
continue
bounds = _parse_bounds(node.get("bounds", ""))
if bounds is None or node.get("clickable") != "true":
continue
if node.get("enabled", "true") == "false":
continue
if node.get("visible-to-user", "true") == "false":
continue
targets.add(bounds)
return targets
class U2PddPurchaseAdapter(PddPurchaseAdapter): class U2PddPurchaseAdapter(PddPurchaseAdapter):
"""PDD 真机采购演练会话,不提供真实下单能力。""" """PDD 真机采购演练会话,不提供真实下单能力。"""
+124 -2
View File
@@ -3,12 +3,17 @@
from pathlib import Path from pathlib import Path
import tempfile import tempfile
import unittest import unittest
import xml.etree.ElementTree as ET
from src.pdd_device_service import PddDeviceService from src.pdd_device_service import PddDeviceService
from src.pdd_purchase_adapter import PddPurchaseError from src.pdd_purchase_adapter import PddPurchaseError
from src.performance_timing import TaskPerformanceTrace from src.performance_timing import TaskPerformanceTrace
from src.pdd_u2_purchase_adapter import U2PddPurchaseAdapter from src.pdd_u2_purchase_adapter import (
from src.pdd_u2_purchase_adapter import U2PddLivePurchaseAdapter U2PddLivePurchaseAdapter,
U2PddPurchaseAdapter,
_final_submit_targets,
_page_kind,
)
GOODS_URL = "https://mobile.yangkeduo.com/goods.html?goods_id=753136429979" GOODS_URL = "https://mobile.yangkeduo.com/goods.html?goods_id=753136429979"
@@ -60,6 +65,36 @@ def panel_xml(quantity: int = 1) -> str:
</hierarchy>""" </hierarchy>"""
def contextual_confirm_panel_xml(quantity: int = 1) -> str:
"""模拟商品 897186799891 的非滚动“确定”规格面板。"""
return f"""<hierarchy>
<node package="com.xunmeng.pinduoduo" bounds="[0,0][1080,2340]">
<node package="com.xunmeng.pinduoduo" class="android.widget.RelativeLayout"
bounds="[0,720][1080,2214]" visible-to-user="true" enabled="true">
<node package="com.xunmeng.pinduoduo" text="确认款式"
bounds="[438,746][642,814]"/>
<node text="券后 ¥8.17" bounds="[324,852][573,919]"/>
<node text="已选择: 黑色 均码" bounds="[324,1009][1044,1062]"/>
<node class="android.widget.EditText" text="{quantity}"
bounds="[408,1080][489,1155]"/>
<node content-desc="减少数量" clickable="true" enabled="true"
visible-to-user="true" bounds="[324,1080][402,1155]"/>
<node content-desc="增加数量" clickable="true" enabled="true"
visible-to-user="true" bounds="[495,1080][573,1155]"/>
<node text="颜色分类" bounds="[36,1214][216,1267]"/>
<node text="黑色" clickable="true" selected="true"
bounds="[36,1295][276,1382]"/>
<node text="尺码" bounds="[36,1684][126,1737]"/>
<node text="均码" clickable="true" selected="true"
bounds="[36,1765][711,1852]"/>
<node text="确定" clickable="true" enabled="true"
visible-to-user="true" bounds="[0,2067][1080,2214]"/>
</node>
</node>
</hierarchy>"""
class FakeEditor: class FakeEditor:
def __init__(self, device) -> None: def __init__(self, device) -> None:
self.device = device self.device = device
@@ -186,6 +221,24 @@ class PanelDoesNotOpenDevice(FakeDevice):
self.clicks.append((x, y)) self.clicks.append((x, y))
class ContextualConfirmPanelDevice(FakeDevice):
def dump_hierarchy(self):
if not self.has_opened:
return '<hierarchy><node package="com.xunmeng.pinduoduo" /></hierarchy>'
if self.mode == "special":
return self.special_xml
if self.mode == "panel":
return contextual_confirm_panel_xml(self.quantity)
return home_xml()
def click(self, x, y):
self.clicks.append((x, y))
if self.mode == "panel" and 495 <= x <= 573 and 1080 <= y <= 1155:
self.quantity += 1
return
self.mode = "panel"
class U2PddPurchaseAdapterTest(unittest.TestCase): class U2PddPurchaseAdapterTest(unittest.TestCase):
def _adapter(self, device, calls): def _adapter(self, device, calls):
def select_color_fn(_device, _xml, target, **_kwargs): def select_color_fn(_device, _xml, target, **_kwargs):
@@ -252,6 +305,75 @@ class U2PddPurchaseAdapterTest(unittest.TestCase):
self.assertEqual(device.editor_values, []) self.assertEqual(device.editor_values, [])
self.assertEqual(device.app_wait_calls, 0) self.assertEqual(device.app_wait_calls, 0)
def test_contextual_confirm_panel_is_reliable_without_clicking_confirm(self):
device = ContextualConfirmPanelDevice()
calls = []
def select_color_fn(_device, _xml, target, **_kwargs):
calls.append(("color", target))
return target == "黑色"
def select_size_fn(_device, _xml, target, **_kwargs):
calls.append(("size", target))
return target == "均码"
adapter = U2PddPurchaseAdapter(
"USB-001",
device_service=PddDeviceService(connector=lambda _serial: device),
sleeper=lambda _seconds: None,
select_color_fn=select_color_fn,
select_size_fn=select_size_fn,
)
adapter.open_goods(GOODS_URL)
adapter.select_options({"color": "黑色", "size": "均码"})
state = adapter.read_state()
adapter.stop_before_submit()
self.assertEqual(state.page_kind, "order_confirmation")
self.assertEqual(state.submit_candidate_count, 1)
self.assertEqual(calls, [("color", "黑色"), ("size", "均码")])
# dry-run 只点击商品页采购入口,不点击底部“确定”。
self.assertEqual(device.clicks, [(790, 2214)])
adapter.close()
def test_plain_confirm_dialog_is_not_a_submit_target(self):
root = ET.fromstring("""<hierarchy>
<node package="com.xunmeng.pinduoduo" bounds="[0,0][1080,2340]">
<node package="com.xunmeng.pinduoduo" text="提示"
bounds="[100,800][980,1400]">
<node package="com.xunmeng.pinduoduo" text="确定"
clickable="true" enabled="true" visible-to-user="true"
bounds="[300,1250][780,1380]"/>
</node>
</node>
</hierarchy>""")
self.assertEqual(_page_kind(root, ""), "unknown")
self.assertEqual(_final_submit_targets(root), [])
def test_live_adapter_clicks_contextual_confirm_only_once(self):
device = ContextualConfirmPanelDevice()
adapter = U2PddLivePurchaseAdapter(
"USB-001",
device_service=PddDeviceService(connector=lambda _serial: device),
sleeper=lambda _seconds: None,
select_color_fn=lambda *_args, **_kwargs: True,
select_size_fn=lambda *_args, **_kwargs: True,
)
adapter.open_goods(GOODS_URL)
adapter.select_options({"color": "黑色", "size": "均码"})
adapter.submit_order_once()
with self.assertRaises(PddPurchaseError) as raised:
adapter.submit_order_once()
self.assertEqual(raised.exception.code, "PURCHASE_SUBMIT_ALREADY_ATTEMPTED")
# 采购入口一次,底部“确定”一次。
self.assertEqual(len(device.clicks), 2)
self.assertEqual(device.clicks[-1], (540, 2140))
adapter.close()
def test_reliable_pdd_tree_skips_repeated_app_current(self): def test_reliable_pdd_tree_skips_repeated_app_current(self):
device = FakeDevice() device = FakeDevice()
adapter = self._adapter(device, []) adapter = self._adapter(device, [])