From fee40284b35dc4192bd4a731578fbcc4c7721a75 Mon Sep 17 00:00:00 2001 From: chengma Date: Fri, 14 Aug 2026 18:10:56 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E9=87=87=E8=B4=AD=E9=80=89=E8=89=B2?= =?UTF-8?q?=E5=89=8D=E5=BD=92=E4=BD=8D=E8=A7=84=E6=A0=BC=E9=9D=A2=E6=9D=BF?= =?UTF-8?q?=20(#230)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- client/src/pdd_u2_purchase_adapter.py | 260 ++++++++++++++++++++ client/test/test_pdd_u2_purchase_adapter.py | 141 +++++++++++ 2 files changed, 401 insertions(+) diff --git a/client/src/pdd_u2_purchase_adapter.py b/client/src/pdd_u2_purchase_adapter.py index 7d19296..faa23d9 100644 --- a/client/src/pdd_u2_purchase_adapter.py +++ b/client/src/pdd_u2_purchase_adapter.py @@ -70,6 +70,14 @@ _PAYMENT_MARKERS = ("输入支付密码", "立即支付", "支付成功", "支 _FINAL_SUBMIT_MARKERS = ("提交订单", "现在买，仅", "确认购买") _OUT_OF_STOCK_MARKERS = ("已售罄", "暂时缺货", "库存不足", "该商品已售罄") _ADDRESS_ACTION_SETTLE_SECONDS = 1.0 +_COLOR_DIMENSION_HEADINGS = frozenset( + {"颜色分类", "顏色分類", "颜色", "顏色", "款式", "花色", "颜色款式"} +) +_SECOND_DIMENSION_HEADINGS = frozenset( + {"尺码", "尺碼", "尺寸", "套餐", "规格", "規格", "型号", "型號"} +) +_PANEL_TOP_MAX_SWIPES = 8 +_PANEL_TOP_STABLE_READS = 2 _SUPPORTED_OPTION_KEYS = frozenset({"color", "size"}) _MAX_QUANTITY_BUTTON_CLICKS = 5 _KEYBOARD_STATUS_MARKERS = ( @@ -167,6 +175,159 @@ def _parent_nodes(root: ET.Element) -> dict[ET.Element, ET.Element]: } +def _dimension_heading(label: str) -> str: + """只接受独立规格标题，明确排除选择摘要和提交提示。""" + + compact = re.sub(r"\s+", "", str(label or "")) + if not compact or compact.startswith( + ("请选择", "請選擇", "已选", "已選", "已选择", "已選擇") + ): + return "" + if compact.startswith("选择") and "后" in compact: + return "" + compact = re.sub(r"[（(]\d+[)）]$", "", compact) + return compact + + +def _screen_bounds(root: ET.Element) -> Optional[Bounds]: + parsed = [ + bounds + for node in root.iter("node") + if (bounds := _parse_bounds(node.get("bounds", ""))) is not None + ] + if not parsed: + return None + return 0, 0, max(item[2] for item in parsed), max(item[3] for item in parsed) + + +def _has_visible_color_region(root: ET.Element) -> bool: + """正式颜色标题下必须同时存在至少一个可靠的可点击选项。""" + + parents = _parent_nodes(root) + headings: list[Bounds] = [] + second_heading_tops: list[int] = [] + for node in root.iter("node"): + heading = _dimension_heading(_label(node)) + bounds = _parse_bounds(node.get("bounds", "")) + if bounds is None or node.get("visible-to-user", "true") == "false": + continue + if heading in _COLOR_DIMENSION_HEADINGS: + headings.append(bounds) + elif heading in _SECOND_DIMENSION_HEADINGS: + second_heading_tops.append(bounds[1]) + + screen = _screen_bounds(root) + if not headings or screen is None: + return False + + excluded = { + "增加数量", + "减少数量", + "提交订单", + "确定", + "打开大图", + } + for heading_bounds in headings: + boundary = min( + ( + top + for top in second_heading_tops + if top > heading_bounds[3] + ), + default=screen[3], + ) + for node in root.iter("node"): + label = _label(node).strip() + bounds = _parse_bounds(node.get("bounds", "")) + if not label or bounds is None: + continue + if bounds[1] < heading_bounds[3] or bounds[1] >= boundary: + continue + if _dimension_heading(label) in ( + _COLOR_DIMENSION_HEADINGS | _SECOND_DIMENSION_HEADINGS + ): + continue + compact = label.replace(" ", "") + if ( + compact.startswith( + ("请选择", "請選擇", "已选", "已選", "选择", "選擇") + ) + or any(marker in compact for marker in excluded) + or re.fullmatch(r"[¥￥]?\d+(?:\.\d{1,2})?", compact) + ): + continue + if _nearest_clickable_bounds(node, parents) is not None: + return True + return False + + +def _needs_color_region_restore(root: ET.Element) -> bool: + """只有第二规格已出现且颜色摘要仍存在时，才判定面板停在中部。""" + + labels = [_label(node).strip() for node in root.iter("node")] + has_color_summary = any( + label.replace(" ", "").startswith( + ("请选择", "請選擇", "已选", "已選", "已选择", "已選擇") + ) + and any(heading in label.replace(" ", "") for heading in _COLOR_DIMENSION_HEADINGS) + for label in labels + ) + has_second_heading = any( + _dimension_heading(label) in _SECOND_DIMENSION_HEADINGS + for label in labels + ) + return has_color_summary and has_second_heading + + +def _purchase_panel_scroll_region(root: ET.Element) -> Optional[Bounds]: + """返回规格面板中唯一可靠的纵向滚动容器。""" + + screen = _screen_bounds(root) + if screen is None: + return None + screen_width = screen[2] - screen[0] + screen_height = screen[3] - screen[1] + candidates: list[Bounds] = [] + for node in root.iter("node"): + if ( + node.get("scrollable") != "true" + or node.get("package") != PDD_PACKAGE_NAME + or node.get("visible-to-user", "true") == "false" + or node.get("enabled", "true") == "false" + ): + continue + bounds = _parse_bounds(node.get("bounds", "")) + if bounds is None: + continue + width = bounds[2] - bounds[0] + height = bounds[3] - bounds[1] + if width >= screen_width * 0.60 and height >= screen_height * 0.20: + candidates.append(bounds) + unique = sorted(set(candidates)) + return unique[0] if len(unique) == 1 else None + + +def _region_signature(root: ET.Element, region: Bounds) -> str: + """生成当前滚动视口的短签名，只用于判断是否已经无法继续归位。""" + + parts = [] + for node in root.iter("node"): + bounds = _parse_bounds(node.get("bounds", "")) + if bounds is None: + continue + center_x = (bounds[0] + bounds[2]) // 2 + center_y = (bounds[1] + bounds[3]) // 2 + if not ( + region[0] <= center_x <= region[2] + and region[1] <= center_y <= region[3] + ): + continue + label = _label(node) + if label: + parts.append(f"{label}|{bounds}") + return hashlib.sha256("\n".join(parts).encode("utf-8")).hexdigest() + + def _nearest_clickable_bounds( node: ET.Element, parents: Mapping[ET.Element, ET.Element] ) -> Optional[Bounds]: @@ -862,6 +1023,7 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter): color = checked.get("color") if color: + panel_xml = self._restore_purchase_panel_color_region(panel_xml) trace = current_performance_trace() stage = ( trace.stage("purchase_select_color") @@ -1258,6 +1420,104 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter): diagnostics=diagnostics, ) + def _restore_purchase_panel_color_region(self, xml_data: str) -> str: + """选择颜色前把规格内容有界归位到正式颜色区域。""" + + root = _parse_xml(xml_data) + if _has_visible_color_region(root): + return xml_data + if not _needs_color_region_restore(root): + # 兼容不暴露正式标题或滚动属性、但原选择器可以处理的旧面板。 + return xml_data + + region = _purchase_panel_scroll_region(root) + if region is None: + self._raise_color_region_not_found( + swipe_count=0, + stable_reads=0, + scroll_region_found=False, + ) + + device = self._require_device() + previous_signature = _region_signature(root, region) + unchanged_reads = 0 + swipe_count = 0 + for _ in range(_PANEL_TOP_MAX_SWIPES): + self._check_cancelled("purchase_select_options") + left, top, right, bottom = region + height = bottom - top + x = (left + right) // 2 + # 手指向下滑，让规格内容返回顶部；坐标始终限制在规格容器内。 + device.swipe( + x, + top + int(height * 0.25), + x, + top + int(height * 0.82), + duration=0.35, + ) + swipe_count += 1 + self._sleep(0.2) + + current_xml = self._dump_hierarchy() + root = _parse_xml(current_xml) + page_kind = _page_kind(root, "") + if page_kind in { + "captcha", + "login_required", + "risk_control", + "payment", + }: + self._raise_special_page(page_kind) + if page_kind != "order_confirmation": + self._raise_color_region_not_found( + swipe_count=swipe_count, + stable_reads=unchanged_reads, + scroll_region_found=True, + ) + if _has_visible_color_region(root): + return current_xml + + current_region = _purchase_panel_scroll_region(root) + if current_region is None: + break + region = current_region + signature = _region_signature(root, region) + if signature == previous_signature: + unchanged_reads += 1 + else: + unchanged_reads = 0 + previous_signature = signature + if unchanged_reads >= _PANEL_TOP_STABLE_READS: + break + + self._raise_color_region_not_found( + swipe_count=swipe_count, + stable_reads=unchanged_reads, + scroll_region_found=True, + ) + + def _raise_color_region_not_found( + self, + *, + swipe_count: int, + stable_reads: int, + scroll_region_found: bool, + ) -> None: + diagnostics: dict[str, Any] = { + "swipe_count": swipe_count, + "stable_reads": stable_reads, + "scroll_region_found": scroll_region_found, + } + artifact = self._save_last_xml("purchase-color-region-not-found") + if artifact is not None: + diagnostics["artifacts"] = [artifact] + raise PddPurchaseError( + "PURCHASE_PANEL_COLOR_REGION_NOT_FOUND", + "规格面板已打开，但无法回到颜色分类区域", + step="purchase_select_options", + diagnostics=diagnostics, + ) + def _dump_hierarchy(self) -> str: raw = self._require_device().dump_hierarchy() xml_data = ( diff --git a/client/test/test_pdd_u2_purchase_adapter.py b/client/test/test_pdd_u2_purchase_adapter.py index 53f8e40..6f67e6d 100644 --- a/client/test/test_pdd_u2_purchase_adapter.py +++ b/client/test/test_pdd_u2_purchase_adapter.py @@ -98,6 +98,53 @@ def contextual_confirm_panel_xml(quantity: int = 1) -> str: """ +def vertically_scrolled_purchase_panel_xml(*, at_color_top: bool) -> str: + """构造规格面板位于颜色顶部或停在中部的脱敏控件树。""" + + if at_color_top: + scroll_content = """ + + + + + + """ + else: + scroll_content = """ + + + + + """ + return f""" + + + + + + + + {scroll_content} + + + + """ + + class FakeEditor: def __init__(self, device) -> None: self.device = device @@ -180,6 +227,32 @@ class FakeDevice: raise AssertionError(f"未预期的 selector: {selector}") +class VerticallyScrolledColorPanelDevice(FakeDevice): + """模拟 PDD 记住规格面板纵向位置。""" + + def __init__(self, *, starts_at_top: bool, restores_on_swipe: bool = True): + super().__init__() + self.at_color_top = starts_at_top + self.restores_on_swipe = restores_on_swipe + self.swipes = [] + + def dump_hierarchy(self): + if not self.has_opened: + return super().dump_hierarchy() + if self.mode == "home": + return home_xml() + if self.mode == "panel": + return vertically_scrolled_purchase_panel_xml( + at_color_top=self.at_color_top + ) + return super().dump_hierarchy() + + def swipe(self, x1, y1, x2, y2, duration=0.0): + self.swipes.append((x1, y1, x2, y2, duration)) + if self.restores_on_swipe: + self.at_color_top = True + + class ColdStartDevice(FakeDevice): def __init__(self) -> None: super().__init__() @@ -927,6 +1000,74 @@ class U2PddPurchaseAdapterTest(unittest.TestCase): self.assertEqual(device.clicks[-1], (540, 2140)) adapter.close() + def test_purchase_restores_color_region_before_selecting_color(self): + clock = FakeClock() + device = VerticallyScrolledColorPanelDevice(starts_at_top=False) + selected_xml = [] + + def select_color_fn(_device, xml_data, _target, **_kwargs): + selected_xml.append(xml_data) + return True + + adapter = self._live_adapter( + device, + [], + sleeper=clock.sleep, + monotonic=clock.monotonic, + select_color_fn=select_color_fn, + ) + adapter.open_goods(GOODS_URL) + + adapter.select_options({"color": "黑色"}) + + self.assertEqual(len(device.swipes), 1) + self.assertLess(device.swipes[0][1], device.swipes[0][3]) + self.assertEqual(len(selected_xml), 1) + self.assertIn('text="颜色分类"', selected_xml[0]) + adapter.close() + + def test_purchase_does_not_scroll_when_color_region_is_already_visible(self): + device = VerticallyScrolledColorPanelDevice(starts_at_top=True) + adapter = self._live_adapter( + device, + [], + select_color_fn=lambda *_args, **_kwargs: True, + ) + adapter.open_goods(GOODS_URL) + + adapter.select_options({"color": "黑色"}) + + self.assertEqual(device.swipes, []) + adapter.close() + + def test_purchase_stops_when_color_region_cannot_be_restored(self): + clock = FakeClock() + device = VerticallyScrolledColorPanelDevice( + starts_at_top=False, + restores_on_swipe=False, + ) + select_calls = [] + adapter = self._live_adapter( + device, + [], + sleeper=clock.sleep, + monotonic=clock.monotonic, + select_color_fn=lambda *_args, **_kwargs: select_calls.append(True), + ) + adapter.open_goods(GOODS_URL) + + with self.assertRaises(PddPurchaseError) as raised: + adapter.select_options({"color": "黑色"}) + + self.assertEqual( + raised.exception.code, + "PURCHASE_PANEL_COLOR_REGION_NOT_FOUND", + ) + self.assertEqual(len(device.swipes), 2) + self.assertEqual(select_calls, []) + self.assertEqual(len(device.clicks), 1) + adapter.close() + def test_address_is_split_at_first_separator_before_tagging(self): self.assertEqual( _tagged_shipping_address(