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()