diff --git a/client/src/pdd_u2_purchase_adapter.py b/client/src/pdd_u2_purchase_adapter.py index f793c0a..02b884f 100644 --- a/client/src/pdd_u2_purchase_adapter.py +++ b/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: diff --git a/client/test/test_pdd_u2_purchase_adapter.py b/client/test/test_pdd_u2_purchase_adapter.py index 7abb56e..aa275ef 100644 --- a/client/test/test_pdd_u2_purchase_adapter.py +++ b/client/test/test_pdd_u2_purchase_adapter.py @@ -544,6 +544,19 @@ def address_confirmation_xml(address: str) -> str: """ +def clipped_address_confirmation_xml(address: str) -> str: + """首次地址卡片被顶部裁剪时的脱敏控件树。""" + + return address_confirmation_xml(address).replace( + '', + "", + ).replace( + '', + '', + ) + + def address_panel_xml(address: str) -> str: return f""" @@ -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, [])