1380 lines
50 KiB
Python
1380 lines
50 KiB
Python
"""PDD 商品采集基础服务。
|
||
|
||
本模块不依赖 Qt、SQLite 或 Admin。页面操作和 XML 解析拆开，解析函数可以
|
||
使用脱敏控件树单独测试；``collect`` 必须由后台工作线程调用。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import re
|
||
import time
|
||
import xml.etree.ElementTree as ET
|
||
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 .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 = ("发起拼单", "立即购买", "单独购买", "免拼购买", "快要抢光")
|
||
_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 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 "颜色" in name or "款式" in name:
|
||
base = "color"
|
||
elif "尺码" in name or "尺寸" in name:
|
||
base = "size"
|
||
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(" ", "")
|
||
return compact in _DIMENSION_NAMES or compact.endswith(
|
||
("分类", "规格", "尺寸", "尺码", "颜色", "型号", "款式", "容量", "类型", "版本", "口味")
|
||
)
|
||
|
||
|
||
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", ""))
|
||
]
|
||
if not scrollables:
|
||
raise PddCollectError("PDD_DATA_SPEC_INCOMPLETE", "规格面板没有可识别的规格区域")
|
||
|
||
outer = max(
|
||
scrollables,
|
||
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
|
||
name = "颜色分类"
|
||
preceding = [item for item in heading_nodes if item[0] <= bounds[1] + 30]
|
||
if preceding:
|
||
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))
|
||
|
||
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_cent, raw_price, list_price_cent = _price_from_nodes(
|
||
root.iter("node"), outer_bounds[1]
|
||
)
|
||
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:
|
||
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 = 12,
|
||
max_spec_swipes: int = 12,
|
||
max_sku_count: int = 200,
|
||
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
|
||
self._artifact_directory = artifact_directory
|
||
self._last_goods_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._check_cancelled()
|
||
|
||
try:
|
||
with self._device_service.connect(self._device_address) as device:
|
||
self._open_goods(device, goods_url)
|
||
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)
|
||
|
||
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 not exc.diagnostics and self._last_goods_xml:
|
||
artifact = self._save_xml("collect-failed", self._last_goods_xml)
|
||
if artifact:
|
||
exc.diagnostics = {
|
||
"artifacts": [artifact],
|
||
"goods_screens_checked": self._goods_screens_checked,
|
||
}
|
||
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) -> None:
|
||
try:
|
||
current = device.app_current()
|
||
if current.get("package") != PDD_PACKAGE_NAME:
|
||
device.app_start(PDD_PACKAGE_NAME)
|
||
if not device.app_wait(PDD_PACKAGE_NAME, timeout=10):
|
||
raise PddCollectError("DEVICE_APP_START_FAILED", "PDD 应用启动失败")
|
||
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
|
||
|
||
deadline = self._monotonic() + self._page_timeout
|
||
while self._monotonic() < deadline:
|
||
self._check_cancelled()
|
||
current = device.app_current()
|
||
xml_data = self._dump_hierarchy(device)
|
||
root = _parse_xml(xml_data)
|
||
last_labels = _all_labels(root)
|
||
pdd_node_count = sum(
|
||
1
|
||
for node in root.iter("node")
|
||
if node.get("package") == PDD_PACKAGE_NAME
|
||
)
|
||
# 部分 OPPO/ColorOS 设备会一直把无线调试设置页报告为焦点，
|
||
# 即使屏幕和无障碍树已经是 PDD。此时以树中真实包名为准。
|
||
is_pdd_hierarchy = pdd_node_count >= 3
|
||
is_pdd_focused = current.get("package") == PDD_PACKAGE_NAME
|
||
if is_pdd_focused or is_pdd_hierarchy:
|
||
_raise_special_page(last_labels)
|
||
combined = " ".join(last_labels)
|
||
loading = "加载中" in combined or "正在加载" in combined
|
||
if not loading and any(marker in combined for marker in _READY_MARKERS):
|
||
return
|
||
self._sleep(0.5)
|
||
raise PddCollectError("PDD_PAGE_TIMEOUT", "等待 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
|
||
while self._monotonic() < deadline:
|
||
self._check_cancelled()
|
||
xml_data = self._dump_hierarchy(device)
|
||
try:
|
||
snapshot = parse_spec_panel(xml_data)
|
||
except PddCollectError as exc:
|
||
if exc.code != "PDD_DATA_SPEC_INCOMPLETE":
|
||
raise
|
||
else:
|
||
if snapshot.dimensions:
|
||
return snapshot
|
||
self._sleep(0.25)
|
||
raise PddCollectError(
|
||
"PDD_PAGE_SPEC_PANEL_TIMEOUT",
|
||
"点击规格入口后，等待规格面板加载超时",
|
||
)
|
||
|
||
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 = self._dump_hierarchy(device)
|
||
first_rows = self._visible_color_rows(first_xml)
|
||
if 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
|
||
|
||
for swipe_count in range(self._max_spec_swipes + 1):
|
||
self._check_cancelled()
|
||
# 每次只处理一个节点；点击可能让列表自动移动，下一项必须
|
||
# 从最新 XML 重新计算，不能继续使用点击前的旧坐标。
|
||
while True:
|
||
xml_data = self._dump_hierarchy(device)
|
||
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(self._dump_hierarchy(device))
|
||
region = self._horizontal_region(root)
|
||
if region is None:
|
||
break
|
||
self._swipe_region(
|
||
device,
|
||
region,
|
||
horizontal=True,
|
||
reverse=not move_right,
|
||
)
|
||
self._sleep(0.35)
|
||
|
||
# 操作采用蛇形以减少无效滑动；输出仍恢复成页面自然的
|
||
# “每行从左到右”顺序，方便 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", "颜色分类", 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
|
||
for _ in range(self._max_spec_swipes):
|
||
self._check_cancelled()
|
||
xml_data = self._dump_hierarchy(device)
|
||
rows = self._visible_color_rows(xml_data)
|
||
signature = tuple(
|
||
(item.text, item.bounds) for row in rows for item in row
|
||
)
|
||
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
|
||
)
|
||
self._sleep(0.35)
|
||
|
||
def _click_and_sample_color(
|
||
self, device: Any, target: str
|
||
) -> ColorPriceSample:
|
||
"""点击最新树中的颜色，等待选择和价格连续两次稳定。"""
|
||
|
||
xml_data = self._dump_hierarchy(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,
|
||
)
|
||
node = self._find_option_node(root, target) if latest_visible 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,
|
||
)
|
||
|
||
previous_price: Optional[tuple[int, Optional[str], Optional[int]]] = None
|
||
stable_price_reads = 0
|
||
for _ in range(8):
|
||
self._check_cancelled()
|
||
self._sleep(0.2)
|
||
latest_xml = self._dump_hierarchy(device)
|
||
latest_root = _parse_xml(latest_xml)
|
||
snapshot = parse_spec_panel(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
|
||
|
||
if snapshot.price_cent is None or not selection_confirmed:
|
||
stable_price_reads = 0
|
||
previous_price = None
|
||
continue
|
||
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,
|
||
)
|
||
|
||
return ColorPriceSample(None, None, None)
|
||
|
||
def _collect_size_dimension(self, device: Any) -> Optional[SpecDimension]:
|
||
"""颜色采价完成后只滚动并收集尺码文字，不点击尺码。"""
|
||
|
||
sizes: dict[str, bool] = {}
|
||
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 = self._dump_hierarchy(device)
|
||
snapshot = parse_spec_panel(xml_data)
|
||
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":
|
||
for value in dimension.values:
|
||
sizes[value.text] = sizes.get(value.text, False) or value.available
|
||
|
||
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 sizes:
|
||
return None
|
||
return SpecDimension(
|
||
"size",
|
||
"尺码",
|
||
tuple(DimensionValue(text, available) for text, available in sizes.items()),
|
||
)
|
||
|
||
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 []
|
||
|
||
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"]
|
||
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
|
||
if any(container in _ancestors(node, parents) for container in scrollables):
|
||
return node
|
||
return None
|
||
|
||
@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)
|