diff --git a/client/src/pdd_u2_purchase_adapter.py b/client/src/pdd_u2_purchase_adapter.py index 307bc28..8d2f1c8 100644 --- a/client/src/pdd_u2_purchase_adapter.py +++ b/client/src/pdd_u2_purchase_adapter.py @@ -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"(? 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( diff --git a/client/test/test_pdd_u2_purchase_adapter.py b/client/test/test_pdd_u2_purchase_adapter.py index d990f78..c0ecb13 100644 --- a/client/test/test_pdd_u2_purchase_adapter.py +++ b/client/test/test_pdd_u2_purchase_adapter.py @@ -314,6 +314,20 @@ def address_edit_xml(address: str) -> str: """ +def coloros_password_overlay_xml() -> str: + return """ + + + + + """ + + 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 """ + + """ + 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, [])