"""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_ORDER_CONFIRMATION, PAGE_PAYMENT, PAGE_RISK_CONTROL, PAGE_UNKNOWN, 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 ( color_selection_failure_reason, 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})?)") _MASKED_PHONE_PATTERN = re.compile(r"(? 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 _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 _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 _parent_nodes(root: ET.Element) -> dict[ET.Element, ET.Element]: return { child: parent for parent in root.iter() for child in parent if child.tag == "node" } def _nearest_clickable_bounds( node: ET.Element, parents: Mapping[ET.Element, ET.Element] ) -> Optional[Bounds]: """从文字节点向上寻找最近的可点击、可见且启用容器。""" current: Optional[ET.Element] = node while current is not None: bounds = _parse_bounds(current.get("bounds", "")) if ( bounds is not None and current.get("clickable") == "true" and current.get("enabled", "true") != "false" and current.get("visible-to-user", "true") != "false" ): return bounds current = parents.get(current) return None def _clickable_text_targets(root: ET.Element, exact_text: str) -> list[Bounds]: """按明确文字返回最近的唯一点击容器候选。""" parents = _parent_nodes(root) targets = { bounds for node in root.iter("node") if exact_text in { node.get("text", "").strip(), node.get("content-desc", "").strip(), } if (bounds := _nearest_clickable_bounds(node, parents)) is not None } return sorted(targets) def _address_entry_targets(root: ET.Element) -> list[Bounds]: """通过脱敏手机号结构定位规格面板顶部的收货地址卡片。""" parents = _parent_nodes(root) targets = { bounds for node in root.iter("node") if _MASKED_PHONE_PATTERN.search(_label(node)) if (bounds := _nearest_clickable_bounds(node, parents)) is not None } return sorted(targets) def _shipping_address_editor(root: ET.Element) -> Optional[tuple[str, Bounds]]: """在修改页中寻找与“详细地址”标签同一行的唯一输入框。""" address_labels = [ bounds for node in root.iter("node") if "详细地址" in _label(node).replace(" ", "") if (bounds := _parse_bounds(node.get("bounds", ""))) is not None ] editors = [ (node.get("text", ""), bounds) for node in root.iter("node") if node.get("class") == "android.widget.EditText" if node.get("enabled", "true") != "false" if node.get("visible-to-user", "true") != "false" if (bounds := _parse_bounds(node.get("bounds", ""))) is not None ] matches = [ editor for editor in editors for label_bounds in address_labels if editor[1][1] <= label_bounds[3] and editor[1][3] >= label_bounds[1] and editor[1][0] >= label_bounds[0] ] return matches[0] if len(matches) == 1 and matches[0][0].strip() else None def _address_page_kind(root: ET.Element) -> str: """识别地址面板、编辑页和返回后的规格确认页。""" labels = _labels(root) if any("修改收货地址" in label.replace(" ", "") for label in labels): return "edit" if _shipping_address_editor(root) is not None: return "edit" if "收货地址" in labels: return "panel" if _page_kind(root, "") == "order_confirmation": return "confirmation" return "unknown" def _tagged_shipping_address(address: str, purchase_number: str) -> str: """只替换程序生成的末尾标记，保留完整真实地址主体。""" checked_address = str(address or "").strip() checked_number = str(purchase_number or "").strip() if not checked_address or not checked_number: raise ValueError("地址和采购编号不能为空") if any(character in checked_number for character in "\r\n\t"): raise ValueError("采购编号不能包含控制字符") address_body = _PROGRAM_ADDRESS_SUFFIX_PATTERN.sub("", checked_address) if not address_body: raise ValueError("地址主体不能为空") return f"{address_body}_{checked_number}" 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: kind = classify_pdd_page(root, current_package).kind contextual_targets = _contextual_confirm_targets(root) if kind in {PAGE_UNKNOWN, PAGE_GOODS} and len(contextual_targets) == 1: return PAGE_ORDER_CONFIRMATION return 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 _quantity_button_targets(root: ET.Element, description: str) -> list[Bounds]: """返回唯一、可点击的数量加减按钮坐标候选。""" targets = set() for node in root.iter("node"): labels = { node.get("text", "").strip(), node.get("content-desc", "").strip(), } bounds = _parse_bounds(node.get("bounds", "")) if description not in labels or bounds is None: continue if node.get("clickable") != "true": continue if node.get("enabled") == "false" or node.get("visible-to-user") == "false": continue targets.add(bounds) return sorted(targets) def _shell_output(response: Any) -> str: """兼容 uiautomator2 ShellResponse 和测试中的普通字符串。""" output = getattr(response, "output", response) if isinstance(output, bytes): return output.decode("utf-8", errors="replace") return str(output or "") def _keyboard_is_shown(device: Any) -> Optional[bool]: """读取安卓输入法状态；无法可靠判断时返回 None。""" try: response = device.shell("dumpsys input_method") except (AttributeError, OSError, RuntimeError): return None normalized = re.sub(r"\s+", "", _shell_output(response)).lower() found_false = False for marker in _KEYBOARD_STATUS_MARKERS: if f"{marker}=true" in normalized: return True if f"{marker}=false" in normalized: found_false = True return False if found_false else None 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) targets.update(_contextual_confirm_targets(root)) return sorted(targets) def _contextual_confirm_targets(root: ET.Element) -> list[Bounds]: """只在强证据规格面板内接受底部“确定”作为最终提交按钮。""" nodes = list(root.iter("node")) if sum( node.get("package") == PDD_PACKAGE_NAME for node in nodes ) < 3: return [] parsed_bounds = [ bounds for node in nodes if (bounds := _parse_bounds(node.get("bounds", ""))) is not None ] if not parsed_bounds: return [] screen_right = max(bounds[2] for bounds in parsed_bounds) screen_bottom = max(bounds[3] for bounds in parsed_bounds) screen_area = screen_right * screen_bottom parents = { child: parent for parent in root.iter() for child in parent if child.tag == "node" } targets: set[Bounds] = set() for confirm in nodes: if _label(confirm).strip() not in {"确定", "確定"}: continue bounds = _parse_bounds(confirm.get("bounds", "")) if bounds is None or confirm.get("clickable") != "true": continue if ( confirm.get("visible-to-user") != "true" or confirm.get("enabled") != "true" ): continue if (bounds[1] + bounds[3]) // 2 < screen_bottom * 0.6: continue panel = parents.get(confirm) while panel is not None: panel_bounds = _parse_bounds(panel.get("bounds", "")) if panel_bounds is not None: area = (panel_bounds[2] - panel_bounds[0]) * ( panel_bounds[3] - panel_bounds[1] ) if (not screen_area or area < screen_area * 0.95) and ( _has_confirmation_panel_evidence(panel) ): targets.add(bounds) break panel = parents.get(panel) return sorted(targets) def _has_confirmation_panel_evidence(panel: ET.Element) -> bool: """确认标题、已选摘要、规格标题和数量控件位于同一面板。""" nodes = list(panel.iter("node")) labels = [_label(node).replace(" ", "") for node in nodes] has_title = any( label in {"确认款式", "確認款式", "确认规格", "確認規格"} for label in labels ) has_summary = any( label.startswith(("已选", "已選", "已选择", "已選擇", "请选择", "請選擇")) for label in labels ) has_dimension = any( re.sub(r"[（(]\d+[)）]$", "", label) in { "颜色分类", "顏色分類", "颜色", "顏色", "尺码", "尺碼", "规格", "規格", "款式", "套餐", } for label in labels ) editors = [ node for node in nodes if node.get("class") == "android.widget.EditText" and node.get("text", "").strip().isdigit() and int(node.get("text", "0")) > 0 ] has_decrease = len(_quantity_targets_in_nodes(nodes, "减少数量")) == 1 has_increase = len(_quantity_targets_in_nodes(nodes, "增加数量")) == 1 return ( has_title and has_summary and has_dimension and len(editors) == 1 and has_decrease and has_increase ) def _quantity_targets_in_nodes( nodes: list[ET.Element], description: str ) -> set[Bounds]: """读取指定面板内唯一、可用的数量按钮。""" targets: set[Bounds] = set() for node in nodes: if description not in { node.get("text", "").strip(), node.get("content-desc", "").strip(), }: continue bounds = _parse_bounds(node.get("bounds", "")) if bounds is None or node.get("clickable") != "true": continue if node.get("enabled", "true") == "false": continue if node.get("visible-to-user", "true") == "false": continue targets.add(bounds) return 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: trace = current_performance_trace() stage = trace.stage("purchase_read_state") if trace else nullcontext() with stage: xml_data = self._dump_hierarchy() root = _parse_xml(xml_data) current_package = _package_hint_from_tree(root) if not current_package: current = device.app_current() current_package = str(current.get("package") or "") kind = _page_kind(root, current_package) 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: trace = current_performance_trace() stage = ( trace.stage("purchase_select_color") if trace else nullcontext() ) with stage: color_selected = self._select_color( device, panel_xml, color, action_delay=0.2 ) if not color_selected: failed_xml = self._dump_hierarchy() failure_reason = color_selection_failure_reason( failed_xml, color ) diagnostics: dict[str, Any] = { "selection_failure": failure_reason, } artifact = self._save_last_xml( "purchase-color-selection-mismatch" ) if artifact is not None: diagnostics["artifacts"] = [artifact] messages = { "target_not_visible": f"没有找到目标颜色：{color}", "safe_target_missing": ( f"目标颜色没有完整可见的安全点击位置：{color}" ), "selection_unconfirmed": ( f"点击颜色后页面没有确认已选中：{color}" ), } raise PddPurchaseError( "PURCHASE_OPTIONS_MISMATCH", messages[failure_reason], step="purchase_select_options", diagnostics=diagnostics, ) size = checked.get("size") if size: latest_xml = self._dump_hierarchy() trace = current_performance_trace() stage = ( trace.stage("purchase_select_size") if trace else nullcontext() ) with stage: size_selected = self._select_size( device, latest_xml, size, action_delay=0.2 ) if not size_selected: 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: root = _parse_xml(self._dump_hierarchy()) current_quantity = _quantity(root) if current_quantity <= 0: raise PddPurchaseError( "PURCHASE_QUANTITY_CONTROL_MISSING", "页面没有唯一可靠的采购数量输入框", step="purchase_set_quantity", ) if current_quantity == quantity: return difference = quantity - current_quantity if abs(difference) <= _MAX_QUANTITY_BUTTON_CLICKS: description = "增加数量" if difference > 0 else "减少数量" targets = _quantity_button_targets(root, description) if len(targets) == 1: self._set_quantity_with_buttons( current_quantity, quantity, description ) return self._set_quantity_with_editor(quantity) except PddPurchaseError: raise except Exception as exc: self._raise_device_or_page_error(exc, "purchase_set_quantity") def _set_quantity_with_buttons( self, current_quantity: int, target_quantity: int, description: str ) -> None: """逐次点击加减按钮，每次都用最新控件树确认数量。""" device = self._require_device() step = 1 if target_quantity > current_quantity else -1 expected = current_quantity while expected != target_quantity: self._check_cancelled("purchase_set_quantity") root = _parse_xml(self._dump_hierarchy()) targets = _quantity_button_targets(root, description) if len(targets) != 1: raise PddPurchaseError( "PURCHASE_QUANTITY_CONTROL_MISSING", f"页面没有唯一可靠的“{description}”按钮", step="purchase_set_quantity", ) left, top, right, bottom = targets[0] device.click((left + right) // 2, (top + bottom) // 2) expected += step self._sleep(0.2) actual = _quantity(_parse_xml(self._dump_hierarchy())) if actual != expected: raise PddPurchaseError( "PURCHASE_QUANTITY_MISMATCH", f"调整采购数量后期望为 {expected}，页面实际为 {actual or '未知'}", step="purchase_set_quantity", ) def _set_quantity_with_editor(self, quantity: int) -> None: """输入框兜底；只在确认输入法显示后按一次返回键。""" device = self._require_device() 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", ) keyboard_shown = _keyboard_is_shown(device) if keyboard_shown is None: raise PddPurchaseError( "PURCHASE_KEYBOARD_DISMISS_FAILED", "无法确认安卓输入法状态，已停止采购以避免误按返回键", step="purchase_set_quantity", retryable=True, ) if keyboard_shown: device.press("back") for _attempt in range(25): self._check_cancelled("purchase_set_quantity") self._sleep(0.2) keyboard_shown = _keyboard_is_shown(device) if keyboard_shown is False: break if keyboard_shown is not False: raise PddPurchaseError( "PURCHASE_KEYBOARD_DISMISS_FAILED", "安卓输入法没有在规定时间内关闭，已停止采购", step="purchase_set_quantity", retryable=True, ) self._validate_panel_after_keyboard(quantity) def _validate_panel_after_keyboard(self, quantity: int) -> None: """输入框操作后重新确认规格面板和最终提交前状态。""" root = _parse_xml(self._dump_hierarchy()) if _page_kind(root, "") != "order_confirmation": raise PddPurchaseError( "PURCHASE_PANEL_LOST_AFTER_KEYBOARD", "关闭输入法后规格面板已经消失，已停止采购", step="purchase_set_quantity", retryable=True, ) if _quantity(root) != quantity: raise PddPurchaseError( "PURCHASE_QUANTITY_MISMATCH", "关闭输入法后采购数量发生变化", step="purchase_set_quantity", ) missing_options = [ value for value in self._requested_options.values() if not _selected(root, value) ] if missing_options: raise PddPurchaseError( "PURCHASE_OPTIONS_MISMATCH", "关闭输入法后已选颜色或尺码发生变化", step="purchase_set_quantity", ) if _price_cent(root) <= 0: raise PddPurchaseError( "PURCHASE_PRICE_MISSING", "关闭输入法后没有识别到有效价格", step="purchase_set_quantity", ) if len(_final_submit_targets(root)) != 1: raise PddPurchaseError( "PURCHASE_SUBMIT_TARGET_AMBIGUOUS", "关闭输入法后没有唯一可靠的下单按钮", step="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 update_shipping_address(self, purchase_number: str) -> None: """更新正式采购地址末尾标记，并返回原规格确认面板。""" self._check_cancelled("purchase_update_shipping_address") device = self._require_device() try: checked_number = str(purchase_number or "").strip() if not checked_number or any( character in checked_number for character in "\r\n\t" ): raise PddPurchaseError( "PURCHASE_NUMBER_INVALID", "采购编号为空或包含无效字符，未修改收货地址", step="purchase_update_shipping_address", ) root = _parse_xml(self._dump_hierarchy()) entry_targets = _address_entry_targets(root) self._click_unique_address_target( entry_targets, "PURCHASE_ADDRESS_ENTRY_AMBIGUOUS", "规格面板没有唯一可靠的收货地址入口", ) panel_root = self._wait_for_address_page("panel") edit_root = self._open_address_edit_page(panel_root) editor_data = _shipping_address_editor(edit_root) if editor_data is None: raise PddPurchaseError( "PURCHASE_ADDRESS_EDITOR_MISSING", "修改页没有唯一可靠的详细地址输入框", step="purchase_update_shipping_address", ) current_address, _editor_bounds = editor_data try: new_address = _tagged_shipping_address( current_address, checked_number ) except ValueError as exc: raise PddPurchaseError( "PURCHASE_ADDRESS_VALUE_INVALID", "当前详细地址或采购编号不符合更新要求", step="purchase_update_shipping_address", ) from exc editor = device( className="android.widget.EditText", text=current_address ) if int(getattr(editor, "count", 0)) != 1: raise PddPurchaseError( "PURCHASE_ADDRESS_EDITOR_AMBIGUOUS", "无法唯一选中详细地址输入框", step="purchase_update_shipping_address", ) editor.set_text(new_address) self._sleep(0.2) latest_edit_root = _parse_xml(self._dump_hierarchy()) latest_editor = _shipping_address_editor(latest_edit_root) if latest_editor is None or latest_editor[0] != new_address: raise PddPurchaseError( "PURCHASE_ADDRESS_INPUT_MISMATCH", "详细地址输入后回读不一致，未保存也未下单", step="purchase_update_shipping_address", ) save_targets = _clickable_text_targets(latest_edit_root, "保存") self._click_unique_address_target( save_targets, "PURCHASE_ADDRESS_SAVE_AMBIGUOUS", "修改页没有唯一可靠的保存按钮", ) saved_root = self._wait_for_address_page( "saved", expected_address=new_address ) back_targets = _clickable_text_targets(saved_root, "返回") self._click_unique_address_target( back_targets, "PURCHASE_ADDRESS_BACK_AMBIGUOUS", "收货地址页没有唯一可靠的返回按钮", ) self._wait_for_address_page("confirmation") except PddPurchaseError: raise except Exception as exc: self._raise_device_or_page_error( exc, "purchase_update_shipping_address" ) def _click_unique_address_target( self, targets: list[Bounds], code: str, message: str ) -> None: """地址流程只允许点击唯一候选，避免误触其他资料。""" if len(targets) != 1: raise PddPurchaseError( code, message, step="purchase_update_shipping_address", diagnostics={"candidate_count": len(targets)}, ) left, top, right, bottom = targets[0] self._require_device().click( (left + right) // 2, (top + bottom) // 2 ) def _open_address_edit_page(self, panel_root: ET.Element) -> ET.Element: """稳定识别“修改”按钮；首次点击丢失时最多安全重试一次。""" deadline = self._monotonic() + self._panel_timeout root = panel_root observed_page = _address_page_kind(root) candidate_count = 0 stable_target: Optional[Bounds] = None stable_reads = 0 first_clicked_target: Optional[Bounds] = None click_attempts = 0 while self._monotonic() < deadline: self._check_cancelled("purchase_update_shipping_address") observed_page = _address_page_kind(root) if observed_page == "edit": return root if observed_page == "panel": targets = _clickable_text_targets(root, "修改") candidate_count = len(targets) if candidate_count != 1: raise PddPurchaseError( "PURCHASE_ADDRESS_EDIT_AMBIGUOUS", "收货地址页没有唯一可靠的修改按钮", step="purchase_update_shipping_address", diagnostics={ "candidate_count": candidate_count, "click_attempts": click_attempts, "observed_page": observed_page, }, ) current_target = targets[0] retry_target_changed = ( first_clicked_target is not None and current_target != first_clicked_target ) if retry_target_changed or click_attempts >= 2: stable_target = None stable_reads = 0 elif current_target == stable_target: stable_reads += 1 else: stable_target = current_target stable_reads = 1 if stable_reads >= 2: self._click_unique_address_target( targets, "PURCHASE_ADDRESS_EDIT_AMBIGUOUS", "收货地址页没有唯一可靠的修改按钮", ) click_attempts += 1 if first_clicked_target is None: first_clicked_target = current_target stable_target = None stable_reads = 0 else: # 页面状态不明确时不能继续点击，避免误触其他页面。 candidate_count = 0 stable_target = None stable_reads = 0 self._sleep(0.2) root = _parse_xml(self._dump_hierarchy()) diagnostics: dict[str, Any] = { "expected_page": "edit", "observed_page": observed_page, "click_attempts": click_attempts, "candidate_count": candidate_count, } artifact = self._save_last_xml("purchase-address-edit-timeout") if artifact is not None: diagnostics["artifacts"] = [artifact] raise PddPurchaseError( "PURCHASE_ADDRESS_PAGE_TIMEOUT", "收货地址页面切换或保存确认超时，未进入不可逆下单阶段", step="purchase_update_shipping_address", diagnostics=diagnostics, ) def _wait_for_address_page( self, kind: str, *, expected_address: str = "" ) -> ET.Element: """等待地址页面切换；诊断信息不包含任何地址文字。""" deadline = self._monotonic() + self._panel_timeout observed_page = "unknown" while self._monotonic() < deadline: self._check_cancelled("purchase_update_shipping_address") root = _parse_xml(self._dump_hierarchy()) labels = _labels(root) observed_page = _address_page_kind(root) if kind == "panel" and "收货地址" in labels: return root if kind == "edit" and any( "修改收货地址" in label.replace(" ", "") for label in labels ): return root if ( kind == "saved" and "收货地址" in labels and expected_address and any(expected_address in label for label in labels) ): return root if ( kind == "confirmation" and _page_kind(root, "") == "order_confirmation" ): return root self._sleep(0.2) diagnostics: dict[str, Any] = { "expected_page": kind, "observed_page": observed_page, } artifact = self._save_last_xml(f"purchase-address-{kind}-timeout") if artifact is not None: diagnostics["artifacts"] = [artifact] raise PddPurchaseError( "PURCHASE_ADDRESS_PAGE_TIMEOUT", "收货地址页面切换或保存确认超时，未进入不可逆下单阶段", step="purchase_update_shipping_address", diagnostics=diagnostics, ) 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", )