cmautobuy/client/src/pdd_u2_purchase_adapter.py

"""uiautomator2 采购 Adapter。

演练 factory 返回没有提交方法的窄接口；live factory 单独返回只允许一次提交的
接口。两条路径都不提供付款、取消订单或绕过安全校验的方法。
"""

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,
    PddDeviceService,
    current_thread_device_service,
)
from .performance_timing import current_performance_trace
from .pdd_page_classifier import (
    ACTION_NETWORK_ERROR,
    ACTION_READY,
    ACTION_REOPEN,
    ACTION_UNAVAILABLE,
    PAGE_CAPTCHA,
    PAGE_GOODS,
    PAGE_HOME,
    PAGE_LOGIN_REQUIRED,
    PAGE_NETWORK_ERROR,
    PAGE_ORDER_CONFIRMATION,
    PAGE_PAYMENT,
    PAGE_RISK_CONTROL,
    PAGE_UNKNOWN,
    GoodsOpenTracker,
    PddPageObservation,
    classify_pdd_page,
)
from .pdd_purchase_adapter import (
    PddLivePurchaseAdapter,
    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 = ("提交订单", "现在买，仅", "确认购买")
_OUT_OF_STOCK_MARKERS = ("已售罄", "暂时缺货", "库存不足", "该商品已售罄")
_SUPPORTED_OPTION_KEYS = frozenset({"color", "size"})
_MAX_QUANTITY_BUTTON_CLICKS = 5
_KEYBOARD_STATUS_MARKERS = (
    "minputshown",
    "misinputviewshown",
    "minputviewshown",
    "inputshown",
)
_DIAGNOSTIC_MARKERS = (
    "增加数量",
    "减少数量",
    "提交订单",
    "现在买",
    "确认购买",
    "颜色分类",
    "款式",
    "尺码",
    "套餐",
    "安全验证",
    "手机号登录",
    "操作频繁",
    "网络不给力",
    "确认款式",
    "确认规格",
    "已选",
    "请选择",
    "确定",
)


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 _package_hint_from_tree(root: ET.Element) -> str:
    """可靠控件树可以直接证明 PDD 在前台，避免慢速前台查询。"""

    pdd_node_count = sum(
        1
        for node in root.iter("node")
        if node.get("package") == PDD_PACKAGE_NAME
    )
    return PDD_PACKAGE_NAME if pdd_node_count >= 3 else ""


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:
        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:
    kind = classify_pdd_page(root, current_package).kind
    contextual_targets = _contextual_confirm_targets(root)
    if kind in {PAGE_UNKNOWN, PAGE_GOODS} and len(contextual_targets) == 1:
        return PAGE_ORDER_CONFIRMATION
    return kind


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 _quantity_button_targets(root: ET.Element, description: str) -> list[Bounds]:
    """返回唯一、可点击的数量加减按钮坐标候选。"""

    targets = set()
    for node in root.iter("node"):
        labels = {
            node.get("text", "").strip(),
            node.get("content-desc", "").strip(),
        }
        bounds = _parse_bounds(node.get("bounds", ""))
        if description not in labels or bounds is None:
            continue
        if node.get("clickable") != "true":
            continue
        if node.get("enabled") == "false" or node.get("visible-to-user") == "false":
            continue
        targets.add(bounds)
    return sorted(targets)


def _shell_output(response: Any) -> str:
    """兼容 uiautomator2 ShellResponse 和测试中的普通字符串。"""

    output = getattr(response, "output", response)
    if isinstance(output, bytes):
        return output.decode("utf-8", errors="replace")
    return str(output or "")


def _keyboard_is_shown(device: Any) -> Optional[bool]:
    """读取安卓输入法状态；无法可靠判断时返回 None。"""

    try:
        response = device.shell("dumpsys input_method")
    except (AttributeError, OSError, RuntimeError):
        return None
    normalized = re.sub(r"\s+", "", _shell_output(response)).lower()
    found_false = False
    for marker in _KEYBOARD_STATUS_MARKERS:
        if f"{marker}=true" in normalized:
            return True
        if f"{marker}=false" in normalized:
            found_false = True
    return False if found_false else None


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]


def _final_submit_targets(root: ET.Element) -> list[Bounds]:
    """返回底部可见、启用且文字明确的唯一提交按钮坐标。"""

    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 []

    targets = set()
    for node in root.iter("node"):
        label = _label(node)
        bounds = _parse_bounds(node.get("bounds", ""))
        if not label or bounds is None:
            continue
        if not any(marker in label for marker in _FINAL_SUBMIT_MARKERS):
            continue
        if node.get("visible-to-user") != "true" or node.get("enabled") != "true":
            continue
        if (bounds[1] + bounds[3]) // 2 < screen_bottom * 0.6:
            continue
        targets.add(bounds)
    targets.update(_contextual_confirm_targets(root))
    return sorted(targets)


def _contextual_confirm_targets(root: ET.Element) -> list[Bounds]:
    """只在强证据规格面板内接受底部“确定”作为最终提交按钮。"""

    nodes = list(root.iter("node"))
    if sum(
        node.get("package") == PDD_PACKAGE_NAME for node in nodes
    ) < 3:
        return []

    parsed_bounds = [
        bounds
        for node in nodes
        if (bounds := _parse_bounds(node.get("bounds", ""))) is not None
    ]
    if not parsed_bounds:
        return []
    screen_right = max(bounds[2] for bounds in parsed_bounds)
    screen_bottom = max(bounds[3] for bounds in parsed_bounds)
    screen_area = screen_right * screen_bottom
    parents = {
        child: parent
        for parent in root.iter()
        for child in parent
        if child.tag == "node"
    }

    targets: set[Bounds] = set()
    for confirm in nodes:
        if _label(confirm).strip() not in {"确定", "確定"}:
            continue
        bounds = _parse_bounds(confirm.get("bounds", ""))
        if bounds is None or confirm.get("clickable") != "true":
            continue
        if (
            confirm.get("visible-to-user") != "true"
            or confirm.get("enabled") != "true"
        ):
            continue
        if (bounds[1] + bounds[3]) // 2 < screen_bottom * 0.6:
            continue

        panel = parents.get(confirm)
        while panel is not None:
            panel_bounds = _parse_bounds(panel.get("bounds", ""))
            if panel_bounds is not None:
                area = (panel_bounds[2] - panel_bounds[0]) * (
                    panel_bounds[3] - panel_bounds[1]
                )
                if (not screen_area or area < screen_area * 0.95) and (
                    _has_confirmation_panel_evidence(panel)
                ):
                    targets.add(bounds)
                    break
            panel = parents.get(panel)
    return sorted(targets)


def _has_confirmation_panel_evidence(panel: ET.Element) -> bool:
    """确认标题、已选摘要、规格标题和数量控件位于同一面板。"""

    nodes = list(panel.iter("node"))
    labels = [_label(node).replace(" ", "") for node in nodes]
    has_title = any(
        label in {"确认款式", "確認款式", "确认规格", "確認規格"}
        for label in labels
    )
    has_summary = any(
        label.startswith(("已选", "已選", "已选择", "已選擇", "请选择", "請選擇"))
        for label in labels
    )
    has_dimension = any(
        re.sub(r"[（(]\d+[)）]$", "", label)
        in {
            "颜色分类",
            "顏色分類",
            "颜色",
            "顏色",
            "尺码",
            "尺碼",
            "规格",
            "規格",
            "款式",
            "套餐",
        }
        for label in labels
    )
    editors = [
        node
        for node in nodes
        if node.get("class") == "android.widget.EditText"
        and node.get("text", "").strip().isdigit()
        and int(node.get("text", "0")) > 0
    ]
    has_decrease = len(_quantity_targets_in_nodes(nodes, "减少数量")) == 1
    has_increase = len(_quantity_targets_in_nodes(nodes, "增加数量")) == 1
    return (
        has_title
        and has_summary
        and has_dimension
        and len(editors) == 1
        and has_decrease
        and has_increase
    )


def _quantity_targets_in_nodes(
    nodes: list[ET.Element], description: str
) -> set[Bounds]:
    """读取指定面板内唯一、可用的数量按钮。"""

    targets: set[Bounds] = set()
    for node in nodes:
        if description not in {
            node.get("text", "").strip(),
            node.get("content-desc", "").strip(),
        }:
            continue
        bounds = _parse_bounds(node.get("bounds", ""))
        if bounds is None or node.get("clickable") != "true":
            continue
        if node.get("enabled", "true") == "false":
            continue
        if node.get("visible-to-user", "true") == "false":
            continue
        targets.add(bounds)
    return targets


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,
        artifact_directory: Optional[Path] = None,
    ) -> 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._artifact_directory = artifact_directory
        self._last_xml: Optional[str] = None
        self._hierarchy_reads = 0
        self._session = None
        self._device = None
        self._goods_id = ""
        self._requested_options: dict[str, str] = {}
        self._submit_attempted = False

    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)
            self._device = self._session.__enter__()
            current = self._session.initial_app_state
            before_open = self._read_observation(current)
            trace = current_performance_trace()
            if current.get("package") != PDD_PACKAGE_NAME:
                stage = trace.stage("pdd_start_or_wait") if trace else nullcontext()
                with stage:
                    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,
                        )
            elif trace is not None:
                trace.record("pdd_start_or_wait", 0, "already_foreground")
            stage = trace.stage("open_url") if trace else nullcontext()
            with stage:
                self._device.open_url(goods_url)
            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:
            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:
            trace = current_performance_trace()
            stage = trace.stage("purchase_read_state") if trace else nullcontext()
            with stage:
                xml_data = self._dump_hierarchy()
                root = _parse_xml(xml_data)
                current_package = _package_hint_from_tree(root)
                if not current_package:
                    current = device.app_current()
                    current_package = str(current.get("package") or "")
                kind = _page_kind(root, current_package)
                labels = _labels(root)
                submit_targets = _final_submit_targets(root)
                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,
                    in_stock=not any(
                        marker in " ".join(labels)
                        for marker in _OUT_OF_STOCK_MARKERS
                    ),
                    submit_candidate_count=len(submit_targets),
                )
        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:
                trace = current_performance_trace()
                stage = (
                    trace.stage("purchase_select_color")
                    if trace else nullcontext()
                )
                with stage:
                    color_selected = self._select_color(
                        device, panel_xml, color, action_delay=0.2
                    )
                if not color_selected:
                    raise PddPurchaseError(
                        "PURCHASE_OPTIONS_MISMATCH",
                        f"没有精确选中颜色：{color}",
                        step="purchase_select_options",
                    )
            size = checked.get("size")
            if size:
                latest_xml = self._dump_hierarchy()
                trace = current_performance_trace()
                stage = (
                    trace.stage("purchase_select_size")
                    if trace else nullcontext()
                )
                with stage:
                    size_selected = self._select_size(
                        device, latest_xml, size, action_delay=0.2
                    )
                if not size_selected:
                    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:
            root = _parse_xml(self._dump_hierarchy())
            current_quantity = _quantity(root)
            if current_quantity <= 0:
                raise PddPurchaseError(
                    "PURCHASE_QUANTITY_CONTROL_MISSING",
                    "页面没有唯一可靠的采购数量输入框",
                    step="purchase_set_quantity",
                )
            if current_quantity == quantity:
                return

            difference = quantity - current_quantity
            if abs(difference) <= _MAX_QUANTITY_BUTTON_CLICKS:
                description = "增加数量" if difference > 0 else "减少数量"
                targets = _quantity_button_targets(root, description)
                if len(targets) == 1:
                    self._set_quantity_with_buttons(
                        current_quantity, quantity, description
                    )
                    return

            self._set_quantity_with_editor(quantity)
        except PddPurchaseError:
            raise
        except Exception as exc:
            self._raise_device_or_page_error(exc, "purchase_set_quantity")

    def _set_quantity_with_buttons(
        self, current_quantity: int, target_quantity: int, description: str
    ) -> None:
        """逐次点击加减按钮，每次都用最新控件树确认数量。"""

        device = self._require_device()
        step = 1 if target_quantity > current_quantity else -1
        expected = current_quantity
        while expected != target_quantity:
            self._check_cancelled("purchase_set_quantity")
            root = _parse_xml(self._dump_hierarchy())
            targets = _quantity_button_targets(root, description)
            if len(targets) != 1:
                raise PddPurchaseError(
                    "PURCHASE_QUANTITY_CONTROL_MISSING",
                    f"页面没有唯一可靠的“{description}”按钮",
                    step="purchase_set_quantity",
                )
            left, top, right, bottom = targets[0]
            device.click((left + right) // 2, (top + bottom) // 2)
            expected += step
            self._sleep(0.2)
            actual = _quantity(_parse_xml(self._dump_hierarchy()))
            if actual != expected:
                raise PddPurchaseError(
                    "PURCHASE_QUANTITY_MISMATCH",
                    f"调整采购数量后期望为 {expected}，页面实际为 {actual or '未知'}",
                    step="purchase_set_quantity",
                )

    def _set_quantity_with_editor(self, quantity: int) -> None:
        """输入框兜底；只在确认输入法显示后按一次返回键。"""

        device = self._require_device()
        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",
            )

        keyboard_shown = _keyboard_is_shown(device)
        if keyboard_shown is None:
            raise PddPurchaseError(
                "PURCHASE_KEYBOARD_DISMISS_FAILED",
                "无法确认安卓输入法状态，已停止采购以避免误按返回键",
                step="purchase_set_quantity",
                retryable=True,
            )
        if keyboard_shown:
            device.press("back")
            for _attempt in range(25):
                self._check_cancelled("purchase_set_quantity")
                self._sleep(0.2)
                keyboard_shown = _keyboard_is_shown(device)
                if keyboard_shown is False:
                    break
            if keyboard_shown is not False:
                raise PddPurchaseError(
                    "PURCHASE_KEYBOARD_DISMISS_FAILED",
                    "安卓输入法没有在规定时间内关闭，已停止采购",
                    step="purchase_set_quantity",
                    retryable=True,
                )

        self._validate_panel_after_keyboard(quantity)

    def _validate_panel_after_keyboard(self, quantity: int) -> None:
        """输入框操作后重新确认规格面板和最终提交前状态。"""

        root = _parse_xml(self._dump_hierarchy())
        if _page_kind(root, "") != "order_confirmation":
            raise PddPurchaseError(
                "PURCHASE_PANEL_LOST_AFTER_KEYBOARD",
                "关闭输入法后规格面板已经消失，已停止采购",
                step="purchase_set_quantity",
                retryable=True,
            )
        if _quantity(root) != quantity:
            raise PddPurchaseError(
                "PURCHASE_QUANTITY_MISMATCH",
                "关闭输入法后采购数量发生变化",
                step="purchase_set_quantity",
            )
        missing_options = [
            value
            for value in self._requested_options.values()
            if not _selected(root, value)
        ]
        if missing_options:
            raise PddPurchaseError(
                "PURCHASE_OPTIONS_MISMATCH",
                "关闭输入法后已选颜色或尺码发生变化",
                step="purchase_set_quantity",
            )
        if _price_cent(root) <= 0:
            raise PddPurchaseError(
                "PURCHASE_PRICE_MISSING",
                "关闭输入法后没有识别到有效价格",
                step="purchase_set_quantity",
            )
        if len(_final_submit_targets(root)) != 1:
            raise PddPurchaseError(
                "PURCHASE_SUBMIT_TARGET_AMBIGUOUS",
                "关闭输入法后没有唯一可靠的下单按钮",
                step="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_goods_page(
        self,
        goods_url: str,
        before_open: Optional[PddPageObservation],
        package_hint: str,
    ) -> str:
        """确认本次深链进入新商品页；稳定首页时只重开一次。"""

        trace = current_performance_trace()
        ready_stage = trace.stage("goods_page_ready") if trace else nullcontext()
        with ready_stage:
            opened_at = self._monotonic()
            deadline = opened_at + self._page_timeout
            tracker = GoodsOpenTracker(before_open, opened_at)
            last_kind = "unknown"
            last_recorded = ""
            first_dump = True
            while self._monotonic() < deadline:
                self._check_cancelled("purchase_open_goods")
                device = self._require_device()
                if first_dump and trace is not None:
                    with trace.stage("first_dump_hierarchy"):
                        xml_data = self._dump_hierarchy()
                    first_dump = False
                else:
                    xml_data = self._dump_hierarchy()
                root = _parse_xml(xml_data)
                # 不在轮询中调用 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(
                        "pdd_page_transition",
                        0,
                        f"attempt_{tracker.attempt}_{last_kind}",
                    )
                    last_recorded = last_kind
                if last_kind in {
                    PAGE_CAPTCHA,
                    PAGE_LOGIN_REQUIRED,
                    PAGE_RISK_CONTROL,
                    PAGE_PAYMENT,
                }:
                    self._raise_special_page(last_kind)
                decision = tracker.observe(observation, self._monotonic())
                if decision.action == ACTION_READY:
                    if trace is not None:
                        trace.checkpoint("end_to_end_total")
                    return xml_data
                if decision.action == ACTION_REOPEN:
                    with (
                        trace.stage("open_url_retry")
                        if trace is not None
                        else nullcontext()
                    ):
                        device.open_url(goods_url)
                    tracker.reopened(self._monotonic())
                    last_recorded = ""
                    continue
                if decision.action == ACTION_UNAVAILABLE:
                    raise PddPurchaseError(
                        "PDD_GOODS_UNAVAILABLE",
                        "商品链接已失效，PDD 无法打开商品详情页并返回了首页",
                        step="purchase_open_goods",
                        retryable=False,
                        diagnostics={
                            "page_kind": PAGE_HOME,
                            "open_attempts": 2,
                        },
                    )
                if decision.action == ACTION_NETWORK_ERROR:
                    raise PddPurchaseError(
                        "PDD_PAGE_NETWORK_ERROR",
                        "PDD 商品页网络或服务异常，请稍后重试",
                        step="purchase_open_goods",
                        retryable=True,
                        diagnostics={"page_kind": PAGE_NETWORK_ERROR},
                    )
                self._sleep(0.25)
            if tracker.stale_goods_seen:
                raise PddPurchaseError(
                    "PDD_GOODS_IDENTITY_UNCONFIRMED",
                    "打开商品链接后仍停留在原商品页，无法确认本次目标商品",
                    step="purchase_open_goods",
                    retryable=True,
                    diagnostics={
                        "page_kind": PAGE_GOODS,
                        "open_attempts": tracker.attempt,
                    },
                )
            raise PddPurchaseError(
                "PDD_PAGE_TIMEOUT",
                f"等待 PDD 商品详情页超时，最后页面为 {last_kind}",
                step="purchase_open_goods",
                retryable=True,
                diagnostics={
                    "page_kind": last_kind,
                    "open_attempts": tracker.attempt,
                },
            )

    def _read_observation(
        self, current: Mapping[str, Any]
    ) -> Optional[PddPageObservation]:
        """读取打开前页面；失败时不阻止后续按新页面重新判断。"""

        try:
            root = _parse_xml(self._dump_hierarchy())
            return classify_pdd_page(
                root, str(current.get("package") or "")
            )
        except (PddPurchaseError, RuntimeError, OSError):
            return None

    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)
            last_kind = _page_kind(root, "")
            if last_kind == "order_confirmation" and any(
                "增加数量" in label for label in _labels(root)
            ):
                return xml_data
            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()
        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:
            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


def create_u2_purchase_adapter(
    device_address: str, cancelled: Callable[[], bool]
) -> PddPurchaseAdapter:
    """为正式 Client 创建一次 dry-run 采购会话。"""

    return U2PddPurchaseAdapter(
        device_address,
        device_service=current_thread_device_service() or PddDeviceService(),
        cancelled=cancelled,
        artifact_directory=data_dir() / "artifacts",
    )


class U2PddLivePurchaseAdapter(U2PddPurchaseAdapter, PddLivePurchaseAdapter):
    """只允许一次最终提交点击的真机 Adapter，不包含付款路径。"""

    def submit_order_once(self) -> None:
        if self._submit_attempted:
            raise PddPurchaseError(
                "PURCHASE_SUBMIT_ALREADY_ATTEMPTED",
                "本次采购已经尝试提交，禁止再次点击",
                step="purchase_submit_once",
            )
        self._check_cancelled("purchase_submit_once")
        device = self._require_device()
        try:
            current = device.app_current()
            root = _parse_xml(self._dump_hierarchy())
            kind = _page_kind(root, str(current.get("package") or ""))
            if kind in {"captcha", "login_required", "risk_control", "payment"}:
                self._raise_special_page(kind)
            if kind != "order_confirmation":
                raise PddPurchaseError(
                    "PURCHASE_CONFIRMATION_LOST",
                    "最终提交前页面已经变化，禁止提交订单",
                    step="purchase_submit_once",
                )
            targets = _final_submit_targets(root)
            if len(targets) != 1:
                raise PddPurchaseError(
                    "PURCHASE_SUBMIT_TARGET_AMBIGUOUS",
                    "最终提交按钮不是唯一可靠目标，禁止提交订单",
                    step="purchase_submit_once",
                    diagnostics={"candidate_count": len(targets)},
                )
            left, top, right, bottom = targets[0]
            self._submit_attempted = True
            device.click((left + right) // 2, (top + bottom) // 2)
        except PddPurchaseError:
            raise
        except Exception as exc:
            self._submit_attempted = True
            self._raise_device_or_page_error(exc, "purchase_submit_once")


def create_u2_live_purchase_adapter(
    device_address: str, cancelled: Callable[[], bool]
) -> PddLivePurchaseAdapter:
    """为已通过本地绑定授权的任务创建一次 live 会话。"""

    return U2PddLivePurchaseAdapter(
        device_address,
        device_service=current_thread_device_service() or PddDeviceService(),
        cancelled=cancelled,
        artifact_directory=data_dir() / "artifacts",
    )