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 collections import Counter
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 = ("个人中心", "我的订单", "待付款")
_ORDER_DETAIL_PAY_LABELS = ("去支付",)
_ORDER_DETAIL_CANCEL_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 _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 _foreground_package_from_tree(root: ET.Element) -> str:
    """控件树只有一个明确业务包时返回它；系统界面节点不参与判断。"""

    ignored_packages = {"", "com.android.systemui"}
    counts = Counter(
        node.get("package", "")
        for node in root.iter("node")
        if node.get("package", "") not in ignored_packages
    )
    if not counts:
        return ""
    highest = max(counts.values())
    packages = [
        package
        for package, count in counts.items()
        if count == highest and count >= 3
    ]
    return packages[0] if len(packages) == 1 else ""


def _label_count(root: ET.Element, targets: tuple[str, ...]) -> int:
    return sum(
        1
        for node in root.iter("node")
        if _label(node) in targets
    )


def _is_single_order_detail(root: ET.Element) -> bool:
    """只在页面上只有一组订单操作时允许跨视口累积字段。"""

    pay_count = _label_count(root, _ORDER_DETAIL_PAY_LABELS)
    cancel_count = _label_count(root, _ORDER_DETAIL_CANCEL_LABELS)
    order_numbers = {
        match.group(1)
        for node in root.iter("node")
        if (match := _ORDER_NO_PATTERN.search(_label(node))) is not None
    }
    return pay_count == 1 and cancel_count <= 1 and len(order_numbers) <= 1


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 _payment_status(text: str) -> str:
    """只在页面有明确状态文字时返回状态。"""

    if any(marker in text for marker in _NON_UNPAID_MARKERS):
        return "other"
    if any(marker in text for marker in _UNPAID_MARKERS):
        return "unpaid"
    return ""


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
        }
        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(card_text),
        )
        previous = parsed.get(order_no)
        if previous is None or previous == candidate:
            parsed[order_no] = candidate
    return tuple(parsed.values())


def _parse_order_detail_evidence(
    root: ET.Element, query: PurchaseReconcileQuery
) -> PurchaseOrderCandidate:
    """读取单个订单详情当前视口；字段可以暂时不完整。"""

    text = _subtree_text(root)
    order_numbers = tuple(dict.fromkeys(_ORDER_NO_PATTERN.findall(text)))
    goods_match = _GOODS_ID_PATTERN.search(text)
    ordered_at, ordered_at_raw = _parse_ordered_at(text)
    options = {
        key: value
        for key, value in query.options.items()
        if str(value).strip() and str(value).strip() in text
    }
    return PurchaseOrderCandidate(
        order_no=order_numbers[0] if len(order_numbers) == 1 else "",
        goods_id=goods_match.group(1) if goods_match else "",
        options=options,
        quantity=_parse_quantity(text),
        total_price_cent=_parse_money_cent(text),
        ordered_at=ordered_at,
        ordered_at_raw=ordered_at_raw,
        payment_status=_payment_status(text),
    )


def _merge_order_evidence(
    previous: PurchaseOrderCandidate,
    current: PurchaseOrderCandidate,
) -> Optional[PurchaseOrderCandidate]:
    """合并同一订单的相邻视口；关键字段冲突时拒绝合并。"""

    scalar_fields = (
        "order_no",
        "goods_id",
        "quantity",
        "total_price_cent",
        "ordered_at",
        "ordered_at_raw",
    )
    values: dict[str, object] = {}
    for field_name in scalar_fields:
        old_value = getattr(previous, field_name)
        new_value = getattr(current, field_name)
        if old_value and new_value and old_value != new_value:
            return None
        values[field_name] = old_value or new_value

    options = dict(previous.options)
    for key, value in current.options.items():
        if key in options and options[key] != value:
            return None
        options[key] = value

    statuses = {
        value
        for value in (previous.payment_status, current.payment_status)
        if value
    }
    if "other" in statuses:
        payment_status = "other"
    else:
        payment_status = "unpaid" if "unpaid" in statuses else ""
    return PurchaseOrderCandidate(
        order_no=str(values["order_no"]),
        goods_id=str(values["goods_id"]),
        options=options,
        quantity=int(values["quantity"]),
        total_price_cent=int(values["total_price_cent"]),
        ordered_at=str(values["ordered_at"]),
        ordered_at_raw=str(values["ordered_at_raw"]),
        payment_status=payment_status,
    )


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

    def __init__(
        self,
        device_address: str,
        *,
        device_service: Optional[PddDeviceService] = None,
        cancelled: Callable[[], bool] = lambda: False,
        settle_seconds: float = 0.4,
        transition_timeout_seconds: float = 30.0,
        max_pages: int = 5,
        monotonic: Callable[[], float] = time.monotonic,
        sleep: Callable[[float], None] = time.sleep,
    ) -> 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._transition_timeout_seconds = max(
            0.0, float(transition_timeout_seconds)
        )
        self._max_pages = max(1, int(max_pages))
        self._monotonic = monotonic
        self._sleep = sleep
        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] = {}
        detail_evidence: Optional[PurchaseOrderCandidate] = None
        detail_context_valid = True
        previous_xml = ""
        pages_scanned = 0
        for _ in range(self._max_pages):
            self._check_cancelled()
            xml_data = device.dump_hierarchy(compressed=False)
            root = _parse_xml(xml_data)
            self._raise_for_special_page(root)
            pages_scanned += 1
            for candidate in parse_order_candidates(xml_data, query):
                previous = found.get(candidate.order_no)
                merged = (
                    _merge_order_evidence(previous, candidate)
                    if previous is not None
                    else candidate
                )
                if merged is not None:
                    found[candidate.order_no] = merged

            if detail_context_valid and _is_single_order_detail(root):
                current_evidence = _parse_order_detail_evidence(root, query)
                if detail_evidence is None:
                    detail_evidence = current_evidence
                else:
                    merged = _merge_order_evidence(
                        detail_evidence, current_evidence
                    )
                    if merged is None:
                        detail_context_valid = False
                        detail_evidence = None
                    else:
                        detail_evidence = merged
            else:
                detail_context_valid = False
                detail_evidence = None

            current_xml = str(xml_data)
            if current_xml == previous_xml:
                break
            previous_xml = current_xml
            width, height = device.window_size()
            device.swipe(
                width // 2,
                int(height * 0.78),
                width // 2,
                int(height * 0.32),
                0.35,
            )
            self._settle()
        if detail_context_valid and detail_evidence and detail_evidence.order_no:
            previous = found.get(detail_evidence.order_no)
            merged = (
                _merge_order_evidence(previous, detail_evidence)
                if previous is not None
                else detail_evidence
            )
            if merged is not None:
                found[detail_evidence.order_no] = merged
        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:
        deadline = self._monotonic() + self._transition_timeout_seconds
        opened_order_area = False
        checked_foreground = False
        started_pdd = False
        pressed_back = False
        clicked_navigation: set[str] = set()
        last_reason = "等待 PDD 待付款页稳定"
        while self._monotonic() <= deadline:
            self._check_cancelled()
            root = _parse_xml(device.dump_hierarchy(compressed=False))
            if self._monotonic() > deadline:
                last_reason = "读取手机控件树超时"
                break
            if not _package_hint_from_tree(root):
                if not checked_foreground:
                    current_package = _foreground_package_from_tree(root)
                    if not current_package:
                        current = device.app_current()
                        current_package = str(current.get("package") or "")
                        if self._monotonic() > deadline:
                            last_reason = "查询手机前台应用超时"
                            break
                    checked_foreground = True
                    if current_package != PDD_PACKAGE_NAME:
                        device.app_start(PDD_PACKAGE_NAME)
                        started_pdd = True
                        last_reason = "已返回 PDD，等待待付款页加载"
                elif started_pdd:
                    last_reason = "PDD 正在从微信中间页恢复"
                else:
                    last_reason = "PDD 控件树正在加载"
                self._settle()
                continue
            self._raise_for_special_page(root)
            combined = _subtree_text(root)
            if _is_single_order_detail(root) or (
                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):
                if not pressed_back:
                    device.press("back")
                    pressed_back = True
                last_reason = "已只读返回，等待待付款页稳定"
                self._settle()
                continue
            if "我的订单" in combined:
                if (
                    "待付款" not in clicked_navigation
                    and self._click_unique_label(device, root, "待付款")
                ):
                    clicked_navigation.add("待付款")
                    opened_order_area = True
                    last_reason = "已打开待付款，等待订单列表加载"
                    self._settle()
                    continue
                if (
                    "我的订单" not in clicked_navigation
                    and self._click_unique_label(device, root, "我的订单")
                ):
                    clicked_navigation.add("我的订单")
                    opened_order_area = True
                    last_reason = "已打开我的订单，等待待付款入口加载"
                    self._settle()
                    continue
            if (
                "个人中心" not in clicked_navigation
                and self._click_unique_label(device, root, "个人中心")
            ):
                clicked_navigation.add("个人中心")
                last_reason = "已打开个人中心，等待我的订单入口加载"
                self._settle()
                continue
            last_reason = "PDD 页面仍在过渡，尚未出现唯一只读订单入口"
            self._settle()
        raise PddPurchaseReconcileError(
            "RECONCILE_ORDER_PAGE_TIMEOUT",
            f"等待 PDD 待付款页稳定超时：{last_reason}",
        )

    @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:
            self._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,
    )