2060 lines
76 KiB
Python
2060 lines
76 KiB
Python
"""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"(?<!\d)\d{3}\*{4}\d{4}(?!\d)")
|
||
_ADDRESS_SEPARATOR_PATTERN = re.compile(r"[-_]")
|
||
_COLOROS_PASSWORD_PACKAGE = "com.coloros.codebook"
|
||
_READY_MARKERS = ("发起拼单", "立即购买", "单独购买", "免拼购买", "快要抢光")
|
||
_LOGIN_MARKERS = ("手机号登录", "登录后继续", "验证码登录", "账号登录")
|
||
_CAPTCHA_MARKERS = ("请完成验证", "拖动滑块", "安全验证", "点击图中")
|
||
_RISK_MARKERS = ("操作频繁", "异常请求", "风险提示", "账号异常")
|
||
_PAYMENT_MARKERS = ("输入支付密码", "立即支付", "支付成功", "支付失败")
|
||
_FINAL_SUBMIT_MARKERS = ("提交订单", "现在买，仅", "确认购买")
|
||
_OUT_OF_STOCK_MARKERS = ("已售罄", "暂时缺货", "库存不足", "该商品已售罄")
|
||
_ADDRESS_ACTION_SETTLE_SECONDS = 1.0
|
||
_COLOR_DIMENSION_HEADINGS = frozenset(
|
||
{"颜色分类", "顏色分類", "颜色", "顏色", "款式", "花色", "颜色款式"}
|
||
)
|
||
_SECOND_DIMENSION_HEADINGS = frozenset(
|
||
{"尺码", "尺碼", "尺寸", "套餐", "规格", "規格", "型号", "型號"}
|
||
)
|
||
_PANEL_TOP_MAX_SWIPES = 8
|
||
_PANEL_TOP_STABLE_READS = 2
|
||
_SUPPORTED_OPTION_KEYS = frozenset({"color", "size"})
|
||
_MAX_QUANTITY_BUTTON_CLICKS = 5
|
||
_KEYBOARD_STATUS_MARKERS = (
|
||
"minputshown",
|
||
"misinputviewshown",
|
||
"minputviewshown",
|
||
"inputshown",
|
||
)
|
||
_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 _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 _dimension_heading(label: str) -> str:
|
||
"""只接受独立规格标题，明确排除选择摘要和提交提示。"""
|
||
|
||
compact = re.sub(r"\s+", "", str(label or ""))
|
||
if not compact or compact.startswith(
|
||
("请选择", "請選擇", "已选", "已選", "已选择", "已選擇")
|
||
):
|
||
return ""
|
||
if compact.startswith("选择") and "后" in compact:
|
||
return ""
|
||
compact = re.sub(r"[（(]\d+[)）]$", "", compact)
|
||
return compact
|
||
|
||
|
||
def _screen_bounds(root: ET.Element) -> Optional[Bounds]:
|
||
parsed = [
|
||
bounds
|
||
for node in root.iter("node")
|
||
if (bounds := _parse_bounds(node.get("bounds", ""))) is not None
|
||
]
|
||
if not parsed:
|
||
return None
|
||
return 0, 0, max(item[2] for item in parsed), max(item[3] for item in parsed)
|
||
|
||
|
||
def _has_visible_color_region(root: ET.Element) -> bool:
|
||
"""正式颜色标题下必须同时存在至少一个可靠的可点击选项。"""
|
||
|
||
parents = _parent_nodes(root)
|
||
headings: list[Bounds] = []
|
||
second_heading_tops: list[int] = []
|
||
for node in root.iter("node"):
|
||
heading = _dimension_heading(_label(node))
|
||
bounds = _parse_bounds(node.get("bounds", ""))
|
||
if bounds is None or node.get("visible-to-user", "true") == "false":
|
||
continue
|
||
if heading in _COLOR_DIMENSION_HEADINGS:
|
||
headings.append(bounds)
|
||
elif heading in _SECOND_DIMENSION_HEADINGS:
|
||
second_heading_tops.append(bounds[1])
|
||
|
||
screen = _screen_bounds(root)
|
||
if not headings or screen is None:
|
||
return False
|
||
|
||
excluded = {
|
||
"增加数量",
|
||
"减少数量",
|
||
"提交订单",
|
||
"确定",
|
||
"打开大图",
|
||
}
|
||
for heading_bounds in headings:
|
||
boundary = min(
|
||
(
|
||
top
|
||
for top in second_heading_tops
|
||
if top > heading_bounds[3]
|
||
),
|
||
default=screen[3],
|
||
)
|
||
for node in root.iter("node"):
|
||
label = _label(node).strip()
|
||
bounds = _parse_bounds(node.get("bounds", ""))
|
||
if not label or bounds is None:
|
||
continue
|
||
if bounds[1] < heading_bounds[3] or bounds[1] >= boundary:
|
||
continue
|
||
if _dimension_heading(label) in (
|
||
_COLOR_DIMENSION_HEADINGS | _SECOND_DIMENSION_HEADINGS
|
||
):
|
||
continue
|
||
compact = label.replace(" ", "")
|
||
if (
|
||
compact.startswith(
|
||
("请选择", "請選擇", "已选", "已選", "选择", "選擇")
|
||
)
|
||
or any(marker in compact for marker in excluded)
|
||
or re.fullmatch(r"[¥￥]?\d+(?:\.\d{1,2})?", compact)
|
||
):
|
||
continue
|
||
if _nearest_clickable_bounds(node, parents) is not None:
|
||
return True
|
||
return False
|
||
|
||
|
||
def _needs_color_region_restore(root: ET.Element) -> bool:
|
||
"""只有第二规格已出现且颜色摘要仍存在时，才判定面板停在中部。"""
|
||
|
||
labels = [_label(node).strip() for node in root.iter("node")]
|
||
has_color_summary = any(
|
||
label.replace(" ", "").startswith(
|
||
("请选择", "請選擇", "已选", "已選", "已选择", "已選擇")
|
||
)
|
||
and any(heading in label.replace(" ", "") for heading in _COLOR_DIMENSION_HEADINGS)
|
||
for label in labels
|
||
)
|
||
has_second_heading = any(
|
||
_dimension_heading(label) in _SECOND_DIMENSION_HEADINGS
|
||
for label in labels
|
||
)
|
||
return has_color_summary and has_second_heading
|
||
|
||
|
||
def _purchase_panel_scroll_region(root: ET.Element) -> Optional[Bounds]:
|
||
"""返回规格面板中唯一可靠的纵向滚动容器。"""
|
||
|
||
screen = _screen_bounds(root)
|
||
if screen is None:
|
||
return None
|
||
screen_width = screen[2] - screen[0]
|
||
screen_height = screen[3] - screen[1]
|
||
candidates: list[Bounds] = []
|
||
for node in root.iter("node"):
|
||
if (
|
||
node.get("scrollable") != "true"
|
||
or node.get("package") != PDD_PACKAGE_NAME
|
||
or node.get("visible-to-user", "true") == "false"
|
||
or node.get("enabled", "true") == "false"
|
||
):
|
||
continue
|
||
bounds = _parse_bounds(node.get("bounds", ""))
|
||
if bounds is None:
|
||
continue
|
||
width = bounds[2] - bounds[0]
|
||
height = bounds[3] - bounds[1]
|
||
if width >= screen_width * 0.60 and height >= screen_height * 0.20:
|
||
candidates.append(bounds)
|
||
unique = sorted(set(candidates))
|
||
return unique[0] if len(unique) == 1 else None
|
||
|
||
|
||
def _region_signature(root: ET.Element, region: Bounds) -> str:
|
||
"""生成当前滚动视口的短签名，只用于判断是否已经无法继续归位。"""
|
||
|
||
parts = []
|
||
for node in root.iter("node"):
|
||
bounds = _parse_bounds(node.get("bounds", ""))
|
||
if bounds is None:
|
||
continue
|
||
center_x = (bounds[0] + bounds[2]) // 2
|
||
center_y = (bounds[1] + bounds[3]) // 2
|
||
if not (
|
||
region[0] <= center_x <= region[2]
|
||
and region[1] <= center_y <= region[3]
|
||
):
|
||
continue
|
||
label = _label(node)
|
||
if label:
|
||
parts.append(f"{label}|{bounds}")
|
||
return hashlib.sha256("\n".join(parts).encode("utf-8")).hexdigest()
|
||
|
||
|
||
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 _node_package(
|
||
node: ET.Element, parents: Mapping[ET.Element, ET.Element]
|
||
) -> str:
|
||
"""返回节点自身或最近祖先声明的包名。"""
|
||
|
||
current: Optional[ET.Element] = node
|
||
while current is not None:
|
||
if package_name := current.get("package", "").strip():
|
||
return package_name
|
||
current = parents.get(current)
|
||
return ""
|
||
|
||
|
||
def _pdd_text_targets(root: ET.Element, exact_text: str) -> list[Bounds]:
|
||
"""只返回 PDD 页面中指定文字的最近点击容器。"""
|
||
|
||
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 _node_package(node, parents) == PDD_PACKAGE_NAME
|
||
if (bounds := _nearest_clickable_bounds(node, parents)) is not None
|
||
}
|
||
return sorted(targets)
|
||
|
||
|
||
def _address_modify_target_info(
|
||
root: ET.Element,
|
||
) -> tuple[list[Bounds], dict[str, int]]:
|
||
"""返回默认地址修改目标及不含地址原文的结构诊断。"""
|
||
|
||
address_cards = [
|
||
card
|
||
for recycler in root.iter("node")
|
||
if recycler.get("class") == "androidx.recyclerview.widget.RecyclerView"
|
||
for card in list(recycler)
|
||
if _pdd_text_targets(card, "修改")
|
||
]
|
||
all_targets = _pdd_text_targets(root, "修改")
|
||
if len(address_cards) <= 1:
|
||
targets = all_targets
|
||
default_card_count = sum(
|
||
1 for card in address_cards if "已设默认" in _labels(card)
|
||
)
|
||
return targets, {
|
||
"address_card_count": len(address_cards),
|
||
"default_card_count": default_card_count,
|
||
"all_modify_target_count": len(all_targets),
|
||
}
|
||
|
||
default_cards = [
|
||
card for card in address_cards if "已设默认" in _labels(card)
|
||
]
|
||
if len(default_cards) != 1:
|
||
targets = []
|
||
else:
|
||
targets = _pdd_text_targets(default_cards[0], "修改")
|
||
return targets, {
|
||
"address_card_count": len(address_cards),
|
||
"default_card_count": len(default_cards),
|
||
"all_modify_target_count": len(all_targets),
|
||
}
|
||
|
||
|
||
def _address_modify_targets(root: ET.Element) -> list[Bounds]:
|
||
"""多地址时只返回唯一“已设默认”卡片内的修改按钮。"""
|
||
|
||
targets, _diagnostics = _address_modify_target_info(root)
|
||
return targets
|
||
|
||
|
||
def _pdd_title_back_targets(root: ET.Element) -> list[Bounds]:
|
||
"""只返回 PDD 地址页标题栏左侧的返回目标。"""
|
||
|
||
return [
|
||
bounds
|
||
for bounds in _pdd_text_targets(root, "返回")
|
||
if bounds[0] <= 240 and bounds[1] <= 320
|
||
]
|
||
|
||
|
||
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 _is_reliable_address_panel(
|
||
root: ET.Element, expected_address: str = ""
|
||
) -> bool:
|
||
"""用标题和唯一导航目标识别地址面板，地址全文只作增强证据。"""
|
||
|
||
labels = _labels(root)
|
||
if "收货地址" not in labels:
|
||
return False
|
||
if len(_address_modify_targets(root)) != 1:
|
||
return False
|
||
if len(_pdd_title_back_targets(root)) != 1:
|
||
return False
|
||
if expected_address and any(expected_address in label for label in labels):
|
||
return True
|
||
return True
|
||
|
||
|
||
def _has_coloros_password_overlay(root: ET.Element) -> bool:
|
||
"""只识别已知 ColorOS 密码本的可见焦点浮层。"""
|
||
|
||
return any(
|
||
node.get("package") == _COLOROS_PASSWORD_PACKAGE
|
||
and node.get("visible-to-user", "true") != "false"
|
||
and node.get("focused", "false") == "true"
|
||
for node in root.iter("node")
|
||
)
|
||
|
||
|
||
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 = _ADDRESS_SEPARATOR_PATTERN.split(
|
||
checked_address, maxsplit=1
|
||
)[0].strip()
|
||
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:
|
||
panel_xml = self._restore_purchase_panel_color_region(panel_xml)
|
||
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 _restore_purchase_panel_color_region(self, xml_data: str) -> str:
|
||
"""选择颜色前把规格内容有界归位到正式颜色区域。"""
|
||
|
||
root = _parse_xml(xml_data)
|
||
if _has_visible_color_region(root):
|
||
return xml_data
|
||
if not _needs_color_region_restore(root):
|
||
# 兼容不暴露正式标题或滚动属性、但原选择器可以处理的旧面板。
|
||
return xml_data
|
||
|
||
region = _purchase_panel_scroll_region(root)
|
||
if region is None:
|
||
self._raise_color_region_not_found(
|
||
swipe_count=0,
|
||
stable_reads=0,
|
||
scroll_region_found=False,
|
||
)
|
||
|
||
device = self._require_device()
|
||
previous_signature = _region_signature(root, region)
|
||
unchanged_reads = 0
|
||
swipe_count = 0
|
||
for _ in range(_PANEL_TOP_MAX_SWIPES):
|
||
self._check_cancelled("purchase_select_options")
|
||
left, top, right, bottom = region
|
||
height = bottom - top
|
||
x = (left + right) // 2
|
||
# 手指向下滑，让规格内容返回顶部；坐标始终限制在规格容器内。
|
||
device.swipe(
|
||
x,
|
||
top + int(height * 0.25),
|
||
x,
|
||
top + int(height * 0.82),
|
||
duration=0.35,
|
||
)
|
||
swipe_count += 1
|
||
self._sleep(0.2)
|
||
|
||
current_xml = self._dump_hierarchy()
|
||
root = _parse_xml(current_xml)
|
||
page_kind = _page_kind(root, "")
|
||
if page_kind in {
|
||
"captcha",
|
||
"login_required",
|
||
"risk_control",
|
||
"payment",
|
||
}:
|
||
self._raise_special_page(page_kind)
|
||
if page_kind != "order_confirmation":
|
||
self._raise_color_region_not_found(
|
||
swipe_count=swipe_count,
|
||
stable_reads=unchanged_reads,
|
||
scroll_region_found=True,
|
||
)
|
||
if _has_visible_color_region(root):
|
||
return current_xml
|
||
|
||
current_region = _purchase_panel_scroll_region(root)
|
||
if current_region is None:
|
||
break
|
||
region = current_region
|
||
signature = _region_signature(root, region)
|
||
if signature == previous_signature:
|
||
unchanged_reads += 1
|
||
else:
|
||
unchanged_reads = 0
|
||
previous_signature = signature
|
||
if unchanged_reads >= _PANEL_TOP_STABLE_READS:
|
||
break
|
||
|
||
self._raise_color_region_not_found(
|
||
swipe_count=swipe_count,
|
||
stable_reads=unchanged_reads,
|
||
scroll_region_found=True,
|
||
)
|
||
|
||
def _raise_color_region_not_found(
|
||
self,
|
||
*,
|
||
swipe_count: int,
|
||
stable_reads: int,
|
||
scroll_region_found: bool,
|
||
) -> None:
|
||
diagnostics: dict[str, Any] = {
|
||
"swipe_count": swipe_count,
|
||
"stable_reads": stable_reads,
|
||
"scroll_region_found": scroll_region_found,
|
||
}
|
||
artifact = self._save_last_xml("purchase-color-region-not-found")
|
||
if artifact is not None:
|
||
diagnostics["artifacts"] = [artifact]
|
||
raise PddPurchaseError(
|
||
"PURCHASE_PANEL_COLOR_REGION_NOT_FOUND",
|
||
"规格面板已打开，但无法回到颜色分类区域",
|
||
step="purchase_select_options",
|
||
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",
|
||
"规格面板没有唯一可靠的收货地址入口",
|
||
)
|
||
self._settle_after_address_action()
|
||
|
||
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._settle_after_address_action()
|
||
save_target = self._wait_for_stable_address_save(new_address)
|
||
self._click_unique_address_target(
|
||
[save_target],
|
||
"PURCHASE_ADDRESS_SAVE_AMBIGUOUS",
|
||
"修改页没有唯一可靠的保存按钮",
|
||
)
|
||
self._settle_after_address_action()
|
||
|
||
saved_root, saved_page = self._wait_after_address_save(new_address)
|
||
if saved_page == "confirmation":
|
||
return
|
||
back_targets = _pdd_title_back_targets(saved_root)
|
||
self._click_unique_address_target(
|
||
back_targets,
|
||
"PURCHASE_ADDRESS_BACK_AMBIGUOUS",
|
||
"收货地址页没有唯一可靠的返回按钮",
|
||
)
|
||
self._settle_after_address_action()
|
||
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 _settle_after_address_action(self) -> None:
|
||
"""关键地址操作后留出固定时间，再继续按页面状态确认。"""
|
||
|
||
self._sleep(_ADDRESS_ACTION_SETTLE_SECONDS)
|
||
|
||
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
|
||
target_diagnostics = {
|
||
"address_card_count": 0,
|
||
"default_card_count": 0,
|
||
"all_modify_target_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, target_diagnostics = _address_modify_target_info(root)
|
||
candidate_count = len(targets)
|
||
if candidate_count != 1:
|
||
stable_target = None
|
||
stable_reads = 0
|
||
else:
|
||
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",
|
||
"收货地址页没有唯一可靠的修改按钮",
|
||
)
|
||
self._settle_after_address_action()
|
||
# 固定稳定等待不能占用原页面切换超时预算。
|
||
deadline += _ADDRESS_ACTION_SETTLE_SECONDS
|
||
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,
|
||
"stable_reads": stable_reads,
|
||
**target_diagnostics,
|
||
}
|
||
artifact = self._save_last_xml("purchase-address-edit-timeout")
|
||
if artifact is not None:
|
||
diagnostics["artifacts"] = [artifact]
|
||
ambiguous = observed_page == "panel" and candidate_count != 1
|
||
raise PddPurchaseError(
|
||
(
|
||
"PURCHASE_ADDRESS_EDIT_AMBIGUOUS"
|
||
if ambiguous
|
||
else "PURCHASE_ADDRESS_PAGE_TIMEOUT"
|
||
),
|
||
(
|
||
"收货地址页没有唯一可靠的修改按钮"
|
||
if ambiguous
|
||
else "收货地址页面切换或保存确认超时，未进入不可逆下单阶段"
|
||
),
|
||
step="purchase_update_shipping_address",
|
||
diagnostics=diagnostics,
|
||
)
|
||
|
||
def _wait_for_stable_address_save(self, expected_address: str) -> Bounds:
|
||
"""地址和唯一保存目标连续稳定后，返回最终复核过的点击位置。"""
|
||
|
||
deadline = self._monotonic() + self._panel_timeout
|
||
stable_target: Optional[Bounds] = None
|
||
stable_reads = 0
|
||
candidate_count = 0
|
||
observed_page = "unknown"
|
||
address_matches = False
|
||
|
||
while self._monotonic() < deadline:
|
||
self._check_cancelled("purchase_update_shipping_address")
|
||
root = _parse_xml(self._dump_hierarchy())
|
||
observed_page = _address_page_kind(root)
|
||
editor = _shipping_address_editor(root)
|
||
address_matches = editor is not None and editor[0] == expected_address
|
||
save_targets = _clickable_text_targets(root, "保存")
|
||
candidate_count = len(save_targets)
|
||
|
||
ready = (
|
||
observed_page == "edit"
|
||
and address_matches
|
||
and candidate_count == 1
|
||
)
|
||
if ready and save_targets[0] == stable_target:
|
||
stable_reads += 1
|
||
elif ready:
|
||
stable_target = save_targets[0]
|
||
stable_reads = 1
|
||
else:
|
||
stable_target = None
|
||
stable_reads = 0
|
||
|
||
if stable_reads >= 2 and stable_target is not None:
|
||
# 给表单一次短暂提交准备时间，然后必须用最新控件树再验证。
|
||
self._sleep(0.3)
|
||
latest_root = _parse_xml(self._dump_hierarchy())
|
||
latest_editor = _shipping_address_editor(latest_root)
|
||
latest_targets = _clickable_text_targets(latest_root, "保存")
|
||
latest_ready = (
|
||
_address_page_kind(latest_root) == "edit"
|
||
and latest_editor is not None
|
||
and latest_editor[0] == expected_address
|
||
and latest_targets == [stable_target]
|
||
)
|
||
if latest_ready:
|
||
return stable_target
|
||
stable_target = None
|
||
stable_reads = 0
|
||
observed_page = _address_page_kind(latest_root)
|
||
address_matches = (
|
||
latest_editor is not None
|
||
and latest_editor[0] == expected_address
|
||
)
|
||
candidate_count = len(latest_targets)
|
||
|
||
self._sleep(0.2)
|
||
|
||
diagnostics: dict[str, Any] = {
|
||
"expected_page": "edit_ready_to_save",
|
||
"observed_page": observed_page,
|
||
"address_matches": address_matches,
|
||
"save_candidate_count": candidate_count,
|
||
"stable_reads": stable_reads,
|
||
}
|
||
artifact = self._save_last_xml("purchase-address-save-not-ready")
|
||
if artifact is not None:
|
||
diagnostics["artifacts"] = [artifact]
|
||
input_mismatch = observed_page == "edit" and not address_matches
|
||
raise PddPurchaseError(
|
||
(
|
||
"PURCHASE_ADDRESS_INPUT_MISMATCH"
|
||
if input_mismatch
|
||
else "PURCHASE_ADDRESS_SAVE_NOT_READY"
|
||
),
|
||
(
|
||
"详细地址输入后回读不一致，未保存也未下单"
|
||
if input_mismatch
|
||
else "地址编辑页尚未稳定，未点击保存也未下单"
|
||
),
|
||
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 _wait_after_address_save(
|
||
self, expected_address: str
|
||
) -> tuple[ET.Element, str]:
|
||
"""等待保存后的地址面板；已知密码本浮层最多按一次返回。"""
|
||
|
||
deadline = self._monotonic() + self._panel_timeout
|
||
observed_page = "unknown"
|
||
overlay_seen = False
|
||
overlay_stable_reads = 0
|
||
system_back_attempts = 0
|
||
|
||
while self._monotonic() < deadline:
|
||
self._check_cancelled("purchase_update_shipping_address")
|
||
root = _parse_xml(self._dump_hierarchy())
|
||
if _has_coloros_password_overlay(root):
|
||
observed_page = "coloros_password_overlay"
|
||
overlay_seen = True
|
||
overlay_stable_reads += 1
|
||
if overlay_stable_reads >= 3 and system_back_attempts == 0:
|
||
# 只关闭已知系统浮层；保存按钮和 PDD 页面都不重试点击。
|
||
self._require_device().press("back")
|
||
self._settle_after_address_action()
|
||
# 固定稳定等待不能占用原浮层关闭确认超时预算。
|
||
deadline += _ADDRESS_ACTION_SETTLE_SECONDS
|
||
system_back_attempts = 1
|
||
overlay_stable_reads = 0
|
||
self._sleep(0.2)
|
||
continue
|
||
|
||
overlay_stable_reads = 0
|
||
if _is_reliable_address_panel(root, expected_address):
|
||
return root, "saved"
|
||
if _page_kind(root, "") == "order_confirmation":
|
||
return root, "confirmation"
|
||
observed_page = _address_page_kind(root)
|
||
self._sleep(0.2)
|
||
|
||
diagnostics: dict[str, Any] = {
|
||
"expected_page": "saved",
|
||
"observed_page": observed_page,
|
||
"coloros_password_overlay_seen": overlay_seen,
|
||
"system_back_attempts": system_back_attempts,
|
||
}
|
||
artifact = self._save_last_xml("purchase-address-saved-timeout")
|
||
if artifact is not None:
|
||
diagnostics["artifacts"] = [artifact]
|
||
save_not_effective = observed_page == "edit"
|
||
raise PddPurchaseError(
|
||
(
|
||
"PURCHASE_ADDRESS_SAVE_NOT_EFFECTIVE"
|
||
if save_not_effective
|
||
else "PURCHASE_ADDRESS_PAGE_TIMEOUT"
|
||
),
|
||
(
|
||
"点击保存后仍停留在地址编辑页，未进入不可逆下单阶段"
|
||
if save_not_effective
|
||
else "收货地址页面切换或保存确认超时，未进入不可逆下单阶段"
|
||
),
|
||
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",
|
||
)
|