diff --git a/client/src/pdd_u2_purchase_adapter.py b/client/src/pdd_u2_purchase_adapter.py index c3ec4c9..89f7273 100644 --- a/client/src/pdd_u2_purchase_adapter.py +++ b/client/src/pdd_u2_purchase_adapter.py @@ -34,8 +34,10 @@ from .pdd_page_classifier import ( PAGE_HOME, PAGE_LOGIN_REQUIRED, PAGE_NETWORK_ERROR, + PAGE_ORDER_CONFIRMATION, PAGE_PAYMENT, PAGE_RISK_CONTROL, + PAGE_UNKNOWN, GoodsOpenTracker, PddPageObservation, 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: - 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: @@ -324,9 +335,136 @@ def _final_submit_targets(root: ET.Element) -> list[Bounds]: if (bounds[1] + bounds[3]) // 2 < screen_bottom * 0.6: continue targets.add(bounds) + targets.update(_contextual_confirm_targets(root)) 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): """PDD 真机采购演练会话，不提供真实下单能力。""" diff --git a/client/test/test_pdd_u2_purchase_adapter.py b/client/test/test_pdd_u2_purchase_adapter.py index 592a050..d569b12 100644 --- a/client/test/test_pdd_u2_purchase_adapter.py +++ b/client/test/test_pdd_u2_purchase_adapter.py @@ -3,12 +3,17 @@ from pathlib import Path import tempfile import unittest +import xml.etree.ElementTree as ET from src.pdd_device_service import PddDeviceService from src.pdd_purchase_adapter import PddPurchaseError from src.performance_timing import TaskPerformanceTrace -from src.pdd_u2_purchase_adapter import U2PddPurchaseAdapter -from src.pdd_u2_purchase_adapter import U2PddLivePurchaseAdapter +from src.pdd_u2_purchase_adapter import ( + U2PddLivePurchaseAdapter, + U2PddPurchaseAdapter, + _final_submit_targets, + _page_kind, +) GOODS_URL = "https://mobile.yangkeduo.com/goods.html?goods_id=753136429979" @@ -60,6 +65,36 @@ def panel_xml(quantity: int = 1) -> str: """ +def contextual_confirm_panel_xml(quantity: int = 1) -> str: + """模拟商品 897186799891 的非滚动“确定”规格面板。""" + + return f""" + + + + + + + + + + + + + + + + """ + + class FakeEditor: def __init__(self, device) -> None: self.device = device @@ -186,6 +221,24 @@ class PanelDoesNotOpenDevice(FakeDevice): self.clicks.append((x, y)) +class ContextualConfirmPanelDevice(FakeDevice): + def dump_hierarchy(self): + if not self.has_opened: + return '' + 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): def _adapter(self, device, calls): def select_color_fn(_device, _xml, target, **_kwargs): @@ -252,6 +305,75 @@ class U2PddPurchaseAdapterTest(unittest.TestCase): self.assertEqual(device.editor_values, []) 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(""" + + + + + + """) + + 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): device = FakeDevice() adapter = self._adapter(device, [])