"""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_ORDER_CONFIRMATION,
    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 _SecondDimensionContext:
    """标题滑出屏幕后，继续识别第二规格所需的稳定特征。"""

    name: str
    horizontal: bool
    container_class: str
    container_resource_id: str
    container_bounds: Bounds
    option_structures: tuple[tuple[str, str, tuple[tuple[str, str], ...]], ...]


@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 any(word in name for word in ("颜色", "花色", "款式")):
        base = "color"
    elif "尺码" in name or "尺寸" in name:
        base = "size"
    elif "套餐" in name:
        # 本项目的稳定结构只有主规格 color 和第二规格 size。商品只有
        # “套餐”一组时，它必须作为主规格采价；前面已有款式/颜色时，
        # 套餐才是第二规格。不能无条件映射成 size，否则单规格商品无法采集。
        base = "size" if "color" in used else "color"
    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(" ", "")
    if compact.startswith(("请选择", "請選擇", "已选", "已選")):
        return False
    # 新版页面会把可选数量写进标题，例如“颜色 (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 属性的自绘规格面板。"""

    all_bounds = [
        bounds
        for node in root.iter("node")
        if (bounds := _parse_bounds(node.get("bounds", ""))) is not None
    ]
    screen_bottom = max((item[3] for item in all_bounds), default=0)
    pdd_node_count = sum(
        node.get("package") == PDD_PACKAGE_NAME
        for node in root.iter("node")
    )

    headings: list[ET.Element] = []
    summaries: list[ET.Element] = []
    panel_cues: list[ET.Element] = []
    confirms: list[ET.Element] = []
    submit_hints: list[ET.Element] = []
    quantity_editors: list[ET.Element] = []
    decreases: list[ET.Element] = []
    increases: list[ET.Element] = []
    for node in root.iter("node"):
        label = _preferred_node_label(node).strip()
        compact = label.replace(" ", "")
        bounds = _parse_bounds(node.get("bounds", ""))
        if bounds is None:
            continue
        is_summary = compact.startswith(
            ("已选", "已選", "请选择", "請選擇")
        )
        if label and not is_summary and _is_dimension_heading(label):
            headings.append(node)
        if is_summary:
            summaries.append(node)
        if compact in ("确认款式", "確認款式", "关闭", "關閉"):
            panel_cues.append(node)
        if compact in ("确定", "確定") and node.get("clickable") == "true":
            confirms.append(node)
        descendant_label = _preferred_or_descendant_label(node).replace(" ", "")
        if (
            node.get("clickable") == "true"
            and node.get("enabled", "true") != "false"
            and node.get("visible-to-user", "true") != "false"
            and screen_bottom
            and (bounds[1] + bounds[3]) // 2 >= screen_bottom * 0.6
            and "提交订单" in descendant_label
            and any(
                word in descendant_label
                for word in ("选择", "颜色", "尺码", "规格")
            )
        ):
            submit_hints.append(node)
        if (
            node.get("class") == "android.widget.EditText"
            and node.get("text", "").strip().isdigit()
            and int(node.get("text", "0")) > 0
        ):
            quantity_editors.append(node)
        if compact == "减少数量" and node.get("clickable") == "true":
            decreases.append(node)
        if compact == "增加数量" and node.get("clickable") == "true":
            increases.append(node)

    if not headings or not summaries or not panel_cues:
        return None
    if confirms:
        action_nodes = [confirms[0]]
    elif (
        pdd_node_count >= 3
        and len(submit_hints) == 1
        and len(quantity_editors) == 1
        and len(decreases) == 1
        and len(increases) == 1
    ):
        action_nodes = [
            submit_hints[0],
            quantity_editors[0],
            decreases[0],
            increases[0],
        ]
    else:
        return None

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

    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 _quick_confirmation_spec_coord(root: ET.Element) -> Optional[tuple[int, int]]:
    """定位快速确认页中的当前主规格值，不接触底部购买按钮。"""

    if classify_pdd_page(root, "").kind != PAGE_ORDER_CONFIRMATION:
        return None

    nodes = list(root.iter("node"))
    if sum(node.get("package") == PDD_PACKAGE_NAME for node in nodes) < 3:
        return None

    bounded_nodes = [
        (node, bounds)
        for node in nodes
        if (bounds := _parse_bounds(node.get("bounds", ""))) is not None
        and node.get("visible-to-user", "true") != "false"
    ]
    screen_bottom = max((bounds[3] for _, bounds in bounded_nodes), default=0)
    if not screen_bottom:
        return None

    labels = [_preferred_or_descendant_label(node).replace(" ", "") for node in nodes]
    has_close = any(label in ("关闭", "關閉") for label in labels)
    has_summary = any(
        label.startswith(("已选", "已選")) for label in labels
    )
    has_payment_area = any(
        word in label for label in labels for word in ("微信支付", "先用后付", "支付方式")
    )
    has_quick_buy_action = any(
        "现在买" in _preferred_or_descendant_label(node).replace(" ", "")
        and (bounds[1] + bounds[3]) // 2 >= screen_bottom * 0.75
        for node, bounds in bounded_nodes
    )
    quantity_editors = [
        node
        for node in nodes
        if node.get("class") == "android.widget.EditText"
        and node.get("text", "").strip().isdigit()
        and int(node.get("text", "0")) > 0
    ]
    decreases = [
        node
        for node in nodes
        if _preferred_or_descendant_label(node).replace(" ", "") == "减少数量"
        and node.get("clickable") == "true"
    ]
    increases = [
        node
        for node in nodes
        if _preferred_or_descendant_label(node).replace(" ", "") == "增加数量"
        and node.get("clickable") == "true"
    ]
    if not (
        has_close
        and has_summary
        and has_payment_area
        and has_quick_buy_action
        and len(quantity_editors) == 1
        and len(decreases) == 1
        and len(increases) == 1
    ):
        return None

    headings = []
    for node, bounds in bounded_nodes:
        label = _preferred_node_label(node).strip()
        if (
            label
            and node.get("clickable") != "true"
            and _is_dimension_heading(label)
        ):
            headings.append((bounds[1], label, bounds))
    headings.sort(key=lambda item: item[0])

    used_keys: set[str] = set()
    primary_index: Optional[int] = None
    for index, (_, label, _) in enumerate(headings):
        key = _dimension_key(label, used_keys)
        used_keys.add(key)
        if key == "color":
            primary_index = index
            break
    if primary_index is None:
        return None

    _, _, heading_bounds = headings[primary_index]
    next_heading_top = (
        headings[primary_index + 1][0]
        if primary_index + 1 < len(headings)
        else int(screen_bottom * 0.75)
    )
    candidates: list[tuple[int, int, Bounds]] = []
    for node, bounds in bounded_nodes:
        if node.get("clickable") != "true" or not _is_available(node):
            continue
        label = _preferred_or_descendant_label(node).strip()
        compact = label.replace(" ", "")
        if not label:
            continue
        if compact.startswith(("已选", "已選", "请选择", "請選擇")):
            continue
        if any(
            word in compact
            for word in (
                "现在买",
                "提交订单",
                "确认购买",
                "支付",
                "增加数量",
                "减少数量",
                "关闭",
            )
        ):
            continue
        if bounds[1] < heading_bounds[3] or bounds[3] > next_heading_top:
            continue
        area = (bounds[2] - bounds[0]) * (bounds[3] - bounds[1])
        candidates.append((bounds[1], area, bounds))

    if not candidates:
        return None
    _, _, target = min(candidates, key=lambda item: (item[0], item[1]))
    left, top, right, bottom = target
    return (left + right) // 2, (top + bottom) // 2


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

    # 嵌套 RecyclerView 会先加入 groups，可能把页面下方的“套餐”排在
    # 上方“款式”之前。规范 key 依赖页面语义顺序，必须按标题 y 坐标恢复。
    heading_tops = {name: top for top, name in heading_nodes}
    groups.sort(key=lambda item: heading_tops.get(item[0], outer_bounds[3]))

    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 = 3,
        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._last_valid_spec_xml: Optional[str] = None
        self._last_invalid_spec_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._last_valid_spec_xml = None
        self._last_invalid_spec_xml = None
        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
            if exc.code == "PDD_PAGE_SPEC_PANEL_LOST":
                diagnostics = dict(exc.diagnostics)
                artifacts = list(diagnostics.get("artifacts") or [])
                for label, xml_data in (
                    ("last-valid-spec", self._last_valid_spec_xml),
                    ("last-invalid-spec", self._last_invalid_spec_xml),
                ):
                    if not xml_data:
                        continue
                    artifact = self._save_xml(label, xml_data)
                    if artifact and artifact not in artifacts:
                        artifacts.append(artifact)
                if artifacts:
                    diagnostics["artifacts"] = artifacts
                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
        saw_confirmation_page = False
        quick_confirmation_recovery_attempted = False
        incomplete_reads = 0
        while self._monotonic() < deadline:
            self._check_cancelled()
            xml_data = self._dump_hierarchy(device)
            root = _parse_xml(xml_data)
            if classify_pdd_page(root, "").kind == PAGE_ORDER_CONFIRMATION:
                saw_confirmation_page = True
            try:
                snapshot = parse_spec_panel(xml_data)
            except PddCollectError as exc:
                if exc.code != "PDD_DATA_SPEC_INCOMPLETE":
                    raise
                self._last_invalid_spec_xml = xml_data
                incomplete_reads += 1
            else:
                if _is_spec_panel_open(xml_data):
                    self._last_valid_spec_xml = xml_data
                    return snapshot
                self._last_invalid_spec_xml = xml_data
                incomplete_reads += 1

            if not quick_confirmation_recovery_attempted:
                coordinate = _quick_confirmation_spec_coord(root)
                if coordinate is not None:
                    # 这里只打开当前主规格，禁止触碰“现在买”等不可逆动作。
                    device.click(*coordinate)
                    quick_confirmation_recovery_attempted = True
            self._sleep(0.25)
        if saw_confirmation_page:
            message = "规格面板已经打开，但页面结构无法识别"
            if quick_confirmation_recovery_attempted:
                message = "快速确认页无法进入可识别的规格选择区域"
            raise PddCollectError(
                "PDD_DATA_SPEC_INCOMPLETE",
                message,
                {
                    "incomplete_spec_reads": incomplete_reads,
                    "quick_confirmation_recovery_attempted": (
                        quick_confirmation_recovery_attempted
                    ),
                },
            )
        raise PddCollectError(
            "PDD_PAGE_SPEC_PANEL_TIMEOUT",
            "点击规格入口后，等待规格面板加载超时",
        )

    def _read_valid_spec_panel(
        self,
        device: Any,
        initial_xml: Optional[str] = None,
    ) -> tuple[str, SpecSnapshot]:
        """跳过规格操作期间的临时空树，返回最新有效面板和解析结果。"""

        deadline = self._monotonic() + min(2.0, self._spec_panel_timeout)
        xml_data = initial_xml
        transient_reads = 0
        # 次数上限避免测试时钟或设备时钟异常导致无限循环。
        for attempt in range(20):
            self._check_cancelled()
            if xml_data is None:
                xml_data = self._dump_hierarchy(device)
            try:
                snapshot = parse_spec_panel(xml_data)
                panel_open = _is_spec_panel_open(xml_data)
            except PddCollectError as exc:
                if exc.code != "PDD_DATA_SPEC_INCOMPLETE":
                    raise
                panel_open = False
                snapshot = None

            if panel_open and snapshot is not None:
                self._last_valid_spec_xml = xml_data
                return xml_data, snapshot

            self._last_invalid_spec_xml = xml_data
            transient_reads += 1
            if attempt >= 19 or self._monotonic() >= deadline:
                break
            self._sleep(0.1)
            xml_data = None

        raise PddCollectError(
            "PDD_PAGE_SPEC_PANEL_LOST",
            "规格面板操作期间控件树持续为空或面板已经消失",
            {"transient_spec_reads": transient_reads},
        )

    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, first_snapshot = self._read_valid_spec_panel(device)
        source_dimension = next(
            (item for item in first_snapshot.dimensions if item.key == "color"),
            None,
        )
        first_rows = self._visible_color_rows(first_xml)
        if source_dimension is None or 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:
                    if pending_xml is not None:
                        xml_data = pending_xml
                    else:
                        xml_data, _ = self._read_valid_spec_panel(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", source_dimension.name, 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._read_valid_spec_panel(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._read_valid_spec_panel(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_xml, snapshot = self._read_valid_spec_panel(
                device, initial_xml=latest_xml
            )
            latest_root = _parse_xml(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, snapshot = self._read_valid_spec_panel(device)
            root = _parse_xml(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_xml, _ = self._read_valid_spec_panel(device)
                        latest_root = _parse_xml(latest_xml)
                        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._read_valid_spec_panel(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")
        self._sleep(self._color_poll_interval)
        try:
            self._read_valid_spec_panel(device)
        except PddCollectError as exc:
            if exc.code != "PDD_PAGE_SPEC_PANEL_LOST":
                raise
            raise PddCollectError(
                "PDD_PAGE_SPEC_PANEL_LOST",
                "点击颜色后进入大图，返回一次仍未恢复规格面板",
                exc.diagnostics,
            ) from exc

    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._read_valid_spec_panel(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] = {}
        size_name = "尺码"
        xml_data: Optional[str] = None
        size_found = False
        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, snapshot = self._read_valid_spec_panel(device)
            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":
                    size_name = dimension.name
                    size_found = True
                    for value in dimension.values:
                        option_text = self._second_dimension_option_text(value.text)
                        if option_text is None:
                            continue
                        sizes[option_text] = (
                            sizes.get(option_text, False) or value.available
                        )

            if size_found:
                break

            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 size_found or xml_data is None:
            return None

        xml_data, start_confirmed = self._move_second_dimension_to_start(
            device, xml_data
        )
        initial_snapshot = parse_spec_panel(xml_data)
        initial_size_dimension = next(
            (item for item in initial_snapshot.dimensions if item.key == "size"),
            None,
        )
        if initial_size_dimension is None:
            raise PddCollectError(
                "PDD_DATA_SPEC_INCOMPLETE",
                f"无法建立第二规格续页上下文：{size_name}",
            )
        static_dimension = self._non_scrollable_second_dimension(
            xml_data, initial_size_dimension
        )
        if static_dimension is not None:
            return static_dimension
        continuation = self._build_second_dimension_context(
            xml_data, initial_size_dimension
        )
        strict_edge_check = "套餐" in size_name and self._max_spec_swipes > 0
        if strict_edge_check and not start_confirmed:
            raise PddCollectError(
                "PDD_DATA_SPEC_INCOMPLETE",
                f"{size_name}无法确认已经到达列表起点",
            )
        # 第一阶段只负责找到分组；正式输出从确认后的起点重新按页面顺序收集。
        sizes = {}
        previous_signature = None
        stable_edge_reads = 0
        end_confirmed = False
        for swipe_count in range(self._max_spec_swipes + 1):
            self._check_cancelled()
            xml_data, snapshot = self._read_valid_spec_panel(
                device, initial_xml=xml_data
            )
            size_dimension = next(
                (item for item in snapshot.dimensions if item.key == "size"),
                None,
            )
            if size_dimension is None:
                size_dimension = self._extract_second_dimension_continuation(
                    xml_data, continuation
                )
            if size_dimension is None:
                # 控件树偶尔会在滚动动画中短暂缺节点，再读取一次后才判定失败。
                xml_data, retry_snapshot = self._read_valid_spec_panel(device)
                size_dimension = next(
                    (item for item in retry_snapshot.dimensions if item.key == "size"),
                    None,
                ) or self._extract_second_dimension_continuation(
                    xml_data, continuation
                )
            if size_dimension is None:
                raise PddCollectError(
                    "PDD_DATA_SPEC_INCOMPLETE",
                    f"滚动过程中持续丢失第二规格：{size_name}",
                )
            size_name = size_dimension.name
            for value in size_dimension.values:
                option_text = self._second_dimension_option_text(value.text)
                if option_text is None:
                    continue
                sizes[option_text] = sizes.get(option_text, False) or value.available

            signature = self._second_dimension_signature(xml_data, size_dimension)
            stable_edge_reads = (
                stable_edge_reads + 1 if signature == previous_signature else 0
            )
            previous_signature = signature
            if stable_edge_reads >= 2:
                end_confirmed = True
                break
            if swipe_count >= self._max_spec_swipes:
                break

            root = _parse_xml(xml_data)
            region, horizontal = self._second_dimension_region(
                root, size_dimension
            )
            if region is None:
                end_confirmed = True
                break
            self._swipe_region(
                device,
                region,
                horizontal=horizontal,
                reverse=False,
            )
            self._sleep(
                self._horizontal_swipe_settle_interval if horizontal else 0.35
            )
            xml_data = None

        expected_count = self._dimension_expected_count(size_name)
        if expected_count is not None and len(sizes) < expected_count:
            raise PddCollectError(
                "PDD_DATA_SPEC_INCOMPLETE",
                f"{size_name}应有 {expected_count} 个选项，实际只采到 {len(sizes)} 个",
                {
                    "dimension_name": size_name,
                    "expected_count": expected_count,
                    "collected_count": len(sizes),
                },
            )
        if expected_count is None and strict_edge_check and not end_confirmed:
            raise PddCollectError(
                "PDD_DATA_SPEC_INCOMPLETE",
                f"{size_name}达到滑动上限，无法确认是否采集完整",
                {"dimension_name": size_name, "collected_count": len(sizes)},
            )
        return SpecDimension(
            "size",
            size_name,
            tuple(DimensionValue(text, available) for text, available in sizes.items()),
        )

    def _non_scrollable_second_dimension(
        self,
        xml_data: str | bytes,
        dimension: SpecDimension,
    ) -> Optional[SpecDimension]:
        """强证据确认非滚动面板时，返回当前树中的完整第二规格。"""

        root = _parse_xml(xml_data)
        region, _ = self._second_dimension_region(root, dimension)
        if region is not None:
            return None

        parents = {child: parent for parent in root.iter() for child in parent}
        if _find_non_scrollable_spec_panel(root, parents) is None:
            return None

        values: dict[str, bool] = {}
        for value in dimension.values:
            option_text = self._second_dimension_option_text(value.text)
            if option_text is None:
                continue
            values[option_text] = values.get(option_text, False) or value.available

        expected_count = self._dimension_expected_count(dimension.name)
        if expected_count is not None and len(values) < expected_count:
            raise PddCollectError(
                "PDD_DATA_SPEC_INCOMPLETE",
                f"{dimension.name}应有 {expected_count} 个选项，实际只采到 {len(values)} 个",
                {
                    "dimension_name": dimension.name,
                    "expected_count": expected_count,
                    "collected_count": len(values),
                },
            )
        if not values:
            return None
        return SpecDimension(
            "size",
            dimension.name,
            tuple(
                DimensionValue(text, available)
                for text, available in values.items()
            ),
        )

    def _move_second_dimension_to_start(
        self, device: Any, xml_data: str
    ) -> tuple[str, bool]:
        """横向第二规格先归位到左端；纵向列表保持当前安全起点。"""

        previous_signature: Optional[tuple[tuple[str, Bounds], ...]] = None
        stable_edge_reads = 0
        for attempt in range(self._max_spec_swipes + 1):
            snapshot = parse_spec_panel(xml_data)
            size_dimension = next(
                (item for item in snapshot.dimensions if item.key == "size"),
                None,
            )
            if size_dimension is None:
                return xml_data, False
            root = _parse_xml(xml_data)
            region, horizontal = self._second_dimension_region(
                root, size_dimension
            )
            if region is None or not horizontal:
                return xml_data, True
            signature = self._second_dimension_signature(xml_data, size_dimension)
            stable_edge_reads = (
                stable_edge_reads + 1 if signature == previous_signature else 0
            )
            previous_signature = signature
            if stable_edge_reads >= 2:
                return xml_data, True
            if attempt >= self._max_spec_swipes:
                return xml_data, False
            self._swipe_region(
                device, region, horizontal=True, reverse=True
            )
            self._sleep(self._horizontal_swipe_settle_interval)
            xml_data, _ = self._read_valid_spec_panel(device)
        return xml_data, False

    def _second_dimension_signature(
        self,
        xml_data: str | bytes,
        dimension: SpecDimension,
    ) -> tuple[tuple[str, Bounds], ...]:
        """返回已确认的第二规格签名，不要求当前屏仍显示规格标题。"""

        root = _parse_xml(xml_data)
        targets = {
            normalized
            for value in dimension.values
            if (normalized := self._second_dimension_option_text(value.text))
            is not None
        }
        result = []
        for node in root.iter("node"):
            if node.get("clickable") != "true":
                continue
            text = self._second_dimension_option_text(
                _preferred_or_descendant_label(node).strip()
            )
            if text not in targets:
                continue
            bounds = _parse_bounds(node.get("bounds", ""))
            if bounds is not None:
                result.append((text, bounds))
        return tuple(result)

    def _build_second_dimension_context(
        self,
        xml_data: str | bytes,
        dimension: SpecDimension,
    ) -> _SecondDimensionContext:
        """记录第二规格容器和选项结构，供标题滑出后的续页使用。"""

        root = _parse_xml(xml_data)
        parents = {child: parent for parent in root.iter() for child in parent}
        region, horizontal = self._second_dimension_region(root, dimension)
        if region is None:
            raise PddCollectError(
                "PDD_DATA_SPEC_INCOMPLETE",
                f"无法定位第二规格滚动区域：{dimension.name}",
            )
        option_nodes = [
            node
            for value in dimension.values
            if (node := self._find_option_node(root, value.text)) is not None
        ]
        containers = [
            ancestor
            for node in option_nodes
            for ancestor in _ancestors(node, parents)
            if ancestor.get("scrollable") == "true"
            and _parse_bounds(ancestor.get("bounds", "")) == region
        ]
        if not containers:
            raise PddCollectError(
                "PDD_DATA_SPEC_INCOMPLETE",
                f"无法锁定第二规格滚动容器：{dimension.name}",
            )
        container = containers[0]
        structures = tuple(
            dict.fromkeys(
                self._option_structure(node, container, parents)
                for node in option_nodes
            )
        )
        return _SecondDimensionContext(
            name=dimension.name,
            horizontal=horizontal,
            container_class=container.get("class", ""),
            container_resource_id=container.get("resource-id", ""),
            container_bounds=region,
            option_structures=structures,
        )

    def _extract_second_dimension_continuation(
        self,
        xml_data: str | bytes,
        context: _SecondDimensionContext,
    ) -> Optional[SpecDimension]:
        """规格标题不可见时，从已锁定容器提取同结构的后续选项。"""

        root = _parse_xml(xml_data)
        parents = {child: parent for parent in root.iter() for child in parent}
        container = self._find_second_dimension_container(root, context)
        if container is None:
            return None
        container_bounds = _parse_bounds(container.get("bounds", ""))
        if container_bounds is None:
            return None

        values: list[DimensionValue] = []
        seen: set[str] = set()
        for node in _top_level_clickable_options(container, parents):
            if self._option_structure(node, container, parents) not in context.option_structures:
                continue
            text = _preferred_or_descendant_label(node).strip()
            bounds = _parse_bounds(node.get("bounds", ""))
            option_text = self._second_dimension_option_text(text)
            if (
                option_text is None
                or option_text in seen
                or bounds is None
            ):
                continue
            if text.endswith(("…", "...")):
                raise PddCollectError(
                    "PDD_DATA_SKU_NAME_TRUNCATED",
                    f"规格名称被截断，无法安全采集：{text}",
                )
            seen.add(option_text)
            values.append(DimensionValue(option_text, _is_available(node)))
        if not values:
            return None
        return SpecDimension("size", context.name, tuple(values))

    @staticmethod
    def _option_structure(
        node: ET.Element,
        container: ET.Element,
        parents: Mapping[ET.Element, ET.Element],
    ) -> tuple[str, str, tuple[tuple[str, str], ...]]:
        """生成不依赖坐标和 XML 对象身份的选项结构特征。"""

        ancestors: list[tuple[str, str]] = []
        current = parents.get(node)
        while current is not None and current is not container:
            ancestors.append(
                (current.get("class", ""), current.get("resource-id", ""))
            )
            current = parents.get(current)
        return (
            node.get("class", ""),
            node.get("resource-id", ""),
            tuple(ancestors),
        )

    @staticmethod
    def _find_second_dimension_container(
        root: ET.Element,
        context: _SecondDimensionContext,
    ) -> Optional[ET.Element]:
        """在新控件树中重新找到首次锁定的滚动容器。"""

        candidates = [
            node
            for node in root.iter("node")
            if node.get("scrollable") == "true"
            and _parse_bounds(node.get("bounds", "")) is not None
            and (
                not context.container_resource_id
                or node.get("resource-id", "") == context.container_resource_id
            )
        ]
        if not candidates:
            return None
        return max(
            candidates,
            key=lambda node: (
                _parse_bounds(node.get("bounds", "")) == context.container_bounds,
                node.get("class", "") == context.container_class,
            ),
        )

    @staticmethod
    def _second_dimension_option_text(text: str) -> Optional[str]:
        """返回去掉尾部价格的规格文字；操作项或纯价格返回 ``None``。"""

        compact = text.replace(" ", "")
        if not compact or compact.startswith(("已选", "请选择")) or any(
            marker in compact
            for marker in (
                "增加数量",
                "减少数量",
                "提交订单",
                "立即购买",
                "单独购买",
                "免拼购买",
                "发起拼单",
                "确认购买",
                "关闭",
            )
        ):
            return None
        if re.fullmatch(
            r"(?:券后|到手价|拼单价|单买价|价格)?[¥￥]\d+(?:\.\d{1,2})?",
            compact,
        ):
            return None
        normalized = re.sub(
            r"\s*[¥￥]\s*\d+(?:\.\d{1,2})?\s*$",
            "",
            text,
        ).strip()
        return normalized or None

    def _second_dimension_region(
        self,
        root: ET.Element,
        dimension: SpecDimension,
    ) -> tuple[Optional[Bounds], bool]:
        """返回第二规格最近的滚动容器及其主要滚动方向。"""

        parents = {child: parent for parent in root.iter() for child in parent}
        containers: dict[ET.Element, int] = {}
        option_bounds: list[Bounds] = []
        for value in dimension.values:
            node = self._find_option_node(root, value.text)
            if node is None:
                continue
            bounds = _parse_bounds(node.get("bounds", ""))
            if bounds is not None:
                option_bounds.append(bounds)
            current = parents.get(node)
            while current is not None:
                if (
                    current.get("scrollable") == "true"
                    and _parse_bounds(current.get("bounds", "")) is not None
                ):
                    containers[current] = containers.get(current, 0) + 1
                    break
                current = parents.get(current)

        if containers:
            container = max(
                containers,
                key=lambda item: (
                    containers[item],
                    len(_ancestors(item, parents)),
                ),
            )
            region = _parse_bounds(container.get("bounds", ""))
        else:
            region = self._vertical_region(root)
        if region is None:
            return None, False

        centers = [
            ((left + right) // 2, (top + bottom) // 2)
            for left, top, right, bottom in option_bounds
        ]
        region_width = region[2] - region[0]
        region_height = region[3] - region[1]
        if region_height >= region_width * 0.75:
            # 外层规格面板通常很高；其中一行尺码横向排布，不代表面板
            # 应横向滚动。只有较矮的独立列表才根据选项分布判断横向。
            horizontal = False
        elif len(centers) >= 2:
            x_span = max(item[0] for item in centers) - min(item[0] for item in centers)
            y_span = max(item[1] for item in centers) - min(item[1] for item in centers)
            horizontal = x_span > y_span
        else:
            horizontal = region_width > region_height * 1.8
        return region, horizontal

    @staticmethod
    def _dimension_expected_count(name: str) -> Optional[int]:
        """读取“套餐（15）”这类标题中的选项数量。"""

        match = re.search(r"[（(]\s*(\d+)\s*[)）]\s*$", name)
        return int(match.group(1)) if match else None

    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)