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