diff --git a/client/src/pdd_collect_service.py b/client/src/pdd_collect_service.py
index 119b215..90f700b 100644
--- a/client/src/pdd_collect_service.py
+++ b/client/src/pdd_collect_service.py
@@ -174,6 +174,18 @@ class SpecSnapshot:
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:
"""当前控件树中完整可点击的规格节点。"""
@@ -1728,6 +1740,8 @@ class PddCollectService:
size_name = dimension.name
size_found = True
for value in dimension.values:
+ if self._is_second_dimension_action(value.text):
+ continue
sizes[value.text] = sizes.get(value.text, False) or value.available
if size_found:
@@ -1760,6 +1774,19 @@ class PddCollectService:
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}",
+ )
+ 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(
@@ -1780,16 +1807,31 @@ class PddCollectService:
(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}",
+ f"滚动过程中持续丢失第二规格：{size_name}",
)
size_name = size_dimension.name
for value in size_dimension.values:
+ if self._is_second_dimension_action(value.text):
+ continue
sizes[value.text] = sizes.get(value.text, False) or value.available
- signature = self._second_dimension_signature(xml_data)
+ signature = self._second_dimension_signature(xml_data, size_dimension)
stable_edge_reads = (
stable_edge_reads + 1 if signature == previous_signature else 0
)
@@ -1862,7 +1904,7 @@ class PddCollectService:
)
if region is None or not horizontal:
return xml_data, True
- signature = self._second_dimension_signature(xml_data)
+ signature = self._second_dimension_signature(xml_data, size_dimension)
stable_edge_reads = (
stable_edge_reads + 1 if signature == previous_signature else 0
)
@@ -1879,20 +1921,15 @@ class PddCollectService:
return xml_data, False
def _second_dimension_signature(
- self, xml_data: str | bytes
+ self,
+ xml_data: str | bytes,
+ dimension: SpecDimension,
) -> tuple[tuple[str, Bounds], ...]:
- """只返回当前可见第二规格，避免其他区域变化干扰到边判断。"""
+ """返回已确认的第二规格签名，不要求当前屏仍显示规格标题。"""
root = _parse_xml(xml_data)
- snapshot = parse_spec_panel(xml_data)
- size_dimension = next(
- (item for item in snapshot.dimensions if item.key == "size"),
- None,
- )
- if size_dimension is None:
- return ()
result = []
- for value in size_dimension.values:
+ for value in dimension.values:
node = self._find_option_node(root, value.text)
if node is None:
continue
@@ -1901,6 +1938,167 @@ class PddCollectService:
result.append((value.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", ""))
+ if (
+ not text
+ or text in seen
+ or bounds is None
+ or self._is_second_dimension_action(text)
+ or bounds[2] - bounds[0]
+ >= (container_bounds[2] - container_bounds[0]) * 0.85
+ ):
+ continue
+ if text.endswith(("…", "...")):
+ raise PddCollectError(
+ "PDD_DATA_SKU_NAME_TRUNCATED",
+ f"规格名称被截断，无法安全采集：{text}",
+ )
+ seen.add(text)
+ values.append(DimensionValue(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 _is_second_dimension_action(text: str) -> bool:
+ """排除规格面板中容易和选项结构相同的操作按钮。"""
+
+ compact = text.replace(" ", "")
+ if "¥" in compact or "￥" in compact:
+ return True
+ return compact.startswith(("已选", "请选择")) or any(
+ marker in compact
+ for marker in (
+ "增加数量",
+ "减少数量",
+ "提交订单",
+ "立即购买",
+ "单独购买",
+ "免拼购买",
+ "发起拼单",
+ "确认购买",
+ "关闭",
+ )
+ )
+
def _second_dimension_region(
self,
root: ET.Element,
diff --git a/client/test/test_pdd_collect_service.py b/client/test/test_pdd_collect_service.py
index 2a409d2..2960c29 100644
--- a/client/test/test_pdd_collect_service.py
+++ b/client/test/test_pdd_collect_service.py
@@ -509,6 +509,59 @@ class HorizontalPackageDevice:
self.page = max(self.page - 1, 0)
+class VerticalLongPackageDevice:
+ """模拟纵向长套餐：滑到第二屏后“套餐”标题不再出现在控件树。"""
+
+ packages = tuple(f"套餐{i}" for i in range(1, 11))
+
+ def __init__(self):
+ self.page = 0
+ self.swipes = []
+ self.clicks = []
+
+ def dump_hierarchy(self):
+ start = min(self.page * 3, len(self.packages) - 4)
+ visible = self.packages[start : start + 4]
+ headings = ""
+ if self.page == 0:
+ headings = """
+
+
+
+ """
+ options = []
+ for index, text in enumerate(visible):
+ top = 1080 + index * 210
+ options.append(
+ f''
+ )
+ return f"""
+
+
+
+ {headings}
+ {''.join(options)}
+
+
+
+
+ """
+
+ def swipe(self, x1, y1, x2, y2, duration=0.35):
+ self.swipes.append((x1, y1, x2, y2, duration))
+ if abs(y2 - y1) <= abs(x2 - x1):
+ return
+ if y1 > y2:
+ self.page = min(self.page + 1, 2)
+ else:
+ self.page = max(self.page - 1, 0)
+
+
class FakeClock:
"""测试用时钟：sleep 只推进虚拟时间，不真的等待。"""
@@ -1258,6 +1311,30 @@ class PddCollectParserTest(unittest.TestCase):
self.assertEqual(raised.exception.diagnostics["expected_count"], 8)
self.assertEqual(raised.exception.diagnostics["collected_count"], 7)
+ def test_vertical_packages_continue_after_heading_scrolls_out(self):
+ device = VerticalLongPackageDevice()
+ service = PddCollectService(
+ PddDeviceService(lambda _serial: device),
+ "USB-001",
+ "client-001",
+ sleeper=lambda _seconds: None,
+ )
+
+ result = service._collect_size_dimension(device)
+
+ self.assertIsNotNone(result)
+ self.assertEqual(result.name, "套餐（10）")
+ self.assertEqual(
+ [value.text for value in result.values],
+ list(device.packages),
+ )
+ self.assertNotIn("增加数量", [value.text for value in result.values])
+ self.assertNotIn("提交订单", [value.text for value in result.values])
+ self.assertEqual(device.clicks, [])
+ self.assertTrue(
+ all(abs(y2 - y1) > abs(x2 - x1) for x1, y1, x2, y2, _ in device.swipes)
+ )
+
def test_visible_click_is_fallback_when_page_exposes_no_selection_state(self):
xml_data = keep_only_one_sku(self.spec_xml)
root = ET.fromstring(xml_data)