diff --git a/client/src/pdd_u2_purchase_adapter.py b/client/src/pdd_u2_purchase_adapter.py index b613727..307bc28 100644 --- a/client/src/pdd_u2_purchase_adapter.py +++ b/client/src/pdd_u2_purchase_adapter.py @@ -245,6 +245,21 @@ def _shipping_address_editor(root: ET.Element) -> Optional[tuple[str, Bounds]]: return matches[0] if len(matches) == 1 and matches[0][0].strip() else None +def _address_page_kind(root: ET.Element) -> str: + """识别地址面板、编辑页和返回后的规格确认页。""" + + labels = _labels(root) + if any("修改收货地址" in label.replace(" ", "") for label in labels): + return "edit" + if _shipping_address_editor(root) is not None: + return "edit" + if "收货地址" in labels: + return "panel" + if _page_kind(root, "") == "order_confirmation": + return "confirmation" + return "unknown" + + def _tagged_shipping_address(address: str, purchase_number: str) -> str: """只替换程序生成的末尾标记，保留完整真实地址主体。""" @@ -1247,14 +1262,7 @@ class U2PddLivePurchaseAdapter(U2PddPurchaseAdapter, PddLivePurchaseAdapter): ) panel_root = self._wait_for_address_page("panel") - modify_targets = _clickable_text_targets(panel_root, "修改") - self._click_unique_address_target( - modify_targets, - "PURCHASE_ADDRESS_EDIT_AMBIGUOUS", - "收货地址页没有唯一可靠的修改按钮", - ) - - edit_root = self._wait_for_address_page("edit") + edit_root = self._open_address_edit_page(panel_root) editor_data = _shipping_address_editor(edit_root) if editor_data is None: raise PddPurchaseError( @@ -1335,16 +1343,101 @@ class U2PddLivePurchaseAdapter(U2PddPurchaseAdapter, PddLivePurchaseAdapter): (left + right) // 2, (top + bottom) // 2 ) + def _open_address_edit_page(self, panel_root: ET.Element) -> ET.Element: + """稳定识别“修改”按钮；首次点击丢失时最多安全重试一次。""" + + deadline = self._monotonic() + self._panel_timeout + root = panel_root + observed_page = _address_page_kind(root) + candidate_count = 0 + stable_target: Optional[Bounds] = None + stable_reads = 0 + first_clicked_target: Optional[Bounds] = None + click_attempts = 0 + + while self._monotonic() < deadline: + self._check_cancelled("purchase_update_shipping_address") + observed_page = _address_page_kind(root) + if observed_page == "edit": + return root + + if observed_page == "panel": + targets = _clickable_text_targets(root, "修改") + candidate_count = len(targets) + if candidate_count != 1: + raise PddPurchaseError( + "PURCHASE_ADDRESS_EDIT_AMBIGUOUS", + "收货地址页没有唯一可靠的修改按钮", + step="purchase_update_shipping_address", + diagnostics={ + "candidate_count": candidate_count, + "click_attempts": click_attempts, + "observed_page": observed_page, + }, + ) + + current_target = targets[0] + retry_target_changed = ( + first_clicked_target is not None + and current_target != first_clicked_target + ) + if retry_target_changed or click_attempts >= 2: + stable_target = None + stable_reads = 0 + elif current_target == stable_target: + stable_reads += 1 + else: + stable_target = current_target + stable_reads = 1 + + if stable_reads >= 2: + self._click_unique_address_target( + targets, + "PURCHASE_ADDRESS_EDIT_AMBIGUOUS", + "收货地址页没有唯一可靠的修改按钮", + ) + click_attempts += 1 + if first_clicked_target is None: + first_clicked_target = current_target + stable_target = None + stable_reads = 0 + else: + # 页面状态不明确时不能继续点击，避免误触其他页面。 + candidate_count = 0 + stable_target = None + stable_reads = 0 + + self._sleep(0.2) + root = _parse_xml(self._dump_hierarchy()) + + diagnostics: dict[str, Any] = { + "expected_page": "edit", + "observed_page": observed_page, + "click_attempts": click_attempts, + "candidate_count": candidate_count, + } + artifact = self._save_last_xml("purchase-address-edit-timeout") + if artifact is not None: + diagnostics["artifacts"] = [artifact] + raise PddPurchaseError( + "PURCHASE_ADDRESS_PAGE_TIMEOUT", + "收货地址页面切换或保存确认超时，未进入不可逆下单阶段", + step="purchase_update_shipping_address", + diagnostics=diagnostics, + ) + def _wait_for_address_page( self, kind: str, *, expected_address: str = "" ) -> ET.Element: """等待地址页面切换；诊断信息不包含任何地址文字。""" deadline = self._monotonic() + self._panel_timeout + observed_page = "unknown" while self._monotonic() < deadline: self._check_cancelled("purchase_update_shipping_address") root = _parse_xml(self._dump_hierarchy()) labels = _labels(root) + observed_page = _address_page_kind(root) if kind == "panel" and "收货地址" in labels: return root if kind == "edit" and any( @@ -1365,11 +1458,18 @@ class U2PddLivePurchaseAdapter(U2PddPurchaseAdapter, PddLivePurchaseAdapter): ): return root self._sleep(0.2) + diagnostics: dict[str, Any] = { + "expected_page": kind, + "observed_page": observed_page, + } + artifact = self._save_last_xml(f"purchase-address-{kind}-timeout") + if artifact is not None: + diagnostics["artifacts"] = [artifact] raise PddPurchaseError( "PURCHASE_ADDRESS_PAGE_TIMEOUT", "收货地址页面切换或保存确认超时，未进入不可逆下单阶段", step="purchase_update_shipping_address", - diagnostics={"expected_page": kind}, + diagnostics=diagnostics, ) def submit_order_once(self) -> None: diff --git a/client/test/test_pdd_u2_purchase_adapter.py b/client/test/test_pdd_u2_purchase_adapter.py index 4837081..d990f78 100644 --- a/client/test/test_pdd_u2_purchase_adapter.py +++ b/client/test/test_pdd_u2_purchase_adapter.py @@ -363,6 +363,88 @@ class AddressFlowDevice(FakeDevice): return super().__call__(**selector) +class LostFirstModifyClickAddressDevice(AddressFlowDevice): + """模拟第一次“修改”点击被页面吞掉。""" + + def __init__(self) -> None: + super().__init__() + self.modify_clicks = 0 + + def click(self, x, y): + if self.mode != "address_panel": + return super().click(x, y) + self.clicks.append((x, y)) + self.modify_clicks += 1 + if self.modify_clicks >= 2: + self.mode = "address_edit" + + +class NeverOpenAddressEditDevice(AddressFlowDevice): + """模拟两次“修改”点击都未打开编辑页。""" + + def __init__(self) -> None: + super().__init__() + self.modify_clicks = 0 + + def click(self, x, y): + if self.mode != "address_panel": + return super().click(x, y) + self.clicks.append((x, y)) + self.modify_clicks += 1 + + +class UnknownAfterModifyAddressDevice(AddressFlowDevice): + """模拟点击“修改”后进入无法识别的中间页面。""" + + def __init__(self) -> None: + super().__init__() + self.modify_clicks = 0 + + def dump_hierarchy(self): + if self.mode == "address_unknown": + return """ + + """ + return super().dump_hierarchy() + + def click(self, x, y): + if self.mode != "address_panel": + return super().click(x, y) + self.clicks.append((x, y)) + self.modify_clicks += 1 + self.mode = "address_unknown" + + +class AmbiguousAfterModifyAddressDevice(AddressFlowDevice): + """模拟点击后地址面板同时出现两个“修改”按钮。""" + + def __init__(self) -> None: + super().__init__() + self.modify_clicks = 0 + + def dump_hierarchy(self): + if self.mode == "address_ambiguous": + xml_data = address_panel_xml(self.address) + return xml_data.replace( + "\n \n ", + """ + + + + + """, + ) + return super().dump_hierarchy() + + def click(self, x, y): + if self.mode != "address_panel": + return super().click(x, y) + self.clicks.append((x, y)) + self.modify_clicks += 1 + self.mode = "address_ambiguous" + + class U2PddPurchaseAdapterTest(unittest.TestCase): def _adapter(self, device, calls): def select_color_fn(_device, _xml, target, **_kwargs): @@ -381,7 +463,7 @@ class U2PddPurchaseAdapterTest(unittest.TestCase): select_size_fn=select_size_fn, ) - def _live_adapter(self, device, calls): + def _live_adapter(self, device, calls, **adapter_options): def select_color_fn(_device, _xml, target, **_kwargs): calls.append(("color", target)) return target == "黑色" @@ -390,12 +472,16 @@ class U2PddPurchaseAdapterTest(unittest.TestCase): calls.append(("size", target)) return target == "3XL【140-165斤】" + options = { + "sleeper": lambda _seconds: None, + "select_color_fn": select_color_fn, + "select_size_fn": select_size_fn, + } + options.update(adapter_options) return U2PddLivePurchaseAdapter( "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, + **options, ) def test_dry_run_reaches_confirmation_without_clicking_submit(self): @@ -548,6 +634,100 @@ class U2PddPurchaseAdapterTest(unittest.TestCase): self.assertEqual(state.price_cent, 503) adapter.close() + def test_address_edit_retries_once_when_first_modify_click_is_lost(self): + device = LostFirstModifyClickAddressDevice() + 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.modify_clicks, 2) + self.assertEqual(device.address, "测试省测试市测试区测试路1号_cg2") + self.assertEqual(device.mode, "panel") + adapter.close() + + def test_address_edit_stops_after_two_lost_modify_clicks(self): + clock = FakeClock() + device = NeverOpenAddressEditDevice() + original_address = device.address + with tempfile.TemporaryDirectory() as directory: + adapter = self._live_adapter( + device, + [], + sleeper=clock.sleep, + monotonic=clock.monotonic, + panel_timeout=1.0, + artifact_directory=Path(directory), + ) + 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" + ) + diagnostics = raised.exception.diagnostics + self.assertEqual(diagnostics["expected_page"], "edit") + self.assertEqual(diagnostics["observed_page"], "panel") + self.assertEqual(diagnostics["click_attempts"], 2) + self.assertEqual(diagnostics["candidate_count"], 1) + self.assertEqual(device.modify_clicks, 2) + self.assertEqual(device.address, original_address) + artifact = diagnostics["artifacts"][0] + content = Path(artifact["path"]).read_text(encoding="utf-8") + self.assertNotIn(original_address, content) + adapter.close() + + def test_address_edit_does_not_retry_on_unknown_page(self): + clock = FakeClock() + device = UnknownAfterModifyAddressDevice() + 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.assertEqual(raised.exception.diagnostics["click_attempts"], 1) + self.assertEqual(device.modify_clicks, 1) + adapter.close() + + def test_address_edit_does_not_retry_when_modify_target_is_ambiguous(self): + device = AmbiguousAfterModifyAddressDevice() + adapter = self._live_adapter(device, []) + 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_EDIT_AMBIGUOUS" + ) + self.assertEqual(raised.exception.diagnostics["candidate_count"], 2) + self.assertEqual(raised.exception.diagnostics["click_attempts"], 1) + self.assertEqual(device.modify_clicks, 1) + adapter.close() + def test_reliable_pdd_tree_skips_repeated_app_current(self): device = FakeDevice() adapter = self._adapter(device, [])