fix: 规格面板地址入口有限下滑识别 (#271)

client/src/pdd_u2_purchase_adapter.py

@@ -14,7 +14,7 @@ from contextlib import nullcontext
 from datetime import datetime, timezone
 from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
 from pathlib import Path
-from typing import Any, Callable, Mapping, Optional
+from typing import Any, Callable, Mapping, NoReturn, Optional
 from urllib.parse import parse_qs, urlparse
 
 from .db import data_dir
@@ -85,6 +85,8 @@ _PAYMENT_MARKERS = ("输入支付密码", "立即支付", "支付成功", "支
 _FINAL_SUBMIT_MARKERS = ("提交订单", "现在买，仅", "确认购买")
 _OUT_OF_STOCK_MARKERS = ("已售罄", "暂时缺货", "库存不足", "该商品已售罄")
 _ADDRESS_ACTION_SETTLE_SECONDS = 1.0
+_ADDRESS_ENTRY_MAX_SWIPES = 3
+_ADDRESS_ENTRY_STABLE_READ_LIMIT = 2
 _COLOR_DIMENSION_HEADINGS = frozenset(
     {"颜色分类", "顏色分類", "颜色", "顏色", "款式", "花色", "颜色款式"}
 )
@@ -2040,7 +2042,7 @@ class U2PddLivePurchaseAdapter(U2PddPurchaseAdapter, PddLivePurchaseAdapter):
                     step="purchase_update_shipping_address",
                 )
 
-            root = _parse_xml(self._dump_hierarchy())
+            root = self._prepare_address_entry()
             entry_targets = _address_entry_targets(root)
             self._click_unique_address_target(
                 entry_targets,
@@ -2107,6 +2109,117 @@ class U2PddLivePurchaseAdapter(U2PddPurchaseAdapter, PddLivePurchaseAdapter):
                 exc, "purchase_update_shipping_address"
             )
 
+    def _prepare_address_entry(self) -> ET.Element:
+        """确保规格面板顶部的收货地址入口可见且唯一。
+
+        地址入口可能因规格面板停留在下方而被裁剪。只有首次识别失败时，
+        才在唯一的规格滚动容器内有限下滑，并在每次操作后重新读取控件树。
+        """
+
+        root = _parse_xml(self._dump_hierarchy())
+        if len(_address_entry_targets(root)) == 1:
+            return root
+
+        region = _purchase_panel_scroll_region(root)
+        if region is None:
+            self._raise_address_entry_ambiguous(
+                swipe_count=0,
+                stable_reads=0,
+                scroll_region_found=False,
+                candidate_count=len(_address_entry_targets(root)),
+            )
+
+        device = self._require_device()
+        previous_signature = _region_signature(root, region)
+        unchanged_reads = 0
+        swipe_count = 0
+        candidate_count = len(_address_entry_targets(root))
+        for _ in range(_ADDRESS_ENTRY_MAX_SWIPES):
+            self._check_cancelled("purchase_update_shipping_address")
+            left, top, right, bottom = region
+            height = bottom - top
+            x = (left + right) // 2
+            # 手指向下滑，使规格面板内容向上移动，显示顶部联系人和地址。
+            device.swipe(
+                x,
+                top + int(height * 0.24),
+                x,
+                top + int(height * 0.78),
+                duration=0.35,
+            )
+            swipe_count += 1
+            self._sleep(0.5)
+
+            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_address_entry_ambiguous(
+                    swipe_count=swipe_count,
+                    stable_reads=unchanged_reads,
+                    scroll_region_found=True,
+                    candidate_count=len(_address_entry_targets(root)),
+                )
+
+            entry_targets = _address_entry_targets(root)
+            candidate_count = len(entry_targets)
+            if candidate_count == 1:
+                return root
+
+            current_region = _purchase_panel_scroll_region(root)
+            if current_region is None:
+                break
+            signature = _region_signature(root, current_region)
+            if signature == previous_signature:
+                unchanged_reads += 1
+                if unchanged_reads >= _ADDRESS_ENTRY_STABLE_READ_LIMIT:
+                    break
+            else:
+                unchanged_reads = 0
+            previous_signature = signature
+            region = current_region
+
+        self._raise_address_entry_ambiguous(
+            swipe_count=swipe_count,
+            stable_reads=unchanged_reads,
+            scroll_region_found=True,
+            candidate_count=candidate_count,
+        )
+        raise AssertionError("unreachable")
+
+    def _raise_address_entry_ambiguous(
+        self,
+        *,
+        swipe_count: int,
+        stable_reads: int,
+        scroll_region_found: bool,
+        candidate_count: int,
+    ) -> NoReturn:
+        """以不含个人信息的诊断结束地址入口识别。"""
+
+        diagnostics: dict[str, Any] = {
+            "candidate_count": candidate_count,
+            "swipe_count": swipe_count,
+            "stable_reads": stable_reads,
+            "scroll_region_found": scroll_region_found,
+        }
+        artifact = self._save_last_xml("purchase-address-entry-ambiguous")
+        if artifact is not None:
+            diagnostics["artifacts"] = [artifact]
+        raise PddPurchaseError(
+            "PURCHASE_ADDRESS_ENTRY_AMBIGUOUS",
+            "规格面板没有唯一可靠的收货地址入口",
+            step="purchase_update_shipping_address",
+            diagnostics=diagnostics,
+        )
+
     def _click_unique_address_target(
         self, targets: list[Bounds], code: str, message: str
     ) -> None:

client/test/test_pdd_u2_purchase_adapter.py

@@ -544,6 +544,19 @@ def address_confirmation_xml(address: str) -> str:
     </hierarchy>"""
 
 
+def clipped_address_confirmation_xml(address: str) -> str:
+    """首次地址卡片被顶部裁剪时的脱敏控件树。"""
+
+    return address_confirmation_xml(address).replace(
+        '<node text="测试用户，000****0000" bounds="[300,330][800,380]"/>',
+        "",
+    ).replace(
+        '<node package="com.xunmeng.pinduoduo" bounds="[0,0][1080,2376]">',
+        '<node package="com.xunmeng.pinduoduo" scrollable="true" '
+        'visible-to-user="true" bounds="[0,0][1080,2079]">',
+    )
+
+
 def address_panel_xml(address: str) -> str:
     return f"""<hierarchy>
       <node package="com.xunmeng.pinduoduo" bounds="[0,0][1080,2376]">
@@ -976,6 +989,25 @@ class LostFirstModifyClickAddressDevice(AddressFlowDevice):
             self.mode = "address_edit"
 
 
+class ClippedAddressFlowDevice(AddressFlowDevice):
+    """模拟规格面板下滑后才显示联系人和手机号。"""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.swipes = []
+        self.address_visible = False
+
+    def dump_hierarchy(self):
+        if self.mode == "panel" and not self.address_visible:
+            return clipped_address_confirmation_xml(self.address)
+        return super().dump_hierarchy()
+
+    def swipe(self, x1, y1, x2, y2, duration=0.0):
+        self.swipes.append((x1, y1, x2, y2, duration))
+        if y2 > y1:
+            self.address_visible = True
+
+
 class NeverOpenAddressEditDevice(AddressFlowDevice):
     """模拟两次“修改”点击都未打开编辑页。"""
 
@@ -1897,6 +1929,24 @@ class U2PddPurchaseAdapterTest(unittest.TestCase):
         self.assertEqual(clock.sleeps.count(1.0), 5)
         adapter.close()
 
+    def test_address_entry_scrolls_down_once_when_contact_is_clipped(self):
+        device = ClippedAddressFlowDevice()
+        adapter = self._live_adapter(device, [])
+        adapter.open_goods(GOODS_URL)
+        adapter.select_options(
+            {"color": "黑色", "size": "3XL【140-165斤】"}
+        )
+
+        adapter.update_shipping_address("cg12")
+
+        self.assertEqual(len(device.swipes), 1)
+        self.assertLess(device.swipes[0][1], device.swipes[0][3])
+        self.assertEqual(
+            device.address, "测试省测试市测试区测试路1号_cg12"
+        )
+        self.assertEqual(device.mode, "panel")
+        adapter.close()
+
     def test_multiple_addresses_modify_only_default_card(self):
         device = MultipleAddressFlowDevice(default_indexes=(0,))
         adapter = self._live_adapter(device, [])