"""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_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"}) _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 _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: 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, 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: 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], 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", )