feat: 接入采购演练真机适配器 (#75)

2026-08-10 01:07:57 +08:00
parent 18f3209a62
commit 7e90af0dc9
6 changed files with 708 additions and 0 deletions
@@ -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,
+    )
@@ -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")