diff --git a/client/src/pdd_u2_purchase_adapter.py b/client/src/pdd_u2_purchase_adapter.py index 3aaf97e..42964ac 100644 --- a/client/src/pdd_u2_purchase_adapter.py +++ b/client/src/pdd_u2_purchase_adapter.py @@ -6,14 +6,17 @@ from __future__ import annotations +import hashlib import re import time import xml.etree.ElementTree as ET from contextlib import nullcontext from decimal import Decimal, InvalidOperation, ROUND_HALF_UP +from pathlib import Path from typing import Any, Callable, Mapping, Optional from urllib.parse import parse_qs, urlparse +from .db import data_dir from .pdd_device_service import ( PDD_PACKAGE_NAME, PddDeviceError, @@ -58,6 +61,19 @@ _PAYMENT_MARKERS = ("输入支付密码", "立即支付", "支付成功", "支 _FINAL_SUBMIT_MARKERS = ("提交订单", "现在买，仅", "确认购买") _OUT_OF_STOCK_MARKERS = ("已售罄", "暂时缺货", "库存不足", "该商品已售罄") _SUPPORTED_OPTION_KEYS = frozenset({"color", "size"}) +_DIAGNOSTIC_MARKERS = ( + "增加数量", + "减少数量", + "提交订单", + "现在买", + "确认购买", + "颜色分类", + "尺码", + "安全验证", + "手机号登录", + "操作频繁", + "网络不给力", +) def _parse_xml(xml_data: str | bytes) -> ET.Element: @@ -83,6 +99,20 @@ def _labels(root: ET.Element) -> list[str]: return [value for node in root.iter("node") if (value := _label(node))] +def _sanitize_diagnostic_xml(xml_data: str) -> str: + """只保留页面判断所需语义，删除商品、账号和收货相关文字。""" + + root = ET.fromstring(xml_data) + for node in root.iter("node"): + for key in ("text", "content-desc", "hint"): + value = node.get(key, "").strip() + if not value: + continue + markers = [marker for marker in _DIAGNOSTIC_MARKERS if marker in value] + node.set(key, " ".join(markers) if markers else "[已脱敏]") + return ET.tostring(root, encoding="unicode") + + def _parse_bounds(value: str) -> Optional[Bounds]: match = _BOUNDS_PATTERN.fullmatch((value or "").strip()) if not match: @@ -248,6 +278,7 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter): panel_timeout: float = 10.0, select_color_fn: Callable[..., bool] = select_color, select_size_fn: Callable[..., bool] = select_size, + artifact_directory: Optional[Path] = None, ) -> None: self._device_address = str(device_address or "").strip() self._device_service = device_service or PddDeviceService() @@ -258,6 +289,9 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter): self._panel_timeout = panel_timeout self._select_color = select_color_fn self._select_size = select_size_fn + self._artifact_directory = artifact_directory + self._last_xml: Optional[str] = None + self._hierarchy_reads = 0 self._session = None self._device = None self._goods_id = "" @@ -266,6 +300,8 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter): def open_goods(self, goods_url: str) -> None: self._goods_id = _goods_id_from_url(goods_url) + self._last_xml = None + self._hierarchy_reads = 0 self._check_cancelled("purchase_open_goods") try: self._session = self._device_service.connect(self._device_address) @@ -289,7 +325,12 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter): stage = trace.stage("open_url") if trace else nullcontext() with stage: self._device.open_url(goods_url) - self._wait_for_goods_page(goods_url, before_open) + package_hint = ( + PDD_PACKAGE_NAME + if current.get("package") == PDD_PACKAGE_NAME + else "" + ) + self._wait_for_goods_page(goods_url, before_open, package_hint) except PddPurchaseError: raise except PddDeviceError as exc: @@ -451,6 +492,7 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter): self, goods_url: str, before_open: Optional[PddPageObservation], + package_hint: str, ) -> str: """确认本次深链进入新商品页；稳定首页时只重开一次。""" @@ -466,7 +508,6 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter): while self._monotonic() < deadline: self._check_cancelled("purchase_open_goods") device = self._require_device() - current = device.app_current() if first_dump and trace is not None: with trace.stage("first_dump_hierarchy"): xml_data = self._dump_hierarchy() @@ -474,9 +515,9 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter): else: xml_data = self._dump_hierarchy() root = _parse_xml(xml_data) - observation = classify_pdd_page( - root, str(current.get("package") or "") - ) + # 不在轮询中调用 app_current()。某些设备会阻塞十秒并错误 + # 报告设置页；最新控件树中的 PDD 包节点才是可靠依据。 + observation = classify_pdd_page(root, package_hint) last_kind = observation.kind if trace is not None and last_kind != last_recorded: trace.record( @@ -564,33 +605,74 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter): def _wait_for_confirmation_panel(self) -> str: deadline = self._monotonic() + self._panel_timeout + last_kind = "unknown" + panel_hierarchy_reads = 0 while self._monotonic() < deadline: self._check_cancelled("purchase_select_options") xml_data = self._dump_hierarchy() + panel_hierarchy_reads += 1 root = _parse_xml(xml_data) - current = self._require_device().app_current() - kind = _page_kind(root, str(current.get("package") or "")) - if kind == "order_confirmation" and any( + last_kind = _page_kind(root, "") + if last_kind == "order_confirmation" and any( "增加数量" in label for label in _labels(root) ): return xml_data - if kind in {"captcha", "login_required", "risk_control", "payment"}: - self._raise_special_page(kind) + if last_kind in { + "captcha", + "login_required", + "risk_control", + "payment", + }: + self._raise_special_page(last_kind) self._sleep(0.2) + diagnostics: dict[str, Any] = { + "page_kind": last_kind, + "hierarchy_reads": self._hierarchy_reads, + "panel_hierarchy_reads": panel_hierarchy_reads, + } + artifact = self._save_last_xml("purchase-panel-timeout") + if artifact is not None: + diagnostics["artifacts"] = [artifact] raise PddPurchaseError( "PURCHASE_PANEL_TIMEOUT", "点击采购入口后没有到达可靠的提交前确认页", step="purchase_select_options", retryable=True, + diagnostics=diagnostics, ) def _dump_hierarchy(self) -> str: raw = self._require_device().dump_hierarchy() - return ( + xml_data = ( raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else str(raw) ) + self._last_xml = xml_data + self._hierarchy_reads += 1 + return xml_data + + def _save_last_xml(self, label: str) -> Optional[Mapping[str, Any]]: + """保存脱敏控件树；没有配置目录或写入失败时只返回空。""" + + if self._artifact_directory is None or not self._last_xml: + return None + try: + sanitized = _sanitize_diagnostic_xml(self._last_xml) + digest = hashlib.sha256(sanitized.encode("utf-8")).hexdigest() + directory = self._artifact_directory / "purchase" / self._goods_id + directory.mkdir(parents=True, exist_ok=True) + path = directory / f"{label}-{digest[:12]}.xml" + if not path.exists(): + path.write_text(sanitized, encoding="utf-8") + return { + "kind": "sanitized_accessibility_xml", + "path": str(path.resolve()), + "sha256": digest, + "hierarchy_reads": self._hierarchy_reads, + } + except (OSError, ET.ParseError): + return None def _require_device(self) -> Any: if self._device is None: @@ -650,6 +732,7 @@ def create_u2_purchase_adapter( device_address, device_service=current_thread_device_service() or PddDeviceService(), cancelled=cancelled, + artifact_directory=data_dir() / "artifacts", ) @@ -704,4 +787,5 @@ def create_u2_live_purchase_adapter( device_address, device_service=current_thread_device_service() or PddDeviceService(), cancelled=cancelled, + artifact_directory=data_dir() / "artifacts", ) diff --git a/client/test/test_pdd_u2_purchase_adapter.py b/client/test/test_pdd_u2_purchase_adapter.py index 5e919df..fe773f4 100644 --- a/client/test/test_pdd_u2_purchase_adapter.py +++ b/client/test/test_pdd_u2_purchase_adapter.py @@ -1,6 +1,7 @@ """uiautomator2 采购演练 Adapter 测试；不连接真实手机。""" from pathlib import Path +import tempfile import unittest from src.pdd_device_service import PddDeviceService @@ -27,9 +28,11 @@ class FakeClock: def home_xml() -> str: return """ - - @@ -73,9 +76,11 @@ class FakeDevice: self.clicks = [] self.opened_urls = [] self.app_wait_calls = 0 + self.app_current_calls = 0 self.has_opened = False def app_current(self): + self.app_current_calls += 1 return {"package": "com.xunmeng.pinduoduo"} def app_start(self, _package): @@ -132,6 +137,24 @@ class StaleGoodsDevice(FakeDevice): return home_xml() +class SlowSettingsFocusDevice(FakeDevice): + """模拟真机：前台查询慢，而且错误报告为设置页。""" + + def __init__(self, clock: FakeClock) -> None: + super().__init__() + self.clock = clock + + def app_current(self): + self.app_current_calls += 1 + self.clock.sleep(11.0) + return {"package": "com.android.settings"} + + +class PanelDoesNotOpenDevice(FakeDevice): + def click(self, x, y): + self.clicks.append((x, y)) + + class U2PddPurchaseAdapterTest(unittest.TestCase): def _adapter(self, device, calls): def select_color_fn(_device, _xml, target, **_kwargs): @@ -253,6 +276,29 @@ class U2PddPurchaseAdapterTest(unittest.TestCase): self.assertEqual(2, len(device.opened_urls)) self.assertEqual([], device.clicks) + def test_slow_wrong_focus_still_completes_home_reopen_state_machine(self): + clock = FakeClock() + device = SlowSettingsFocusDevice(clock) + device.mode = "special" + device.special_xml = ( + FIXTURES / "pdd_home_page.xml" + ).read_text(encoding="utf-8") + adapter = U2PddPurchaseAdapter( + "USB-001", + device_service=PddDeviceService(connector=lambda _serial: device), + sleeper=clock.sleep, + monotonic=clock.monotonic, + page_timeout=10.0, + ) + + with self.assertRaises(PddPurchaseError) as raised: + adapter.open_goods(GOODS_URL) + + self.assertEqual("PDD_GOODS_UNAVAILABLE", raised.exception.code) + self.assertEqual(1, device.app_current_calls) + self.assertEqual(2, len(device.opened_urls)) + adapter.close() + def test_stale_goods_page_is_not_accepted_as_target(self): device = StaleGoodsDevice() clock = FakeClock() @@ -271,6 +317,59 @@ class U2PddPurchaseAdapterTest(unittest.TestCase): self.assertEqual([], device.clicks) adapter.close() + def test_slow_wrong_focus_does_not_consume_panel_timeout(self): + clock = FakeClock() + device = SlowSettingsFocusDevice(clock) + adapter = U2PddPurchaseAdapter( + "USB-001", + device_service=PddDeviceService(connector=lambda _serial: device), + sleeper=clock.sleep, + monotonic=clock.monotonic, + panel_timeout=1.0, + select_color_fn=lambda *_args, **_kwargs: True, + select_size_fn=lambda *_args, **_kwargs: True, + ) + + adapter.open_goods(GOODS_URL) + adapter.select_options( + {"color": "黑色", "size": "3XL【140-165斤】"} + ) + + self.assertEqual(1, device.app_current_calls) + self.assertEqual(11.0, clock.now) + self.assertEqual(1, len(device.clicks)) + adapter.close() + + def test_panel_timeout_saves_sanitized_last_hierarchy(self): + clock = FakeClock() + device = PanelDoesNotOpenDevice() + with tempfile.TemporaryDirectory() as directory: + adapter = U2PddPurchaseAdapter( + "USB-001", + device_service=PddDeviceService( + connector=lambda _serial: device + ), + sleeper=clock.sleep, + monotonic=clock.monotonic, + panel_timeout=0.5, + artifact_directory=Path(directory), + ) + adapter.open_goods(GOODS_URL) + + with self.assertRaises(PddPurchaseError) as raised: + adapter.select_options({"color": "黑色"}) + + diagnostics = raised.exception.diagnostics + self.assertEqual("goods", diagnostics["page_kind"]) + self.assertGreaterEqual(diagnostics["hierarchy_reads"], 2) + self.assertGreaterEqual(diagnostics["panel_hierarchy_reads"], 1) + artifact = diagnostics["artifacts"][0] + content = Path(artifact["path"]).read_text(encoding="utf-8") + self.assertNotIn("5.03", content) + self.assertNotIn("立即购买", content) + self.assertIn("[已脱敏]", content) + adapter.close() + def test_invalid_non_pdd_url_is_rejected_before_connect(self): device = FakeDevice() adapter = self._adapter(device, [])