2559 lines
97 KiB
Python
2559 lines
97 KiB
Python
"""PDD 商品采集基础服务。
|
||
|
||
本模块不依赖 Qt、SQLite 或 Admin。页面操作和 XML 解析拆开，解析函数可以
|
||
使用脱敏控件树单独测试；``collect`` 必须由后台工作线程调用。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import math
|
||
import re
|
||
import time
|
||
import xml.etree.ElementTree as ET
|
||
from contextlib import nullcontext
|
||
from dataclasses import dataclass
|
||
from datetime import datetime, timezone
|
||
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
|
||
from pathlib import Path
|
||
from typing import Any, Callable, Iterable, Mapping, Optional, Sequence
|
||
from urllib.parse import parse_qs, urlparse
|
||
|
||
from .pdd_device_service import PDD_PACKAGE_NAME, PddDeviceError, PddDeviceService
|
||
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,
|
||
GoodsOpenTracker,
|
||
PddPageObservation,
|
||
classify_pdd_page,
|
||
)
|
||
from .util.get_size_panle_coord import get_size_panel_coord
|
||
|
||
|
||
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})?)")
|
||
_QUANTITY_PATTERN = re.compile(r"(\d+(?:\.\d+)?)\s*(万|亿)?\s*(\+)?")
|
||
_DIMENSION_NAMES = (
|
||
"颜色分类",
|
||
"颜色",
|
||
"花色",
|
||
"款式",
|
||
"尺码",
|
||
"尺寸",
|
||
"套餐",
|
||
"规格",
|
||
"型号",
|
||
)
|
||
_LOGIN_MARKERS = ("手机号登录", "登录后继续", "验证码登录", "账号登录")
|
||
_CAPTCHA_MARKERS = ("请完成验证", "拖动滑块", "安全验证", "点击图中")
|
||
_READY_MARKERS = ("发起拼单", "立即购买", "单独购买", "免拼购买", "快要抢光")
|
||
_OUT_OF_STOCK_PAGE_MARKERS = (
|
||
"商品已售罄，推荐以下相似商品",
|
||
"商品已售罄，看看相似商品",
|
||
"该商品已售罄",
|
||
"该商品暂时缺货",
|
||
"商品暂时缺货",
|
||
)
|
||
_SHOP_NAME_EXCLUDES = frozenset({"进店", "关注", "店铺", "收藏", "客服"})
|
||
_SHOP_ROW_TOLERANCE = 40
|
||
|
||
|
||
def _is_device_disconnect(error: BaseException) -> bool:
|
||
details = str(error).lower()
|
||
return any(
|
||
marker in details
|
||
for marker in (
|
||
"device not found",
|
||
"device offline",
|
||
"disconnected",
|
||
"closed transport",
|
||
)
|
||
)
|
||
|
||
|
||
class PddCollectError(RuntimeError):
|
||
"""采集失败，并携带稳定错误码。"""
|
||
|
||
def __init__(
|
||
self,
|
||
code: str,
|
||
message: str,
|
||
diagnostics: Optional[Mapping[str, Any]] = None,
|
||
) -> None:
|
||
super().__init__(message)
|
||
self.code = code
|
||
self.message = message
|
||
self.diagnostics = dict(diagnostics or {})
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class QuantityMetric:
|
||
"""销量或评价数量，保留原文和是否为近似值。"""
|
||
|
||
value: Optional[int]
|
||
raw: Optional[str]
|
||
approximate: bool
|
||
|
||
def to_dict(self) -> dict[str, Any]:
|
||
return {
|
||
"value": self.value,
|
||
"raw": self.raw,
|
||
"approximate": self.approximate,
|
||
}
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class DimensionValue:
|
||
text: str
|
||
available: bool
|
||
|
||
def to_dict(self) -> dict[str, Any]:
|
||
return {"text": self.text, "available": self.available}
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class SpecDimension:
|
||
key: str
|
||
name: str
|
||
values: tuple[DimensionValue, ...]
|
||
|
||
def to_dict(self) -> dict[str, Any]:
|
||
return {
|
||
"key": self.key,
|
||
"name": self.name,
|
||
"values": [value.to_dict() for value in self.values],
|
||
}
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class SkuResult:
|
||
options: Mapping[str, str]
|
||
price_cent: Optional[int]
|
||
available: bool
|
||
raw_price: Optional[str]
|
||
price_observed_at: Mapping[str, str]
|
||
list_price_cent: Optional[int] = None
|
||
|
||
def to_dict(self) -> dict[str, Any]:
|
||
return {
|
||
"options": dict(self.options),
|
||
"price_cent": self.price_cent,
|
||
"list_price_cent": self.list_price_cent,
|
||
"price_observed_at": dict(self.price_observed_at),
|
||
"currency": "CNY",
|
||
"available": self.available,
|
||
"raw_price": self.raw_price,
|
||
}
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class GoodsSnapshot:
|
||
title: Optional[str]
|
||
shop_name: Optional[str]
|
||
sales: QuantityMetric
|
||
reviews: QuantityMetric
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class SpecSnapshot:
|
||
dimensions: tuple[SpecDimension, ...]
|
||
selected_text: Optional[str]
|
||
price_cent: Optional[int]
|
||
raw_price: Optional[str]
|
||
list_price_cent: Optional[int]
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class _SecondDimensionContext:
|
||
"""标题滑出屏幕后，继续识别第二规格所需的稳定特征。"""
|
||
|
||
name: str
|
||
horizontal: bool
|
||
container_class: str
|
||
container_resource_id: str
|
||
container_bounds: Bounds
|
||
option_structures: tuple[tuple[str, str, tuple[tuple[str, str], ...]], ...]
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class VisibleSpecOption:
|
||
"""当前控件树中完整可点击的规格节点。"""
|
||
|
||
text: str
|
||
available: bool
|
||
bounds: Bounds
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ColorPriceSample:
|
||
"""点击一个颜色后稳定读取到的颜色级价格。"""
|
||
|
||
price_cent: Optional[int]
|
||
raw_price: Optional[str]
|
||
list_price_cent: Optional[int]
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class CollectResult:
|
||
"""与 ``pdd_data`` v1 对应的采集结果。"""
|
||
|
||
goods_id: str
|
||
goods_url: str
|
||
title: str
|
||
shop_name: Optional[str]
|
||
sales: QuantityMetric
|
||
reviews: QuantityMetric
|
||
dimensions: tuple[SpecDimension, ...]
|
||
skus: tuple[SkuResult, ...]
|
||
captured_at: str
|
||
client_id: str
|
||
device_address: str
|
||
artifacts: tuple[Mapping[str, Any], ...] = ()
|
||
|
||
def to_pdd_data(self) -> dict[str, Any]:
|
||
return {
|
||
"schema_version": 1,
|
||
"goods_id": self.goods_id,
|
||
"goods_url": self.goods_url,
|
||
"title": self.title,
|
||
"shop_name": self.shop_name,
|
||
"price_granularity": "color",
|
||
"metrics": {
|
||
"sales": self.sales.to_dict(),
|
||
"reviews": self.reviews.to_dict(),
|
||
},
|
||
"dimensions": [item.to_dict() for item in self.dimensions],
|
||
"skus": [item.to_dict() for item in self.skus],
|
||
"purchase": None,
|
||
"captured_at": self.captured_at,
|
||
"source": {
|
||
"client_id": self.client_id,
|
||
"device_address": self.device_address,
|
||
"pdd_package": PDD_PACKAGE_NAME,
|
||
},
|
||
"artifacts": [dict(item) for item in self.artifacts],
|
||
}
|
||
|
||
|
||
def _parse_xml(xml_data: str | bytes) -> ET.Element:
|
||
try:
|
||
return ET.fromstring(xml_data)
|
||
except (ET.ParseError, TypeError) as exc:
|
||
raise PddCollectError(
|
||
"PDD_DATA_XML_INVALID",
|
||
"PDD 返回的无障碍控件树不是有效 XML",
|
||
) from exc
|
||
|
||
|
||
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 _node_label(node: ET.Element) -> str:
|
||
return " ".join(
|
||
part.strip()
|
||
for part in (node.get("text", ""), node.get("content-desc", ""))
|
||
if part.strip()
|
||
)
|
||
|
||
|
||
def _preferred_node_label(node: ET.Element) -> str:
|
||
"""规格名被 text 截断时，优先采用完整的无障碍描述。"""
|
||
|
||
text = (node.get("text") or "").strip()
|
||
description = (node.get("content-desc") or "").strip()
|
||
return description or text
|
||
|
||
|
||
def _own_or_descendant_label(node: ET.Element) -> str:
|
||
own = _node_label(node)
|
||
if own:
|
||
return own
|
||
for child in node.iter("node"):
|
||
label = _node_label(child)
|
||
if label and label != "打开大图":
|
||
return label
|
||
return ""
|
||
|
||
|
||
def _preferred_or_descendant_label(node: ET.Element) -> str:
|
||
own = _preferred_node_label(node)
|
||
if own:
|
||
return own
|
||
for child in node.iter("node"):
|
||
label = _preferred_node_label(child)
|
||
if label and label != "打开大图":
|
||
return label
|
||
return ""
|
||
|
||
|
||
def _all_labels(root: ET.Element) -> list[str]:
|
||
return [label for node in root.iter("node") if (label := _node_label(node))]
|
||
|
||
|
||
def parse_quantity(raw: Optional[str]) -> QuantityMetric:
|
||
"""把“已拼1.2万+”等文字转成整数，同时保留原文。"""
|
||
|
||
if not raw:
|
||
return QuantityMetric(None, None, False)
|
||
match = _QUANTITY_PATTERN.search(raw.replace(",", ""))
|
||
if match is None:
|
||
return QuantityMetric(None, raw, False)
|
||
multiplier = {None: 1, "万": 10_000, "亿": 100_000_000}[match.group(2)]
|
||
value = int(Decimal(match.group(1)) * multiplier)
|
||
approximate = bool(match.group(2) or match.group(3))
|
||
return QuantityMetric(value, raw, approximate)
|
||
|
||
|
||
def _find_metric(labels: Sequence[str], pattern: re.Pattern[str]) -> QuantityMetric:
|
||
for label in labels:
|
||
match = pattern.search(label)
|
||
if match:
|
||
return parse_quantity(match.group(0))
|
||
return QuantityMetric(None, None, False)
|
||
|
||
|
||
def _shop_name_by_enter_anchor(root: ET.Element) -> Optional[str]:
|
||
"""以“进店”为锚点，读取同一行左侧的店铺名。"""
|
||
|
||
anchors = []
|
||
for node in root.iter("node"):
|
||
if _preferred_node_label(node) != "进店":
|
||
continue
|
||
bounds = _parse_bounds(node.get("bounds", ""))
|
||
if bounds is not None:
|
||
anchors.append(bounds)
|
||
if not anchors:
|
||
return None
|
||
|
||
anchor_left, anchor_top, _, _ = min(
|
||
anchors, key=lambda bounds: (bounds[1], bounds[0])
|
||
)
|
||
candidates = []
|
||
for node in root.iter("node"):
|
||
if node.get("class") != "android.widget.TextView":
|
||
continue
|
||
text = _preferred_node_label(node)
|
||
if not text or text in _SHOP_NAME_EXCLUDES:
|
||
continue
|
||
if "已拼" in text or "评价" in text or not 2 <= len(text) <= 30:
|
||
continue
|
||
bounds = _parse_bounds(node.get("bounds", ""))
|
||
if bounds is None:
|
||
continue
|
||
left, top, _, _ = bounds
|
||
if left >= anchor_left or abs(top - anchor_top) > _SHOP_ROW_TOLERANCE:
|
||
continue
|
||
candidates.append((abs(top - anchor_top), -left, text))
|
||
|
||
if not candidates:
|
||
return None
|
||
return min(candidates)[2]
|
||
|
||
|
||
def parse_goods_page(xml_data: str | bytes) -> GoodsSnapshot:
|
||
"""解析一棵商品页控件树中的标题、店铺、销量和评价。"""
|
||
|
||
root = _parse_xml(xml_data)
|
||
labels = _all_labels(root)
|
||
_raise_special_page(labels)
|
||
|
||
title: Optional[str] = None
|
||
for node in root.iter("node"):
|
||
if node.get("class") == "androidx.viewpager.widget.ViewPager":
|
||
candidate = (node.get("content-desc") or node.get("text") or "").strip()
|
||
if len(candidate) >= 6:
|
||
title = candidate
|
||
break
|
||
if title is None:
|
||
line_parts: dict[int, list[tuple[int, str]]] = {}
|
||
for node in root.iter("node"):
|
||
label = _preferred_node_label(node)
|
||
bounds = _parse_bounds(node.get("bounds", ""))
|
||
if not label or bounds is None:
|
||
continue
|
||
line_key = bounds[1] // 24
|
||
line_parts.setdefault(line_key, []).append((bounds[0], label))
|
||
joined_lines = [
|
||
"".join(text for _, text in sorted(parts))
|
||
for parts in line_parts.values()
|
||
]
|
||
candidates = [
|
||
label
|
||
for label in labels + joined_lines
|
||
if len(label) >= 12
|
||
and not any(word in label for word in ("通知", "支付", "已拼", "评价"))
|
||
]
|
||
title = max(candidates, key=len, default=None)
|
||
|
||
sales = _find_metric(labels, re.compile(r"已拼\s*\d+(?:\.\d+)?\s*(?:万|亿)?\s*\+?\s*(?:件|人)?"))
|
||
reviews_pattern = re.compile(
|
||
r"(?:商品评价\s*[（(]\s*\d+(?:\.\d+)?\s*(?:万|亿)?\s*\+?\s*[)）]"
|
||
r"|\d+(?:\.\d+)?\s*(?:万|亿)?\s*\+?\s*条?评价"
|
||
r"|评价\s*\d+(?:\.\d+)?\s*(?:万|亿)?\s*\+?)"
|
||
)
|
||
reviews = _find_metric(labels, reviews_pattern)
|
||
if reviews.raw is None:
|
||
empty_reviews = next(
|
||
(label for label in labels if label in ("暂无评价", "暂无评论")),
|
||
None,
|
||
)
|
||
if empty_reviews:
|
||
reviews = QuantityMetric(0, empty_reviews, False)
|
||
|
||
shop_name: Optional[str] = None
|
||
for label in labels:
|
||
match = re.search(r"(?:店铺|商家)[:：]\s*(.+)", label)
|
||
if match and match.group(1).strip():
|
||
shop_name = match.group(1).strip()
|
||
break
|
||
if len(label) > 2 and label.endswith(("旗舰店", "专卖店", "专营店")):
|
||
shop_name = label
|
||
break
|
||
if shop_name is None:
|
||
shop_name = _shop_name_by_enter_anchor(root)
|
||
return GoodsSnapshot(title, shop_name, sales, reviews)
|
||
|
||
|
||
def _is_available(node: ET.Element) -> bool:
|
||
label = _own_or_descendant_label(node)
|
||
return (
|
||
node.get("enabled", "true") == "true"
|
||
and node.get("visible-to-user", "true") == "true"
|
||
and not any(word in label for word in ("售罄", "缺货", "不可选"))
|
||
)
|
||
|
||
|
||
def _dimension_key(name: str, used: set[str]) -> str:
|
||
if any(word in name for word in ("颜色", "花色", "款式")):
|
||
base = "color"
|
||
elif "尺码" in name or "尺寸" in name:
|
||
base = "size"
|
||
elif "套餐" in name:
|
||
# 本项目的稳定结构只有主规格 color 和第二规格 size。商品只有
|
||
# “套餐”一组时，它必须作为主规格采价；前面已有款式/颜色时，
|
||
# 套餐才是第二规格。不能无条件映射成 size，否则单规格商品无法采集。
|
||
base = "size" if "color" in used else "color"
|
||
elif "型号" in name:
|
||
base = "model"
|
||
else:
|
||
digest = hashlib.sha1(name.encode("utf-8")).hexdigest()[:8]
|
||
base = f"dimension_{digest}"
|
||
key = base
|
||
suffix = 2
|
||
while key in used:
|
||
key = f"{base}_{suffix}"
|
||
suffix += 1
|
||
return key
|
||
|
||
|
||
def _is_dimension_heading(label: str) -> bool:
|
||
compact = label.replace(" ", "")
|
||
if compact.startswith(("请选择", "請選擇", "已选", "已選")):
|
||
return False
|
||
# 新版页面会把可选数量写进标题，例如“颜色 (6)”或“颜色（6）”。
|
||
# 数量不是规格名称的一部分，只在判断标题类型时去掉，最终展示仍保留原文。
|
||
compact = re.sub(r"[（(]\d+[)）]$", "", compact)
|
||
# “确认款式”是新版规格弹层的标题，不是一个可选择的“款式”维度。
|
||
if compact in ("确认款式", "確認款式"):
|
||
return False
|
||
return compact in _DIMENSION_NAMES or compact.endswith(
|
||
(
|
||
"分类",
|
||
"规格",
|
||
"尺寸",
|
||
"尺码",
|
||
"颜色",
|
||
"型号",
|
||
"款式",
|
||
"套餐",
|
||
"容量",
|
||
"类型",
|
||
"版本",
|
||
"口味",
|
||
)
|
||
)
|
||
|
||
|
||
def _find_non_scrollable_spec_panel(
|
||
root: ET.Element,
|
||
parents: Mapping[ET.Element, ET.Element],
|
||
) -> Optional[ET.Element]:
|
||
"""用多项强证据定位不暴露 scrollable 属性的自绘规格面板。"""
|
||
|
||
all_bounds = [
|
||
bounds
|
||
for node in root.iter("node")
|
||
if (bounds := _parse_bounds(node.get("bounds", ""))) is not None
|
||
]
|
||
screen_bottom = max((item[3] for item in all_bounds), default=0)
|
||
pdd_node_count = sum(
|
||
node.get("package") == PDD_PACKAGE_NAME
|
||
for node in root.iter("node")
|
||
)
|
||
|
||
headings: list[ET.Element] = []
|
||
summaries: list[ET.Element] = []
|
||
panel_cues: list[ET.Element] = []
|
||
confirms: list[ET.Element] = []
|
||
submit_hints: list[ET.Element] = []
|
||
quantity_editors: list[ET.Element] = []
|
||
decreases: list[ET.Element] = []
|
||
increases: list[ET.Element] = []
|
||
for node in root.iter("node"):
|
||
label = _preferred_node_label(node).strip()
|
||
compact = label.replace(" ", "")
|
||
bounds = _parse_bounds(node.get("bounds", ""))
|
||
if bounds is None:
|
||
continue
|
||
is_summary = compact.startswith(
|
||
("已选", "已選", "请选择", "請選擇")
|
||
)
|
||
if label and not is_summary and _is_dimension_heading(label):
|
||
headings.append(node)
|
||
if is_summary:
|
||
summaries.append(node)
|
||
if compact in ("确认款式", "確認款式", "关闭", "關閉"):
|
||
panel_cues.append(node)
|
||
if compact in ("确定", "確定") and node.get("clickable") == "true":
|
||
confirms.append(node)
|
||
descendant_label = _preferred_or_descendant_label(node).replace(" ", "")
|
||
if (
|
||
node.get("clickable") == "true"
|
||
and node.get("enabled", "true") != "false"
|
||
and node.get("visible-to-user", "true") != "false"
|
||
and screen_bottom
|
||
and (bounds[1] + bounds[3]) // 2 >= screen_bottom * 0.6
|
||
and "提交订单" in descendant_label
|
||
and any(
|
||
word in descendant_label
|
||
for word in ("选择", "颜色", "尺码", "规格")
|
||
)
|
||
):
|
||
submit_hints.append(node)
|
||
if (
|
||
node.get("class") == "android.widget.EditText"
|
||
and node.get("text", "").strip().isdigit()
|
||
and int(node.get("text", "0")) > 0
|
||
):
|
||
quantity_editors.append(node)
|
||
if compact == "减少数量" and node.get("clickable") == "true":
|
||
decreases.append(node)
|
||
if compact == "增加数量" and node.get("clickable") == "true":
|
||
increases.append(node)
|
||
|
||
if not headings or not summaries or not panel_cues:
|
||
return None
|
||
if confirms:
|
||
action_nodes = [confirms[0]]
|
||
elif (
|
||
pdd_node_count >= 3
|
||
and len(submit_hints) == 1
|
||
and len(quantity_editors) == 1
|
||
and len(decreases) == 1
|
||
and len(increases) == 1
|
||
):
|
||
action_nodes = [
|
||
submit_hints[0],
|
||
quantity_editors[0],
|
||
decreases[0],
|
||
increases[0],
|
||
]
|
||
else:
|
||
return None
|
||
|
||
required = [*headings, summaries[0], panel_cues[0], *action_nodes]
|
||
common = set([required[0], *_ancestors(required[0], parents)])
|
||
for node in required[1:]:
|
||
common.intersection_update([node, *_ancestors(node, parents)])
|
||
|
||
screen_area = 0
|
||
if all_bounds:
|
||
screen_area = max(item[2] for item in all_bounds) * max(
|
||
item[3] for item in all_bounds
|
||
)
|
||
|
||
candidates: list[tuple[int, ET.Element]] = []
|
||
for node in common:
|
||
bounds = _parse_bounds(node.get("bounds", ""))
|
||
if bounds is None or node.get("visible-to-user", "true") != "true":
|
||
continue
|
||
area = (bounds[2] - bounds[0]) * (bounds[3] - bounds[1])
|
||
if screen_area and area >= screen_area * 0.95:
|
||
continue
|
||
candidates.append((area, node))
|
||
if not candidates:
|
||
return None
|
||
return min(candidates, key=lambda item: item[0])[1]
|
||
|
||
|
||
def _is_spec_panel_open(xml_data: str | bytes) -> bool:
|
||
"""判断规格面板是否已经出现，不要求当前视口已露出规格选项。"""
|
||
|
||
root = _parse_xml(xml_data)
|
||
labels = _all_labels(root)
|
||
_raise_special_page(labels)
|
||
parents = {child: parent for parent in root.iter() for child in parent}
|
||
if _find_non_scrollable_spec_panel(root, parents) is not None:
|
||
return True
|
||
scrollable_regions = [
|
||
node
|
||
for node in root.iter("node")
|
||
if node.get("scrollable") == "true"
|
||
and _parse_bounds(node.get("bounds", "")) is not None
|
||
]
|
||
has_scrollable_region = bool(scrollable_regions)
|
||
if not has_scrollable_region:
|
||
return False
|
||
|
||
compact_labels = [label.replace(" ", "") for label in labels]
|
||
has_selection_summary = any(
|
||
label.startswith(("请选择", "請選擇", "已选", "已選"))
|
||
for label in compact_labels
|
||
)
|
||
has_submit_hint = any(
|
||
"提交订单" in label
|
||
and any(word in label for word in ("选择", "颜色", "尺码", "规格"))
|
||
for label in compact_labels
|
||
)
|
||
has_dimension_heading = any(
|
||
_is_dimension_heading(_preferred_node_label(node).strip())
|
||
and any(
|
||
region in _ancestors(node, parents)
|
||
for region in scrollable_regions
|
||
)
|
||
for node in root.iter("node")
|
||
)
|
||
return has_selection_summary or has_submit_hint or has_dimension_heading
|
||
|
||
|
||
def _top_level_clickable_options(
|
||
container: ET.Element,
|
||
parents: Mapping[ET.Element, ET.Element],
|
||
) -> list[ET.Element]:
|
||
result: list[ET.Element] = []
|
||
for node in container.iter("node"):
|
||
label = _own_or_descendant_label(node).strip()
|
||
bounds = _parse_bounds(node.get("bounds", ""))
|
||
if node.get("clickable") != "true" or not label or bounds is None:
|
||
continue
|
||
if label.startswith("#") or label in (
|
||
"打开大图",
|
||
"增加数量",
|
||
"减少数量",
|
||
"关闭",
|
||
):
|
||
continue
|
||
parent = parents.get(node)
|
||
if parent is not None and parent is not container:
|
||
parent_label = _own_or_descendant_label(parent).strip()
|
||
if parent.get("clickable") == "true" and parent_label == label:
|
||
continue
|
||
result.append(node)
|
||
return result
|
||
|
||
|
||
def _price_from_nodes(
|
||
nodes: Iterable[ET.Element],
|
||
spec_top: int,
|
||
) -> tuple[Optional[int], Optional[str], Optional[int]]:
|
||
candidates: list[tuple[int, int, str]] = []
|
||
for node in nodes:
|
||
label = _node_label(node)
|
||
bounds = _parse_bounds(node.get("bounds", ""))
|
||
match = _PRICE_PATTERN.search(label)
|
||
if match is None or bounds is None or bounds[1] >= spec_top:
|
||
continue
|
||
try:
|
||
cents = int(
|
||
(Decimal(match.group(1)) * 100).quantize(
|
||
Decimal("1"),
|
||
rounding=ROUND_HALF_UP,
|
||
)
|
||
)
|
||
except InvalidOperation:
|
||
continue
|
||
candidates.append((bounds[1], cents, match.group(0).replace(" ", "")))
|
||
if not candidates:
|
||
return None, None, None
|
||
first_top = min(item[0] for item in candidates)
|
||
first_row = [item for item in candidates if abs(item[0] - first_top) <= 12]
|
||
_, cents, raw = min(first_row, key=lambda item: item[1])
|
||
list_prices = [item[1] for item in first_row if item[1] > cents]
|
||
return cents, raw, max(list_prices, default=None)
|
||
|
||
|
||
def parse_spec_panel(xml_data: str | bytes) -> SpecSnapshot:
|
||
"""解析规格面板当前视口中的规格值、已选摘要和当前价格。"""
|
||
|
||
root = _parse_xml(xml_data)
|
||
labels = _all_labels(root)
|
||
_raise_special_page(labels)
|
||
parents = {child: parent for parent in root.iter() for child in parent}
|
||
scrollables = [
|
||
node
|
||
for node in root.iter("node")
|
||
if node.get("scrollable") == "true" and _parse_bounds(node.get("bounds", ""))
|
||
]
|
||
non_scrollable_outer = _find_non_scrollable_spec_panel(root, parents)
|
||
if non_scrollable_outer is None and not _is_spec_panel_open(xml_data):
|
||
raise PddCollectError(
|
||
"PDD_DATA_SPEC_INCOMPLETE", "规格面板没有可识别的规格区域"
|
||
)
|
||
panel_scrollables = []
|
||
if non_scrollable_outer is not None:
|
||
panel_scrollables = [
|
||
node
|
||
for node in scrollables
|
||
if node is non_scrollable_outer
|
||
or non_scrollable_outer in _ancestors(node, parents)
|
||
]
|
||
outer_candidates = panel_scrollables or (
|
||
[non_scrollable_outer] if non_scrollable_outer is not None else scrollables
|
||
)
|
||
if not outer_candidates:
|
||
raise PddCollectError("PDD_DATA_SPEC_INCOMPLETE", "规格面板没有可识别的规格区域")
|
||
|
||
outer = max(
|
||
outer_candidates,
|
||
key=lambda node: (
|
||
(_parse_bounds(node.get("bounds", "")) or (0, 0, 0, 0))[2]
|
||
- (_parse_bounds(node.get("bounds", "")) or (0, 0, 0, 0))[0]
|
||
) * (
|
||
(_parse_bounds(node.get("bounds", "")) or (0, 0, 0, 0))[3]
|
||
- (_parse_bounds(node.get("bounds", "")) or (0, 0, 0, 0))[1]
|
||
),
|
||
)
|
||
outer_bounds = _parse_bounds(outer.get("bounds", ""))
|
||
assert outer_bounds is not None
|
||
|
||
heading_nodes: list[tuple[int, str]] = []
|
||
for node in outer.iter("node"):
|
||
label = _node_label(node).strip()
|
||
bounds = _parse_bounds(node.get("bounds", ""))
|
||
if label and bounds and _is_dimension_heading(label):
|
||
heading_nodes.append((bounds[1], label))
|
||
heading_nodes.sort()
|
||
|
||
groups: list[tuple[str, list[ET.Element]]] = []
|
||
nested_scrollables = [
|
||
node
|
||
for node in scrollables
|
||
if node is not outer and outer in _ancestors(node, parents)
|
||
]
|
||
nested_options: set[ET.Element] = set()
|
||
for nested in nested_scrollables:
|
||
bounds = _parse_bounds(nested.get("bounds", ""))
|
||
if bounds is None:
|
||
continue
|
||
preceding = [item for item in heading_nodes if item[0] <= bounds[1] + 30]
|
||
if not preceding:
|
||
# 普通推荐列表也是 RecyclerView。没有真实规格标题时，不能
|
||
# 默认把其中的商品卡片解释成颜色。
|
||
continue
|
||
name = preceding[-1][1]
|
||
options = _top_level_clickable_options(nested, parents)
|
||
if options:
|
||
groups.append((name, options))
|
||
nested_options.update(options)
|
||
|
||
for index, (top, name) in enumerate(heading_nodes):
|
||
bottom = heading_nodes[index + 1][0] if index + 1 < len(heading_nodes) else outer_bounds[3]
|
||
options = []
|
||
for node in _top_level_clickable_options(outer, parents):
|
||
bounds = _parse_bounds(node.get("bounds", ""))
|
||
if node in nested_options or bounds is None:
|
||
continue
|
||
width = bounds[2] - bounds[0]
|
||
if top < bounds[1] < bottom and width < (outer_bounds[2] - outer_bounds[0]) * 0.85:
|
||
options.append(node)
|
||
if options:
|
||
groups.append((name, options))
|
||
|
||
# 嵌套 RecyclerView 会先加入 groups，可能把页面下方的“套餐”排在
|
||
# 上方“款式”之前。规范 key 依赖页面语义顺序，必须按标题 y 坐标恢复。
|
||
heading_tops = {name: top for top, name in heading_nodes}
|
||
groups.sort(key=lambda item: heading_tops.get(item[0], outer_bounds[3]))
|
||
|
||
used_keys: set[str] = set()
|
||
dimensions: list[SpecDimension] = []
|
||
for name, option_nodes in groups:
|
||
values: list[DimensionValue] = []
|
||
seen: set[str] = set()
|
||
for node in option_nodes:
|
||
text = _preferred_or_descendant_label(node).strip()
|
||
if text.endswith(("…", "...")):
|
||
raise PddCollectError(
|
||
"PDD_DATA_SKU_NAME_TRUNCATED",
|
||
f"规格名称被截断，无法安全采集：{text}",
|
||
)
|
||
if not text or text in seen:
|
||
continue
|
||
seen.add(text)
|
||
values.append(DimensionValue(text, _is_available(node)))
|
||
if not values:
|
||
continue
|
||
key = _dimension_key(name, used_keys)
|
||
used_keys.add(key)
|
||
dimensions.append(SpecDimension(key, name, tuple(values)))
|
||
|
||
selected_text = next((label for label in labels if label.startswith("已选")), None)
|
||
price_boundary = heading_nodes[0][0] if heading_nodes else outer_bounds[1]
|
||
price_cent, raw_price, list_price_cent = _price_from_nodes(
|
||
root.iter("node"), price_boundary
|
||
)
|
||
return SpecSnapshot(
|
||
tuple(dimensions), selected_text, price_cent, raw_price, list_price_cent
|
||
)
|
||
|
||
|
||
def _ancestors(
|
||
node: ET.Element,
|
||
parents: Mapping[ET.Element, ET.Element],
|
||
) -> list[ET.Element]:
|
||
result: list[ET.Element] = []
|
||
current = parents.get(node)
|
||
while current is not None:
|
||
result.append(current)
|
||
current = parents.get(current)
|
||
return result
|
||
|
||
|
||
def _raise_special_page(labels: Sequence[str]) -> None:
|
||
compact_labels = [label.replace(" ", "") for label in labels]
|
||
if any(
|
||
marker in label
|
||
for label in compact_labels
|
||
for marker in _OUT_OF_STOCK_PAGE_MARKERS
|
||
):
|
||
raise PddCollectError(
|
||
"PDD_GOODS_UNAVAILABLE",
|
||
"商品已售罄，PDD 没有提供可采集的规格",
|
||
{"page_reason": "out_of_stock"},
|
||
)
|
||
combined = " ".join(labels)
|
||
if any(marker in combined for marker in _CAPTCHA_MARKERS):
|
||
raise PddCollectError("PDD_PAGE_CAPTCHA", "PDD 出现安全验证，需要人工处理")
|
||
if any(marker in combined for marker in _LOGIN_MARKERS):
|
||
raise PddCollectError("PDD_PAGE_LOGIN_REQUIRED", "PDD 登录已失效，需要人工重新登录")
|
||
|
||
|
||
def _validate_goods_url(goods_url: str, task_goods_id: str) -> str:
|
||
parsed = urlparse(goods_url)
|
||
host = (parsed.hostname or "").lower()
|
||
allowed_host = host == "yangkeduo.com" or host.endswith(".yangkeduo.com")
|
||
allowed_host = allowed_host or host == "pinduoduo.com" or host.endswith(
|
||
".pinduoduo.com"
|
||
)
|
||
if parsed.scheme not in ("http", "https") or not allowed_host:
|
||
raise PddCollectError(
|
||
"PDD_DATA_GOODS_URL_INVALID",
|
||
"商品链接不是受支持的 PDD 链接",
|
||
)
|
||
|
||
url_goods_id = parse_qs(parsed.query).get("goods_id", [""])[0].strip()
|
||
if task_goods_id and url_goods_id and task_goods_id != url_goods_id:
|
||
raise PddCollectError(
|
||
"PDD_DATA_GOODS_ID_MISMATCH",
|
||
"任务商品编号与商品链接中的 goods_id 不一致",
|
||
)
|
||
value = task_goods_id or url_goods_id
|
||
if not value or not value.isdigit():
|
||
raise PddCollectError("PDD_DATA_GOODS_ID_MISSING", "商品链接中没有 goods_id")
|
||
return value
|
||
|
||
|
||
class PddCollectService:
|
||
"""打开商品页并采集结构化商品与 SKU 数据。"""
|
||
|
||
def __init__(
|
||
self,
|
||
device_service: PddDeviceService,
|
||
device_address: str,
|
||
client_id: str,
|
||
*,
|
||
sleeper: Callable[[float], None] = time.sleep,
|
||
monotonic: Callable[[], float] = time.monotonic,
|
||
now: Callable[[], datetime] = lambda: datetime.now(timezone.utc),
|
||
cancelled: Callable[[], bool] = lambda: False,
|
||
page_timeout: float = 30.0,
|
||
spec_panel_timeout: float = 10.0,
|
||
overall_timeout: float = 600.0,
|
||
max_goods_page_swipes: int = 3,
|
||
max_spec_swipes: int = 12,
|
||
max_sku_count: int = 200,
|
||
color_poll_interval: float = 0.1,
|
||
color_selection_timeout: float = 0.6,
|
||
color_price_timeout: float = 1.2,
|
||
horizontal_swipe_settle_interval: float = 0.1,
|
||
artifact_directory: Optional[Path] = None,
|
||
) -> None:
|
||
self._device_service = device_service
|
||
self._device_address = device_address
|
||
self._client_id = client_id
|
||
self._sleep = sleeper
|
||
self._monotonic = monotonic
|
||
self._now = now
|
||
self._cancelled = cancelled
|
||
self._page_timeout = page_timeout
|
||
self._spec_panel_timeout = spec_panel_timeout
|
||
self._overall_timeout = overall_timeout
|
||
self._overall_deadline: Optional[float] = None
|
||
self._max_goods_page_swipes = max_goods_page_swipes
|
||
self._max_spec_swipes = max_spec_swipes
|
||
self._max_sku_count = max_sku_count
|
||
timing_values = {
|
||
"颜色轮询间隔": color_poll_interval,
|
||
"颜色选中超时": color_selection_timeout,
|
||
"颜色价格超时": color_price_timeout,
|
||
"水平滑动稳定间隔": horizontal_swipe_settle_interval,
|
||
}
|
||
for name, value in timing_values.items():
|
||
if value <= 0:
|
||
raise ValueError(f"{name}必须大于 0 秒")
|
||
self._color_poll_interval = color_poll_interval
|
||
self._color_selection_timeout = color_selection_timeout
|
||
self._color_price_timeout = color_price_timeout
|
||
self._horizontal_swipe_settle_interval = (
|
||
horizontal_swipe_settle_interval
|
||
)
|
||
self._artifact_directory = artifact_directory
|
||
self._last_goods_xml: Optional[str] = None
|
||
self._last_valid_spec_xml: Optional[str] = None
|
||
self._last_invalid_spec_xml: Optional[str] = None
|
||
self._goods_screens_checked = 0
|
||
self._artifacts: list[Mapping[str, Any]] = []
|
||
|
||
def collect(self, task: Any) -> CollectResult:
|
||
"""执行采集；``task`` 至少提供 ``goods_url`` 和可选 ``goods_id``。"""
|
||
|
||
goods_url = str(getattr(task, "goods_url", "") or "").strip()
|
||
if not goods_url:
|
||
raise PddCollectError("PDD_DATA_GOODS_URL_MISSING", "采集任务缺少商品链接")
|
||
goods_id = str(getattr(task, "goods_id", "") or "").strip()
|
||
goods_id = _validate_goods_url(goods_url, goods_id)
|
||
self._overall_deadline = self._monotonic() + self._overall_timeout
|
||
self._last_valid_spec_xml = None
|
||
self._last_invalid_spec_xml = None
|
||
self._check_cancelled()
|
||
|
||
try:
|
||
session = self._device_service.connect(self._device_address)
|
||
with session as device:
|
||
self._open_goods(
|
||
device, goods_url, session.initial_app_state
|
||
)
|
||
goods = self._collect_goods_details(device)
|
||
if not goods.title:
|
||
raise PddCollectError("PDD_DATA_TITLE_MISSING", "商品页没有可识别的标题")
|
||
if not goods.sales.raw:
|
||
raise PddCollectError("PDD_DATA_SALES_MISSING", "商品页没有采集到已拼数量")
|
||
if (
|
||
(not goods.shop_name or not goods.reviews.raw)
|
||
and self._last_goods_xml
|
||
):
|
||
artifact = self._save_xml(
|
||
"goods-metadata-incomplete", self._last_goods_xml
|
||
)
|
||
if artifact:
|
||
self._artifacts.append(artifact)
|
||
|
||
home_xml = self._dump_hierarchy(device)
|
||
coordinate = get_size_panel_coord(home_xml)
|
||
if coordinate is None:
|
||
raise PddCollectError(
|
||
"PDD_PAGE_SPEC_ENTRY_MISSING",
|
||
"商品页没有找到可靠的规格入口",
|
||
)
|
||
device.click(*coordinate)
|
||
self._wait_spec_panel(device)
|
||
|
||
self._clear_default_size_selection(device)
|
||
color_dimension, color_samples = self._collect_color_prices(device)
|
||
size_dimension = self._collect_size_dimension(device)
|
||
dimensions = (color_dimension,)
|
||
if size_dimension is not None:
|
||
dimensions += (size_dimension,)
|
||
if not dimensions or any(not item.values for item in dimensions):
|
||
raise PddCollectError(
|
||
"PDD_DATA_SPEC_INCOMPLETE",
|
||
"规格面板没有采集到完整的规格维度",
|
||
)
|
||
skus = self._build_color_price_skus(
|
||
dimensions, color_samples
|
||
)
|
||
if not skus:
|
||
raise PddCollectError(
|
||
"PDD_DATA_SPEC_INCOMPLETE",
|
||
"规格面板没有生成可提交的规格组合",
|
||
)
|
||
except PddCollectError as exc:
|
||
if self._last_goods_xml:
|
||
artifact = self._save_xml("collect-failed", self._last_goods_xml)
|
||
if artifact:
|
||
diagnostics = dict(exc.diagnostics)
|
||
diagnostics.setdefault("artifacts", []).append(artifact)
|
||
diagnostics.setdefault(
|
||
"goods_screens_checked", self._goods_screens_checked
|
||
)
|
||
exc.diagnostics = diagnostics
|
||
if exc.code == "PDD_PAGE_SPEC_PANEL_LOST":
|
||
diagnostics = dict(exc.diagnostics)
|
||
artifacts = list(diagnostics.get("artifacts") or [])
|
||
for label, xml_data in (
|
||
("last-valid-spec", self._last_valid_spec_xml),
|
||
("last-invalid-spec", self._last_invalid_spec_xml),
|
||
):
|
||
if not xml_data:
|
||
continue
|
||
artifact = self._save_xml(label, xml_data)
|
||
if artifact and artifact not in artifacts:
|
||
artifacts.append(artifact)
|
||
if artifacts:
|
||
diagnostics["artifacts"] = artifacts
|
||
exc.diagnostics = diagnostics
|
||
raise
|
||
except PddDeviceError as exc:
|
||
raise PddCollectError(exc.code, exc.message) from exc
|
||
except Exception as exc:
|
||
if _is_device_disconnect(exc):
|
||
raise PddCollectError(
|
||
"DEVICE_DISCONNECTED",
|
||
f"Android 设备在采集过程中断开：{exc}",
|
||
) from exc
|
||
raise PddCollectError("PDD_PAGE_UNKNOWN", f"PDD 采集过程中发生未知错误：{exc}") from exc
|
||
|
||
return CollectResult(
|
||
goods_id=goods_id,
|
||
goods_url=goods_url,
|
||
title=goods.title,
|
||
shop_name=goods.shop_name,
|
||
sales=goods.sales,
|
||
reviews=goods.reviews,
|
||
dimensions=dimensions,
|
||
skus=skus,
|
||
captured_at=self._now().astimezone(timezone.utc).isoformat().replace("+00:00", "Z"),
|
||
client_id=self._client_id,
|
||
device_address=self._device_address,
|
||
artifacts=tuple(self._artifacts),
|
||
)
|
||
|
||
def _check_cancelled(self) -> None:
|
||
if self._cancelled():
|
||
raise PddCollectError("PDD_CANCELLED", "采集任务已安全取消")
|
||
if (
|
||
self._overall_deadline is not None
|
||
and self._monotonic() >= self._overall_deadline
|
||
):
|
||
raise PddCollectError("PDD_PAGE_OVERALL_TIMEOUT", "PDD 采集超过 10 分钟")
|
||
|
||
def _open_goods(
|
||
self,
|
||
device: Any,
|
||
goods_url: str,
|
||
initial_app_state: Optional[Mapping[str, Any]] = None,
|
||
) -> None:
|
||
trace = current_performance_trace()
|
||
try:
|
||
current = (
|
||
dict(initial_app_state)
|
||
if initial_app_state is not None
|
||
else device.app_current()
|
||
)
|
||
package_hint = (
|
||
PDD_PACKAGE_NAME
|
||
if current.get("package") == PDD_PACKAGE_NAME
|
||
else ""
|
||
)
|
||
before_open = self._read_page_observation(device, current)
|
||
if current.get("package") != PDD_PACKAGE_NAME:
|
||
stage = trace.stage("pdd_start_or_wait") if trace else nullcontext()
|
||
with stage:
|
||
device.app_start(PDD_PACKAGE_NAME)
|
||
if not device.app_wait(PDD_PACKAGE_NAME, timeout=10):
|
||
raise PddCollectError(
|
||
"DEVICE_APP_START_FAILED", "PDD 应用启动失败"
|
||
)
|
||
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:
|
||
device.open_url(goods_url)
|
||
except PddCollectError:
|
||
raise
|
||
except Exception as exc:
|
||
if _is_device_disconnect(exc):
|
||
raise PddCollectError(
|
||
"DEVICE_DISCONNECTED",
|
||
f"Android 设备在打开商品页时断开：{exc}",
|
||
) from exc
|
||
raise PddCollectError("DEVICE_APP_START_FAILED", f"无法打开 PDD 商品链接：{exc}") from exc
|
||
|
||
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)
|
||
first_dump = True
|
||
last_kind = "unknown"
|
||
last_recorded = ""
|
||
while self._monotonic() < deadline:
|
||
self._check_cancelled()
|
||
if first_dump and trace is not None:
|
||
with trace.stage("first_dump_hierarchy"):
|
||
xml_data = self._dump_hierarchy(device)
|
||
first_dump = False
|
||
else:
|
||
xml_data = self._dump_hierarchy(device)
|
||
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
|
||
self._raise_classified_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
|
||
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 PddCollectError(
|
||
"PDD_GOODS_UNAVAILABLE",
|
||
"商品链接已失效，PDD 无法打开商品详情页并返回了首页",
|
||
{"page_kind": PAGE_HOME, "open_attempts": 2},
|
||
)
|
||
if decision.action == ACTION_NETWORK_ERROR:
|
||
raise PddCollectError(
|
||
"PDD_PAGE_NETWORK_ERROR",
|
||
"PDD 商品页网络或服务异常，请稍后重试",
|
||
{"page_kind": PAGE_NETWORK_ERROR},
|
||
)
|
||
self._sleep(0.25)
|
||
if tracker.stale_goods_seen:
|
||
raise PddCollectError(
|
||
"PDD_GOODS_IDENTITY_UNCONFIRMED",
|
||
"打开商品链接后仍停留在原商品页，无法确认本次目标商品",
|
||
{"page_kind": PAGE_GOODS, "open_attempts": tracker.attempt},
|
||
)
|
||
raise PddCollectError(
|
||
"PDD_PAGE_TIMEOUT",
|
||
f"等待 PDD 商品详情页加载超时，最后页面为 {last_kind}",
|
||
{"page_kind": last_kind, "open_attempts": tracker.attempt},
|
||
)
|
||
|
||
def _read_page_observation(
|
||
self, device: Any, current: Mapping[str, Any]
|
||
) -> Optional[PddPageObservation]:
|
||
"""读取深链打开前页面；读取失败不妨碍后续打开目标链接。"""
|
||
|
||
try:
|
||
root = _parse_xml(self._dump_hierarchy(device))
|
||
return classify_pdd_page(
|
||
root, str(current.get("package") or "")
|
||
)
|
||
except (PddCollectError, RuntimeError, OSError):
|
||
return None
|
||
|
||
@staticmethod
|
||
def _raise_classified_special_page(kind: str) -> None:
|
||
if kind == PAGE_CAPTCHA:
|
||
raise PddCollectError(
|
||
"PDD_PAGE_CAPTCHA", "PDD 出现安全验证，需要人工处理"
|
||
)
|
||
if kind == PAGE_LOGIN_REQUIRED:
|
||
raise PddCollectError(
|
||
"PDD_PAGE_LOGIN_REQUIRED", "PDD 登录已失效，需要人工重新登录"
|
||
)
|
||
if kind in {PAGE_RISK_CONTROL, PAGE_PAYMENT}:
|
||
raise PddCollectError(
|
||
"PDD_PAGE_UNKNOWN", "PDD 出现风控或支付页面，已停止采集"
|
||
)
|
||
|
||
def _collect_goods_details(self, device: Any) -> GoodsSnapshot:
|
||
"""保留首屏结果，并向下浏览补采评价数量和店铺名。"""
|
||
|
||
self._check_cancelled()
|
||
xml_data = self._dump_hierarchy(device)
|
||
goods = parse_goods_page(xml_data)
|
||
if goods.shop_name and goods.reviews.raw is not None:
|
||
return goods
|
||
|
||
unchanged_reads = 0
|
||
for _ in range(self._max_goods_page_swipes):
|
||
self._check_cancelled()
|
||
self._swipe_goods_page(device, xml_data)
|
||
self._sleep(0.8)
|
||
self._check_cancelled()
|
||
|
||
next_xml = self._dump_hierarchy(device)
|
||
snapshot = parse_goods_page(next_xml)
|
||
goods = GoodsSnapshot(
|
||
title=goods.title,
|
||
shop_name=goods.shop_name or snapshot.shop_name,
|
||
sales=goods.sales,
|
||
reviews=(
|
||
goods.reviews
|
||
if goods.reviews.raw is not None
|
||
else snapshot.reviews
|
||
),
|
||
)
|
||
if goods.shop_name and goods.reviews.raw is not None:
|
||
break
|
||
|
||
unchanged_reads = unchanged_reads + 1 if next_xml == xml_data else 0
|
||
xml_data = next_xml
|
||
if unchanged_reads >= 2:
|
||
break
|
||
return goods
|
||
|
||
@staticmethod
|
||
def _swipe_goods_page(device: Any, xml_data: str | bytes) -> None:
|
||
"""在商品详情区域慢速向上滑动约 30%，浏览页面下方内容。"""
|
||
|
||
root = _parse_xml(xml_data)
|
||
scrollable_bounds = [
|
||
bounds
|
||
for node in root.iter("node")
|
||
if node.get("scrollable") == "true"
|
||
and (bounds := _parse_bounds(node.get("bounds", ""))) is not None
|
||
]
|
||
all_bounds = [
|
||
bounds
|
||
for node in root.iter("node")
|
||
if (bounds := _parse_bounds(node.get("bounds", ""))) is not None
|
||
]
|
||
candidates = scrollable_bounds or all_bounds
|
||
if not candidates:
|
||
raise PddCollectError(
|
||
"PDD_DATA_PAGE_BOUNDS_MISSING",
|
||
"商品详情页没有可用于滚动的有效区域",
|
||
)
|
||
left, top, right, bottom = max(
|
||
candidates,
|
||
key=lambda bounds: (bounds[2] - bounds[0]) * (bounds[3] - bounds[1]),
|
||
)
|
||
x = (left + right) // 2
|
||
height = bottom - top
|
||
device.swipe(
|
||
x,
|
||
top + int(height * 0.65),
|
||
x,
|
||
top + int(height * 0.35),
|
||
duration=0.6,
|
||
)
|
||
|
||
def _wait_spec_panel(self, device: Any) -> SpecSnapshot:
|
||
"""等待点击后的规格面板真正出现，不能只依赖固定延时。"""
|
||
|
||
deadline = self._monotonic() + self._spec_panel_timeout
|
||
saw_confirmation_page = False
|
||
incomplete_reads = 0
|
||
while self._monotonic() < deadline:
|
||
self._check_cancelled()
|
||
xml_data = self._dump_hierarchy(device)
|
||
root = _parse_xml(xml_data)
|
||
if classify_pdd_page(root, "").kind == PAGE_ORDER_CONFIRMATION:
|
||
saw_confirmation_page = True
|
||
try:
|
||
snapshot = parse_spec_panel(xml_data)
|
||
except PddCollectError as exc:
|
||
if exc.code != "PDD_DATA_SPEC_INCOMPLETE":
|
||
raise
|
||
self._last_invalid_spec_xml = xml_data
|
||
incomplete_reads += 1
|
||
else:
|
||
if _is_spec_panel_open(xml_data):
|
||
self._last_valid_spec_xml = xml_data
|
||
return snapshot
|
||
self._last_invalid_spec_xml = xml_data
|
||
incomplete_reads += 1
|
||
self._sleep(0.25)
|
||
if saw_confirmation_page:
|
||
raise PddCollectError(
|
||
"PDD_DATA_SPEC_INCOMPLETE",
|
||
"规格面板已经打开，但页面结构无法识别",
|
||
{"incomplete_spec_reads": incomplete_reads},
|
||
)
|
||
raise PddCollectError(
|
||
"PDD_PAGE_SPEC_PANEL_TIMEOUT",
|
||
"点击规格入口后，等待规格面板加载超时",
|
||
)
|
||
|
||
def _read_valid_spec_panel(
|
||
self,
|
||
device: Any,
|
||
initial_xml: Optional[str] = None,
|
||
) -> tuple[str, SpecSnapshot]:
|
||
"""跳过规格操作期间的临时空树，返回最新有效面板和解析结果。"""
|
||
|
||
deadline = self._monotonic() + min(2.0, self._spec_panel_timeout)
|
||
xml_data = initial_xml
|
||
transient_reads = 0
|
||
# 次数上限避免测试时钟或设备时钟异常导致无限循环。
|
||
for attempt in range(20):
|
||
self._check_cancelled()
|
||
if xml_data is None:
|
||
xml_data = self._dump_hierarchy(device)
|
||
try:
|
||
snapshot = parse_spec_panel(xml_data)
|
||
panel_open = _is_spec_panel_open(xml_data)
|
||
except PddCollectError as exc:
|
||
if exc.code != "PDD_DATA_SPEC_INCOMPLETE":
|
||
raise
|
||
panel_open = False
|
||
snapshot = None
|
||
|
||
if panel_open and snapshot is not None:
|
||
self._last_valid_spec_xml = xml_data
|
||
return xml_data, snapshot
|
||
|
||
self._last_invalid_spec_xml = xml_data
|
||
transient_reads += 1
|
||
if attempt >= 19 or self._monotonic() >= deadline:
|
||
break
|
||
self._sleep(0.1)
|
||
xml_data = None
|
||
|
||
raise PddCollectError(
|
||
"PDD_PAGE_SPEC_PANEL_LOST",
|
||
"规格面板操作期间控件树持续为空或面板已经消失",
|
||
{"transient_spec_reads": transient_reads},
|
||
)
|
||
|
||
def _save_xml(self, label: str, xml_data: str) -> Optional[Mapping[str, Any]]:
|
||
"""保存本地诊断 XML；测试未提供目录时不写文件。"""
|
||
|
||
if self._artifact_directory is None:
|
||
return None
|
||
try:
|
||
digest = hashlib.sha256(xml_data.encode("utf-8")).hexdigest()
|
||
directory = self._artifact_directory / self._client_id
|
||
directory.mkdir(parents=True, exist_ok=True)
|
||
path = directory / f"{label}-{digest[:12]}.xml"
|
||
if not path.exists():
|
||
path.write_text(xml_data, encoding="utf-8")
|
||
return {
|
||
"kind": "accessibility_xml",
|
||
"path": str(path.resolve()),
|
||
"sha256": digest,
|
||
"screens_checked": self._goods_screens_checked,
|
||
}
|
||
except OSError:
|
||
return None
|
||
|
||
def _dump_hierarchy(self, device: Any) -> str:
|
||
"""读取并记住最新控件树，失败诊断必须指向最后一次操作。"""
|
||
|
||
raw_xml = device.dump_hierarchy()
|
||
xml_data = (
|
||
raw_xml.decode("utf-8", errors="replace")
|
||
if isinstance(raw_xml, bytes)
|
||
else str(raw_xml)
|
||
)
|
||
self._last_goods_xml = xml_data
|
||
self._goods_screens_checked += 1
|
||
return xml_data
|
||
|
||
def _collect_color_prices(
|
||
self, device: Any
|
||
) -> tuple[SpecDimension, Mapping[str, ColorPriceSample]]:
|
||
"""按行蛇形遍历颜色，并在每次点击后立即读取颜色级价格。"""
|
||
|
||
self._move_color_list_to_left(device)
|
||
first_xml, first_snapshot = self._read_valid_spec_panel(device)
|
||
source_dimension = next(
|
||
(item for item in first_snapshot.dimensions if item.key == "color"),
|
||
None,
|
||
)
|
||
first_rows = self._visible_color_rows(first_xml)
|
||
if source_dimension is None or not first_rows:
|
||
raise PddCollectError(
|
||
"PDD_DATA_SPEC_INCOMPLETE", "规格面板没有可识别的颜色分类"
|
||
)
|
||
|
||
row_count = len(first_rows)
|
||
visited: set[str] = set()
|
||
ordered_colors: list[DimensionValue] = []
|
||
samples: dict[str, ColorPriceSample] = {}
|
||
|
||
for row_index in range(row_count):
|
||
move_right = row_index % 2 == 0
|
||
current_row_colors: list[DimensionValue] = []
|
||
stable_edge_reads = 0
|
||
previous_signature: Optional[tuple[tuple[str, Bounds], ...]] = None
|
||
pending_xml: Optional[str] = None
|
||
|
||
for swipe_count in range(self._max_spec_swipes + 1):
|
||
self._check_cancelled()
|
||
# 每次只处理一个节点；点击可能让列表自动移动，下一项必须
|
||
# 从最新 XML 重新计算，不能继续使用点击前的旧坐标。
|
||
while True:
|
||
if pending_xml is not None:
|
||
xml_data = pending_xml
|
||
else:
|
||
xml_data, _ = self._read_valid_spec_panel(device)
|
||
pending_xml = None
|
||
rows = self._visible_color_rows(xml_data)
|
||
if row_index >= len(rows):
|
||
raise PddCollectError(
|
||
"PDD_DATA_SPEC_INCOMPLETE",
|
||
f"颜色第 {row_index + 1} 行在滑动后消失",
|
||
)
|
||
current_row = sorted(
|
||
rows[row_index],
|
||
key=lambda item: item.bounds[0],
|
||
reverse=not move_right,
|
||
)
|
||
visible = next(
|
||
(item for item in current_row if item.text not in visited),
|
||
None,
|
||
)
|
||
if visible is None:
|
||
break
|
||
visited.add(visible.text)
|
||
color = DimensionValue(visible.text, visible.available)
|
||
current_row_colors.append(color)
|
||
if visible.available:
|
||
samples[visible.text] = self._click_and_sample_color(
|
||
device, visible.text
|
||
)
|
||
else:
|
||
samples[visible.text] = ColorPriceSample(
|
||
None, None, None
|
||
)
|
||
|
||
signature = tuple((item.text, item.bounds) for item in current_row)
|
||
if signature == previous_signature:
|
||
stable_edge_reads += 1
|
||
else:
|
||
stable_edge_reads = 0
|
||
previous_signature = signature
|
||
if stable_edge_reads >= 2 or swipe_count >= self._max_spec_swipes:
|
||
break
|
||
|
||
root = _parse_xml(xml_data)
|
||
region = self._horizontal_region(root)
|
||
if region is None:
|
||
break
|
||
self._swipe_region(
|
||
device,
|
||
region,
|
||
horizontal=True,
|
||
reverse=not move_right,
|
||
)
|
||
pending_xml = self._wait_for_horizontal_change(
|
||
device, signature
|
||
)
|
||
|
||
# 操作采用蛇形以减少无效滑动；输出仍恢复成页面自然的
|
||
# “每行从左到右”顺序，方便 Admin 下拉框稳定展示。
|
||
ordered_colors.extend(
|
||
current_row_colors if move_right else reversed(current_row_colors)
|
||
)
|
||
|
||
if not ordered_colors:
|
||
raise PddCollectError(
|
||
"PDD_DATA_SPEC_INCOMPLETE", "规格面板没有采集到颜色"
|
||
)
|
||
return (
|
||
SpecDimension("color", source_dimension.name, tuple(ordered_colors)),
|
||
samples,
|
||
)
|
||
|
||
def _move_color_list_to_left(self, device: Any) -> None:
|
||
"""把颜色列表归位到左端；连续两次视口不变才认为到边。"""
|
||
|
||
previous_signature: Optional[tuple[tuple[str, Bounds], ...]] = None
|
||
stable_edge_reads = 0
|
||
xml_data, _ = self._read_valid_spec_panel(device)
|
||
for _ in range(self._max_spec_swipes):
|
||
self._check_cancelled()
|
||
signature = self._color_view_signature(xml_data)
|
||
if signature == previous_signature:
|
||
stable_edge_reads += 1
|
||
else:
|
||
stable_edge_reads = 0
|
||
previous_signature = signature
|
||
if stable_edge_reads >= 2:
|
||
return
|
||
|
||
root = _parse_xml(xml_data)
|
||
region = self._horizontal_region(root)
|
||
if region is None:
|
||
return
|
||
self._swipe_region(
|
||
device, region, horizontal=True, reverse=True
|
||
)
|
||
xml_data = self._wait_for_horizontal_change(device, signature)
|
||
|
||
def _click_and_sample_color(
|
||
self, device: Any, target: str
|
||
) -> ColorPriceSample:
|
||
"""点击最新树中的颜色，等待选择和价格连续两次稳定。"""
|
||
|
||
xml_data, _ = self._read_valid_spec_panel(device)
|
||
root = _parse_xml(xml_data)
|
||
latest_visible = next(
|
||
(
|
||
item
|
||
for row in self._visible_color_rows(xml_data)
|
||
for item in row
|
||
if item.text == target
|
||
),
|
||
None,
|
||
)
|
||
state_node = self._find_option_node(root, target) if latest_visible else None
|
||
node = (
|
||
self._find_safe_option_click_node(state_node, target)
|
||
if state_node is not None
|
||
else None
|
||
)
|
||
if node is None or not _is_available(node):
|
||
return ColorPriceSample(None, None, None)
|
||
bounds = _parse_bounds(node.get("bounds", ""))
|
||
assert bounds is not None
|
||
device.click(
|
||
(bounds[0] + bounds[2]) // 2,
|
||
(bounds[1] + bounds[3]) // 2,
|
||
)
|
||
|
||
selection_deadline = self._monotonic() + self._color_selection_timeout
|
||
selection_sleep_limit = math.ceil(
|
||
self._color_selection_timeout / self._color_poll_interval
|
||
)
|
||
selection_sleeps = 0
|
||
price_deadline: Optional[float] = None
|
||
price_sleep_limit = math.ceil(
|
||
self._color_price_timeout / self._color_poll_interval
|
||
)
|
||
price_sleeps = 0
|
||
previous_price: Optional[tuple[int, Optional[str], Optional[int]]] = None
|
||
stable_price_reads = 0
|
||
while True:
|
||
self._check_cancelled()
|
||
latest_xml = self._dump_hierarchy(device)
|
||
if self._is_big_image_viewer(latest_xml):
|
||
self._recover_spec_panel_from_big_image(device)
|
||
return ColorPriceSample(None, None, None)
|
||
latest_xml, snapshot = self._read_valid_spec_panel(
|
||
device, initial_xml=latest_xml
|
||
)
|
||
latest_root = _parse_xml(latest_xml)
|
||
latest_node = self._find_option_node(latest_root, target)
|
||
selection_exposed = self._color_selection_state_exposed(latest_xml)
|
||
selected = bool(
|
||
latest_node is not None and self._node_is_selected(latest_node)
|
||
) or bool(snapshot.selected_text and target in snapshot.selected_text)
|
||
# 老版本页面不暴露 selected/checked，也没有“已选”摘要时，
|
||
# 只能以完整可见可点击节点未报错的点击作为降级证据。
|
||
selection_confirmed = selected or not selection_exposed
|
||
|
||
now = self._monotonic()
|
||
if not selection_confirmed:
|
||
stable_price_reads = 0
|
||
previous_price = None
|
||
selection_timed_out = (
|
||
now >= selection_deadline
|
||
or selection_sleeps >= selection_sleep_limit
|
||
)
|
||
if selection_timed_out:
|
||
return ColorPriceSample(None, None, None)
|
||
else:
|
||
if price_deadline is None:
|
||
price_deadline = now + self._color_price_timeout
|
||
if snapshot.price_cent is None:
|
||
stable_price_reads = 0
|
||
previous_price = None
|
||
else:
|
||
current_price = (
|
||
snapshot.price_cent,
|
||
snapshot.raw_price,
|
||
snapshot.list_price_cent,
|
||
)
|
||
if current_price == previous_price:
|
||
stable_price_reads += 1
|
||
else:
|
||
stable_price_reads = 1
|
||
previous_price = current_price
|
||
if stable_price_reads >= 2:
|
||
return ColorPriceSample(
|
||
snapshot.price_cent,
|
||
snapshot.raw_price,
|
||
snapshot.list_price_cent,
|
||
)
|
||
if now >= price_deadline or price_sleeps >= price_sleep_limit:
|
||
return ColorPriceSample(None, None, None)
|
||
|
||
self._sleep(self._color_poll_interval)
|
||
if selection_confirmed:
|
||
price_sleeps += 1
|
||
else:
|
||
selection_sleeps += 1
|
||
|
||
def _clear_default_size_selection(self, device: Any) -> None:
|
||
"""逐屏寻找默认尺码；只在唯一选中项明确时点击一次取消。"""
|
||
|
||
previous_signature: Optional[
|
||
tuple[tuple[str, tuple[str, ...]], ...]
|
||
] = None
|
||
stable_edge_reads = 0
|
||
for swipe_count in range(self._max_spec_swipes + 1):
|
||
self._check_cancelled()
|
||
xml_data, snapshot = self._read_valid_spec_panel(device)
|
||
root = _parse_xml(xml_data)
|
||
size_dimension = next(
|
||
(item for item in snapshot.dimensions if item.key == "size"),
|
||
None,
|
||
)
|
||
if size_dimension is not None:
|
||
selected_values = []
|
||
for value in size_dimension.values:
|
||
state_node = self._find_option_node(root, value.text)
|
||
if state_node is not None and self._node_is_selected(state_node):
|
||
selected_values.append((value.text, state_node))
|
||
|
||
if len(selected_values) == 1:
|
||
target, state_node = selected_values[0]
|
||
click_node = self._find_safe_option_click_node(state_node, target)
|
||
if click_node is not None:
|
||
bounds = _parse_bounds(click_node.get("bounds", ""))
|
||
assert bounds is not None
|
||
device.click(
|
||
(bounds[0] + bounds[2]) // 2,
|
||
(bounds[1] + bounds[3]) // 2,
|
||
)
|
||
self._sleep(self._color_poll_interval)
|
||
latest_xml, _ = self._read_valid_spec_panel(device)
|
||
latest_root = _parse_xml(latest_xml)
|
||
latest_node = self._find_option_node(latest_root, target)
|
||
if latest_node is None or not self._node_is_selected(
|
||
latest_node
|
||
):
|
||
break
|
||
# 页面不支持取消时也不能重复点击，否则可能重新选中或误操作。
|
||
break
|
||
if not selected_values:
|
||
break
|
||
|
||
signature = tuple(
|
||
(item.key, tuple(value.text for value in item.values))
|
||
for item in snapshot.dimensions
|
||
)
|
||
stable_edge_reads = (
|
||
stable_edge_reads + 1 if signature == previous_signature else 0
|
||
)
|
||
previous_signature = signature
|
||
if stable_edge_reads >= 2 or swipe_count >= self._max_spec_swipes:
|
||
break
|
||
region = self._vertical_region(root)
|
||
if region is None:
|
||
break
|
||
self._swipe_region(device, region, horizontal=False, reverse=False)
|
||
self._sleep(0.35)
|
||
|
||
# 查找尺码时页面可能已经滚到底部，逐色采价前必须回到顶部。
|
||
self._move_spec_panel_to_top(device)
|
||
|
||
def _move_spec_panel_to_top(self, device: Any) -> None:
|
||
"""把规格面板纵向恢复到顶部；连续两次不变化即停止。"""
|
||
|
||
previous_signature: Optional[tuple[tuple[str, Bounds], ...]] = None
|
||
stable_edge_reads = 0
|
||
for _ in range(self._max_spec_swipes):
|
||
self._check_cancelled()
|
||
xml_data, _ = self._read_valid_spec_panel(device)
|
||
root = _parse_xml(xml_data)
|
||
signature = tuple(
|
||
(
|
||
_preferred_node_label(node),
|
||
bounds,
|
||
)
|
||
for node in root.iter("node")
|
||
if (bounds := _parse_bounds(node.get("bounds", ""))) is not None
|
||
and (
|
||
_is_dimension_heading(_preferred_node_label(node))
|
||
or node.get("clickable") == "true"
|
||
)
|
||
)
|
||
stable_edge_reads = (
|
||
stable_edge_reads + 1 if signature == previous_signature else 0
|
||
)
|
||
previous_signature = signature
|
||
if stable_edge_reads >= 2:
|
||
return
|
||
region = self._vertical_region(root)
|
||
if region is None:
|
||
return
|
||
self._swipe_region(device, region, horizontal=False, reverse=True)
|
||
self._sleep(0.35)
|
||
|
||
@staticmethod
|
||
def _is_big_image_viewer(xml_data: str | bytes) -> bool:
|
||
"""用 ViewPager 和页码共同识别误入的商品大图页。"""
|
||
|
||
root = _parse_xml(xml_data)
|
||
has_pager = any(
|
||
(node.get("class") or "").endswith("ViewPager")
|
||
for node in root.iter("node")
|
||
)
|
||
has_page_number = any(
|
||
re.fullmatch(r"\d+\s*/\s*\d+", _preferred_node_label(node))
|
||
for node in root.iter("node")
|
||
)
|
||
has_dimension = any(
|
||
_is_dimension_heading(_preferred_node_label(node))
|
||
for node in root.iter("node")
|
||
)
|
||
return has_pager and has_page_number and not has_dimension
|
||
|
||
def _recover_spec_panel_from_big_image(self, device: Any) -> None:
|
||
"""误入大图后只返回一次，并确认规格面板已经恢复。"""
|
||
|
||
device.press("back")
|
||
self._sleep(self._color_poll_interval)
|
||
try:
|
||
self._read_valid_spec_panel(device)
|
||
except PddCollectError as exc:
|
||
if exc.code != "PDD_PAGE_SPEC_PANEL_LOST":
|
||
raise
|
||
raise PddCollectError(
|
||
"PDD_PAGE_SPEC_PANEL_LOST",
|
||
"点击颜色后进入大图，返回一次仍未恢复规格面板",
|
||
exc.diagnostics,
|
||
) from exc
|
||
|
||
def _color_view_signature(
|
||
self, xml_data: str | bytes
|
||
) -> tuple[tuple[str, Bounds], ...]:
|
||
"""返回颜色视口签名，用于判断水平滑动是否已经更新页面。"""
|
||
|
||
return tuple(
|
||
(item.text, item.bounds)
|
||
for row in self._visible_color_rows(xml_data)
|
||
for item in row
|
||
)
|
||
|
||
def _wait_for_horizontal_change(
|
||
self,
|
||
device: Any,
|
||
previous_signature: tuple[tuple[str, Bounds], ...],
|
||
) -> str:
|
||
"""水平滑动后短暂等待；首次未变化时只补等一次。"""
|
||
|
||
latest_xml = ""
|
||
for _ in range(2):
|
||
self._check_cancelled()
|
||
self._sleep(self._horizontal_swipe_settle_interval)
|
||
latest_xml, _ = self._read_valid_spec_panel(device)
|
||
if self._color_view_signature(latest_xml) != previous_signature:
|
||
break
|
||
return latest_xml
|
||
|
||
def _collect_size_dimension(self, device: Any) -> Optional[SpecDimension]:
|
||
"""完整遍历第二规格并收集文字，全程不点击尺码或套餐。"""
|
||
|
||
sizes: dict[str, bool] = {}
|
||
size_name = "尺码"
|
||
xml_data: Optional[str] = None
|
||
size_found = False
|
||
previous_signature: Optional[
|
||
tuple[tuple[str, tuple[str, ...]], ...]
|
||
] = None
|
||
stable_edge_reads = 0
|
||
|
||
# 颜色采价结束时页面通常仍在顶部，先纵向找到第二规格。这个阶段
|
||
# 保留原有行为，避免改变普通衣服“颜色 + 尺码”的稳定流程。
|
||
for swipe_count in range(self._max_spec_swipes + 1):
|
||
self._check_cancelled()
|
||
xml_data, snapshot = self._read_valid_spec_panel(device)
|
||
for dimension in snapshot.dimensions:
|
||
if dimension.key not in ("color", "size"):
|
||
raise PddCollectError(
|
||
"PDD_DATA_SPEC_UNSUPPORTED",
|
||
f"发现尚未支持的规格维度：{dimension.name}",
|
||
)
|
||
if dimension.key == "size":
|
||
size_name = dimension.name
|
||
size_found = True
|
||
for value in dimension.values:
|
||
option_text = self._second_dimension_option_text(value.text)
|
||
if option_text is None:
|
||
continue
|
||
sizes[option_text] = (
|
||
sizes.get(option_text, False) or value.available
|
||
)
|
||
|
||
if size_found:
|
||
break
|
||
|
||
signature = tuple(
|
||
(item.key, tuple(value.text for value in item.values))
|
||
for item in snapshot.dimensions
|
||
)
|
||
if signature == previous_signature:
|
||
stable_edge_reads += 1
|
||
else:
|
||
stable_edge_reads = 0
|
||
previous_signature = signature
|
||
if stable_edge_reads >= 2 or swipe_count >= self._max_spec_swipes:
|
||
break
|
||
|
||
root = _parse_xml(xml_data)
|
||
region = self._vertical_region(root)
|
||
if region is None:
|
||
break
|
||
self._swipe_region(
|
||
device, region, horizontal=False, reverse=False
|
||
)
|
||
self._sleep(0.35)
|
||
|
||
if not size_found or xml_data is None:
|
||
return None
|
||
|
||
xml_data, start_confirmed = self._move_second_dimension_to_start(
|
||
device, xml_data
|
||
)
|
||
initial_snapshot = parse_spec_panel(xml_data)
|
||
initial_size_dimension = next(
|
||
(item for item in initial_snapshot.dimensions if item.key == "size"),
|
||
None,
|
||
)
|
||
if initial_size_dimension is None:
|
||
raise PddCollectError(
|
||
"PDD_DATA_SPEC_INCOMPLETE",
|
||
f"无法建立第二规格续页上下文：{size_name}",
|
||
)
|
||
static_dimension = self._non_scrollable_second_dimension(
|
||
xml_data, initial_size_dimension
|
||
)
|
||
if static_dimension is not None:
|
||
return static_dimension
|
||
continuation = self._build_second_dimension_context(
|
||
xml_data, initial_size_dimension
|
||
)
|
||
strict_edge_check = "套餐" in size_name and self._max_spec_swipes > 0
|
||
if strict_edge_check and not start_confirmed:
|
||
raise PddCollectError(
|
||
"PDD_DATA_SPEC_INCOMPLETE",
|
||
f"{size_name}无法确认已经到达列表起点",
|
||
)
|
||
# 第一阶段只负责找到分组；正式输出从确认后的起点重新按页面顺序收集。
|
||
sizes = {}
|
||
previous_signature = None
|
||
stable_edge_reads = 0
|
||
end_confirmed = False
|
||
for swipe_count in range(self._max_spec_swipes + 1):
|
||
self._check_cancelled()
|
||
xml_data, snapshot = self._read_valid_spec_panel(
|
||
device, initial_xml=xml_data
|
||
)
|
||
size_dimension = next(
|
||
(item for item in snapshot.dimensions if item.key == "size"),
|
||
None,
|
||
)
|
||
if size_dimension is None:
|
||
size_dimension = self._extract_second_dimension_continuation(
|
||
xml_data, continuation
|
||
)
|
||
if size_dimension is None:
|
||
# 控件树偶尔会在滚动动画中短暂缺节点，再读取一次后才判定失败。
|
||
xml_data, retry_snapshot = self._read_valid_spec_panel(device)
|
||
size_dimension = next(
|
||
(item for item in retry_snapshot.dimensions if item.key == "size"),
|
||
None,
|
||
) or self._extract_second_dimension_continuation(
|
||
xml_data, continuation
|
||
)
|
||
if size_dimension is None:
|
||
raise PddCollectError(
|
||
"PDD_DATA_SPEC_INCOMPLETE",
|
||
f"滚动过程中持续丢失第二规格：{size_name}",
|
||
)
|
||
size_name = size_dimension.name
|
||
for value in size_dimension.values:
|
||
option_text = self._second_dimension_option_text(value.text)
|
||
if option_text is None:
|
||
continue
|
||
sizes[option_text] = sizes.get(option_text, False) or value.available
|
||
|
||
signature = self._second_dimension_signature(xml_data, size_dimension)
|
||
stable_edge_reads = (
|
||
stable_edge_reads + 1 if signature == previous_signature else 0
|
||
)
|
||
previous_signature = signature
|
||
if stable_edge_reads >= 2:
|
||
end_confirmed = True
|
||
break
|
||
if swipe_count >= self._max_spec_swipes:
|
||
break
|
||
|
||
root = _parse_xml(xml_data)
|
||
region, horizontal = self._second_dimension_region(
|
||
root, size_dimension
|
||
)
|
||
if region is None:
|
||
end_confirmed = True
|
||
break
|
||
self._swipe_region(
|
||
device,
|
||
region,
|
||
horizontal=horizontal,
|
||
reverse=False,
|
||
)
|
||
self._sleep(
|
||
self._horizontal_swipe_settle_interval if horizontal else 0.35
|
||
)
|
||
xml_data = None
|
||
|
||
expected_count = self._dimension_expected_count(size_name)
|
||
if expected_count is not None and len(sizes) < expected_count:
|
||
raise PddCollectError(
|
||
"PDD_DATA_SPEC_INCOMPLETE",
|
||
f"{size_name}应有 {expected_count} 个选项，实际只采到 {len(sizes)} 个",
|
||
{
|
||
"dimension_name": size_name,
|
||
"expected_count": expected_count,
|
||
"collected_count": len(sizes),
|
||
},
|
||
)
|
||
if expected_count is None and strict_edge_check and not end_confirmed:
|
||
raise PddCollectError(
|
||
"PDD_DATA_SPEC_INCOMPLETE",
|
||
f"{size_name}达到滑动上限，无法确认是否采集完整",
|
||
{"dimension_name": size_name, "collected_count": len(sizes)},
|
||
)
|
||
return SpecDimension(
|
||
"size",
|
||
size_name,
|
||
tuple(DimensionValue(text, available) for text, available in sizes.items()),
|
||
)
|
||
|
||
def _non_scrollable_second_dimension(
|
||
self,
|
||
xml_data: str | bytes,
|
||
dimension: SpecDimension,
|
||
) -> Optional[SpecDimension]:
|
||
"""强证据确认非滚动面板时，返回当前树中的完整第二规格。"""
|
||
|
||
root = _parse_xml(xml_data)
|
||
region, _ = self._second_dimension_region(root, dimension)
|
||
if region is not None:
|
||
return None
|
||
|
||
parents = {child: parent for parent in root.iter() for child in parent}
|
||
if _find_non_scrollable_spec_panel(root, parents) is None:
|
||
return None
|
||
|
||
values: dict[str, bool] = {}
|
||
for value in dimension.values:
|
||
option_text = self._second_dimension_option_text(value.text)
|
||
if option_text is None:
|
||
continue
|
||
values[option_text] = values.get(option_text, False) or value.available
|
||
|
||
expected_count = self._dimension_expected_count(dimension.name)
|
||
if expected_count is not None and len(values) < expected_count:
|
||
raise PddCollectError(
|
||
"PDD_DATA_SPEC_INCOMPLETE",
|
||
f"{dimension.name}应有 {expected_count} 个选项，实际只采到 {len(values)} 个",
|
||
{
|
||
"dimension_name": dimension.name,
|
||
"expected_count": expected_count,
|
||
"collected_count": len(values),
|
||
},
|
||
)
|
||
if not values:
|
||
return None
|
||
return SpecDimension(
|
||
"size",
|
||
dimension.name,
|
||
tuple(
|
||
DimensionValue(text, available)
|
||
for text, available in values.items()
|
||
),
|
||
)
|
||
|
||
def _move_second_dimension_to_start(
|
||
self, device: Any, xml_data: str
|
||
) -> tuple[str, bool]:
|
||
"""横向第二规格先归位到左端；纵向列表保持当前安全起点。"""
|
||
|
||
previous_signature: Optional[tuple[tuple[str, Bounds], ...]] = None
|
||
stable_edge_reads = 0
|
||
for attempt in range(self._max_spec_swipes + 1):
|
||
snapshot = parse_spec_panel(xml_data)
|
||
size_dimension = next(
|
||
(item for item in snapshot.dimensions if item.key == "size"),
|
||
None,
|
||
)
|
||
if size_dimension is None:
|
||
return xml_data, False
|
||
root = _parse_xml(xml_data)
|
||
region, horizontal = self._second_dimension_region(
|
||
root, size_dimension
|
||
)
|
||
if region is None or not horizontal:
|
||
return xml_data, True
|
||
signature = self._second_dimension_signature(xml_data, size_dimension)
|
||
stable_edge_reads = (
|
||
stable_edge_reads + 1 if signature == previous_signature else 0
|
||
)
|
||
previous_signature = signature
|
||
if stable_edge_reads >= 2:
|
||
return xml_data, True
|
||
if attempt >= self._max_spec_swipes:
|
||
return xml_data, False
|
||
self._swipe_region(
|
||
device, region, horizontal=True, reverse=True
|
||
)
|
||
self._sleep(self._horizontal_swipe_settle_interval)
|
||
xml_data, _ = self._read_valid_spec_panel(device)
|
||
return xml_data, False
|
||
|
||
def _second_dimension_signature(
|
||
self,
|
||
xml_data: str | bytes,
|
||
dimension: SpecDimension,
|
||
) -> tuple[tuple[str, Bounds], ...]:
|
||
"""返回已确认的第二规格签名，不要求当前屏仍显示规格标题。"""
|
||
|
||
root = _parse_xml(xml_data)
|
||
targets = {
|
||
normalized
|
||
for value in dimension.values
|
||
if (normalized := self._second_dimension_option_text(value.text))
|
||
is not None
|
||
}
|
||
result = []
|
||
for node in root.iter("node"):
|
||
if node.get("clickable") != "true":
|
||
continue
|
||
text = self._second_dimension_option_text(
|
||
_preferred_or_descendant_label(node).strip()
|
||
)
|
||
if text not in targets:
|
||
continue
|
||
bounds = _parse_bounds(node.get("bounds", ""))
|
||
if bounds is not None:
|
||
result.append((text, bounds))
|
||
return tuple(result)
|
||
|
||
def _build_second_dimension_context(
|
||
self,
|
||
xml_data: str | bytes,
|
||
dimension: SpecDimension,
|
||
) -> _SecondDimensionContext:
|
||
"""记录第二规格容器和选项结构，供标题滑出后的续页使用。"""
|
||
|
||
root = _parse_xml(xml_data)
|
||
parents = {child: parent for parent in root.iter() for child in parent}
|
||
region, horizontal = self._second_dimension_region(root, dimension)
|
||
if region is None:
|
||
raise PddCollectError(
|
||
"PDD_DATA_SPEC_INCOMPLETE",
|
||
f"无法定位第二规格滚动区域：{dimension.name}",
|
||
)
|
||
option_nodes = [
|
||
node
|
||
for value in dimension.values
|
||
if (node := self._find_option_node(root, value.text)) is not None
|
||
]
|
||
containers = [
|
||
ancestor
|
||
for node in option_nodes
|
||
for ancestor in _ancestors(node, parents)
|
||
if ancestor.get("scrollable") == "true"
|
||
and _parse_bounds(ancestor.get("bounds", "")) == region
|
||
]
|
||
if not containers:
|
||
raise PddCollectError(
|
||
"PDD_DATA_SPEC_INCOMPLETE",
|
||
f"无法锁定第二规格滚动容器：{dimension.name}",
|
||
)
|
||
container = containers[0]
|
||
structures = tuple(
|
||
dict.fromkeys(
|
||
self._option_structure(node, container, parents)
|
||
for node in option_nodes
|
||
)
|
||
)
|
||
return _SecondDimensionContext(
|
||
name=dimension.name,
|
||
horizontal=horizontal,
|
||
container_class=container.get("class", ""),
|
||
container_resource_id=container.get("resource-id", ""),
|
||
container_bounds=region,
|
||
option_structures=structures,
|
||
)
|
||
|
||
def _extract_second_dimension_continuation(
|
||
self,
|
||
xml_data: str | bytes,
|
||
context: _SecondDimensionContext,
|
||
) -> Optional[SpecDimension]:
|
||
"""规格标题不可见时，从已锁定容器提取同结构的后续选项。"""
|
||
|
||
root = _parse_xml(xml_data)
|
||
parents = {child: parent for parent in root.iter() for child in parent}
|
||
container = self._find_second_dimension_container(root, context)
|
||
if container is None:
|
||
return None
|
||
container_bounds = _parse_bounds(container.get("bounds", ""))
|
||
if container_bounds is None:
|
||
return None
|
||
|
||
values: list[DimensionValue] = []
|
||
seen: set[str] = set()
|
||
for node in _top_level_clickable_options(container, parents):
|
||
if self._option_structure(node, container, parents) not in context.option_structures:
|
||
continue
|
||
text = _preferred_or_descendant_label(node).strip()
|
||
bounds = _parse_bounds(node.get("bounds", ""))
|
||
option_text = self._second_dimension_option_text(text)
|
||
if (
|
||
option_text is None
|
||
or option_text in seen
|
||
or bounds is None
|
||
):
|
||
continue
|
||
if text.endswith(("…", "...")):
|
||
raise PddCollectError(
|
||
"PDD_DATA_SKU_NAME_TRUNCATED",
|
||
f"规格名称被截断，无法安全采集：{text}",
|
||
)
|
||
seen.add(option_text)
|
||
values.append(DimensionValue(option_text, _is_available(node)))
|
||
if not values:
|
||
return None
|
||
return SpecDimension("size", context.name, tuple(values))
|
||
|
||
@staticmethod
|
||
def _option_structure(
|
||
node: ET.Element,
|
||
container: ET.Element,
|
||
parents: Mapping[ET.Element, ET.Element],
|
||
) -> tuple[str, str, tuple[tuple[str, str], ...]]:
|
||
"""生成不依赖坐标和 XML 对象身份的选项结构特征。"""
|
||
|
||
ancestors: list[tuple[str, str]] = []
|
||
current = parents.get(node)
|
||
while current is not None and current is not container:
|
||
ancestors.append(
|
||
(current.get("class", ""), current.get("resource-id", ""))
|
||
)
|
||
current = parents.get(current)
|
||
return (
|
||
node.get("class", ""),
|
||
node.get("resource-id", ""),
|
||
tuple(ancestors),
|
||
)
|
||
|
||
@staticmethod
|
||
def _find_second_dimension_container(
|
||
root: ET.Element,
|
||
context: _SecondDimensionContext,
|
||
) -> Optional[ET.Element]:
|
||
"""在新控件树中重新找到首次锁定的滚动容器。"""
|
||
|
||
candidates = [
|
||
node
|
||
for node in root.iter("node")
|
||
if node.get("scrollable") == "true"
|
||
and _parse_bounds(node.get("bounds", "")) is not None
|
||
and (
|
||
not context.container_resource_id
|
||
or node.get("resource-id", "") == context.container_resource_id
|
||
)
|
||
]
|
||
if not candidates:
|
||
return None
|
||
return max(
|
||
candidates,
|
||
key=lambda node: (
|
||
_parse_bounds(node.get("bounds", "")) == context.container_bounds,
|
||
node.get("class", "") == context.container_class,
|
||
),
|
||
)
|
||
|
||
@staticmethod
|
||
def _second_dimension_option_text(text: str) -> Optional[str]:
|
||
"""返回去掉尾部价格的规格文字；操作项或纯价格返回 ``None``。"""
|
||
|
||
compact = text.replace(" ", "")
|
||
if not compact or compact.startswith(("已选", "请选择")) or any(
|
||
marker in compact
|
||
for marker in (
|
||
"增加数量",
|
||
"减少数量",
|
||
"提交订单",
|
||
"立即购买",
|
||
"单独购买",
|
||
"免拼购买",
|
||
"发起拼单",
|
||
"确认购买",
|
||
"关闭",
|
||
)
|
||
):
|
||
return None
|
||
if re.fullmatch(
|
||
r"(?:券后|到手价|拼单价|单买价|价格)?[¥￥]\d+(?:\.\d{1,2})?",
|
||
compact,
|
||
):
|
||
return None
|
||
normalized = re.sub(
|
||
r"\s*[¥￥]\s*\d+(?:\.\d{1,2})?\s*$",
|
||
"",
|
||
text,
|
||
).strip()
|
||
return normalized or None
|
||
|
||
def _second_dimension_region(
|
||
self,
|
||
root: ET.Element,
|
||
dimension: SpecDimension,
|
||
) -> tuple[Optional[Bounds], bool]:
|
||
"""返回第二规格最近的滚动容器及其主要滚动方向。"""
|
||
|
||
parents = {child: parent for parent in root.iter() for child in parent}
|
||
containers: dict[ET.Element, int] = {}
|
||
option_bounds: list[Bounds] = []
|
||
for value in dimension.values:
|
||
node = self._find_option_node(root, value.text)
|
||
if node is None:
|
||
continue
|
||
bounds = _parse_bounds(node.get("bounds", ""))
|
||
if bounds is not None:
|
||
option_bounds.append(bounds)
|
||
current = parents.get(node)
|
||
while current is not None:
|
||
if (
|
||
current.get("scrollable") == "true"
|
||
and _parse_bounds(current.get("bounds", "")) is not None
|
||
):
|
||
containers[current] = containers.get(current, 0) + 1
|
||
break
|
||
current = parents.get(current)
|
||
|
||
if containers:
|
||
container = max(
|
||
containers,
|
||
key=lambda item: (
|
||
containers[item],
|
||
len(_ancestors(item, parents)),
|
||
),
|
||
)
|
||
region = _parse_bounds(container.get("bounds", ""))
|
||
else:
|
||
region = self._vertical_region(root)
|
||
if region is None:
|
||
return None, False
|
||
|
||
centers = [
|
||
((left + right) // 2, (top + bottom) // 2)
|
||
for left, top, right, bottom in option_bounds
|
||
]
|
||
region_width = region[2] - region[0]
|
||
region_height = region[3] - region[1]
|
||
if region_height >= region_width * 0.75:
|
||
# 外层规格面板通常很高；其中一行尺码横向排布，不代表面板
|
||
# 应横向滚动。只有较矮的独立列表才根据选项分布判断横向。
|
||
horizontal = False
|
||
elif len(centers) >= 2:
|
||
x_span = max(item[0] for item in centers) - min(item[0] for item in centers)
|
||
y_span = max(item[1] for item in centers) - min(item[1] for item in centers)
|
||
horizontal = x_span > y_span
|
||
else:
|
||
horizontal = region_width > region_height * 1.8
|
||
return region, horizontal
|
||
|
||
@staticmethod
|
||
def _dimension_expected_count(name: str) -> Optional[int]:
|
||
"""读取“套餐（15）”这类标题中的选项数量。"""
|
||
|
||
match = re.search(r"[（(]\s*(\d+)\s*[)）]\s*$", name)
|
||
return int(match.group(1)) if match else None
|
||
|
||
def _build_color_price_skus(
|
||
self,
|
||
dimensions: tuple[SpecDimension, ...],
|
||
samples: Mapping[str, ColorPriceSample],
|
||
) -> tuple[SkuResult, ...]:
|
||
"""按颜色价格生成展示组合；采价证据只记录真实选中的颜色。"""
|
||
|
||
colors = dimensions[0].values
|
||
sizes = dimensions[1].values if len(dimensions) > 1 else (None,)
|
||
combination_count = len(colors) * len(sizes)
|
||
if combination_count > self._max_sku_count:
|
||
raise PddCollectError(
|
||
"PDD_DATA_TOO_MANY_SKUS",
|
||
f"规格组合共 {combination_count} 个，超过安全上限 {self._max_sku_count}",
|
||
)
|
||
|
||
results: list[SkuResult] = []
|
||
for color in colors:
|
||
sample = samples[color.text]
|
||
for size in sizes:
|
||
options = {"color": color.text}
|
||
if size is not None:
|
||
options["size"] = size.text
|
||
# 是否可用来自规格控件状态，不能用“是否采到价格”代替。
|
||
available = color.available and (size is None or size.available)
|
||
results.append(
|
||
SkuResult(
|
||
options,
|
||
sample.price_cent,
|
||
available,
|
||
sample.raw_price,
|
||
{"color": color.text}
|
||
if sample.price_cent is not None
|
||
else {},
|
||
sample.list_price_cent,
|
||
)
|
||
)
|
||
return tuple(results)
|
||
|
||
def _visible_color_rows(
|
||
self, xml_data: str | bytes
|
||
) -> list[list[VisibleSpecOption]]:
|
||
"""把当前视口中完整显示的颜色节点按中心 y 聚类成行。"""
|
||
|
||
root = _parse_xml(xml_data)
|
||
snapshot = parse_spec_panel(xml_data)
|
||
color_dimension = next(
|
||
(item for item in snapshot.dimensions if item.key == "color"),
|
||
None,
|
||
)
|
||
if color_dimension is None:
|
||
return []
|
||
|
||
candidates: list[VisibleSpecOption] = []
|
||
for value in color_dimension.values:
|
||
node = self._find_option_node(root, value.text)
|
||
if node is None:
|
||
continue
|
||
bounds = _parse_bounds(node.get("bounds", ""))
|
||
if bounds is None:
|
||
continue
|
||
candidates.append(
|
||
VisibleSpecOption(
|
||
value.text,
|
||
value.available,
|
||
bounds,
|
||
)
|
||
)
|
||
if not candidates:
|
||
return []
|
||
|
||
parents = {child: parent for parent in root.iter() for child in parent}
|
||
non_scrollable_panel = _find_non_scrollable_spec_panel(root, parents)
|
||
if non_scrollable_panel is not None:
|
||
# 新版面板会按文字长度设置按钮宽度。它没有横向滚动视口，
|
||
# 短按钮不代表被屏幕边缘截断，因此保留所有安全范围内的节点。
|
||
complete = candidates
|
||
else:
|
||
widest = max(item.bounds[2] - item.bounds[0] for item in candidates)
|
||
minimum_width = max(48, int(widest * 0.6))
|
||
complete = [
|
||
item
|
||
for item in candidates
|
||
if item.bounds[2] - item.bounds[0] >= minimum_width
|
||
]
|
||
complete.sort(key=lambda item: ((item.bounds[1] + item.bounds[3]) // 2, item.bounds[0]))
|
||
|
||
rows: list[list[VisibleSpecOption]] = []
|
||
row_centers: list[int] = []
|
||
for item in complete:
|
||
center_y = (item.bounds[1] + item.bounds[3]) // 2
|
||
matching_index = next(
|
||
(
|
||
index
|
||
for index, row_center in enumerate(row_centers)
|
||
if abs(center_y - row_center) <= 80
|
||
),
|
||
None,
|
||
)
|
||
if matching_index is None:
|
||
rows.append([item])
|
||
row_centers.append(center_y)
|
||
else:
|
||
rows[matching_index].append(item)
|
||
for row in rows:
|
||
row.sort(key=lambda item: item.bounds[0])
|
||
return rows
|
||
|
||
@staticmethod
|
||
def _node_selection_state_exposed(node: ET.Element) -> bool:
|
||
return any(
|
||
item.get("selected") is not None or item.get("checked") is not None
|
||
for item in node.iter()
|
||
)
|
||
|
||
@staticmethod
|
||
def _node_is_selected(node: ET.Element) -> bool:
|
||
return any(
|
||
item.get("selected") == "true" or item.get("checked") == "true"
|
||
for item in node.iter()
|
||
)
|
||
|
||
def _color_selection_state_exposed(self, xml_data: str | bytes) -> bool:
|
||
root = _parse_xml(xml_data)
|
||
snapshot = parse_spec_panel(xml_data)
|
||
if snapshot.selected_text:
|
||
return True
|
||
color_dimension = next(
|
||
(item for item in snapshot.dimensions if item.key == "color"),
|
||
None,
|
||
)
|
||
if color_dimension is None:
|
||
return False
|
||
return any(
|
||
node is not None and self._node_selection_state_exposed(node)
|
||
for value in color_dimension.values
|
||
if (node := self._find_option_node(root, value.text)) is not None
|
||
)
|
||
|
||
@staticmethod
|
||
def _find_option_node(root: ET.Element, target: str) -> Optional[ET.Element]:
|
||
parents = {child: parent for parent in root.iter() for child in parent}
|
||
scrollables = [node for node in root.iter("node") if node.get("scrollable") == "true"]
|
||
non_scrollable_panel = _find_non_scrollable_spec_panel(root, parents)
|
||
right_edges = [
|
||
bounds[2]
|
||
for node in root.iter("node")
|
||
if (bounds := _parse_bounds(node.get("bounds", ""))) is not None
|
||
]
|
||
screen_right = max(right_edges, default=0)
|
||
for node in root.iter("node"):
|
||
if _preferred_or_descendant_label(node).strip() != target:
|
||
continue
|
||
bounds = _parse_bounds(node.get("bounds", ""))
|
||
if node.get("clickable") != "true" or bounds is None:
|
||
continue
|
||
center_x = (bounds[0] + bounds[2]) // 2
|
||
if bounds[2] - bounds[0] < 48:
|
||
continue
|
||
if not 24 <= center_x <= screen_right - 24:
|
||
continue
|
||
ancestors = _ancestors(node, parents)
|
||
in_scrollable = any(container in ancestors for container in scrollables)
|
||
in_non_scrollable_panel = non_scrollable_panel is not None and (
|
||
node is non_scrollable_panel or non_scrollable_panel in ancestors
|
||
)
|
||
if in_scrollable or in_non_scrollable_panel:
|
||
return node
|
||
return None
|
||
|
||
@staticmethod
|
||
def _find_safe_option_click_node(
|
||
state_node: ET.Element,
|
||
target: str,
|
||
) -> Optional[ET.Element]:
|
||
"""返回规格文字的安全点击节点，图片卡片绝不使用整卡中心。"""
|
||
|
||
candidates: list[tuple[int, int, ET.Element]] = []
|
||
for node in state_node.iter():
|
||
if _preferred_node_label(node).strip() != target:
|
||
continue
|
||
bounds = _parse_bounds(node.get("bounds", ""))
|
||
class_name = node.get("class") or ""
|
||
if bounds is None or class_name.endswith("ImageView"):
|
||
continue
|
||
if node.get("visible-to-user", "true") != "true":
|
||
continue
|
||
area = (bounds[2] - bounds[0]) * (bounds[3] - bounds[1])
|
||
text_rank = 0 if class_name.endswith("TextView") else 1
|
||
candidates.append((text_rank, area, node))
|
||
if candidates:
|
||
return min(candidates, key=lambda item: (item[0], item[1]))[2]
|
||
|
||
has_image_area = any(
|
||
(node.get("class") or "").endswith("ImageView")
|
||
or _preferred_node_label(node).strip() in ("打开大图", "查看大图")
|
||
for node in state_node.iter()
|
||
)
|
||
return None if has_image_area else state_node
|
||
|
||
@staticmethod
|
||
def _horizontal_region(root: ET.Element) -> Optional[Bounds]:
|
||
parents = {child: parent for parent in root.iter() for child in parent}
|
||
candidates: list[tuple[int, Bounds]] = []
|
||
for node in root.iter("node"):
|
||
if node.get("scrollable") != "true":
|
||
continue
|
||
bounds = _parse_bounds(node.get("bounds", ""))
|
||
if bounds is None:
|
||
continue
|
||
depth = len(_ancestors(node, parents))
|
||
if any(child.get("clickable") == "true" for child in node):
|
||
candidates.append((depth, bounds))
|
||
return max(candidates, key=lambda item: item[0])[1] if candidates else None
|
||
|
||
@staticmethod
|
||
def _vertical_region(root: ET.Element) -> Optional[Bounds]:
|
||
candidates: list[Bounds] = []
|
||
for node in root.iter("node"):
|
||
bounds = _parse_bounds(node.get("bounds", ""))
|
||
if node.get("scrollable") == "true" and bounds is not None:
|
||
candidates.append(bounds)
|
||
if not candidates:
|
||
return None
|
||
return max(
|
||
candidates,
|
||
key=lambda item: (item[3] - item[1]) * (item[2] - item[0]),
|
||
)
|
||
|
||
@staticmethod
|
||
def _swipe_region(device: Any, bounds: Bounds, *, horizontal: bool, reverse: bool) -> None:
|
||
left, top, right, bottom = bounds
|
||
if horizontal:
|
||
y = (top + bottom) // 2
|
||
near_left = left + int((right - left) * 0.25)
|
||
near_right = left + int((right - left) * 0.75)
|
||
start_x, end_x = (
|
||
(near_left, near_right) if reverse else (near_right, near_left)
|
||
)
|
||
device.swipe(start_x, y, end_x, y, duration=0.35)
|
||
else:
|
||
x = (left + right) // 2
|
||
near_top = top + int((bottom - top) * 0.25)
|
||
near_bottom = top + int((bottom - top) * 0.75)
|
||
start_y, end_y = (
|
||
(near_top, near_bottom) if reverse else (near_bottom, near_top)
|
||
)
|
||
device.swipe(x, start_y, x, end_y, duration=0.35)
|