cmautobuy/client/src/pdd_collect_service.py

"""PDD 商品采集基础服务。

本模块不依赖 Qt、SQLite 或 Admin。页面操作和 XML 解析拆开，解析函数可以
使用脱敏控件树单独测试；``collect`` 必须由后台工作线程调用。
"""

from __future__ import annotations

import hashlib
import math
import re
import time
import xml.etree.ElementTree as ET
from contextlib import nullcontext
from dataclasses import dataclass
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
from pathlib import Path
from typing import Any, Callable, Iterable, Mapping, Optional, Sequence
from urllib.parse import parse_qs, urlparse

from .pdd_device_service import PDD_PACKAGE_NAME, PddDeviceError, PddDeviceService
from .performance_timing import current_performance_trace
from .pdd_page_classifier import (
    ACTION_NETWORK_ERROR,
    ACTION_READY,
    ACTION_REOPEN,
    ACTION_UNAVAILABLE,
    PAGE_CAPTCHA,
    PAGE_GOODS,
    PAGE_HOME,
    PAGE_LOGIN_REQUIRED,
    PAGE_NETWORK_ERROR,
    PAGE_PAYMENT,
    PAGE_RISK_CONTROL,
    GoodsOpenTracker,
    PddPageObservation,
    classify_pdd_page,
)
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 = ("发起拼单", "立即购买", "单独购买", "免拼购买", "快要抢光")
_OUT_OF_STOCK_PAGE_MARKERS = (
    "商品已售罄，推荐以下相似商品",
    "商品已售罄，看看相似商品",
    "该商品已售罄",
    "该商品暂时缺货",
    "商品暂时缺货",
)
_SHOP_NAME_EXCLUDES = frozenset({"进店", "关注", "店铺", "收藏", "客服"})
_SHOP_ROW_TOLERANCE = 40


def _is_device_disconnect(error: BaseException) -> bool:
    details = str(error).lower()
    return any(
        marker in details
        for marker in (
            "device not found",
            "device offline",
            "disconnected",
            "closed transport",
        )
    )


class PddCollectError(RuntimeError):
    """采集失败，并携带稳定错误码。"""

    def __init__(
        self,
        code: str,
        message: str,
        diagnostics: Optional[Mapping[str, Any]] = None,
    ) -> None:
        super().__init__(message)
        self.code = code
        self.message = message
        self.diagnostics = dict(diagnostics or {})


@dataclass(frozen=True)
class QuantityMetric:
    """销量或评价数量，保留原文和是否为近似值。"""

    value: Optional[int]
    raw: Optional[str]
    approximate: bool

    def to_dict(self) -> dict[str, Any]:
        return {
            "value": self.value,
            "raw": self.raw,
            "approximate": self.approximate,
        }


@dataclass(frozen=True)
class DimensionValue:
    text: str
    available: bool

    def to_dict(self) -> dict[str, Any]:
        return {"text": self.text, "available": self.available}


@dataclass(frozen=True)
class SpecDimension:
    key: str
    name: str
    values: tuple[DimensionValue, ...]

    def to_dict(self) -> dict[str, Any]:
        return {
            "key": self.key,
            "name": self.name,
            "values": [value.to_dict() for value in self.values],
        }


@dataclass(frozen=True)
class SkuResult:
    options: Mapping[str, str]
    price_cent: Optional[int]
    available: bool
    raw_price: Optional[str]
    price_observed_at: Mapping[str, str]
    list_price_cent: Optional[int] = None

    def to_dict(self) -> dict[str, Any]:
        return {
            "options": dict(self.options),
            "price_cent": self.price_cent,
            "list_price_cent": self.list_price_cent,
            "price_observed_at": dict(self.price_observed_at),
            "currency": "CNY",
            "available": self.available,
            "raw_price": self.raw_price,
        }


@dataclass(frozen=True)
class GoodsSnapshot:
    title: Optional[str]
    shop_name: Optional[str]
    sales: QuantityMetric
    reviews: QuantityMetric


@dataclass(frozen=True)
class SpecSnapshot:
    dimensions: tuple[SpecDimension, ...]
    selected_text: Optional[str]
    price_cent: Optional[int]
    raw_price: Optional[str]
    list_price_cent: Optional[int]


@dataclass(frozen=True)
class VisibleSpecOption:
    """当前控件树中完整可点击的规格节点。"""

    text: str
    available: bool
    bounds: Bounds


@dataclass(frozen=True)
class ColorPriceSample:
    """点击一个颜色后稳定读取到的颜色级价格。"""

    price_cent: Optional[int]
    raw_price: Optional[str]
    list_price_cent: Optional[int]


@dataclass(frozen=True)
class CollectResult:
    """与 ``pdd_data`` v1 对应的采集结果。"""

    goods_id: str
    goods_url: str
    title: str
    shop_name: Optional[str]
    sales: QuantityMetric
    reviews: QuantityMetric
    dimensions: tuple[SpecDimension, ...]
    skus: tuple[SkuResult, ...]
    captured_at: str
    client_id: str
    device_address: str
    artifacts: tuple[Mapping[str, Any], ...] = ()

    def to_pdd_data(self) -> dict[str, Any]:
        return {
            "schema_version": 1,
            "goods_id": self.goods_id,
            "goods_url": self.goods_url,
            "title": self.title,
            "shop_name": self.shop_name,
            "price_granularity": "color",
            "metrics": {
                "sales": self.sales.to_dict(),
                "reviews": self.reviews.to_dict(),
            },
            "dimensions": [item.to_dict() for item in self.dimensions],
            "skus": [item.to_dict() for item in self.skus],
            "purchase": None,
            "captured_at": self.captured_at,
            "source": {
                "client_id": self.client_id,
                "device_address": self.device_address,
                "pdd_package": PDD_PACKAGE_NAME,
            },
            "artifacts": [dict(item) for item in self.artifacts],
        }


def _parse_xml(xml_data: str | bytes) -> ET.Element:
    try:
        return ET.fromstring(xml_data)
    except (ET.ParseError, TypeError) as exc:
        raise PddCollectError(
            "PDD_DATA_XML_INVALID",
            "PDD 返回的无障碍控件树不是有效 XML",
        ) from exc


def _parse_bounds(value: str) -> Optional[Bounds]:
    match = _BOUNDS_PATTERN.fullmatch((value or "").strip())
    if not match:
        return None
    left, top, right, bottom = map(int, match.groups())
    if right <= left or bottom <= top:
        return None
    return left, top, right, bottom


def _node_label(node: ET.Element) -> str:
    return " ".join(
        part.strip()
        for part in (node.get("text", ""), node.get("content-desc", ""))
        if part.strip()
    )


def _preferred_node_label(node: ET.Element) -> str:
    """规格名被 text 截断时，优先采用完整的无障碍描述。"""

    text = (node.get("text") or "").strip()
    description = (node.get("content-desc") or "").strip()
    return description or text


def _own_or_descendant_label(node: ET.Element) -> str:
    own = _node_label(node)
    if own:
        return own
    for child in node.iter("node"):
        label = _node_label(child)
        if label and label != "打开大图":
            return label
    return ""


def _preferred_or_descendant_label(node: ET.Element) -> str:
    own = _preferred_node_label(node)
    if own:
        return own
    for child in node.iter("node"):
        label = _preferred_node_label(child)
        if label and label != "打开大图":
            return label
    return ""


def _all_labels(root: ET.Element) -> list[str]:
    return [label for node in root.iter("node") if (label := _node_label(node))]


def parse_quantity(raw: Optional[str]) -> QuantityMetric:
    """把“已拼1.2万+”等文字转成整数，同时保留原文。"""

    if not raw:
        return QuantityMetric(None, None, False)
    match = _QUANTITY_PATTERN.search(raw.replace(",", ""))
    if match is None:
        return QuantityMetric(None, raw, False)
    multiplier = {None: 1, "万": 10_000, "亿": 100_000_000}[match.group(2)]
    value = int(Decimal(match.group(1)) * multiplier)
    approximate = bool(match.group(2) or match.group(3))
    return QuantityMetric(value, raw, approximate)


def _find_metric(labels: Sequence[str], pattern: re.Pattern[str]) -> QuantityMetric:
    for label in labels:
        match = pattern.search(label)
        if match:
            return parse_quantity(match.group(0))
    return QuantityMetric(None, None, False)


def _shop_name_by_enter_anchor(root: ET.Element) -> Optional[str]:
    """以“进店”为锚点，读取同一行左侧的店铺名。"""

    anchors = []
    for node in root.iter("node"):
        if _preferred_node_label(node) != "进店":
            continue
        bounds = _parse_bounds(node.get("bounds", ""))
        if bounds is not None:
            anchors.append(bounds)
    if not anchors:
        return None

    anchor_left, anchor_top, _, _ = min(
        anchors, key=lambda bounds: (bounds[1], bounds[0])
    )
    candidates = []
    for node in root.iter("node"):
        if node.get("class") != "android.widget.TextView":
            continue
        text = _preferred_node_label(node)
        if not text or text in _SHOP_NAME_EXCLUDES:
            continue
        if "已拼" in text or "评价" in text or not 2 <= len(text) <= 30:
            continue
        bounds = _parse_bounds(node.get("bounds", ""))
        if bounds is None:
            continue
        left, top, _, _ = bounds
        if left >= anchor_left or abs(top - anchor_top) > _SHOP_ROW_TOLERANCE:
            continue
        candidates.append((abs(top - anchor_top), -left, text))

    if not candidates:
        return None
    return min(candidates)[2]


def parse_goods_page(xml_data: str | bytes) -> GoodsSnapshot:
    """解析一棵商品页控件树中的标题、店铺、销量和评价。"""

    root = _parse_xml(xml_data)
    labels = _all_labels(root)
    _raise_special_page(labels)

    title: Optional[str] = None
    for node in root.iter("node"):
        if node.get("class") == "androidx.viewpager.widget.ViewPager":
            candidate = (node.get("content-desc") or node.get("text") or "").strip()
            if len(candidate) >= 6:
                title = candidate
                break
    if title is None:
        line_parts: dict[int, list[tuple[int, str]]] = {}
        for node in root.iter("node"):
            label = _preferred_node_label(node)
            bounds = _parse_bounds(node.get("bounds", ""))
            if not label or bounds is None:
                continue
            line_key = bounds[1] // 24
            line_parts.setdefault(line_key, []).append((bounds[0], label))
        joined_lines = [
            "".join(text for _, text in sorted(parts))
            for parts in line_parts.values()
        ]
        candidates = [
            label
            for label in labels + joined_lines
            if len(label) >= 12
            and not any(word in label for word in ("通知", "支付", "已拼", "评价"))
        ]
        title = max(candidates, key=len, default=None)

    sales = _find_metric(labels, re.compile(r"已拼\s*\d+(?:\.\d+)?\s*(?:万|亿)?\s*\+?\s*(?:件|人)?"))
    reviews_pattern = re.compile(
        r"(?:商品评价\s*[（(]\s*\d+(?:\.\d+)?\s*(?:万|亿)?\s*\+?\s*[)）]"
        r"|\d+(?:\.\d+)?\s*(?:万|亿)?\s*\+?\s*条?评价"
        r"|评价\s*\d+(?:\.\d+)?\s*(?:万|亿)?\s*\+?)"
    )
    reviews = _find_metric(labels, reviews_pattern)
    if reviews.raw is None:
        empty_reviews = next(
            (label for label in labels if label in ("暂无评价", "暂无评论")),
            None,
        )
        if empty_reviews:
            reviews = QuantityMetric(0, empty_reviews, False)

    shop_name: Optional[str] = None
    for label in labels:
        match = re.search(r"(?:店铺|商家)[:：]\s*(.+)", label)
        if match and match.group(1).strip():
            shop_name = match.group(1).strip()
            break
        if len(label) > 2 and label.endswith(("旗舰店", "专卖店", "专营店")):
            shop_name = label
            break
    if shop_name is None:
        shop_name = _shop_name_by_enter_anchor(root)
    return GoodsSnapshot(title, shop_name, sales, reviews)


def _is_available(node: ET.Element) -> bool:
    label = _own_or_descendant_label(node)
    return (
        node.get("enabled", "true") == "true"
        and node.get("visible-to-user", "true") == "true"
        and not any(word in label for word in ("售罄", "缺货", "不可选"))
    )


def _dimension_key(name: str, used: set[str]) -> str:
    if "颜色" in name or "款式" in name:
        base = "color"
    elif "尺码" in name or "尺寸" in name:
        base = "size"
    elif "型号" in name:
        base = "model"
    else:
        digest = hashlib.sha1(name.encode("utf-8")).hexdigest()[:8]
        base = f"dimension_{digest}"
    key = base
    suffix = 2
    while key in used:
        key = f"{base}_{suffix}"
        suffix += 1
    return key


def _is_dimension_heading(label: str) -> bool:
    compact = label.replace(" ", "")
    # 新版页面会把可选数量写进标题，例如“颜色 (6)”或“颜色（6）”。
    # 数量不是规格名称的一部分，只在判断标题类型时去掉，最终展示仍保留原文。
    compact = re.sub(r"[（(]\d+[)）]$", "", compact)
    # “确认款式”是新版规格弹层的标题，不是一个可选择的“款式”维度。
    if compact in ("确认款式", "確認款式"):
        return False
    return compact in _DIMENSION_NAMES or compact.endswith(
        ("分类", "规格", "尺寸", "尺码", "颜色", "型号", "款式", "容量", "类型", "版本", "口味")
    )


def _find_non_scrollable_spec_panel(
    root: ET.Element,
    parents: Mapping[ET.Element, ET.Element],
) -> Optional[ET.Element]:
    """用多项强证据定位不暴露 scrollable 属性的自绘规格面板。"""

    headings: list[ET.Element] = []
    summaries: list[ET.Element] = []
    panel_cues: list[ET.Element] = []
    confirms: list[ET.Element] = []
    for node in root.iter("node"):
        label = _preferred_node_label(node).strip()
        compact = label.replace(" ", "")
        if not label or _parse_bounds(node.get("bounds", "")) is None:
            continue
        if _is_dimension_heading(label):
            headings.append(node)
        if compact.startswith(("已选", "已選", "请选择", "請選擇")):
            summaries.append(node)
        if compact in ("确认款式", "確認款式", "关闭", "關閉"):
            panel_cues.append(node)
        if compact in ("确定", "確定") and node.get("clickable") == "true":
            confirms.append(node)

    if not headings or not summaries or not panel_cues or not confirms:
        return None

    required = [*headings, summaries[0], panel_cues[0], confirms[0]]
    common = set([required[0], *_ancestors(required[0], parents)])
    for node in required[1:]:
        common.intersection_update([node, *_ancestors(node, parents)])

    all_bounds = [
        bounds
        for node in root.iter("node")
        if (bounds := _parse_bounds(node.get("bounds", ""))) is not None
    ]
    screen_area = 0
    if all_bounds:
        screen_area = max(item[2] for item in all_bounds) * max(
            item[3] for item in all_bounds
        )

    candidates: list[tuple[int, ET.Element]] = []
    for node in common:
        bounds = _parse_bounds(node.get("bounds", ""))
        if bounds is None or node.get("visible-to-user", "true") != "true":
            continue
        area = (bounds[2] - bounds[0]) * (bounds[3] - bounds[1])
        if screen_area and area >= screen_area * 0.95:
            continue
        candidates.append((area, node))
    if not candidates:
        return None
    return min(candidates, key=lambda item: item[0])[1]


def _is_spec_panel_open(xml_data: str | bytes) -> bool:
    """判断规格面板是否已经出现，不要求当前视口已露出规格选项。"""

    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}
    if _find_non_scrollable_spec_panel(root, parents) is not None:
        return True
    scrollable_regions = [
        node
        for node in root.iter("node")
        if node.get("scrollable") == "true"
        and _parse_bounds(node.get("bounds", "")) is not None
    ]
    has_scrollable_region = bool(scrollable_regions)
    if not has_scrollable_region:
        return False

    compact_labels = [label.replace(" ", "") for label in labels]
    has_selection_summary = any(
        label.startswith(("请选择", "請選擇", "已选", "已選"))
        for label in compact_labels
    )
    has_submit_hint = any(
        "提交订单" in label
        and any(word in label for word in ("选择", "颜色", "尺码", "规格"))
        for label in compact_labels
    )
    has_dimension_heading = any(
        _is_dimension_heading(_preferred_node_label(node).strip())
        and any(
            region in _ancestors(node, parents)
            for region in scrollable_regions
        )
        for node in root.iter("node")
    )
    return has_selection_summary or has_submit_hint or has_dimension_heading


def _top_level_clickable_options(
    container: ET.Element,
    parents: Mapping[ET.Element, ET.Element],
) -> list[ET.Element]:
    result: list[ET.Element] = []
    for node in container.iter("node"):
        label = _own_or_descendant_label(node).strip()
        bounds = _parse_bounds(node.get("bounds", ""))
        if node.get("clickable") != "true" or not label or bounds is None:
            continue
        if label.startswith("#") or label in (
            "打开大图",
            "增加数量",
            "减少数量",
            "关闭",
        ):
            continue
        parent = parents.get(node)
        if parent is not None and parent is not container:
            parent_label = _own_or_descendant_label(parent).strip()
            if parent.get("clickable") == "true" and parent_label == label:
                continue
        result.append(node)
    return result


def _price_from_nodes(
    nodes: Iterable[ET.Element],
    spec_top: int,
) -> tuple[Optional[int], Optional[str], Optional[int]]:
    candidates: list[tuple[int, int, str]] = []
    for node in nodes:
        label = _node_label(node)
        bounds = _parse_bounds(node.get("bounds", ""))
        match = _PRICE_PATTERN.search(label)
        if match is None or bounds is None or bounds[1] >= spec_top:
            continue
        try:
            cents = int(
                (Decimal(match.group(1)) * 100).quantize(
                    Decimal("1"),
                    rounding=ROUND_HALF_UP,
                )
            )
        except InvalidOperation:
            continue
        candidates.append((bounds[1], cents, match.group(0).replace(" ", "")))
    if not candidates:
        return None, None, None
    first_top = min(item[0] for item in candidates)
    first_row = [item for item in candidates if abs(item[0] - first_top) <= 12]
    _, cents, raw = min(first_row, key=lambda item: item[1])
    list_prices = [item[1] for item in first_row if item[1] > cents]
    return cents, raw, max(list_prices, default=None)


def parse_spec_panel(xml_data: str | bytes) -> SpecSnapshot:
    """解析规格面板当前视口中的规格值、已选摘要和当前价格。"""

    root = _parse_xml(xml_data)
    labels = _all_labels(root)
    _raise_special_page(labels)
    parents = {child: parent for parent in root.iter() for child in parent}
    scrollables = [
        node
        for node in root.iter("node")
        if node.get("scrollable") == "true" and _parse_bounds(node.get("bounds", ""))
    ]
    non_scrollable_outer = _find_non_scrollable_spec_panel(root, parents)
    if non_scrollable_outer is None and not _is_spec_panel_open(xml_data):
        raise PddCollectError(
            "PDD_DATA_SPEC_INCOMPLETE", "规格面板没有可识别的规格区域"
        )
    panel_scrollables = []
    if non_scrollable_outer is not None:
        panel_scrollables = [
            node
            for node in scrollables
            if node is non_scrollable_outer
            or non_scrollable_outer in _ancestors(node, parents)
        ]
    outer_candidates = panel_scrollables or (
        [non_scrollable_outer] if non_scrollable_outer is not None else scrollables
    )
    if not outer_candidates:
        raise PddCollectError("PDD_DATA_SPEC_INCOMPLETE", "规格面板没有可识别的规格区域")

    outer = max(
        outer_candidates,
        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
        preceding = [item for item in heading_nodes if item[0] <= bounds[1] + 30]
        if not preceding:
            # 普通推荐列表也是 RecyclerView。没有真实规格标题时，不能
            # 默认把其中的商品卡片解释成颜色。
            continue
        name = preceding[-1][1]
        options = _top_level_clickable_options(nested, parents)
        if options:
            groups.append((name, options))
            nested_options.update(options)

    for index, (top, name) in enumerate(heading_nodes):
        bottom = heading_nodes[index + 1][0] if index + 1 < len(heading_nodes) else outer_bounds[3]
        options = []
        for node in _top_level_clickable_options(outer, parents):
            bounds = _parse_bounds(node.get("bounds", ""))
            if node in nested_options or bounds is None:
                continue
            width = bounds[2] - bounds[0]
            if top < bounds[1] < bottom and width < (outer_bounds[2] - outer_bounds[0]) * 0.85:
                options.append(node)
        if options:
            groups.append((name, options))

    used_keys: set[str] = set()
    dimensions: list[SpecDimension] = []
    for name, option_nodes in groups:
        values: list[DimensionValue] = []
        seen: set[str] = set()
        for node in option_nodes:
            text = _preferred_or_descendant_label(node).strip()
            if text.endswith(("…", "...")):
                raise PddCollectError(
                    "PDD_DATA_SKU_NAME_TRUNCATED",
                    f"规格名称被截断，无法安全采集：{text}",
                )
            if not text or text in seen:
                continue
            seen.add(text)
            values.append(DimensionValue(text, _is_available(node)))
        if not values:
            continue
        key = _dimension_key(name, used_keys)
        used_keys.add(key)
        dimensions.append(SpecDimension(key, name, tuple(values)))

    selected_text = next((label for label in labels if label.startswith("已选")), None)
    price_boundary = heading_nodes[0][0] if heading_nodes else outer_bounds[1]
    price_cent, raw_price, list_price_cent = _price_from_nodes(
        root.iter("node"), price_boundary
    )
    return SpecSnapshot(
        tuple(dimensions), selected_text, price_cent, raw_price, list_price_cent
    )


def _ancestors(
    node: ET.Element,
    parents: Mapping[ET.Element, ET.Element],
) -> list[ET.Element]:
    result: list[ET.Element] = []
    current = parents.get(node)
    while current is not None:
        result.append(current)
        current = parents.get(current)
    return result


def _raise_special_page(labels: Sequence[str]) -> None:
    compact_labels = [label.replace(" ", "") for label in labels]
    if any(
        marker in label
        for label in compact_labels
        for marker in _OUT_OF_STOCK_PAGE_MARKERS
    ):
        raise PddCollectError(
            "PDD_GOODS_UNAVAILABLE",
            "商品已售罄，PDD 没有提供可采集的规格",
            {"page_reason": "out_of_stock"},
        )
    combined = " ".join(labels)
    if any(marker in combined for marker in _CAPTCHA_MARKERS):
        raise PddCollectError("PDD_PAGE_CAPTCHA", "PDD 出现安全验证，需要人工处理")
    if any(marker in combined for marker in _LOGIN_MARKERS):
        raise PddCollectError("PDD_PAGE_LOGIN_REQUIRED", "PDD 登录已失效，需要人工重新登录")


def _validate_goods_url(goods_url: str, task_goods_id: str) -> str:
    parsed = urlparse(goods_url)
    host = (parsed.hostname or "").lower()
    allowed_host = host == "yangkeduo.com" or host.endswith(".yangkeduo.com")
    allowed_host = allowed_host or host == "pinduoduo.com" or host.endswith(
        ".pinduoduo.com"
    )
    if parsed.scheme not in ("http", "https") or not allowed_host:
        raise PddCollectError(
            "PDD_DATA_GOODS_URL_INVALID",
            "商品链接不是受支持的 PDD 链接",
        )

    url_goods_id = parse_qs(parsed.query).get("goods_id", [""])[0].strip()
    if task_goods_id and url_goods_id and task_goods_id != url_goods_id:
        raise PddCollectError(
            "PDD_DATA_GOODS_ID_MISMATCH",
            "任务商品编号与商品链接中的 goods_id 不一致",
        )
    value = task_goods_id or url_goods_id
    if not value or not value.isdigit():
        raise PddCollectError("PDD_DATA_GOODS_ID_MISSING", "商品链接中没有 goods_id")
    return value


class PddCollectService:
    """打开商品页并采集结构化商品与 SKU 数据。"""

    def __init__(
        self,
        device_service: PddDeviceService,
        device_address: str,
        client_id: str,
        *,
        sleeper: Callable[[float], None] = time.sleep,
        monotonic: Callable[[], float] = time.monotonic,
        now: Callable[[], datetime] = lambda: datetime.now(timezone.utc),
        cancelled: Callable[[], bool] = lambda: False,
        page_timeout: float = 30.0,
        spec_panel_timeout: float = 10.0,
        overall_timeout: float = 600.0,
        max_goods_page_swipes: int = 6,
        max_spec_swipes: int = 12,
        max_sku_count: int = 200,
        color_poll_interval: float = 0.1,
        color_selection_timeout: float = 0.6,
        color_price_timeout: float = 1.2,
        horizontal_swipe_settle_interval: float = 0.1,
        artifact_directory: Optional[Path] = None,
    ) -> None:
        self._device_service = device_service
        self._device_address = device_address
        self._client_id = client_id
        self._sleep = sleeper
        self._monotonic = monotonic
        self._now = now
        self._cancelled = cancelled
        self._page_timeout = page_timeout
        self._spec_panel_timeout = spec_panel_timeout
        self._overall_timeout = overall_timeout
        self._overall_deadline: Optional[float] = None
        self._max_goods_page_swipes = max_goods_page_swipes
        self._max_spec_swipes = max_spec_swipes
        self._max_sku_count = max_sku_count
        timing_values = {
            "颜色轮询间隔": color_poll_interval,
            "颜色选中超时": color_selection_timeout,
            "颜色价格超时": color_price_timeout,
            "水平滑动稳定间隔": horizontal_swipe_settle_interval,
        }
        for name, value in timing_values.items():
            if value <= 0:
                raise ValueError(f"{name}必须大于 0 秒")
        self._color_poll_interval = color_poll_interval
        self._color_selection_timeout = color_selection_timeout
        self._color_price_timeout = color_price_timeout
        self._horizontal_swipe_settle_interval = (
            horizontal_swipe_settle_interval
        )
        self._artifact_directory = artifact_directory
        self._last_goods_xml: Optional[str] = None
        self._goods_screens_checked = 0
        self._artifacts: list[Mapping[str, Any]] = []

    def collect(self, task: Any) -> CollectResult:
        """执行采集；``task`` 至少提供 ``goods_url`` 和可选 ``goods_id``。"""

        goods_url = str(getattr(task, "goods_url", "") or "").strip()
        if not goods_url:
            raise PddCollectError("PDD_DATA_GOODS_URL_MISSING", "采集任务缺少商品链接")
        goods_id = str(getattr(task, "goods_id", "") or "").strip()
        goods_id = _validate_goods_url(goods_url, goods_id)
        self._overall_deadline = self._monotonic() + self._overall_timeout
        self._check_cancelled()

        try:
            session = self._device_service.connect(self._device_address)
            with session as device:
                self._open_goods(
                    device, goods_url, session.initial_app_state
                )
                goods = self._collect_goods_details(device)
                if not goods.title:
                    raise PddCollectError("PDD_DATA_TITLE_MISSING", "商品页没有可识别的标题")
                if not goods.sales.raw:
                    raise PddCollectError("PDD_DATA_SALES_MISSING", "商品页没有采集到已拼数量")
                if (
                    (not goods.shop_name or not goods.reviews.raw)
                    and self._last_goods_xml
                ):
                    artifact = self._save_xml(
                        "goods-metadata-incomplete", self._last_goods_xml
                    )
                    if artifact:
                        self._artifacts.append(artifact)

                home_xml = self._dump_hierarchy(device)
                coordinate = get_size_panel_coord(home_xml)
                if coordinate is None:
                    raise PddCollectError(
                        "PDD_PAGE_SPEC_ENTRY_MISSING",
                        "商品页没有找到可靠的规格入口",
                    )
                device.click(*coordinate)
                self._wait_spec_panel(device)

                self._clear_default_size_selection(device)
                color_dimension, color_samples = self._collect_color_prices(device)
                size_dimension = self._collect_size_dimension(device)
                dimensions = (color_dimension,)
                if size_dimension is not None:
                    dimensions += (size_dimension,)
                if not dimensions or any(not item.values for item in dimensions):
                    raise PddCollectError(
                        "PDD_DATA_SPEC_INCOMPLETE",
                        "规格面板没有采集到完整的规格维度",
                    )
                skus = self._build_color_price_skus(
                    dimensions, color_samples
                )
                if not skus:
                    raise PddCollectError(
                        "PDD_DATA_SPEC_INCOMPLETE",
                        "规格面板没有生成可提交的规格组合",
                    )
        except PddCollectError as exc:
            if self._last_goods_xml:
                artifact = self._save_xml("collect-failed", self._last_goods_xml)
                if artifact:
                    diagnostics = dict(exc.diagnostics)
                    diagnostics.setdefault("artifacts", []).append(artifact)
                    diagnostics.setdefault(
                        "goods_screens_checked", self._goods_screens_checked
                    )
                    exc.diagnostics = diagnostics
            raise
        except PddDeviceError as exc:
            raise PddCollectError(exc.code, exc.message) from exc
        except Exception as exc:
            if _is_device_disconnect(exc):
                raise PddCollectError(
                    "DEVICE_DISCONNECTED",
                    f"Android 设备在采集过程中断开：{exc}",
                ) from exc
            raise PddCollectError("PDD_PAGE_UNKNOWN", f"PDD 采集过程中发生未知错误：{exc}") from exc

        return CollectResult(
            goods_id=goods_id,
            goods_url=goods_url,
            title=goods.title,
            shop_name=goods.shop_name,
            sales=goods.sales,
            reviews=goods.reviews,
            dimensions=dimensions,
            skus=skus,
            captured_at=self._now().astimezone(timezone.utc).isoformat().replace("+00:00", "Z"),
            client_id=self._client_id,
            device_address=self._device_address,
            artifacts=tuple(self._artifacts),
        )

    def _check_cancelled(self) -> None:
        if self._cancelled():
            raise PddCollectError("PDD_CANCELLED", "采集任务已安全取消")
        if (
            self._overall_deadline is not None
            and self._monotonic() >= self._overall_deadline
        ):
            raise PddCollectError("PDD_PAGE_OVERALL_TIMEOUT", "PDD 采集超过 10 分钟")

    def _open_goods(
        self,
        device: Any,
        goods_url: str,
        initial_app_state: Optional[Mapping[str, Any]] = None,
    ) -> None:
        trace = current_performance_trace()
        try:
            current = (
                dict(initial_app_state)
                if initial_app_state is not None
                else device.app_current()
            )
            package_hint = (
                PDD_PACKAGE_NAME
                if current.get("package") == PDD_PACKAGE_NAME
                else ""
            )
            before_open = self._read_page_observation(device, current)
            if current.get("package") != PDD_PACKAGE_NAME:
                stage = trace.stage("pdd_start_or_wait") if trace else nullcontext()
                with stage:
                    device.app_start(PDD_PACKAGE_NAME)
                    if not device.app_wait(PDD_PACKAGE_NAME, timeout=10):
                        raise PddCollectError(
                            "DEVICE_APP_START_FAILED", "PDD 应用启动失败"
                        )
            elif trace is not None:
                trace.record("pdd_start_or_wait", 0, "already_foreground")
            stage = trace.stage("open_url") if trace else nullcontext()
            with stage:
                device.open_url(goods_url)
        except PddCollectError:
            raise
        except Exception as exc:
            if _is_device_disconnect(exc):
                raise PddCollectError(
                    "DEVICE_DISCONNECTED",
                    f"Android 设备在打开商品页时断开：{exc}",
                ) from exc
            raise PddCollectError("DEVICE_APP_START_FAILED", f"无法打开 PDD 商品链接：{exc}") from exc

        ready_stage = trace.stage("goods_page_ready") if trace else nullcontext()
        with ready_stage:
            opened_at = self._monotonic()
            deadline = opened_at + self._page_timeout
            tracker = GoodsOpenTracker(before_open, opened_at)
            first_dump = True
            last_kind = "unknown"
            last_recorded = ""
            while self._monotonic() < deadline:
                self._check_cancelled()
                if first_dump and trace is not None:
                    with trace.stage("first_dump_hierarchy"):
                        xml_data = self._dump_hierarchy(device)
                    first_dump = False
                else:
                    xml_data = self._dump_hierarchy(device)
                root = _parse_xml(xml_data)
                # 部分设备的 app_current() 一次会阻塞十秒以上，并错误报告
                # 设置页。循环中直接使用最新控件树，PDD 节点包名足以分类。
                observation = classify_pdd_page(root, package_hint)
                last_kind = observation.kind
                if trace is not None and last_kind != last_recorded:
                    trace.record(
                        "pdd_page_transition",
                        0,
                        f"attempt_{tracker.attempt}_{last_kind}",
                    )
                    last_recorded = last_kind
                self._raise_classified_special_page(last_kind)
                decision = tracker.observe(observation, self._monotonic())
                if decision.action == ACTION_READY:
                    if trace is not None:
                        trace.checkpoint("end_to_end_total")
                    return
                if decision.action == ACTION_REOPEN:
                    with (
                        trace.stage("open_url_retry")
                        if trace is not None
                        else nullcontext()
                    ):
                        device.open_url(goods_url)
                    tracker.reopened(self._monotonic())
                    last_recorded = ""
                    continue
                if decision.action == ACTION_UNAVAILABLE:
                    raise PddCollectError(
                        "PDD_GOODS_UNAVAILABLE",
                        "商品链接已失效，PDD 无法打开商品详情页并返回了首页",
                        {"page_kind": PAGE_HOME, "open_attempts": 2},
                    )
                if decision.action == ACTION_NETWORK_ERROR:
                    raise PddCollectError(
                        "PDD_PAGE_NETWORK_ERROR",
                        "PDD 商品页网络或服务异常，请稍后重试",
                        {"page_kind": PAGE_NETWORK_ERROR},
                    )
                self._sleep(0.25)
            if tracker.stale_goods_seen:
                raise PddCollectError(
                    "PDD_GOODS_IDENTITY_UNCONFIRMED",
                    "打开商品链接后仍停留在原商品页，无法确认本次目标商品",
                    {"page_kind": PAGE_GOODS, "open_attempts": tracker.attempt},
                )
            raise PddCollectError(
                "PDD_PAGE_TIMEOUT",
                f"等待 PDD 商品详情页加载超时，最后页面为 {last_kind}",
                {"page_kind": last_kind, "open_attempts": tracker.attempt},
            )

    def _read_page_observation(
        self, device: Any, current: Mapping[str, Any]
    ) -> Optional[PddPageObservation]:
        """读取深链打开前页面；读取失败不妨碍后续打开目标链接。"""

        try:
            root = _parse_xml(self._dump_hierarchy(device))
            return classify_pdd_page(
                root, str(current.get("package") or "")
            )
        except (PddCollectError, RuntimeError, OSError):
            return None

    @staticmethod
    def _raise_classified_special_page(kind: str) -> None:
        if kind == PAGE_CAPTCHA:
            raise PddCollectError(
                "PDD_PAGE_CAPTCHA", "PDD 出现安全验证，需要人工处理"
            )
        if kind == PAGE_LOGIN_REQUIRED:
            raise PddCollectError(
                "PDD_PAGE_LOGIN_REQUIRED", "PDD 登录已失效，需要人工重新登录"
            )
        if kind in {PAGE_RISK_CONTROL, PAGE_PAYMENT}:
            raise PddCollectError(
                "PDD_PAGE_UNKNOWN", "PDD 出现风控或支付页面，已停止采集"
            )

    def _collect_goods_details(self, device: Any) -> GoodsSnapshot:
        """保留首屏结果，并向下浏览补采评价数量和店铺名。"""

        self._check_cancelled()
        xml_data = self._dump_hierarchy(device)
        goods = parse_goods_page(xml_data)
        if goods.shop_name and goods.reviews.raw is not None:
            return goods

        unchanged_reads = 0
        for _ in range(self._max_goods_page_swipes):
            self._check_cancelled()
            self._swipe_goods_page(device, xml_data)
            self._sleep(0.8)
            self._check_cancelled()

            next_xml = self._dump_hierarchy(device)
            snapshot = parse_goods_page(next_xml)
            goods = GoodsSnapshot(
                title=goods.title,
                shop_name=goods.shop_name or snapshot.shop_name,
                sales=goods.sales,
                reviews=(
                    goods.reviews
                    if goods.reviews.raw is not None
                    else snapshot.reviews
                ),
            )
            if goods.shop_name and goods.reviews.raw is not None:
                break

            unchanged_reads = unchanged_reads + 1 if next_xml == xml_data else 0
            xml_data = next_xml
            if unchanged_reads >= 2:
                break
        return goods

    @staticmethod
    def _swipe_goods_page(device: Any, xml_data: str | bytes) -> None:
        """在商品详情区域慢速向上滑动约 30%，浏览页面下方内容。"""

        root = _parse_xml(xml_data)
        scrollable_bounds = [
            bounds
            for node in root.iter("node")
            if node.get("scrollable") == "true"
            and (bounds := _parse_bounds(node.get("bounds", ""))) is not None
        ]
        all_bounds = [
            bounds
            for node in root.iter("node")
            if (bounds := _parse_bounds(node.get("bounds", ""))) is not None
        ]
        candidates = scrollable_bounds or all_bounds
        if not candidates:
            raise PddCollectError(
                "PDD_DATA_PAGE_BOUNDS_MISSING",
                "商品详情页没有可用于滚动的有效区域",
            )
        left, top, right, bottom = max(
            candidates,
            key=lambda bounds: (bounds[2] - bounds[0]) * (bounds[3] - bounds[1]),
        )
        x = (left + right) // 2
        height = bottom - top
        device.swipe(
            x,
            top + int(height * 0.65),
            x,
            top + int(height * 0.35),
            duration=0.6,
        )

    def _wait_spec_panel(self, device: Any) -> SpecSnapshot:
        """等待点击后的规格面板真正出现，不能只依赖固定延时。"""

        deadline = self._monotonic() + self._spec_panel_timeout
        while self._monotonic() < deadline:
            self._check_cancelled()
            xml_data = self._dump_hierarchy(device)
            try:
                snapshot = parse_spec_panel(xml_data)
            except PddCollectError as exc:
                if exc.code != "PDD_DATA_SPEC_INCOMPLETE":
                    raise
            else:
                if _is_spec_panel_open(xml_data):
                    return snapshot
            self._sleep(0.25)
        raise PddCollectError(
            "PDD_PAGE_SPEC_PANEL_TIMEOUT",
            "点击规格入口后，等待规格面板加载超时",
        )

    def _save_xml(self, label: str, xml_data: str) -> Optional[Mapping[str, Any]]:
        """保存本地诊断 XML；测试未提供目录时不写文件。"""

        if self._artifact_directory is None:
            return None
        try:
            digest = hashlib.sha256(xml_data.encode("utf-8")).hexdigest()
            directory = self._artifact_directory / self._client_id
            directory.mkdir(parents=True, exist_ok=True)
            path = directory / f"{label}-{digest[:12]}.xml"
            if not path.exists():
                path.write_text(xml_data, encoding="utf-8")
            return {
                "kind": "accessibility_xml",
                "path": str(path.resolve()),
                "sha256": digest,
                "screens_checked": self._goods_screens_checked,
            }
        except OSError:
            return None

    def _dump_hierarchy(self, device: Any) -> str:
        """读取并记住最新控件树，失败诊断必须指向最后一次操作。"""

        raw_xml = device.dump_hierarchy()
        xml_data = (
            raw_xml.decode("utf-8", errors="replace")
            if isinstance(raw_xml, bytes)
            else str(raw_xml)
        )
        self._last_goods_xml = xml_data
        self._goods_screens_checked += 1
        return xml_data

    def _collect_color_prices(
        self, device: Any
    ) -> tuple[SpecDimension, Mapping[str, ColorPriceSample]]:
        """按行蛇形遍历颜色，并在每次点击后立即读取颜色级价格。"""

        self._move_color_list_to_left(device)
        first_xml = self._dump_hierarchy(device)
        first_rows = self._visible_color_rows(first_xml)
        if not first_rows:
            raise PddCollectError(
                "PDD_DATA_SPEC_INCOMPLETE", "规格面板没有可识别的颜色分类"
            )

        row_count = len(first_rows)
        visited: set[str] = set()
        ordered_colors: list[DimensionValue] = []
        samples: dict[str, ColorPriceSample] = {}

        for row_index in range(row_count):
            move_right = row_index % 2 == 0
            current_row_colors: list[DimensionValue] = []
            stable_edge_reads = 0
            previous_signature: Optional[tuple[tuple[str, Bounds], ...]] = None
            pending_xml: Optional[str] = None

            for swipe_count in range(self._max_spec_swipes + 1):
                self._check_cancelled()
                # 每次只处理一个节点；点击可能让列表自动移动，下一项必须
                # 从最新 XML 重新计算，不能继续使用点击前的旧坐标。
                while True:
                    xml_data = pending_xml or self._dump_hierarchy(device)
                    pending_xml = None
                    rows = self._visible_color_rows(xml_data)
                    if row_index >= len(rows):
                        raise PddCollectError(
                            "PDD_DATA_SPEC_INCOMPLETE",
                            f"颜色第 {row_index + 1} 行在滑动后消失",
                        )
                    current_row = sorted(
                        rows[row_index],
                        key=lambda item: item.bounds[0],
                        reverse=not move_right,
                    )
                    visible = next(
                        (item for item in current_row if item.text not in visited),
                        None,
                    )
                    if visible is None:
                        break
                    visited.add(visible.text)
                    color = DimensionValue(visible.text, visible.available)
                    current_row_colors.append(color)
                    if visible.available:
                        samples[visible.text] = self._click_and_sample_color(
                            device, visible.text
                        )
                    else:
                        samples[visible.text] = ColorPriceSample(
                            None, None, None
                        )

                signature = tuple((item.text, item.bounds) for item in current_row)
                if signature == previous_signature:
                    stable_edge_reads += 1
                else:
                    stable_edge_reads = 0
                previous_signature = signature
                if stable_edge_reads >= 2 or swipe_count >= self._max_spec_swipes:
                    break

                root = _parse_xml(xml_data)
                region = self._horizontal_region(root)
                if region is None:
                    break
                self._swipe_region(
                    device,
                    region,
                    horizontal=True,
                    reverse=not move_right,
                )
                pending_xml = self._wait_for_horizontal_change(
                    device, signature
                )

            # 操作采用蛇形以减少无效滑动；输出仍恢复成页面自然的
            # “每行从左到右”顺序，方便 Admin 下拉框稳定展示。
            ordered_colors.extend(
                current_row_colors if move_right else reversed(current_row_colors)
            )

        if not ordered_colors:
            raise PddCollectError(
                "PDD_DATA_SPEC_INCOMPLETE", "规格面板没有采集到颜色"
            )
        return (
            SpecDimension("color", "颜色分类", tuple(ordered_colors)),
            samples,
        )

    def _move_color_list_to_left(self, device: Any) -> None:
        """把颜色列表归位到左端；连续两次视口不变才认为到边。"""

        previous_signature: Optional[tuple[tuple[str, Bounds], ...]] = None
        stable_edge_reads = 0
        xml_data = self._dump_hierarchy(device)
        for _ in range(self._max_spec_swipes):
            self._check_cancelled()
            signature = self._color_view_signature(xml_data)
            if signature == previous_signature:
                stable_edge_reads += 1
            else:
                stable_edge_reads = 0
            previous_signature = signature
            if stable_edge_reads >= 2:
                return

            root = _parse_xml(xml_data)
            region = self._horizontal_region(root)
            if region is None:
                return
            self._swipe_region(
                device, region, horizontal=True, reverse=True
            )
            xml_data = self._wait_for_horizontal_change(device, signature)

    def _click_and_sample_color(
        self, device: Any, target: str
    ) -> ColorPriceSample:
        """点击最新树中的颜色，等待选择和价格连续两次稳定。"""

        xml_data = self._dump_hierarchy(device)
        root = _parse_xml(xml_data)
        latest_visible = next(
            (
                item
                for row in self._visible_color_rows(xml_data)
                for item in row
                if item.text == target
            ),
            None,
        )
        state_node = self._find_option_node(root, target) if latest_visible else None
        node = (
            self._find_safe_option_click_node(state_node, target)
            if state_node is not None
            else None
        )
        if node is None or not _is_available(node):
            return ColorPriceSample(None, None, None)
        bounds = _parse_bounds(node.get("bounds", ""))
        assert bounds is not None
        device.click(
            (bounds[0] + bounds[2]) // 2,
            (bounds[1] + bounds[3]) // 2,
        )

        selection_deadline = self._monotonic() + self._color_selection_timeout
        selection_sleep_limit = math.ceil(
            self._color_selection_timeout / self._color_poll_interval
        )
        selection_sleeps = 0
        price_deadline: Optional[float] = None
        price_sleep_limit = math.ceil(
            self._color_price_timeout / self._color_poll_interval
        )
        price_sleeps = 0
        previous_price: Optional[tuple[int, Optional[str], Optional[int]]] = None
        stable_price_reads = 0
        while True:
            self._check_cancelled()
            latest_xml = self._dump_hierarchy(device)
            if self._is_big_image_viewer(latest_xml):
                self._recover_spec_panel_from_big_image(device)
                return ColorPriceSample(None, None, None)
            latest_root = _parse_xml(latest_xml)
            snapshot = parse_spec_panel(latest_xml)
            latest_node = self._find_option_node(latest_root, target)
            selection_exposed = self._color_selection_state_exposed(latest_xml)
            selected = bool(
                latest_node is not None and self._node_is_selected(latest_node)
            ) or bool(snapshot.selected_text and target in snapshot.selected_text)
            # 老版本页面不暴露 selected/checked，也没有“已选”摘要时，
            # 只能以完整可见可点击节点未报错的点击作为降级证据。
            selection_confirmed = selected or not selection_exposed

            now = self._monotonic()
            if not selection_confirmed:
                stable_price_reads = 0
                previous_price = None
                selection_timed_out = (
                    now >= selection_deadline
                    or selection_sleeps >= selection_sleep_limit
                )
                if selection_timed_out:
                    return ColorPriceSample(None, None, None)
            else:
                if price_deadline is None:
                    price_deadline = now + self._color_price_timeout
                if snapshot.price_cent is None:
                    stable_price_reads = 0
                    previous_price = None
                else:
                    current_price = (
                        snapshot.price_cent,
                        snapshot.raw_price,
                        snapshot.list_price_cent,
                    )
                    if current_price == previous_price:
                        stable_price_reads += 1
                    else:
                        stable_price_reads = 1
                        previous_price = current_price
                    if stable_price_reads >= 2:
                        return ColorPriceSample(
                            snapshot.price_cent,
                            snapshot.raw_price,
                            snapshot.list_price_cent,
                        )
                if now >= price_deadline or price_sleeps >= price_sleep_limit:
                    return ColorPriceSample(None, None, None)

            self._sleep(self._color_poll_interval)
            if selection_confirmed:
                price_sleeps += 1
            else:
                selection_sleeps += 1

    def _clear_default_size_selection(self, device: Any) -> None:
        """逐屏寻找默认尺码；只在唯一选中项明确时点击一次取消。"""

        previous_signature: Optional[
            tuple[tuple[str, tuple[str, ...]], ...]
        ] = None
        stable_edge_reads = 0
        for swipe_count in range(self._max_spec_swipes + 1):
            self._check_cancelled()
            xml_data = self._dump_hierarchy(device)
            root = _parse_xml(xml_data)
            snapshot = parse_spec_panel(xml_data)
            size_dimension = next(
                (item for item in snapshot.dimensions if item.key == "size"),
                None,
            )
            if size_dimension is not None:
                selected_values = []
                for value in size_dimension.values:
                    state_node = self._find_option_node(root, value.text)
                    if state_node is not None and self._node_is_selected(state_node):
                        selected_values.append((value.text, state_node))

                if len(selected_values) == 1:
                    target, state_node = selected_values[0]
                    click_node = self._find_safe_option_click_node(state_node, target)
                    if click_node is not None:
                        bounds = _parse_bounds(click_node.get("bounds", ""))
                        assert bounds is not None
                        device.click(
                            (bounds[0] + bounds[2]) // 2,
                            (bounds[1] + bounds[3]) // 2,
                        )
                        self._sleep(self._color_poll_interval)
                        latest_root = _parse_xml(self._dump_hierarchy(device))
                        latest_node = self._find_option_node(latest_root, target)
                        if latest_node is None or not self._node_is_selected(
                            latest_node
                        ):
                            break
                    # 页面不支持取消时也不能重复点击，否则可能重新选中或误操作。
                    break
                if not selected_values:
                    break

            signature = tuple(
                (item.key, tuple(value.text for value in item.values))
                for item in snapshot.dimensions
            )
            stable_edge_reads = (
                stable_edge_reads + 1 if signature == previous_signature else 0
            )
            previous_signature = signature
            if stable_edge_reads >= 2 or swipe_count >= self._max_spec_swipes:
                break
            region = self._vertical_region(root)
            if region is None:
                break
            self._swipe_region(device, region, horizontal=False, reverse=False)
            self._sleep(0.35)

        # 查找尺码时页面可能已经滚到底部，逐色采价前必须回到顶部。
        self._move_spec_panel_to_top(device)

    def _move_spec_panel_to_top(self, device: Any) -> None:
        """把规格面板纵向恢复到顶部；连续两次不变化即停止。"""

        previous_signature: Optional[tuple[tuple[str, Bounds], ...]] = None
        stable_edge_reads = 0
        for _ in range(self._max_spec_swipes):
            self._check_cancelled()
            xml_data = self._dump_hierarchy(device)
            root = _parse_xml(xml_data)
            signature = tuple(
                (
                    _preferred_node_label(node),
                    bounds,
                )
                for node in root.iter("node")
                if (bounds := _parse_bounds(node.get("bounds", ""))) is not None
                and (
                    _is_dimension_heading(_preferred_node_label(node))
                    or node.get("clickable") == "true"
                )
            )
            stable_edge_reads = (
                stable_edge_reads + 1 if signature == previous_signature else 0
            )
            previous_signature = signature
            if stable_edge_reads >= 2:
                return
            region = self._vertical_region(root)
            if region is None:
                return
            self._swipe_region(device, region, horizontal=False, reverse=True)
            self._sleep(0.35)

    @staticmethod
    def _is_big_image_viewer(xml_data: str | bytes) -> bool:
        """用 ViewPager 和页码共同识别误入的商品大图页。"""

        root = _parse_xml(xml_data)
        has_pager = any(
            (node.get("class") or "").endswith("ViewPager")
            for node in root.iter("node")
        )
        has_page_number = any(
            re.fullmatch(r"\d+\s*/\s*\d+", _preferred_node_label(node))
            for node in root.iter("node")
        )
        has_dimension = any(
            _is_dimension_heading(_preferred_node_label(node))
            for node in root.iter("node")
        )
        return has_pager and has_page_number and not has_dimension

    def _recover_spec_panel_from_big_image(self, device: Any) -> None:
        """误入大图后只返回一次，并确认规格面板已经恢复。"""

        device.press("back")
        for _ in range(3):
            self._check_cancelled()
            self._sleep(self._color_poll_interval)
            xml_data = self._dump_hierarchy(device)
            try:
                if _is_spec_panel_open(xml_data):
                    return
            except PddCollectError:
                break
        raise PddCollectError(
            "PDD_PAGE_SPEC_PANEL_LOST",
            "点击颜色后进入大图，返回一次仍未恢复规格面板",
        )

    def _color_view_signature(
        self, xml_data: str | bytes
    ) -> tuple[tuple[str, Bounds], ...]:
        """返回颜色视口签名，用于判断水平滑动是否已经更新页面。"""

        return tuple(
            (item.text, item.bounds)
            for row in self._visible_color_rows(xml_data)
            for item in row
        )

    def _wait_for_horizontal_change(
        self,
        device: Any,
        previous_signature: tuple[tuple[str, Bounds], ...],
    ) -> str:
        """水平滑动后短暂等待；首次未变化时只补等一次。"""

        latest_xml = ""
        for _ in range(2):
            self._check_cancelled()
            self._sleep(self._horizontal_swipe_settle_interval)
            latest_xml = self._dump_hierarchy(device)
            if self._color_view_signature(latest_xml) != previous_signature:
                break
        return latest_xml

    def _collect_size_dimension(self, device: Any) -> Optional[SpecDimension]:
        """颜色采价完成后只滚动并收集尺码文字，不点击尺码。"""

        sizes: dict[str, bool] = {}
        previous_signature: Optional[
            tuple[tuple[str, tuple[str, ...]], ...]
        ] = None
        stable_edge_reads = 0

        for swipe_count in range(self._max_spec_swipes + 1):
            self._check_cancelled()
            xml_data = self._dump_hierarchy(device)
            snapshot = parse_spec_panel(xml_data)
            for dimension in snapshot.dimensions:
                if dimension.key not in ("color", "size"):
                    raise PddCollectError(
                        "PDD_DATA_SPEC_UNSUPPORTED",
                        f"发现尚未支持的规格维度：{dimension.name}",
                    )
                if dimension.key == "size":
                    for value in dimension.values:
                        sizes[value.text] = sizes.get(value.text, False) or value.available

            signature = tuple(
                (item.key, tuple(value.text for value in item.values))
                for item in snapshot.dimensions
            )
            if signature == previous_signature:
                stable_edge_reads += 1
            else:
                stable_edge_reads = 0
            previous_signature = signature
            if stable_edge_reads >= 2 or swipe_count >= self._max_spec_swipes:
                break

            root = _parse_xml(xml_data)
            region = self._vertical_region(root)
            if region is None:
                break
            self._swipe_region(
                device, region, horizontal=False, reverse=False
            )
            self._sleep(0.35)

        if not sizes:
            return None
        return SpecDimension(
            "size",
            "尺码",
            tuple(DimensionValue(text, available) for text, available in sizes.items()),
        )

    def _build_color_price_skus(
        self,
        dimensions: tuple[SpecDimension, ...],
        samples: Mapping[str, ColorPriceSample],
    ) -> tuple[SkuResult, ...]:
        """按颜色价格生成展示组合；采价证据只记录真实选中的颜色。"""

        colors = dimensions[0].values
        sizes = dimensions[1].values if len(dimensions) > 1 else (None,)
        combination_count = len(colors) * len(sizes)
        if combination_count > self._max_sku_count:
            raise PddCollectError(
                "PDD_DATA_TOO_MANY_SKUS",
                f"规格组合共 {combination_count} 个，超过安全上限 {self._max_sku_count}",
            )

        results: list[SkuResult] = []
        for color in colors:
            sample = samples[color.text]
            for size in sizes:
                options = {"color": color.text}
                if size is not None:
                    options["size"] = size.text
                # 是否可用来自规格控件状态，不能用“是否采到价格”代替。
                available = color.available and (size is None or size.available)
                results.append(
                    SkuResult(
                        options,
                        sample.price_cent,
                        available,
                        sample.raw_price,
                        {"color": color.text}
                        if sample.price_cent is not None
                        else {},
                        sample.list_price_cent,
                    )
                )
        return tuple(results)

    def _visible_color_rows(
        self, xml_data: str | bytes
    ) -> list[list[VisibleSpecOption]]:
        """把当前视口中完整显示的颜色节点按中心 y 聚类成行。"""

        root = _parse_xml(xml_data)
        snapshot = parse_spec_panel(xml_data)
        color_dimension = next(
            (item for item in snapshot.dimensions if item.key == "color"),
            None,
        )
        if color_dimension is None:
            return []

        candidates: list[VisibleSpecOption] = []
        for value in color_dimension.values:
            node = self._find_option_node(root, value.text)
            if node is None:
                continue
            bounds = _parse_bounds(node.get("bounds", ""))
            if bounds is None:
                continue
            candidates.append(
                VisibleSpecOption(
                    value.text,
                    value.available,
                    bounds,
                )
            )
        if not candidates:
            return []

        parents = {child: parent for parent in root.iter() for child in parent}
        non_scrollable_panel = _find_non_scrollable_spec_panel(root, parents)
        if non_scrollable_panel is not None:
            # 新版面板会按文字长度设置按钮宽度。它没有横向滚动视口，
            # 短按钮不代表被屏幕边缘截断，因此保留所有安全范围内的节点。
            complete = candidates
        else:
            widest = max(item.bounds[2] - item.bounds[0] for item in candidates)
            minimum_width = max(48, int(widest * 0.6))
            complete = [
                item
                for item in candidates
                if item.bounds[2] - item.bounds[0] >= minimum_width
            ]
        complete.sort(key=lambda item: ((item.bounds[1] + item.bounds[3]) // 2, item.bounds[0]))

        rows: list[list[VisibleSpecOption]] = []
        row_centers: list[int] = []
        for item in complete:
            center_y = (item.bounds[1] + item.bounds[3]) // 2
            matching_index = next(
                (
                    index
                    for index, row_center in enumerate(row_centers)
                    if abs(center_y - row_center) <= 80
                ),
                None,
            )
            if matching_index is None:
                rows.append([item])
                row_centers.append(center_y)
            else:
                rows[matching_index].append(item)
        for row in rows:
            row.sort(key=lambda item: item.bounds[0])
        return rows

    @staticmethod
    def _node_selection_state_exposed(node: ET.Element) -> bool:
        return any(
            item.get("selected") is not None or item.get("checked") is not None
            for item in node.iter()
        )

    @staticmethod
    def _node_is_selected(node: ET.Element) -> bool:
        return any(
            item.get("selected") == "true" or item.get("checked") == "true"
            for item in node.iter()
        )

    def _color_selection_state_exposed(self, xml_data: str | bytes) -> bool:
        root = _parse_xml(xml_data)
        snapshot = parse_spec_panel(xml_data)
        if snapshot.selected_text:
            return True
        color_dimension = next(
            (item for item in snapshot.dimensions if item.key == "color"),
            None,
        )
        if color_dimension is None:
            return False
        return any(
            node is not None and self._node_selection_state_exposed(node)
            for value in color_dimension.values
            if (node := self._find_option_node(root, value.text)) is not None
        )

    @staticmethod
    def _find_option_node(root: ET.Element, target: str) -> Optional[ET.Element]:
        parents = {child: parent for parent in root.iter() for child in parent}
        scrollables = [node for node in root.iter("node") if node.get("scrollable") == "true"]
        non_scrollable_panel = _find_non_scrollable_spec_panel(root, parents)
        right_edges = [
            bounds[2]
            for node in root.iter("node")
            if (bounds := _parse_bounds(node.get("bounds", ""))) is not None
        ]
        screen_right = max(right_edges, default=0)
        for node in root.iter("node"):
            if _preferred_or_descendant_label(node).strip() != target:
                continue
            bounds = _parse_bounds(node.get("bounds", ""))
            if node.get("clickable") != "true" or bounds is None:
                continue
            center_x = (bounds[0] + bounds[2]) // 2
            if bounds[2] - bounds[0] < 48:
                continue
            if not 24 <= center_x <= screen_right - 24:
                continue
            ancestors = _ancestors(node, parents)
            in_scrollable = any(container in ancestors for container in scrollables)
            in_non_scrollable_panel = non_scrollable_panel is not None and (
                node is non_scrollable_panel or non_scrollable_panel in ancestors
            )
            if in_scrollable or in_non_scrollable_panel:
                return node
        return None

    @staticmethod
    def _find_safe_option_click_node(
        state_node: ET.Element,
        target: str,
    ) -> Optional[ET.Element]:
        """返回规格文字的安全点击节点，图片卡片绝不使用整卡中心。"""

        candidates: list[tuple[int, int, ET.Element]] = []
        for node in state_node.iter():
            if _preferred_node_label(node).strip() != target:
                continue
            bounds = _parse_bounds(node.get("bounds", ""))
            class_name = node.get("class") or ""
            if bounds is None or class_name.endswith("ImageView"):
                continue
            if node.get("visible-to-user", "true") != "true":
                continue
            area = (bounds[2] - bounds[0]) * (bounds[3] - bounds[1])
            text_rank = 0 if class_name.endswith("TextView") else 1
            candidates.append((text_rank, area, node))
        if candidates:
            return min(candidates, key=lambda item: (item[0], item[1]))[2]

        has_image_area = any(
            (node.get("class") or "").endswith("ImageView")
            or _preferred_node_label(node).strip() in ("打开大图", "查看大图")
            for node in state_node.iter()
        )
        return None if has_image_area else state_node

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