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

fix: 记录规格面板误超时诊断 (#115)

Browse Source
This commit is contained in:
chengma
2026-08-10 18:38:56 +08:00
parent 6530b11125
commit 24f4ad1ba9
2 changed files with 196 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/pdd_u2_purchase_adapter.py
+95 -11
View File
@@ -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",
)
client/test/test_pdd_u2_purchase_adapter.py
+101 -2
View File
@@ -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 """<hierarchy>
<node package="com.xunmeng.pinduoduo" bounds="[0,0][1080,2376]">
<node clickable="true" enabled="true" visible-to-user="true"
<node package="com.xunmeng.pinduoduo" clickable="true"
enabled="true" visible-to-user="true"
bounds="[500,2100][1080,2328]">
<node text="立即购买 ¥5.03" enabled="true" visible-to-user="true"
<node package="com.xunmeng.pinduoduo" text="立即购买 ¥5.03"
enabled="true" visible-to-user="true"
bounds="[600,2150][1000,2280]"/>
</node>
</node>
@@ -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, [])