cmautobuy/client/src/pdd_u2_purchase_adapter.py

"""uiautomator2 采购 Adapter。

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

from __future__ import annotations

import re
import time
import xml.etree.ElementTree as ET
from contextlib import nullcontext
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,
    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_PAYMENT,
    PAGE_RISK_CONTROL,
    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"})


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:
    return classify_pdd_page(root, current_package).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 _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)
    return sorted(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,
    ) -> 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] = {}
        self._submit_attempted = False

    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._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)
            self._wait_for_goods_page(goods_url, before_open)
        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 ""))
            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 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_goods_page(
        self,
        goods_url: str,
        before_open: Optional[PddPageObservation],
    ) -> 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()
                current = device.app_current()
                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)
                observation = classify_pdd_page(
                    root, str(current.get("package") or "")
                )
                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
        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


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,
    )


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,
    )