cmautobuy/client/src/pdd_collect_service.py

"""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 _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,
        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 = _validate_goods_url(goods_url, goods_id)
        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)