cmautobuy/client/src/pdd_u2_purchase_reconcile_adapter.py

"""uiautomator2 只读订单核对 Adapter。

只允许启动 PDD、切换到“个人中心/我的订单/待付款”、返回和滚动读取。
代码中没有提交订单、取消订单或付款入口。
"""

from __future__ import annotations

import re
import time
import xml.etree.ElementTree as ET
from datetime import datetime, timedelta, timezone
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
from typing import Any, Callable, Optional

from .pdd_device_service import (
    PDD_PACKAGE_NAME,
    PddDeviceService,
    current_thread_device_service,
)
from .pdd_purchase_reconcile_adapter import (
    PddPurchaseReconcileAdapter,
    PurchaseOrderCandidate,
    PurchaseReconcileQuery,
    PurchaseReconcileScan,
)


Bounds = tuple[int, int, int, int]
_BOUNDS_PATTERN = re.compile(r"^\[(\d+),(\d+)\]\[(\d+),(\d+)\]$")
_ORDER_NO_PATTERN = re.compile(
    r"(?:订单编号|订单号)\s*[:：]?\s*([A-Za-z0-9-]{6,64})"
)
_GOODS_ID_PATTERN = re.compile(
    r"(?:goods_id=|商品编号\s*[:：]?\s*)(\d{6,})",
    re.IGNORECASE,
)
_QUANTITY_PATTERN = re.compile(r"(?:共\s*(\d+)\s*件|[xX×]\s*(\d+))")
_TOTAL_PATTERN = re.compile(
    r"(?:需付款|应付款|合计|实付款)\s*[:：]?\s*[¥￥]?\s*(\d+(?:\.\d{1,2})?)"
)
_ORDER_TIME_PATTERN = re.compile(
    r"(?:下单时间|创建时间)\s*[:：]?\s*"
    r"(20\d{2}[-/.年]\d{1,2}[-/.月]\d{1,2}日?\s+\d{1,2}:\d{2}(?::\d{2})?)"
)
_LOGIN_MARKERS = ("手机号登录", "登录后继续", "验证码登录", "账号登录")
_CAPTCHA_MARKERS = ("请完成验证", "拖动滑块", "安全验证", "点击图中")
_RISK_MARKERS = ("操作频繁", "异常请求", "风险提示", "账号异常")
_PAYMENT_ACTION_MARKERS = ("立即支付", "确认支付", "输入支付密码")
_UNPAID_MARKERS = ("待付款", "待支付")
_NON_UNPAID_MARKERS = ("已付款", "交易成功", "交易完成", "已取消", "退款")
_SAFE_NAVIGATION_LABELS = ("个人中心", "我的订单", "待付款")


class PddPurchaseReconcileError(RuntimeError):
    """只读核单无法安全继续。"""

    def __init__(self, code: str, message: str) -> None:
        super().__init__(message)
        self.code = code


def _parse_xml(xml_data: str | bytes) -> ET.Element:
    try:
        return ET.fromstring(xml_data)
    except (ET.ParseError, TypeError) as exc:
        raise PddPurchaseReconcileError(
            "RECONCILE_XML_INVALID", "订单页无障碍控件树无效"
        ) 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 _subtree_text(node: ET.Element) -> str:
    return " ".join(
        label for item in node.iter("node") if (label := _label(item))
    )


def _parse_bounds(value: str) -> Optional[Bounds]:
    match = _BOUNDS_PATTERN.fullmatch(str(value or "").strip())
    if match is None:
        return None
    left, top, right, bottom = map(int, match.groups())
    if right <= left or bottom <= top:
        return None
    return left, top, right, bottom


def _parse_money_cent(text: str) -> int:
    match = _TOTAL_PATTERN.search(text)
    if match is None:
        return 0
    try:
        value = Decimal(match.group(1)).quantize(
            Decimal("0.01"), rounding=ROUND_HALF_UP
        )
    except InvalidOperation:
        return 0
    return int(value * 100)


def _parse_quantity(text: str) -> int:
    match = _QUANTITY_PATTERN.search(text)
    if match is None:
        return 0
    value = match.group(1) or match.group(2) or "0"
    return int(value)


def _parse_ordered_at(text: str) -> tuple[str, str]:
    match = _ORDER_TIME_PATTERN.search(text)
    if match is None:
        return "", ""
    raw = match.group(1)
    normalized = (
        raw.replace("年", "-")
        .replace("月", "-")
        .replace("日", "")
        .replace("/", "-")
        .replace(".", "-")
    )
    formats = ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M")
    for time_format in formats:
        try:
            local = datetime.strptime(normalized, time_format).replace(
                tzinfo=timezone(timedelta(hours=8))
            )
            ordered_at = local.astimezone(timezone.utc).isoformat(
                timespec="seconds"
            ).replace("+00:00", "Z")
            return ordered_at, raw
        except ValueError:
            continue
    return "", raw


def parse_order_candidates(
    xml_data: str | bytes, query: PurchaseReconcileQuery
) -> tuple[PurchaseOrderCandidate, ...]:
    """从脱敏控件树提取候选，只保留核单必需字段。"""

    root = _parse_xml(xml_data)
    parents = {
        child: parent
        for parent in root.iter("node")
        for child in parent
        if child.tag == "node"
    }
    parsed: dict[str, PurchaseOrderCandidate] = {}
    for node in root.iter("node"):
        node_label = _label(node)
        order_match = _ORDER_NO_PATTERN.search(node_label)
        if order_match is None:
            continue
        order_no = order_match.group(1)
        current = node
        card_text = node_label
        for _ in range(8):
            text = _subtree_text(current)
            if any(marker in text for marker in _UNPAID_MARKERS):
                card_text = text
                if (
                    _TOTAL_PATTERN.search(text)
                    and _QUANTITY_PATTERN.search(text)
                    and _ORDER_TIME_PATTERN.search(text)
                ):
                    break
            parent = parents.get(current)
            if parent is None:
                break
            current = parent

        goods_match = _GOODS_ID_PATTERN.search(card_text)
        ordered_at, ordered_at_raw = _parse_ordered_at(card_text)
        options = {
            key: value
            for key, value in query.options.items()
            if str(value).strip() and str(value).strip() in card_text
        }
        payment_status = (
            "unpaid"
            if any(marker in card_text for marker in _UNPAID_MARKERS)
            and not any(marker in card_text for marker in _NON_UNPAID_MARKERS)
            else "other"
        )
        candidate = PurchaseOrderCandidate(
            order_no=order_no,
            goods_id=goods_match.group(1) if goods_match else "",
            options=options,
            quantity=_parse_quantity(card_text),
            total_price_cent=_parse_money_cent(card_text),
            ordered_at=ordered_at,
            ordered_at_raw=ordered_at_raw,
            payment_status=payment_status,
        )
        previous = parsed.get(order_no)
        if previous is None or previous == candidate:
            parsed[order_no] = candidate
    return tuple(parsed.values())


class U2PddPurchaseReconcileAdapter(PddPurchaseReconcileAdapter):
    """只导航和滚动读取待付款订单，不包含任何订单写操作。"""

    def __init__(
        self,
        device_address: str,
        *,
        device_service: Optional[PddDeviceService] = None,
        cancelled: Callable[[], bool] = lambda: False,
        settle_seconds: float = 0.4,
        max_pages: int = 5,
    ) -> None:
        self._device_address = str(device_address or "").strip()
        self._device_service = (
            device_service
            or current_thread_device_service()
            or PddDeviceService()
        )
        self._cancelled = cancelled
        self._settle_seconds = max(0.0, float(settle_seconds))
        self._max_pages = max(1, int(max_pages))
        self._session = None
        self._device = None

    def read_order_candidates(
        self, query: PurchaseReconcileQuery
    ) -> PurchaseReconcileScan:
        self._check_cancelled()
        device = self._connect()
        self._open_unpaid_orders(device)
        found: dict[str, PurchaseOrderCandidate] = {}
        previous_signature = ""
        pages_scanned = 0
        for _ in range(self._max_pages):
            self._check_cancelled()
            xml_data = device.dump_hierarchy(compressed=False)
            self._raise_for_special_page(_parse_xml(xml_data))
            pages_scanned += 1
            for candidate in parse_order_candidates(xml_data, query):
                found.setdefault(candidate.order_no, candidate)
            signature = "|".join(sorted(found)) + f":{len(str(xml_data))}"
            if signature == previous_signature:
                break
            previous_signature = signature
            width, height = device.window_size()
            device.swipe(
                width // 2,
                int(height * 0.78),
                width // 2,
                int(height * 0.32),
                0.35,
            )
            self._settle()
        return PurchaseReconcileScan(
            candidates=tuple(found.values()),
            diagnostics={
                "pages_scanned": pages_scanned,
                "candidate_count": len(found),
            },
        )

    def close(self) -> None:
        session = self._session
        self._session = None
        self._device = None
        if session is not None:
            session.__exit__(None, None, None)

    def _connect(self) -> Any:
        if self._device is None:
            self._session = self._device_service.connect(self._device_address)
            self._device = self._session.__enter__()
        return self._device

    def _open_unpaid_orders(self, device: Any) -> None:
        current = device.app_current()
        if str(current.get("package") or "") != PDD_PACKAGE_NAME:
            device.app_start(PDD_PACKAGE_NAME)
            self._settle()

        opened_order_area = False
        for _ in range(6):
            self._check_cancelled()
            current = device.app_current()
            if str(current.get("package") or "") != PDD_PACKAGE_NAME:
                raise PddPurchaseReconcileError(
                    "RECONCILE_WRONG_APP", "核单时 PDD 不在前台"
                )
            root = _parse_xml(device.dump_hierarchy(compressed=False))
            self._raise_for_special_page(root)
            combined = _subtree_text(root)
            if opened_order_area and (
                _ORDER_NO_PATTERN.search(combined)
                or "暂无订单" in combined
                or "暂无相关订单" in combined
            ):
                return
            if any(marker in combined for marker in _PAYMENT_ACTION_MARKERS):
                device.press("back")
                self._settle()
                continue
            if "我的订单" in combined:
                if self._click_unique_label(device, root, "待付款"):
                    opened_order_area = True
                    self._settle()
                    continue
                if self._click_unique_label(device, root, "我的订单"):
                    opened_order_area = True
                    self._settle()
                    continue
            if self._click_unique_label(device, root, "个人中心"):
                self._settle()
                continue
            raise PddPurchaseReconcileError(
                "RECONCILE_ORDER_ENTRY_NOT_FOUND",
                "没有找到唯一的“个人中心/我的订单/待付款”只读入口",
            )
        raise PddPurchaseReconcileError(
            "RECONCILE_ORDER_PAGE_TIMEOUT", "打开待付款订单列表超时"
        )

    @staticmethod
    def _click_unique_label(
        device: Any, root: ET.Element, target: str
    ) -> bool:
        if target not in _SAFE_NAVIGATION_LABELS:
            raise ValueError("只允许使用白名单只读导航入口")
        targets = []
        for node in root.iter("node"):
            labels = {
                node.get("text", "").strip(),
                node.get("content-desc", "").strip(),
            }
            if target not in labels:
                continue
            if (
                node.get("visible-to-user") != "true"
                or node.get("enabled") != "true"
            ):
                continue
            bounds = _parse_bounds(node.get("bounds", ""))
            if bounds is not None:
                targets.append(bounds)
        unique = list(dict.fromkeys(targets))
        if len(unique) != 1:
            return False
        left, top, right, bottom = unique[0]
        device.click((left + right) // 2, (top + bottom) // 2)
        return True

    @staticmethod
    def _raise_for_special_page(root: ET.Element) -> None:
        combined = _subtree_text(root)
        markers = (
            ("RECONCILE_CAPTCHA", _CAPTCHA_MARKERS),
            ("RECONCILE_LOGIN_REQUIRED", _LOGIN_MARKERS),
            ("RECONCILE_RISK_CONTROL", _RISK_MARKERS),
        )
        for code, values in markers:
            if any(value in combined for value in values):
                raise PddPurchaseReconcileError(
                    code, "PDD 核单遇到登录、验证或风控页面，请人工处理"
                )

    def _check_cancelled(self) -> None:
        if self._cancelled():
            raise PddPurchaseReconcileError(
                "RECONCILE_CANCELLED", "用户已停止只读订单核对"
            )

    def _settle(self) -> None:
        if self._settle_seconds:
            time.sleep(self._settle_seconds)


def create_u2_purchase_reconcile_adapter(
    device_address: str, cancelled: Callable[[], bool]
) -> PddPurchaseReconcileAdapter:
    """为正式 Client 创建只读订单核对会话。"""

    return U2PddPurchaseReconcileAdapter(
        device_address,
        device_service=current_thread_device_service() or PddDeviceService(),
        cancelled=cancelled,
    )