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

fix: 等待多地址列表稳定加载 (#189)

Browse Source
This commit is contained in:
chengma
2026-08-12 17:08:03 +08:00
parent 3f5ee8303d
commit 41d1b1383f
2 changed files with 219 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/pdd_u2_purchase_adapter.py
+72 -40
View File
@@ -233,8 +233,10 @@ def _pdd_text_targets(root: ET.Element, exact_text: str) -> list[Bounds]:
return sorted(targets)
def _address_modify_targets(root: ET.Element) -> list[Bounds]:
"""多地址时只返回唯一“已设默认”卡片内的修改按钮。"""
def _address_modify_target_info(
root: ET.Element,
) -> tuple[list[Bounds], dict[str, int]]:
"""返回默认地址修改目标及不含地址原文的结构诊断。"""
address_cards = [
card
@@ -243,15 +245,37 @@ def _address_modify_targets(root: ET.Element) -> list[Bounds]:
for card in list(recycler)
if _pdd_text_targets(card, "修改")
]
all_targets = _pdd_text_targets(root, "修改")
if len(address_cards) <= 1:
return _pdd_text_targets(root, "修改")
targets = all_targets
default_card_count = sum(
1 for card in address_cards if "已设默认" in _labels(card)
)
return targets, {
"address_card_count": len(address_cards),
"default_card_count": default_card_count,
"all_modify_target_count": len(all_targets),
}
default_cards = [
card for card in address_cards if "已设默认" in _labels(card)
]
if len(default_cards) != 1:
return []
return _pdd_text_targets(default_cards[0], "修改")
targets = []
else:
targets = _pdd_text_targets(default_cards[0], "修改")
return targets, {
"address_card_count": len(address_cards),
"default_card_count": len(default_cards),
"all_modify_target_count": len(all_targets),
}
def _address_modify_targets(root: ET.Element) -> list[Bounds]:
"""多地址时只返回唯一“已设默认”卡片内的修改按钮。"""
targets, _diagnostics = _address_modify_target_info(root)
return targets
def _pdd_title_back_targets(root: ET.Element) -> list[Bounds]:
@@ -1444,6 +1468,11 @@ class U2PddLivePurchaseAdapter(U2PddPurchaseAdapter, PddLivePurchaseAdapter):
root = panel_root
observed_page = _address_page_kind(root)
candidate_count = 0
target_diagnostics = {
"address_card_count": 0,
"default_card_count": 0,
"all_modify_target_count": 0,
}
stable_target: Optional[Bounds] = None
stable_reads = 0
first_clicked_target: Optional[Bounds] = None
@@ -1456,45 +1485,37 @@ class U2PddLivePurchaseAdapter(U2PddPurchaseAdapter, PddLivePurchaseAdapter):
return root
if observed_page == "panel":
targets = _address_modify_targets(root)
targets, target_diagnostics = _address_modify_target_info(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",
"收货地址页没有唯一可靠的修改按钮",
current_target = targets[0]
retry_target_changed = (
first_clicked_target is not None
and current_target != first_clicked_target
)
click_attempts += 1
if first_clicked_target is None:
first_clicked_target = current_target
stable_target = None
stable_reads = 0
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
@@ -1509,13 +1530,24 @@ class U2PddLivePurchaseAdapter(U2PddPurchaseAdapter, PddLivePurchaseAdapter):
"observed_page": observed_page,
"click_attempts": click_attempts,
"candidate_count": candidate_count,
"stable_reads": stable_reads,
**target_diagnostics,
}
artifact = self._save_last_xml("purchase-address-edit-timeout")
if artifact is not None:
diagnostics["artifacts"] = [artifact]
ambiguous = observed_page == "panel" and candidate_count != 1
raise PddPurchaseError(
"PURCHASE_ADDRESS_PAGE_TIMEOUT",
"收货地址页面切换或保存确认超时，未进入不可逆下单阶段",
(
"PURCHASE_ADDRESS_EDIT_AMBIGUOUS"
if ambiguous
else "PURCHASE_ADDRESS_PAGE_TIMEOUT"
),
(
"收货地址页没有唯一可靠的修改按钮"
if ambiguous
else "收货地址页面切换或保存确认超时，未进入不可逆下单阶段"
),
step="purchase_update_shipping_address",
diagnostics=diagnostics,
)
client/test/test_pdd_u2_purchase_adapter.py
+147 -3
View File
@@ -460,6 +460,50 @@ class MultipleAddressFlowDevice(AddressFlowDevice):
super().click(x, y)
class DelayedMultipleAddressFlowDevice(MultipleAddressFlowDevice):
"""模拟地址标题先出现，地址列表稍后才进入控件树。"""
def __init__(self, loading_reads=2) -> None:
super().__init__(default_indexes=(0,))
self.loading_reads = loading_reads
self.address_panel_reads = 0
def dump_hierarchy(self):
if self.mode == "address_panel":
self.address_panel_reads += 1
if self.address_panel_reads <= self.loading_reads:
return """<hierarchy>
<node package="com.xunmeng.pinduoduo"
bounds="[0,0][1080,2376]">
<node package="com.xunmeng.pinduoduo" clickable="true"
bounds="[0,102][217,226]">
<node package="com.xunmeng.pinduoduo" text="返回"
bounds="[38,132][98,195]"/>
</node>
<node package="com.xunmeng.pinduoduo" text="收货地址"
bounds="[418,102][602,226]"/>
</node>
</hierarchy>"""
return super().dump_hierarchy()
class AlternatingDefaultAddressDevice(MultipleAddressFlowDevice):
"""模拟默认地址目标坐标持续变化。"""
def __init__(self) -> None:
super().__init__(default_indexes=(0,))
self.address_panel_reads = 0
def dump_hierarchy(self):
if self.mode == "address_panel":
self.address_panel_reads += 1
indexes = (self.address_panel_reads % 2,)
return multiple_address_panel_xml(
self.address, default_indexes=indexes
)
return super().dump_hierarchy()
class AmbiguousDefaultAddressDevice(MultipleAddressFlowDevice):
"""模拟默认地址卡片内部出现两个修改目标。"""
@@ -864,6 +908,54 @@ class U2PddPurchaseAdapterTest(unittest.TestCase):
self.assertEqual(device.mode, "panel")
adapter.close()
def test_address_list_can_load_after_panel_title(self):
device = DelayedMultipleAddressFlowDevice(loading_reads=2)
adapter = self._live_adapter(device, [])
adapter.open_goods(GOODS_URL)
adapter.select_options(
{"color": "黑色", "size": "3XL【140-165斤】"}
)
adapter.update_shipping_address("cg11")
self.assertGreaterEqual(device.address_panel_reads, 4)
self.assertEqual(device.address_modify_clicks, [(948, 571)])
self.assertEqual(device.mode, "panel")
adapter.close()
def test_address_list_missing_until_timeout_has_structural_diagnostics(self):
clock = FakeClock()
device = DelayedMultipleAddressFlowDevice(loading_reads=100)
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("cg11")
diagnostics = raised.exception.diagnostics
self.assertEqual(
raised.exception.code, "PURCHASE_ADDRESS_EDIT_AMBIGUOUS"
)
self.assertEqual(diagnostics["address_card_count"], 0)
self.assertEqual(diagnostics["default_card_count"], 0)
self.assertEqual(diagnostics["candidate_count"], 0)
self.assertEqual(diagnostics["stable_reads"], 0)
self.assertEqual(device.address_modify_clicks, [])
artifact = diagnostics["artifacts"][0]
self.assertTrue(Path(artifact["path"]).is_file())
adapter.close()
def test_default_address_can_be_second_card(self):
device = MultipleAddressFlowDevice(default_indexes=(1,))
adapter = self._live_adapter(device, [])
@@ -885,7 +977,14 @@ class U2PddPurchaseAdapterTest(unittest.TestCase):
device = MultipleAddressFlowDevice(
default_indexes=default_indexes
)
adapter = self._live_adapter(device, [])
clock = FakeClock()
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斤】"}
@@ -900,11 +999,25 @@ class U2PddPurchaseAdapterTest(unittest.TestCase):
)
self.assertEqual(device.address_modify_clicks, [])
self.assertEqual(device.address_back_clicks, [])
self.assertEqual(
raised.exception.diagnostics["address_card_count"], 2
)
self.assertEqual(
raised.exception.diagnostics["default_card_count"],
len(default_indexes),
)
adapter.close()
def test_default_address_with_multiple_modify_targets_stops_safely(self):
device = AmbiguousDefaultAddressDevice()
adapter = self._live_adapter(device, [])
clock = FakeClock()
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斤】"}
@@ -920,6 +1033,30 @@ class U2PddPurchaseAdapterTest(unittest.TestCase):
self.assertEqual(device.address_modify_clicks, [])
adapter.close()
def test_changing_default_target_never_clicks_until_timeout(self):
clock = FakeClock()
device = AlternatingDefaultAddressDevice()
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("cg11")
self.assertEqual(raised.exception.code, "PURCHASE_ADDRESS_PAGE_TIMEOUT")
self.assertEqual(raised.exception.diagnostics["candidate_count"], 1)
self.assertLessEqual(raised.exception.diagnostics["stable_reads"], 1)
self.assertEqual(device.address_modify_clicks, [])
adapter.close()
def test_saved_address_panel_can_hide_full_address(self):
device = HiddenSavedAddressDevice()
adapter = self._live_adapter(device, [])
@@ -1095,8 +1232,15 @@ class U2PddPurchaseAdapterTest(unittest.TestCase):
adapter.close()
def test_address_edit_does_not_retry_when_modify_target_is_ambiguous(self):
clock = FakeClock()
device = AmbiguousAfterModifyAddressDevice()
adapter = self._live_adapter(device, [])
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斤】"}