diff --git a/client/src/pdd_u2_purchase_adapter.py b/client/src/pdd_u2_purchase_adapter.py
new file mode 100644
index 0000000..bb12bff
--- /dev/null
+++ b/client/src/pdd_u2_purchase_adapter.py
@@ -0,0 +1,514 @@
+"""uiautomator2 采购演练 Adapter。
+
+本模块只到 PDD 最终提交订单按钮前。代码中没有点击提交订单
+或付款的方法。
+"""
+
+from __future__ import annotations
+
+import re
+import time
+import xml.etree.ElementTree as ET
+from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
+from typing import Any, Callable, Mapping, Optional
+from urllib.parse import parse_qs, urlparse
+
+from .pdd_device_service import (
+ PDD_PACKAGE_NAME,
+ PddDeviceError,
+ PddDeviceService,
+)
+from .pdd_purchase_adapter import (
+ PddPurchaseAdapter,
+ PddPurchaseError,
+ PurchasePageState,
+)
+from .util.get_size_panle_coord import get_size_panel_coord
+from .util.select_color_size import select_color, select_size
+
+
+Bounds = tuple[int, int, int, int]
+_BOUNDS_PATTERN = re.compile(r"^\[(\d+),(\d+)\]\[(\d+),(\d+)\]$")
+_PRICE_PATTERN = re.compile(r"[¥¥]\s*(\d+(?:\.\d{1,2})?)")
+_READY_MARKERS = ("发起拼单", "立即购买", "单独购买", "免拼购买", "快要抢光")
+_LOGIN_MARKERS = ("手机号登录", "登录后继续", "验证码登录", "账号登录")
+_CAPTCHA_MARKERS = ("请完成验证", "拖动滑块", "安全验证", "点击图中")
+_RISK_MARKERS = ("操作频繁", "异常请求", "风险提示", "账号异常")
+_PAYMENT_MARKERS = ("输入支付密码", "立即支付", "支付成功", "支付失败")
+_FINAL_SUBMIT_MARKERS = ("提交订单", "现在买,仅", "确认购买")
+_SUPPORTED_OPTION_KEYS = frozenset({"color", "size"})
+
+
+def _parse_xml(xml_data: str | bytes) -> ET.Element:
+ try:
+ return ET.fromstring(xml_data)
+ except (ET.ParseError, TypeError) as exc:
+ raise PddPurchaseError(
+ "PDD_DATA_XML_INVALID",
+ "PDD 返回的无障碍控件树不是有效 XML",
+ step="purchase_page_check",
+ ) from exc
+
+
+def _label(node: ET.Element) -> str:
+ return " ".join(
+ value.strip()
+ for value in (node.get("text", ""), node.get("content-desc", ""))
+ if value.strip()
+ )
+
+
+def _labels(root: ET.Element) -> list[str]:
+ return [value for node in root.iter("node") if (value := _label(node))]
+
+
+def _parse_bounds(value: str) -> Optional[Bounds]:
+ match = _BOUNDS_PATTERN.fullmatch((value or "").strip())
+ if not match:
+ return None
+ left, top, right, bottom = map(int, match.groups())
+ if right <= left or bottom <= top:
+ return None
+ return left, top, right, bottom
+
+
+def _goods_id_from_url(goods_url: str) -> str:
+ parsed = urlparse(goods_url)
+ host = (parsed.hostname or "").lower()
+ supported_host = (
+ host == "yangkeduo.com"
+ or host.endswith(".yangkeduo.com")
+ or host == "pinduoduo.com"
+ or host.endswith(".pinduoduo.com")
+ )
+ goods_id = parse_qs(parsed.query).get("goods_id", [""])[0].strip()
+ if parsed.scheme not in {"http", "https"} or not supported_host:
+ raise PddPurchaseError(
+ "PURCHASE_GOODS_URL_INVALID",
+ "采购商品链接不是受支持的 PDD 链接",
+ step="purchase_open_goods",
+ )
+ if not goods_id.isdigit():
+ raise PddPurchaseError(
+ "PURCHASE_GOODS_ID_MISSING",
+ "采购商品链接缺少有效 goods_id",
+ step="purchase_open_goods",
+ )
+ return goods_id
+
+
+def _page_kind(root: ET.Element, current_package: str) -> str:
+ combined = " ".join(_labels(root))
+ if any(marker in combined for marker in _CAPTCHA_MARKERS):
+ return "captcha"
+ if any(marker in combined for marker in _LOGIN_MARKERS):
+ return "login_required"
+ if any(marker in combined for marker in _RISK_MARKERS):
+ return "risk_control"
+ if any(marker in combined for marker in _PAYMENT_MARKERS):
+ return "payment"
+ if any(marker in combined for marker in _FINAL_SUBMIT_MARKERS):
+ return "order_confirmation"
+ if current_package == PDD_PACKAGE_NAME and any(
+ marker in combined for marker in _READY_MARKERS
+ ):
+ return "goods"
+ return "unknown"
+
+
+def _selected(root: ET.Element, target: str) -> bool:
+ """从最新控件树确认一个规格已选中。"""
+
+ checked_target = target.strip().casefold()
+ parents = {
+ child: parent
+ for parent in root.iter()
+ for child in parent
+ if child.tag == "node"
+ }
+ for node in root.iter("node"):
+ labels = {
+ node.get("text", "").strip().casefold(),
+ node.get("content-desc", "").strip().casefold(),
+ }
+ if checked_target not in labels:
+ continue
+ current: Optional[ET.Element] = node
+ while current is not None:
+ if (
+ current.get("selected") == "true"
+ or current.get("checked") == "true"
+ ):
+ return True
+ current = parents.get(current)
+
+ # 选中节点滑出屏幕后,PDD 仍会在顶部“已选”摘要中保留完整文字。
+ return any(
+ label.startswith("已选") and target.strip() in label
+ for label in _labels(root)
+ )
+
+
+def _quantity(root: ET.Element) -> int:
+ values = []
+ for node in root.iter("node"):
+ if node.get("class") != "android.widget.EditText":
+ continue
+ text = node.get("text", "").strip()
+ if text.isdigit() and int(text) > 0:
+ values.append(int(text))
+ return values[0] if len(values) == 1 else 0
+
+
+def _price_cent(root: ET.Element) -> int:
+ """优先读取页面上半部的当前单价,不把划线价当成当前价。"""
+
+ parsed_nodes: list[tuple[int, int, str]] = []
+ screen_bottom = 0
+ for node in root.iter("node"):
+ bounds = _parse_bounds(node.get("bounds", ""))
+ if bounds is not None:
+ screen_bottom = max(screen_bottom, bounds[3])
+ if screen_bottom <= 0:
+ return 0
+
+ for node in root.iter("node"):
+ label = _label(node)
+ match = _PRICE_PATTERN.search(label)
+ bounds = _parse_bounds(node.get("bounds", ""))
+ if match is None or bounds is None:
+ continue
+ center_y = (bounds[1] + bounds[3]) // 2
+ if center_y >= screen_bottom * 0.75:
+ continue
+ # “折后/到手/快卖完”比单独的划线价更可信。
+ semantic_score = 1 if any(
+ word in label for word in ("折后", "到手", "快卖完", "现价")
+ ) else 0
+ try:
+ cents = int(
+ (Decimal(match.group(1)) * 100).quantize(
+ Decimal("1"), rounding=ROUND_HALF_UP
+ )
+ )
+ except (InvalidOperation, ValueError):
+ continue
+ if cents > 0:
+ parsed_nodes.append((semantic_score, -center_y, cents))
+ if not parsed_nodes:
+ return 0
+ return max(parsed_nodes)[2]
+
+
+class U2PddPurchaseAdapter(PddPurchaseAdapter):
+ """PDD 真机采购演练会话,不提供真实下单能力。"""
+
+ def __init__(
+ self,
+ device_address: str,
+ *,
+ device_service: Optional[PddDeviceService] = None,
+ cancelled: Callable[[], bool] = lambda: False,
+ sleeper: Callable[[float], None] = time.sleep,
+ monotonic: Callable[[], float] = time.monotonic,
+ page_timeout: float = 30.0,
+ panel_timeout: float = 10.0,
+ select_color_fn: Callable[..., bool] = select_color,
+ select_size_fn: Callable[..., bool] = select_size,
+ ) -> None:
+ self._device_address = str(device_address or "").strip()
+ self._device_service = device_service or PddDeviceService()
+ self._cancelled = cancelled
+ self._sleep = sleeper
+ self._monotonic = monotonic
+ self._page_timeout = page_timeout
+ self._panel_timeout = panel_timeout
+ self._select_color = select_color_fn
+ self._select_size = select_size_fn
+ self._session = None
+ self._device = None
+ self._goods_id = ""
+ self._requested_options: dict[str, str] = {}
+
+ def open_goods(self, goods_url: str) -> None:
+ self._goods_id = _goods_id_from_url(goods_url)
+ self._check_cancelled("purchase_open_goods")
+ try:
+ self._session = self._device_service.connect(self._device_address)
+ self._device = self._session.__enter__()
+ current = self._device.app_current()
+ if current.get("package") != PDD_PACKAGE_NAME:
+ self._device.app_start(PDD_PACKAGE_NAME)
+ if not self._device.app_wait(PDD_PACKAGE_NAME, timeout=10):
+ raise PddPurchaseError(
+ "DEVICE_APP_START_FAILED",
+ "PDD 应用启动失败",
+ step="purchase_open_goods",
+ retryable=True,
+ )
+ self._device.open_url(goods_url)
+ self._wait_for_page("goods", self._page_timeout)
+ except PddPurchaseError:
+ raise
+ except PddDeviceError as exc:
+ raise PddPurchaseError(
+ exc.code,
+ exc.message,
+ step="purchase_open_goods",
+ retryable=True,
+ ) from exc
+ except Exception as exc:
+ self._raise_device_or_page_error(exc, "purchase_open_goods")
+
+ def read_state(self) -> PurchasePageState:
+ device = self._require_device()
+ self._check_cancelled("purchase_page_check")
+ try:
+ current = device.app_current()
+ xml_data = self._dump_hierarchy()
+ root = _parse_xml(xml_data)
+ kind = _page_kind(root, str(current.get("package") or ""))
+ selected = {
+ key: value
+ for key, value in self._requested_options.items()
+ if _selected(root, value)
+ }
+ return PurchasePageState(
+ page_kind=kind,
+ goods_id=self._goods_id,
+ selected_options=selected,
+ quantity=_quantity(root),
+ price_cent=_price_cent(root),
+ candidate_count=1 if kind != "unknown" else 0,
+ )
+ except PddPurchaseError:
+ raise
+ except Exception as exc:
+ self._raise_device_or_page_error(exc, "purchase_page_check")
+
+ def select_options(self, options: Mapping[str, str]) -> None:
+ device = self._require_device()
+ checked = {
+ str(key).strip(): str(value).strip()
+ for key, value in options.items()
+ }
+ unsupported = sorted(set(checked) - _SUPPORTED_OPTION_KEYS)
+ if unsupported or not checked or any(not value for value in checked.values()):
+ names = "、".join(unsupported) if unsupported else "空规格"
+ raise PddPurchaseError(
+ "PURCHASE_OPTIONS_UNSUPPORTED",
+ f"当前真机 Adapter 不支持规格:{names}",
+ step="purchase_select_options",
+ )
+ self._requested_options = checked
+ self._check_cancelled("purchase_select_options")
+ try:
+ home_xml = self._dump_hierarchy()
+ coord = get_size_panel_coord(home_xml)
+ if coord is None:
+ raise PddPurchaseError(
+ "PURCHASE_PANEL_ENTRY_MISSING",
+ "商品页没有可靠的采购规格入口",
+ step="purchase_select_options",
+ )
+ device.click(*coord)
+ panel_xml = self._wait_for_confirmation_panel()
+
+ color = checked.get("color")
+ if color and not self._select_color(
+ device, panel_xml, color, action_delay=0.2
+ ):
+ raise PddPurchaseError(
+ "PURCHASE_OPTIONS_MISMATCH",
+ f"没有精确选中颜色:{color}",
+ step="purchase_select_options",
+ )
+ size = checked.get("size")
+ if size:
+ latest_xml = self._dump_hierarchy()
+ if not self._select_size(
+ device, latest_xml, size, action_delay=0.2
+ ):
+ raise PddPurchaseError(
+ "PURCHASE_OPTIONS_MISMATCH",
+ f"没有精确选中尺码:{size}",
+ step="purchase_select_options",
+ )
+ except PddPurchaseError:
+ raise
+ except Exception as exc:
+ self._raise_device_or_page_error(exc, "purchase_select_options")
+
+ def set_quantity(self, quantity: int) -> None:
+ device = self._require_device()
+ if isinstance(quantity, bool) or not isinstance(quantity, int) or quantity <= 0:
+ raise PddPurchaseError(
+ "PURCHASE_QUANTITY_INVALID",
+ "采购数量必须是大于 0 的整数",
+ step="purchase_set_quantity",
+ )
+ self._check_cancelled("purchase_set_quantity")
+ try:
+ editor = device(className="android.widget.EditText")
+ if int(getattr(editor, "count", 0)) != 1:
+ raise PddPurchaseError(
+ "PURCHASE_QUANTITY_CONTROL_MISSING",
+ "页面没有唯一可靠的采购数量输入框",
+ step="purchase_set_quantity",
+ )
+ editor.set_text(str(quantity))
+ self._sleep(0.2)
+ if _quantity(_parse_xml(self._dump_hierarchy())) != quantity:
+ raise PddPurchaseError(
+ "PURCHASE_QUANTITY_MISMATCH",
+ "PDD 页面没有保存目标采购数量",
+ step="purchase_set_quantity",
+ )
+ except PddPurchaseError:
+ raise
+ except Exception as exc:
+ self._raise_device_or_page_error(exc, "purchase_set_quantity")
+
+ def enter_confirmation(self) -> None:
+ """只确认已到最终提交前页面,不点击底部提交按钮。"""
+
+ state = self.read_state()
+ if state.page_kind != "order_confirmation":
+ raise PddPurchaseError(
+ "PURCHASE_CONFIRMATION_NOT_REACHED",
+ "当前没有到达最终提交前确认页",
+ step="purchase_enter_confirmation",
+ )
+
+ def stop_before_submit(self) -> None:
+ """再次确认最终提交按钮存在,然后不做任何点击。"""
+
+ state = self.read_state()
+ if state.page_kind != "order_confirmation":
+ raise PddPurchaseError(
+ "PURCHASE_CONFIRMATION_LOST",
+ "最终提交前确认页已变化,已停止演练",
+ step="purchase_dry_run_stopped",
+ )
+
+ def close(self) -> None:
+ session = self._session
+ self._session = None
+ self._device = None
+ if session is not None:
+ session.__exit__(None, None, None)
+
+ def _wait_for_page(self, expected: str, timeout: float) -> str:
+ deadline = self._monotonic() + timeout
+ last_kind = "unknown"
+ while self._monotonic() < deadline:
+ self._check_cancelled("purchase_open_goods")
+ device = self._require_device()
+ current = device.app_current()
+ xml_data = self._dump_hierarchy()
+ root = _parse_xml(xml_data)
+ last_kind = _page_kind(root, str(current.get("package") or ""))
+ if last_kind == expected:
+ return xml_data
+ if last_kind in {"captcha", "login_required", "risk_control", "payment"}:
+ self._raise_special_page(last_kind)
+ self._sleep(0.25)
+ raise PddPurchaseError(
+ "PDD_PAGE_TIMEOUT",
+ f"等待 PDD {expected} 页面超时,最后页面为 {last_kind}",
+ step="purchase_open_goods",
+ retryable=True,
+ )
+
+ def _wait_for_confirmation_panel(self) -> str:
+ deadline = self._monotonic() + self._panel_timeout
+ while self._monotonic() < deadline:
+ self._check_cancelled("purchase_select_options")
+ xml_data = self._dump_hierarchy()
+ 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(
+ "增加数量" in label for label in _labels(root)
+ ):
+ return xml_data
+ if kind in {"captcha", "login_required", "risk_control", "payment"}:
+ self._raise_special_page(kind)
+ self._sleep(0.2)
+ raise PddPurchaseError(
+ "PURCHASE_PANEL_TIMEOUT",
+ "点击采购入口后没有到达可靠的提交前确认页",
+ step="purchase_select_options",
+ retryable=True,
+ )
+
+ def _dump_hierarchy(self) -> str:
+ raw = self._require_device().dump_hierarchy()
+ return (
+ raw.decode("utf-8", errors="replace")
+ if isinstance(raw, bytes)
+ else str(raw)
+ )
+
+ def _require_device(self) -> Any:
+ if self._device is None:
+ raise PddPurchaseError(
+ "DEVICE_SESSION_CLOSED",
+ "采购演练设备会话未建立或已关闭",
+ step="purchase_page_check",
+ retryable=True,
+ )
+ return self._device
+
+ def _check_cancelled(self, step: str) -> None:
+ if self._cancelled():
+ raise PddPurchaseError(
+ "PURCHASE_CANCELLED",
+ "用户已请求停止采购演练",
+ step=step,
+ )
+
+ @staticmethod
+ def _raise_special_page(kind: str) -> None:
+ mapping = {
+ "captcha": ("PDD_PAGE_CAPTCHA", "PDD 出现安全验证,请人工处理"),
+ "login_required": (
+ "PDD_PAGE_LOGIN_REQUIRED",
+ "PDD 登录状态失效,请人工登录",
+ ),
+ "risk_control": ("PDD_PAGE_RISK_CONTROL", "PDD 出现风控页,请人工处理"),
+ "payment": ("PDD_PAGE_PAYMENT", "PDD 已进入支付页,已立即停止"),
+ }
+ code, message = mapping[kind]
+ raise PddPurchaseError(code, message, step="purchase_page_check")
+
+ @staticmethod
+ def _raise_device_or_page_error(error: Exception, step: str) -> None:
+ details = str(error)
+ lowered = details.lower()
+ disconnected = any(
+ marker in lowered
+ for marker in ("offline", "device not found", "disconnected", "closed transport")
+ )
+ raise PddPurchaseError(
+ "DEVICE_DISCONNECTED" if disconnected else "PURCHASE_ADAPTER_ERROR",
+ f"采购演练在 {step} 失败:{details}",
+ step=step,
+ retryable=disconnected,
+ diagnostics={"exception_type": type(error).__name__},
+ ) from error
+
+
+_PURCHASE_DEVICE_SERVICE = PddDeviceService()
+
+
+def create_u2_purchase_adapter(
+ device_address: str, cancelled: Callable[[], bool]
+) -> PddPurchaseAdapter:
+ """为正式 Client 创建一次 dry-run 采购会话。"""
+
+ return U2PddPurchaseAdapter(
+ device_address,
+ device_service=_PURCHASE_DEVICE_SERVICE,
+ cancelled=cancelled,
+ )
diff --git a/client/src/ui_main.py b/client/src/ui_main.py
index 857c9b5..819e708 100644
--- a/client/src/ui_main.py
+++ b/client/src/ui_main.py
@@ -29,6 +29,7 @@ from qfluentwidgets import (
from .pdd_ui import PDDTaskPage
from .pdd_ui_event import PDDTaskPageEvent
+from .pdd_u2_purchase_adapter import create_u2_purchase_adapter
from .settings_ui import SettingsPage
from .task_repository import TaskRepository
@@ -54,6 +55,7 @@ class MainWindow(FluentWindow):
self.pddTaskPage,
task_repository or TaskRepository(),
self,
+ purchase_adapter_factory=create_u2_purchase_adapter,
)
self.addSubInterface(self.pddTaskPage, FIF.HOME, "pdd")
diff --git a/client/test/test_pdd_u2_purchase_adapter.py b/client/test/test_pdd_u2_purchase_adapter.py
new file mode 100644
index 0000000..50e8652
--- /dev/null
+++ b/client/test/test_pdd_u2_purchase_adapter.py
@@ -0,0 +1,173 @@
+"""uiautomator2 采购演练 Adapter 测试;不连接真实手机。"""
+
+import unittest
+
+from src.pdd_device_service import PddDeviceService
+from src.pdd_purchase_adapter import PddPurchaseError
+from src.pdd_u2_purchase_adapter import U2PddPurchaseAdapter
+
+
+GOODS_URL = "https://mobile.yangkeduo.com/goods.html?goods_id=753136429979"
+
+
+def home_xml() -> str:
+ return """
+
+
+
+
+
+ """
+
+
+def panel_xml(quantity: int = 1) -> str:
+ return f"""
+
+
+
+
+
+
+
+
+
+
+
+
+ """
+
+
+class FakeEditor:
+ def __init__(self, device) -> None:
+ self.device = device
+ self.count = 1
+
+ def set_text(self, value: str) -> None:
+ self.device.quantity = int(value)
+
+
+class FakeDevice:
+ def __init__(self, initial_xml: str = "") -> None:
+ self.mode = "special" if initial_xml else "home"
+ self.special_xml = initial_xml
+ self.quantity = 1
+ self.clicks = []
+ self.opened_urls = []
+
+ def app_current(self):
+ return {"package": "com.xunmeng.pinduoduo"}
+
+ def app_start(self, _package):
+ pass
+
+ def app_wait(self, _package, timeout=10):
+ return True
+
+ def open_url(self, url):
+ self.opened_urls.append(url)
+
+ def dump_hierarchy(self):
+ if self.mode == "special":
+ return self.special_xml
+ return home_xml() if self.mode == "home" else panel_xml(self.quantity)
+
+ def click(self, x, y):
+ self.clicks.append((x, y))
+ self.mode = "panel"
+
+ def __call__(self, **selector):
+ if selector == {"className": "android.widget.EditText"}:
+ return FakeEditor(self)
+ raise AssertionError(f"未预期的 selector: {selector}")
+
+
+class U2PddPurchaseAdapterTest(unittest.TestCase):
+ def _adapter(self, device, calls):
+ def select_color_fn(_device, _xml, target, **_kwargs):
+ calls.append(("color", target))
+ return target == "黑色"
+
+ def select_size_fn(_device, _xml, target, **_kwargs):
+ calls.append(("size", target))
+ return target == "3XL【140-165斤】"
+
+ return U2PddPurchaseAdapter(
+ "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,
+ )
+
+ def test_dry_run_reaches_confirmation_without_clicking_submit(self):
+ device = FakeDevice()
+ calls = []
+ adapter = self._adapter(device, calls)
+
+ adapter.open_goods(GOODS_URL)
+ goods = adapter.read_state()
+ adapter.select_options(
+ {"color": "黑色", "size": "3XL【140-165斤】"}
+ )
+ adapter.set_quantity(2)
+ confirmation = adapter.read_state()
+ adapter.enter_confirmation()
+ adapter.stop_before_submit()
+ adapter.close()
+
+ self.assertEqual(goods.page_kind, "goods")
+ self.assertEqual(goods.goods_id, "753136429979")
+ self.assertEqual(
+ confirmation.selected_options,
+ {"color": "黑色", "size": "3XL【140-165斤】"},
+ )
+ self.assertEqual(confirmation.quantity, 2)
+ self.assertEqual(confirmation.price_cent, 503)
+ self.assertEqual(confirmation.page_kind, "order_confirmation")
+ self.assertEqual(calls, [("color", "黑色"), ("size", "3XL【140-165斤】")])
+ # 只点击一次商品页采购入口;最终“提交订单”从不点击。
+ self.assertEqual(len(device.clicks), 1)
+
+ def test_unsupported_dynamic_option_stops_before_click(self):
+ device = FakeDevice()
+ adapter = self._adapter(device, [])
+ adapter.open_goods(GOODS_URL)
+
+ with self.assertRaises(PddPurchaseError) as raised:
+ adapter.select_options({"bundle": "标准版"})
+
+ self.assertEqual(raised.exception.code, "PURCHASE_OPTIONS_UNSUPPORTED")
+ self.assertEqual(device.clicks, [])
+ adapter.close()
+
+ def test_captcha_stops_without_waiting_or_clicking(self):
+ device = FakeDevice(
+ ''
+ )
+ adapter = self._adapter(device, [])
+
+ with self.assertRaises(PddPurchaseError) as raised:
+ adapter.open_goods(GOODS_URL)
+
+ self.assertEqual(raised.exception.code, "PDD_PAGE_CAPTCHA")
+ self.assertEqual(device.clicks, [])
+ adapter.close()
+
+ def test_invalid_non_pdd_url_is_rejected_before_connect(self):
+ device = FakeDevice()
+ adapter = self._adapter(device, [])
+
+ with self.assertRaises(PddPurchaseError) as raised:
+ adapter.open_goods("https://example.test/goods?goods_id=1")
+
+ self.assertEqual(raised.exception.code, "PURCHASE_GOODS_URL_INVALID")
+ self.assertEqual(device.opened_urls, [])
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/client/test/test_pdd_ui_event.py b/client/test/test_pdd_ui_event.py
index 5513c69..51ec3c5 100644
--- a/client/test/test_pdd_ui_event.py
+++ b/client/test/test_pdd_ui_event.py
@@ -21,6 +21,7 @@ from src.pdd_ui_event import (
)
from src.mock_admin_gateway import MockAdminGateway
from src.pdd_purchase_adapter import PddPurchaseAdapter, PurchasePageState
+from src.pdd_u2_purchase_adapter import create_u2_purchase_adapter
from src.settings_repository import SettingsRepository
from src.task_models import NewClaimedTask, TaskStatus, TaskSummary, TaskType
from src.task_repository import TaskRepository
@@ -461,6 +462,10 @@ class PDDTaskPageEventTest(unittest.TestCase):
)
self.assertIsInstance(window.pddTaskPageEvent, PDDTaskPageEvent)
+ self.assertIs(
+ window.pddTaskPageEvent._purchase_adapter_factory,
+ create_u2_purchase_adapter,
+ )
self.assertEqual(window.pddTaskPage.taskModel.data_row_count(), 0)
window.close()
window.deleteLater()
diff --git a/docs/client/02-architecture.md b/docs/client/02-architecture.md
index 14aaf80..028b62a 100644
--- a/docs/client/02-architecture.md
+++ b/docs/client/02-architecture.md
@@ -365,6 +365,17 @@ reconcile_purchase(task, run) -> PurchaseResult | ManualReview
只读核对另用 `PddPurchaseReconcileAdapter`,只暴露 `read_order_match`
和 `close`,不暴露选规格、设数量、下单或付款方法。
+正式 Client 由 `pdd_u2_purchase_adapter.py` 实现上述演练接口,并在
+`ui_main.py` 注入工厂。Adapter 通过 `PddDeviceService` 独占连接,每次判断
+都重新读取当前包名和控件树。当前 Admin 下发的 `color` 和 `size`
+使用精确文字匹配;其他动态维度直接停止,不做相似匹配。原生 PDD
+控件树不暴露商品编号,因此商品编号来自 Adapter 本次已校验并打开的
+PDD URL,包名和页面类型仍以最新控件树确认。
+
+部分 PDD 页面在点击商品页购买入口后直接进入包含规格和数量的
+订单确认页。Adapter 只点击这一次可逆入口;`enter_confirmation`
+和 `stop_before_submit` 只读确认“提交订单”等最终按钮存在,**不点击它**。
+
现有 `wait_goods_page`、规格面板坐标、颜色尺码选择和下单按钮定位函数可以迁移到该适配层。实验脚本中的硬编码商品、设备、文件路径和 `print` 不得进入正式服务。
PDD 页面可能出现登录失效、验证码、控件树不完整、A/B 页面、库存变化和价格变化。适配层必须返回结构化错误,不得把这些情况统一返回 `False`。
diff --git a/docs/client/06-quality-security.md b/docs/client/06-quality-security.md
index a7dc077..2b37fe6 100644
--- a/docs/client/06-quality-security.md
+++ b/docs/client/06-quality-security.md
@@ -94,6 +94,9 @@ Adapter 不提供提交订单或付款方法。开发者不能把自动化测试
- 已落库 Outbox 只补交,不重跑手机流程。
- 停止、设备断开、验证码、登录失效和结果不确定都保留稳定错误和诊断。
- `purchase_mode=live` 会在数据对象创建时被拒绝。
+- 正式 uiautomator2 采购 Adapter 只点击商品页采购入口和目标规格,
+ `enter_confirmation` 与 `stop_before_submit` 均为只读检查。
+- 单元测试必须断言到达最终确认页后没有产生第二次底部按钮点击。
上述只证明演练和恢复代码的默认安全性,**不代表真实下单已放行**。