"""PDD 商品采集基础服务。 本模块不依赖 Qt、SQLite 或 Admin。页面操作和 XML 解析拆开，解析函数可以 使用脱敏控件树单独测试；``collect`` 必须由后台工作线程调用。 """ from __future__ import annotations import hashlib import itertools 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 = ("发起拼单", "立即购买", "单独购买", "免拼购买", "快要抢光") 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 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 _screen_bounds(root: ET.Element) -> Optional[Bounds]: bounds_list = [ bounds for node in root.iter("node") if (bounds := _parse_bounds(node.get("bounds", ""))) is not None ] if not bounds_list: return None return ( min(item[0] for item in bounds_list), min(item[1] for item in bounds_list), max(item[2] for item in bounds_list), max(item[3] for item in bounds_list), ) 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 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"(?:\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 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 merge_dimensions(snapshots: Iterable[SpecSnapshot]) -> tuple[SpecDimension, ...]: """合并滚动过程中多棵控件树看到的规格值。""" order: list[str] = [] names: dict[str, str] = {} values: dict[str, dict[str, bool]] = {} for snapshot in snapshots: for dimension in snapshot.dimensions: if dimension.key not in values: order.append(dimension.key) names[dimension.key] = dimension.name values[dimension.key] = {} for value in dimension.values: was_available = values[dimension.key].get(value.text, False) values[dimension.key][value.text] = was_available or value.available return tuple( SpecDimension( key, names[key], tuple(DimensionValue(text, available) for text, available in values[key].items()), ) for key in order ) 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 def _combine_goods_snapshots(snapshots: Sequence[GoodsSnapshot]) -> GoodsSnapshot: return GoodsSnapshot( next((item.title for item in snapshots if item.title), None), next((item.shop_name for item in snapshots if item.shop_name), None), next( (item.sales for item in snapshots if item.sales.raw), QuantityMetric(None, None, False), ), next( (item.reviews for item in snapshots if item.reviews.raw), QuantityMetric(None, None, False), ), ) 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, overall_timeout: float = 600.0, max_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._overall_timeout = overall_timeout self._overall_deadline: Optional[float] = None self._max_page_swipes = max_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.reviews.raw: raise PddCollectError("PDD_DATA_REVIEWS_MISSING", "商品页没有采集到评价数量") if not goods.shop_name and self._last_goods_xml: artifact = self._save_xml("shop-not-found", self._last_goods_xml) if artifact: self._artifacts.append(artifact) home_xml = device.dump_hierarchy() coordinate = get_size_panel_coord(home_xml) if coordinate is None: raise PddCollectError( "PDD_PAGE_SPEC_ENTRY_MISSING", "商品页没有找到可靠的规格入口", ) device.click(*coordinate) self._sleep(0.5) snapshots = self._discover_dimensions(device) dimensions = merge_dimensions(snapshots) if not dimensions or any(not item.values for item in dimensions): raise PddCollectError( "PDD_DATA_SPEC_INCOMPLETE", "规格面板没有采集到完整的规格维度", ) skus = self._collect_skus(device, dimensions) has_available_price = any( item.price_cent is not None for item in skus if item.available ) if not skus or not has_available_price: raise PddCollectError("PDD_DATA_PRICE_MISSING", "没有采集到可用 SKU 的价格") 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 = device.dump_hierarchy() self._last_goods_xml = str(xml_data) self._goods_screens_checked += 1 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: snapshots: list[GoodsSnapshot] = [] previous_signature: Optional[tuple[str, ...]] = None unchanged = 0 for _ in range(self._max_page_swipes + 1): self._check_cancelled() xml_data = device.dump_hierarchy() self._last_goods_xml = str(xml_data) self._goods_screens_checked += 1 root = _parse_xml(xml_data) labels = tuple(_all_labels(root)) snapshots.append(parse_goods_page(xml_data)) combined = _combine_goods_snapshots(snapshots) is_complete = ( combined.title and combined.sales.raw and combined.reviews.raw ) if is_complete: break if labels == previous_signature: unchanged += 1 else: unchanged = 0 if unchanged >= 1: break previous_signature = labels screen = _screen_bounds(root) if screen is None: break left, top, right, bottom = screen x = (left + right) // 2 device.swipe( x, top + int((bottom - top) * 0.75), x, top + int((bottom - top) * 0.30), duration=0.35, ) self._sleep(0.35) return _combine_goods_snapshots(snapshots) 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 _discover_dimensions(self, device: Any) -> list[SpecSnapshot]: snapshots: list[SpecSnapshot] = [] # 当前选中项可能让列表自动停在中间。因此横向和纵向都扫描两个方向， # 不能假设打开面板时正好位于列表起点。 for horizontal, region_getter in ( (True, self._horizontal_region), (False, self._vertical_region), ): for reverse in (False, True): previous_signature: Optional[ tuple[tuple[str, tuple[str, ...]], ...] ] = None for _ in range(self._max_spec_swipes + 1): self._check_cancelled() xml_data = device.dump_hierarchy() snapshot = parse_spec_panel(xml_data) snapshots.append(snapshot) signature = tuple( (item.name, tuple(value.text for value in item.values)) for item in snapshot.dimensions ) root = _parse_xml(xml_data) region = region_getter(root) if region is None or signature == previous_signature: break previous_signature = signature self._swipe_region( device, region, horizontal=horizontal, reverse=reverse, ) self._sleep(0.35) return snapshots def _collect_skus( self, device: Any, dimensions: tuple[SpecDimension, ...], ) -> tuple[SkuResult, ...]: combinations = list(itertools.product(*(item.values for item in dimensions))) if len(combinations) > self._max_sku_count: raise PddCollectError( "PDD_DATA_TOO_MANY_SKUS", f"规格组合共 {len(combinations)} 个，超过安全上限 {self._max_sku_count}", ) color_index = next( (index for index, item in enumerate(dimensions) if item.key == "color"), None, ) if color_index is None: raise PddCollectError( "PDD_DATA_SPEC_INCOMPLETE", "规格面板没有可识别的颜色分类" ) samples: dict[ str, tuple[Optional[int], Optional[str], Optional[int], dict[str, str]], ] = {} color_dimension = dimensions[color_index] for color in color_dimension.values: self._check_cancelled() if not color.available: samples[color.text] = (None, None, None, {"color": color.text}) continue selected = self._select_option(device, color.text, color_dimension.key) if not selected: samples[color.text] = (None, None, None, {"color": color.text}) continue snapshot = parse_spec_panel(device.dump_hierarchy()) summary = snapshot.selected_text or "" observed: dict[str, str] = {} for dimension in dimensions: match = next( (value.text for value in dimension.values if value.text in summary), None, ) if match: observed[dimension.key] = match confirmed = ( observed.get(color_dimension.key) == color.text and len(observed) == len(dimensions) ) samples[color.text] = ( snapshot.price_cent if confirmed else None, snapshot.raw_price if confirmed else None, snapshot.list_price_cent if confirmed else None, observed, ) results: list[SkuResult] = [] for values in combinations: options = { dimension.key: value.text for dimension, value in zip(dimensions, values) } price, raw_price, list_price, observed = samples[values[color_index].text] available = all(value.available for value in values) and price is not None results.append( SkuResult( options, price if available else None, available, raw_price if available else None, observed, list_price if available else None, ) ) return tuple(results) def _select_combination( self, device: Any, pairs: Sequence[tuple[SpecDimension, DimensionValue]], ) -> bool: for dimension, value in pairs: if not self._select_option(device, value.text, dimension.key): return False return True def _select_option(self, device: Any, target: str, dimension_key: str) -> bool: prefer_horizontal = dimension_key == "color" or dimension_key.startswith( "color_" ) for reverse in (False, True): previous_signature: Optional[tuple[str, ...]] = None for _ in range(self._max_spec_swipes + 1): self._check_cancelled() xml_data = device.dump_hierarchy() root = _parse_xml(xml_data) node = self._find_option_node(root, target) if node is not None: if not _is_available(node): return False bounds = _parse_bounds(node.get("bounds", "")) assert bounds is not None device.click( (bounds[0] + bounds[2]) // 2, (bounds[1] + bounds[3]) // 2, ) self._sleep(0.25) labels = _all_labels(_parse_xml(device.dump_hierarchy())) return target in " ".join(labels) signature = tuple(_all_labels(root)) if signature == previous_signature: break previous_signature = signature horizontal = self._horizontal_region(root) vertical = self._vertical_region(root) if horizontal is not None and (prefer_horizontal or vertical is None): self._swipe_region( device, horizontal, horizontal=True, reverse=reverse, ) elif vertical is not None: self._swipe_region( device, vertical, horizontal=False, reverse=reverse, ) else: return False self._sleep(0.25) return False @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)