chengma/cmautobuy
2
0
Fork 0
Code Issues 89 Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 稳定进入收货地址编辑页 (#179)

Browse Source
This commit is contained in:
chengma
2026-08-12 11:01:27 +08:00
parent 944d2cd262
commit 7c569813e1
2 changed files with 293 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/pdd_u2_purchase_adapter.py
+109 -9
View File
@@ -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:
client/test/test_pdd_u2_purchase_adapter.py
+184 -4
View File
@@ -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 """<hierarchy>
<node package="com.xunmeng.pinduoduo" text="其他页面"
bounds="[0,0][1080,2376]"/>
</hierarchy>"""
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(
"</node>\n </node>\n </hierarchy>",
"""</node>
<node clickable="true" enabled="true" bounds="[700,600][850,700]">
<node text="修改" bounds="[730,620][820,680]"/>
</node>
</node>
</hierarchy>""",
)
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, [])