From 4aff22c009ba71e0d31ba226bde3a243636719ef Mon Sep 17 00:00:00 2001 From: chengma Date: Fri, 7 Aug 2026 16:12:39 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=9E=E7=8E=B0=20PDD=20=E5=95=86?= =?UTF-8?q?=E5=93=81=E9=87=87=E9=9B=86=E5=9F=BA=E7=A1=80=E6=9C=8D=E5=8A=A1?= =?UTF-8?q?=20(#28)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- client/src/pdd_collect_service.py | 910 ++++++++++++++++++++++++ client/src/pdd_device_service.py | 139 ++++ client/test/fixtures/pdd_goods_page.xml | 13 + client/test/fixtures/pdd_spec_panel.xml | 21 + client/test/test_pdd_collect_service.py | 213 ++++++ client/test/test_pdd_device_service.py | 69 ++ 6 files changed, 1365 insertions(+) create mode 100644 client/src/pdd_collect_service.py create mode 100644 client/src/pdd_device_service.py create mode 100644 client/test/fixtures/pdd_goods_page.xml create mode 100644 client/test/fixtures/pdd_spec_panel.xml create mode 100644 client/test/test_pdd_collect_service.py create mode 100644 client/test/test_pdd_device_service.py diff --git a/client/src/pdd_collect_service.py b/client/src/pdd_collect_service.py new file mode 100644 index 0000000..351165c --- /dev/null +++ b/client/src/pdd_collect_service.py @@ -0,0 +1,910 @@ +"""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 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 = ("发起拼单", "立即购买", "单独购买", "免拼购买", "快要抢光") + + +class PddCollectError(RuntimeError): + """采集失败，并携带稳定错误码。""" + + def __init__(self, code: str, message: str) -> None: + super().__init__(message) + self.code = code + self.message = message + + +@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] + + def to_dict(self) -> dict[str, Any]: + return { + "options": dict(self.options), + "price_cent": self.price_cent, + "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] + + +@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 + + def to_pdd_data(self) -> dict[str, Any]: + return { + "schema_version": 1, + "goods": { + "goods_id": self.goods_id, + "url": self.goods_url, + "title": self.title, + }, + "shop": {"name": self.shop_name}, + "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": [], + } + + +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 _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 _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: + candidates = [ + label + for label in labels + 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]]: + 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 + _, cents, raw = min(candidates, key=lambda item: (item[0], item[1])) + return cents, raw + + +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 = _own_or_descendant_label(node).strip() + 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 = _price_from_nodes(root.iter("node"), outer_bounds[1]) + return SpecSnapshot(tuple(dimensions), selected_text, price_cent, raw_price) + + +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 _goods_id_from_url(goods_url: str) -> str: + value = parse_qs(urlparse(goods_url).query).get("goods_id", [""])[0].strip() + if not value: + 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, + max_page_swipes: int = 12, + max_spec_swipes: int = 12, + max_sku_count: int = 200, + ) -> 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._max_page_swipes = max_page_swipes + self._max_spec_swipes = max_spec_swipes + self._max_sku_count = max_sku_count + + 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 = goods_id or _goods_id_from_url(goods_url) + 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.shop_name: + raise PddCollectError("PDD_DATA_SHOP_MISSING", "商品页没有采集到店铺名称") + if not goods.sales.raw: + raise PddCollectError("PDD_DATA_SALES_MISSING", "商品页没有采集到已拼数量") + if not goods.reviews.raw: + raise PddCollectError("PDD_DATA_REVIEWS_MISSING", "商品页没有采集到评价数量") + + 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: + raise + except PddDeviceError as exc: + raise PddCollectError(exc.code, exc.message) from exc + except Exception as 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, + ) + + def _check_cancelled(self) -> None: + if self._cancelled(): + raise PddCollectError("PDD_CANCELLED", "采集任务已安全取消") + + 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: + raise PddCollectError("DEVICE_APP_START_FAILED", f"无法打开 PDD 商品链接：{exc}") from exc + + deadline = self._monotonic() + self._page_timeout + last_labels: list[str] = [] + while self._monotonic() < deadline: + self._check_cancelled() + current = device.app_current() + if current.get("package") == PDD_PACKAGE_NAME: + xml_data = device.dump_hierarchy() + root = _parse_xml(xml_data) + last_labels = _all_labels(root) + _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_special_page(last_labels) + 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() + 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.shop_name + 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 _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}", + ) + + results: list[SkuResult] = [] + for values in combinations: + self._check_cancelled() + options = {dimension.key: value.text for dimension, value in zip(dimensions, values)} + pairs = list(zip(dimensions, values)) + selected = self._select_combination(device, pairs) + if not selected and len(pairs) > 1: + # 某个选项可能只是在当前搭配下禁用。反向选择一次，可以先改变 + # 依赖维度，再重新判断目标组合是否真的缺货。 + selected = self._select_combination(device, list(reversed(pairs))) + if not selected: + results.append(SkuResult(options, None, False, None)) + continue + snapshot = parse_spec_panel(device.dump_hierarchy()) + summary = snapshot.selected_text or "" + confirmed = all(value.text in summary for value in values) + results.append( + SkuResult( + options, + snapshot.price_cent if confirmed else None, + confirmed and snapshot.price_cent is not None, + snapshot.raw_price if confirmed 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 _own_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) diff --git a/client/src/pdd_device_service.py b/client/src/pdd_device_service.py new file mode 100644 index 0000000..008313a --- /dev/null +++ b/client/src/pdd_device_service.py @@ -0,0 +1,139 @@ +"""uiautomator2 设备连接边界。 + +每次自动化任务通过 ``connect`` 获取一个会话。会话只能在创建它的线程中 +使用，退出 ``with`` 后立即释放，避免多个任务同时控制同一台手机。 +""" + +from __future__ import annotations + +import threading +from contextlib import AbstractContextManager +from typing import Any, Callable, Optional + + +PDD_PACKAGE_NAME = "com.xunmeng.pinduoduo" + + +class PddDeviceError(RuntimeError): + """设备连接失败，并携带供任务状态使用的稳定错误码。""" + + def __init__(self, code: str, message: str) -> None: + super().__init__(message) + self.code = code + self.message = message + + +class PddDeviceSession(AbstractContextManager[Any]): + """一个只能由创建线程使用的 uiautomator2 Device 会话。""" + + def __init__( + self, + device: Any, + serial: str, + release: Callable[[], None], + ) -> None: + self._device = device + self.serial = serial + self._release = release + self._owner_thread_id = threading.get_ident() + self._closed = False + + @property + def device(self) -> Any: + """返回 Device；跨线程或会话关闭后访问会明确失败。""" + + if self._closed: + raise PddDeviceError("DEVICE_SESSION_CLOSED", "设备会话已经关闭") + if threading.get_ident() != self._owner_thread_id: + raise PddDeviceError( + "DEVICE_THREAD_VIOLATION", + "uiautomator2 Device 只能在创建会话的工作线程中使用", + ) + return self._device + + def __enter__(self) -> Any: + return self.device + + def __exit__(self, exc_type: Any, exc: Any, traceback: Any) -> None: + if not self._closed: + self._closed = True + self._release() + return None + + +class PddDeviceService: + """连接并校验一台已保存的 Android 设备。""" + + def __init__( + self, + connector: Optional[Callable[[str], Any]] = None, + ) -> None: + self._connector = connector or self._default_connector + self._state_lock = threading.Lock() + self._active_serial: Optional[str] = None + + @staticmethod + def _default_connector(serial: str) -> Any: + try: + import uiautomator2 as u2 + except ImportError as exc: + raise PddDeviceError( + "DEVICE_U2_MISSING", + "未安装 uiautomator2，请先安装 client/requirements.txt", + ) from exc + return u2.connect(serial) + + @staticmethod + def _validate_serial(serial: str) -> str: + value = str(serial or "").strip() + if not value or any(character.isspace() for character in value): + raise PddDeviceError("DEVICE_ADDRESS_INVALID", "Android 设备号无效") + return value + + def connect(self, serial: str) -> PddDeviceSession: + """连接设备并返回单线程独占会话。 + + 调用方必须使用 ``with service.connect(serial) as device``。同一个服务 + 已有活动会话时会拒绝第二次连接。 + """ + + checked_serial = self._validate_serial(serial) + with self._state_lock: + if self._active_serial is not None: + raise PddDeviceError( + "DEVICE_IN_USE", + f"设备 {self._active_serial} 正在执行其他自动化任务", + ) + self._active_serial = checked_serial + + try: + device = self._connector(checked_serial) + current = device.app_current() + if not isinstance(current, dict): + raise RuntimeError("uiautomator2 未返回有效的设备状态") + except PddDeviceError: + self._release(checked_serial) + raise + except Exception as exc: + self._release(checked_serial) + details = str(exc).lower() + code = ( + "DEVICE_OFFLINE" + if any(word in details for word in ("offline", "not found", "disconnected")) + else "DEVICE_CONNECT_FAILED" + ) + raise PddDeviceError( + code, + f"无法连接 Android 设备 {checked_serial}：{exc}", + ) from exc + + return PddDeviceSession( + device, + checked_serial, + lambda: self._release(checked_serial), + ) + + def _release(self, serial: str) -> None: + with self._state_lock: + if self._active_serial == serial: + self._active_serial = None diff --git a/client/test/fixtures/pdd_goods_page.xml b/client/test/fixtures/pdd_goods_page.xml new file mode 100644 index 0000000..f755603 --- /dev/null +++ b/client/test/fixtures/pdd_goods_page.xml @@ -0,0 +1,13 @@ + + + + + + + + + + + + + diff --git a/client/test/fixtures/pdd_spec_panel.xml b/client/test/fixtures/pdd_spec_panel.xml new file mode 100644 index 0000000..518254b --- /dev/null +++ b/client/test/fixtures/pdd_spec_panel.xml @@ -0,0 +1,21 @@ + + + + + + + + + + + + + + + + + + + + + diff --git a/client/test/test_pdd_collect_service.py b/client/test/test_pdd_collect_service.py new file mode 100644 index 0000000..27cad43 --- /dev/null +++ b/client/test/test_pdd_collect_service.py @@ -0,0 +1,213 @@ +"""PDD 商品采集与 XML 解析测试，不连接真机。""" + +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +import unittest +import xml.etree.ElementTree as ET + +from src.pdd_collect_service import ( + PddCollectError, + PddCollectService, + parse_goods_page, + parse_quantity, + parse_spec_panel, +) +from src.pdd_device_service import PddDeviceService + + +FIXTURES = Path(__file__).parent / "fixtures" + + +@dataclass(frozen=True) +class FakeTask: + goods_url: str + goods_id: str | None = None + + +class FakeCollectDevice: + def __init__(self, home_xml: str, spec_xml: str): + self.home_xml = home_xml + self.spec_xml = spec_xml + self.panel_open = False + self.clicks = [] + self.swipes = [] + + def app_current(self): + return {"package": "com.xunmeng.pinduoduo"} + + def app_start(self, _package): + raise AssertionError("PDD 已经在前台，不应重复启动") + + def app_wait(self, _package, timeout=10): + return 123 + + def open_url(self, url): + self.opened_url = url + + def dump_hierarchy(self): + return self.spec_xml if self.panel_open else self.home_xml + + def click(self, x, y): + self.clicks.append((x, y)) + if y > 1600: + self.panel_open = True + + def swipe(self, *args, **kwargs): + self.swipes.append((args, kwargs)) + + +class LoadingDevice(FakeCollectDevice): + def dump_hierarchy(self): + return '' + + +def keep_only_one_sku(xml_data: str) -> str: + """从脱敏固件中删除蓝色和 L，只保留一个组合。""" + + root = ET.fromstring(xml_data) + for parent in root.iter(): + for child in list(parent): + own_label = ( + child.get("text") or child.get("content-desc") or "" + ).strip() + labels = " ".join( + (node.get("text") or node.get("content-desc") or "").strip() + for node in child.iter("node") + ) + if own_label == "蓝色" or labels.strip() == "L": + parent.remove(child) + return ET.tostring(root, encoding="unicode") + + +class PddCollectParserTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.home_xml = (FIXTURES / "pdd_goods_page.xml").read_text(encoding="utf-8") + cls.spec_xml = (FIXTURES / "pdd_spec_panel.xml").read_text(encoding="utf-8") + + def test_quantity_keeps_raw_value_and_approximate_flag(self): + result = parse_quantity("已拼1.2万+件") + self.assertEqual(result.value, 12000) + self.assertEqual(result.raw, "已拼1.2万+件") + self.assertTrue(result.approximate) + + def test_parse_goods_page(self): + result = parse_goods_page(self.home_xml) + self.assertEqual(result.title, "测试纯棉短袖商品") + self.assertEqual(result.shop_name, "测试服饰旗舰店") + self.assertEqual(result.sales.value, 12000) + self.assertTrue(result.sales.approximate) + self.assertEqual(result.reviews.value, 2356) + + def test_parse_spec_panel_uses_generic_dimensions_and_cents(self): + result = parse_spec_panel(self.spec_xml) + self.assertEqual(result.price_cent, 1000) + self.assertEqual([item.key for item in result.dimensions], ["color", "size"]) + self.assertEqual( + [value.text for value in result.dimensions[0].values], + ["红色", "蓝色"], + ) + self.assertFalse(result.dimensions[0].values[1].available) + self.assertEqual( + [value.text for value in result.dimensions[1].values], + ["M", "L"], + ) + + def test_login_and_invalid_xml_have_different_error_codes(self): + with self.assertRaises(PddCollectError) as login: + parse_goods_page('') + with self.assertRaises(PddCollectError) as invalid: + parse_goods_page("") + self.assertEqual(login.exception.code, "PDD_PAGE_LOGIN_REQUIRED") + self.assertEqual(invalid.exception.code, "PDD_DATA_XML_INVALID") + + def test_captcha_and_incomplete_spec_have_specific_codes(self): + with self.assertRaises(PddCollectError) as captcha: + parse_goods_page('') + with self.assertRaises(PddCollectError) as incomplete: + parse_spec_panel('') + self.assertEqual(captcha.exception.code, "PDD_PAGE_CAPTCHA") + self.assertEqual(incomplete.exception.code, "PDD_DATA_SPEC_INCOMPLETE") + + def test_collect_returns_versioned_pdd_data_without_real_device(self): + # 收窄到一个可用组合，便于验证完整流程而不模拟真实页面切换。 + one_sku_xml = keep_only_one_sku(self.spec_xml) + device = FakeCollectDevice(self.home_xml, one_sku_xml) + service = PddCollectService( + PddDeviceService(lambda _serial: device), + "USB-001", + "client-001", + sleeper=lambda _seconds: None, + now=lambda: datetime(2026, 8, 7, 8, 0, tzinfo=timezone.utc), + max_page_swipes=1, + max_spec_swipes=1, + ) + + result = service.collect( + FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123") + ) + data = result.to_pdd_data() + + self.assertEqual(data["schema_version"], 1) + self.assertEqual(data["goods"]["goods_id"], "123") + self.assertEqual(data["metrics"]["sales"]["value"], 12000) + self.assertEqual(data["dimensions"][0]["key"], "color") + self.assertEqual(data["skus"][0]["options"], {"color": "红色", "size": "M"}) + self.assertEqual(data["skus"][0]["price_cent"], 1000) + self.assertEqual(data["source"]["device_address"], "USB-001") + self.assertIsNone(data["purchase"]) + + def test_missing_goods_id_is_structured_error(self): + service = PddCollectService( + PddDeviceService(lambda _serial: object()), + "USB-001", + "client-001", + ) + with self.assertRaises(PddCollectError) as raised: + service.collect(FakeTask("https://example.com/goods.html")) + self.assertEqual(raised.exception.code, "PDD_DATA_GOODS_ID_MISSING") + + def test_page_timeout_has_specific_error_code(self): + device = LoadingDevice(self.home_xml, self.spec_xml) + ticks = iter((0.0, 2.0)) + service = PddCollectService( + PddDeviceService(lambda _serial: device), + "USB-001", + "client-001", + sleeper=lambda _seconds: None, + monotonic=lambda: next(ticks), + page_timeout=1.0, + ) + + with self.assertRaises(PddCollectError) as raised: + service.collect( + FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123") + ) + self.assertEqual(raised.exception.code, "PDD_PAGE_TIMEOUT") + + def test_incomplete_goods_data_does_not_return_partial_success(self): + incomplete_home = self.home_xml.replace( + '', + "", + ) + device = FakeCollectDevice(incomplete_home, self.spec_xml) + service = PddCollectService( + PddDeviceService(lambda _serial: device), + "USB-001", + "client-001", + sleeper=lambda _seconds: None, + max_page_swipes=0, + ) + + with self.assertRaises(PddCollectError) as raised: + service.collect( + FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123") + ) + self.assertEqual(raised.exception.code, "PDD_DATA_SHOP_MISSING") + + +if __name__ == "__main__": + unittest.main() diff --git a/client/test/test_pdd_device_service.py b/client/test/test_pdd_device_service.py new file mode 100644 index 0000000..fb147aa --- /dev/null +++ b/client/test/test_pdd_device_service.py @@ -0,0 +1,69 @@ +"""uiautomator2 设备连接边界测试，不连接真机。""" + +import threading +import unittest + +from src.pdd_device_service import PddDeviceError, PddDeviceService + + +class FakeDevice: + def app_current(self): + return {"package": "com.xunmeng.pinduoduo"} + + +class PddDeviceServiceTest(unittest.TestCase): + def test_connect_validates_and_releases_exclusive_session(self): + connected = [] + service = PddDeviceService(lambda serial: connected.append(serial) or FakeDevice()) + + with service.connect("USB-001") as device: + self.assertIsInstance(device, FakeDevice) + with self.assertRaisesRegex(PddDeviceError, "正在执行"): + service.connect("USB-002") + + with service.connect("USB-002"): + pass + self.assertEqual(connected, ["USB-001", "USB-002"]) + + def test_invalid_serial_is_rejected_before_connect(self): + with self.assertRaisesRegex(PddDeviceError, "设备号无效") as raised: + PddDeviceService(lambda _serial: FakeDevice()).connect("bad serial") + self.assertEqual(raised.exception.code, "DEVICE_ADDRESS_INVALID") + + def test_offline_connector_error_has_specific_code(self): + def fail(_serial): + raise OSError("offline") + + with self.assertRaises(PddDeviceError) as raised: + PddDeviceService(fail).connect("USB-001") + self.assertEqual(raised.exception.code, "DEVICE_OFFLINE") + self.assertIn("offline", raised.exception.message) + + def test_other_connector_error_becomes_connect_failed(self): + def fail(_serial): + raise OSError("permission denied") + + with self.assertRaises(PddDeviceError) as raised: + PddDeviceService(fail).connect("USB-001") + self.assertEqual(raised.exception.code, "DEVICE_CONNECT_FAILED") + + def test_device_object_cannot_be_used_from_another_thread(self): + session = PddDeviceService(lambda _serial: FakeDevice()).connect("USB-001") + errors = [] + + def read_device(): + try: + session.device + except PddDeviceError as exc: + errors.append(exc.code) + + worker = threading.Thread(target=read_device) + worker.start() + worker.join() + session.__exit__(None, None, None) + + self.assertEqual(errors, ["DEVICE_THREAD_VIOLATION"]) + + +if __name__ == "__main__": + unittest.main()