fix: 兼容套餐标题滑出后的续页采集 (#168)

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,

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 = """
+              <node text="款式" bounds="[36,720][180,780]" />
+              <node text="口罩款" clickable="true" selected="true"
+                    bounds="[36,800][300,900]" />
+              <node text="套餐（10）" bounds="[36,980][240,1040]" />
+            """
+        options = []
+        for index, text in enumerate(visible):
+            top = 1080 + index * 210
+            options.append(
+                f'<node text="{text}" clickable="true" '
+                f'bounds="[36,{top}][500,{top + 150}]" />'
+            )
+        return f"""<hierarchy>
+          <node bounds="[0,0][1080,2340]">
+            <node class="android.widget.ScrollView" scrollable="true"
+                  bounds="[0,650][1080,2200]">
+              <node text="已选：套餐1" bounds="[36,660][500,710]" />
+              {headings}
+              {''.join(options)}
+              <node text="增加数量" clickable="true"
+                    bounds="[36,1950][220,2050]" />
+              <node text="提交订单" clickable="true"
+                    bounds="[300,2080][1040,2190]" />
+            </node>
+          </node>
+        </hierarchy>"""
+
+    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)