fix: 兼容地址保存系统浮层 (#180)

2026-08-12 11:14:01 +08:00
parent 4df023a455
commit 669f144a9d
2 changed files with 284 additions and 11 deletions
@@ -60,10 +60,8 @@ Bounds = tuple[int, int, int, int]
 _BOUNDS_PATTERN = re.compile(r"^\[(\d+),(\d+)\]\[(\d+),(\d+)\]$")
 _PRICE_PATTERN = re.compile(r"[¥￥]\s*(\d+(?:\.\d{1,2})?)")
 _MASKED_PHONE_PATTERN = re.compile(r"(?<!\d)\d{3}\*{4}\d{4}(?!\d)")
-_PROGRAM_ADDRESS_SUFFIX_PATTERN = re.compile(
-    r"(?:[-_](?:cg\d+|(?:am|pm)?[0-9a-f]{4,12}))$",
-    re.IGNORECASE,
-)
+_ADDRESS_SEPARATOR_PATTERN = re.compile(r"[-_]")
+_COLOROS_PASSWORD_PACKAGE = "com.coloros.codebook"
 _READY_MARKERS = ("发起拼单", "立即购买", "单独购买", "免拼购买", "快要抢光")
 _LOGIN_MARKERS = ("手机号登录", "登录后继续", "验证码登录", "账号登录")
 _CAPTCHA_MARKERS = ("请完成验证", "拖动滑块", "安全验证", "点击图中")
@@ -260,8 +258,36 @@ def _address_page_kind(root: ET.Element) -> str:
    return "unknown"


+def _is_reliable_address_panel(
+    root: ET.Element, expected_address: str = ""
+) -> bool:
+    """用标题和唯一导航目标识别地址面板，地址全文只作增强证据。"""
+
+    labels = _labels(root)
+    if "收货地址" not in labels:
+        return False
+    if len(_clickable_text_targets(root, "修改")) != 1:
+        return False
+    if len(_clickable_text_targets(root, "返回")) != 1:
+        return False
+    if expected_address and any(expected_address in label for label in labels):
+        return True
+    return True
+
+
+def _has_coloros_password_overlay(root: ET.Element) -> bool:
+    """只识别已知 ColorOS 密码本的可见焦点浮层。"""
+
+    return any(
+        node.get("package") == _COLOROS_PASSWORD_PACKAGE
+        and node.get("visible-to-user", "true") != "false"
+        and node.get("focused", "false") == "true"
+        for node in root.iter("node")
+    )
+
+
 def _tagged_shipping_address(address: str, purchase_number: str) -> str:
-    """只替换程序生成的末尾标记，保留完整真实地址主体。"""
+    """按首个连字符或下划线截取地址主体，再追加采购编号。"""

    checked_address = str(address or "").strip()
    checked_number = str(purchase_number or "").strip()
@@ -269,7 +295,9 @@ def _tagged_shipping_address(address: str, purchase_number: str) -> str:
        raise ValueError("地址和采购编号不能为空")
    if any(character in checked_number for character in "\r\n\t"):
        raise ValueError("采购编号不能包含控制字符")
-    address_body = _PROGRAM_ADDRESS_SUFFIX_PATTERN.sub("", checked_address)
+    address_body = _ADDRESS_SEPARATOR_PATTERN.split(
+        checked_address, maxsplit=1
+    )[0].strip()
    if not address_body:
        raise ValueError("地址主体不能为空")
    return f"{address_body}_{checked_number}"
@@ -1309,9 +1337,9 @@ class U2PddLivePurchaseAdapter(U2PddPurchaseAdapter, PddLivePurchaseAdapter):
                "修改页没有唯一可靠的保存按钮",
            )

-            saved_root = self._wait_for_address_page(
-                "saved", expected_address=new_address
-            )
+            saved_root, saved_page = self._wait_after_address_save(new_address)
+            if saved_page == "confirmation":
+                return
            back_targets = _clickable_text_targets(saved_root, "返回")
            self._click_unique_address_target(
                back_targets,
@@ -1472,6 +1500,56 @@ class U2PddLivePurchaseAdapter(U2PddPurchaseAdapter, PddLivePurchaseAdapter):
            diagnostics=diagnostics,
        )

+    def _wait_after_address_save(
+        self, expected_address: str
+    ) -> tuple[ET.Element, str]:
+        """等待保存后的地址面板；已知密码本浮层最多按一次返回。"""
+
+        deadline = self._monotonic() + self._panel_timeout
+        observed_page = "unknown"
+        overlay_seen = False
+        overlay_stable_reads = 0
+        system_back_attempts = 0
+
+        while self._monotonic() < deadline:
+            self._check_cancelled("purchase_update_shipping_address")
+            root = _parse_xml(self._dump_hierarchy())
+            if _has_coloros_password_overlay(root):
+                observed_page = "coloros_password_overlay"
+                overlay_seen = True
+                overlay_stable_reads += 1
+                if overlay_stable_reads >= 3 and system_back_attempts == 0:
+                    # 只关闭已知系统浮层；保存按钮和 PDD 页面都不重试点击。
+                    self._require_device().press("back")
+                    system_back_attempts = 1
+                    overlay_stable_reads = 0
+                self._sleep(0.2)
+                continue
+
+            overlay_stable_reads = 0
+            if _is_reliable_address_panel(root, expected_address):
+                return root, "saved"
+            if _page_kind(root, "") == "order_confirmation":
+                return root, "confirmation"
+            observed_page = _address_page_kind(root)
+            self._sleep(0.2)
+
+        diagnostics: dict[str, Any] = {
+            "expected_page": "saved",
+            "observed_page": observed_page,
+            "coloros_password_overlay_seen": overlay_seen,
+            "system_back_attempts": system_back_attempts,
+        }
+        artifact = self._save_last_xml("purchase-address-saved-timeout")
+        if artifact is not None:
+            diagnostics["artifacts"] = [artifact]
+        raise PddPurchaseError(
+            "PURCHASE_ADDRESS_PAGE_TIMEOUT",
+            "收货地址页面切换或保存确认超时，未进入不可逆下单阶段",
+            step="purchase_update_shipping_address",
+            diagnostics=diagnostics,
+        )
+
    def submit_order_once(self) -> None:
        if self._submit_attempted:
            raise PddPurchaseError(
@@ -314,6 +314,20 @@ def address_edit_xml(address: str) -> str:
    </hierarchy>"""


+def coloros_password_overlay_xml() -> str:
+    return """<hierarchy>
+      <node package="com.xunmeng.pinduoduo" class="android.widget.FrameLayout"
+            bounds="[0,0][1080,2376]"/>
+      <node package="com.coloros.codebook" class="android.widget.FrameLayout"
+            clickable="true" focusable="true" focused="true"
+            visible-to-user="true" bounds="[0,0][1080,2328]">
+        <node package="com.coloros.codebook" text="系统提示"
+              class="android.widget.TextView" visible-to-user="true"
+              bounds="[96,1965][516,2022]"/>
+      </node>
+    </hierarchy>"""
+
+
 class FakeAddressEditor:
    def __init__(self, device, matched: bool) -> None:
        self.device = device
@@ -363,6 +377,82 @@ class AddressFlowDevice(FakeDevice):
        return super().__call__(**selector)


+class AddressSaveOverlayDevice(AddressFlowDevice):
+    """模拟保存后出现 ColorOS 密码本浮层。"""
+
+    def __init__(self, clear_after_reads=None) -> None:
+        super().__init__()
+        self.clear_after_reads = clear_after_reads
+        self.overlay_reads = 0
+        self.save_clicks = 0
+
+    def dump_hierarchy(self):
+        if self.mode == "address_overlay":
+            self.overlay_reads += 1
+            if (
+                self.clear_after_reads is not None
+                and self.overlay_reads > self.clear_after_reads
+            ):
+                self.mode = "address_saved"
+                return super().dump_hierarchy()
+            return coloros_password_overlay_xml()
+        return super().dump_hierarchy()
+
+    def click(self, x, y):
+        if self.mode == "address_edit":
+            self.clicks.append((x, y))
+            self.save_clicks += 1
+            self.mode = "address_overlay"
+            return
+        super().click(x, y)
+
+    def press(self, key):
+        self.presses.append(key)
+        if self.mode == "address_overlay" and key == "back":
+            self.mode = "address_saved"
+
+
+class PersistentAddressSaveOverlayDevice(AddressSaveOverlayDevice):
+    """模拟系统返回键也没有关闭密码本浮层。"""
+
+    def press(self, key):
+        self.presses.append(key)
+
+
+class UnknownAfterAddressSaveDevice(AddressFlowDevice):
+    """模拟保存后进入非白名单未知浮层。"""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.save_clicks = 0
+
+    def dump_hierarchy(self):
+        if self.mode == "address_save_unknown":
+            return """<hierarchy>
+              <node package="com.example.unknown" text="未知页面"
+                    focused="true" visible-to-user="true"
+                    bounds="[0,0][1080,2376]"/>
+            </hierarchy>"""
+        return super().dump_hierarchy()
+
+    def click(self, x, y):
+        if self.mode == "address_edit":
+            self.clicks.append((x, y))
+            self.save_clicks += 1
+            self.mode = "address_save_unknown"
+            return
+        super().click(x, y)
+
+
+class HiddenSavedAddressDevice(AddressFlowDevice):
+    """模拟地址面板不在控件树中暴露完整地址。"""
+
+    def dump_hierarchy(self):
+        if self.mode == "address_saved":
+            return address_panel_xml("")
+        return super().dump_hierarchy()
+
+
 class LostFirstModifyClickAddressDevice(AddressFlowDevice):
    """模拟第一次“修改”点击被页面吞掉。"""

@@ -593,7 +683,7 @@ class U2PddPurchaseAdapterTest(unittest.TestCase):
        self.assertEqual(device.clicks[-1], (540, 2140))
        adapter.close()

-    def test_program_address_suffix_is_replaced_without_cutting_body(self):
+    def test_address_is_split_at_first_separator_before_tagging(self):
        self.assertEqual(
            _tagged_shipping_address(
                "测试省测试市测试区测试路1号-1dd212", "cg2"
@@ -610,7 +700,13 @@ class U2PddPurchaseAdapterTest(unittest.TestCase):
            _tagged_shipping_address(
                "测试省-开发区测试路1号", "cg4"
            ),
-            "测试省-开发区测试路1号_cg4",
+            "测试省_cg4",
+        )
+        self.assertEqual(
+            _tagged_shipping_address(
+                "测试省测试市测试路1号_旧采购编号", "cg4"
+            ),
+            "测试省测试市测试路1号_cg4",
        )

    def test_live_adapter_updates_address_and_returns_to_confirmation(self):
@@ -634,6 +730,105 @@ class U2PddPurchaseAdapterTest(unittest.TestCase):
        self.assertEqual(state.price_cent, 503)
        adapter.close()

+    def test_saved_address_panel_can_hide_full_address(self):
+        device = HiddenSavedAddressDevice()
+        adapter = self._live_adapter(device, [])
+        adapter.open_goods(GOODS_URL)
+        adapter.select_options(
+            {"color": "黑色", "size": "3XL【140-165斤】"}
+        )
+
+        adapter.update_shipping_address("cg2")
+
+        self.assertEqual(device.mode, "panel")
+        self.assertEqual(device.presses, [])
+        adapter.close()
+
+    def test_coloros_overlay_can_disappear_without_system_back(self):
+        device = AddressSaveOverlayDevice(clear_after_reads=2)
+        adapter = self._live_adapter(device, [])
+        adapter.open_goods(GOODS_URL)
+        adapter.select_options(
+            {"color": "黑色", "size": "3XL【140-165斤】"}
+        )
+
+        adapter.update_shipping_address("cg2")
+
+        self.assertEqual(device.save_clicks, 1)
+        self.assertEqual(device.presses, [])
+        self.assertEqual(device.mode, "panel")
+        adapter.close()
+
+    def test_persistent_coloros_overlay_is_dismissed_with_one_system_back(self):
+        device = AddressSaveOverlayDevice()
+        adapter = self._live_adapter(device, [])
+        adapter.open_goods(GOODS_URL)
+        adapter.select_options(
+            {"color": "黑色", "size": "3XL【140-165斤】"}
+        )
+
+        adapter.update_shipping_address("cg2")
+
+        self.assertEqual(device.save_clicks, 1)
+        self.assertEqual(device.presses, ["back"])
+        self.assertEqual(device.mode, "panel")
+        adapter.close()
+
+    def test_coloros_overlay_is_never_dismissed_more_than_once(self):
+        clock = FakeClock()
+        device = PersistentAddressSaveOverlayDevice()
+        adapter = self._live_adapter(
+            device,
+            [],
+            sleeper=clock.sleep,
+            monotonic=clock.monotonic,
+            panel_timeout=1.0,
+        )
+        adapter.open_goods(GOODS_URL)
+        adapter.select_options(
+            {"color": "黑色", "size": "3XL【140-165斤】"}
+        )
+
+        with self.assertRaises(PddPurchaseError) as raised:
+            adapter.update_shipping_address("cg2")
+
+        self.assertEqual(raised.exception.code, "PURCHASE_ADDRESS_PAGE_TIMEOUT")
+        self.assertTrue(
+            raised.exception.diagnostics["coloros_password_overlay_seen"]
+        )
+        self.assertEqual(raised.exception.diagnostics["system_back_attempts"], 1)
+        self.assertEqual(device.save_clicks, 1)
+        self.assertEqual(device.presses, ["back"])
+        adapter.close()
+
+    def test_unknown_page_after_save_does_not_press_back_or_click_again(self):
+        clock = FakeClock()
+        device = UnknownAfterAddressSaveDevice()
+        adapter = self._live_adapter(
+            device,
+            [],
+            sleeper=clock.sleep,
+            monotonic=clock.monotonic,
+            panel_timeout=1.0,
+        )
+        adapter.open_goods(GOODS_URL)
+        adapter.select_options(
+            {"color": "黑色", "size": "3XL【140-165斤】"}
+        )
+
+        with self.assertRaises(PddPurchaseError) as raised:
+            adapter.update_shipping_address("cg2")
+
+        self.assertEqual(raised.exception.code, "PURCHASE_ADDRESS_PAGE_TIMEOUT")
+        self.assertEqual(raised.exception.diagnostics["observed_page"], "unknown")
+        self.assertFalse(
+            raised.exception.diagnostics["coloros_password_overlay_seen"]
+        )
+        self.assertEqual(raised.exception.diagnostics["system_back_attempts"], 0)
+        self.assertEqual(device.save_clicks, 1)
+        self.assertEqual(device.presses, [])
+        adapter.close()
+
    def test_address_edit_retries_once_when_first_modify_click_is_lost(self):
        device = LostFirstModifyClickAddressDevice()
        adapter = self._live_adapter(device, [])