fix: 安全采集图片颜色并取消默认尺码 (#160)

client/src/pdd_collect_service.py

@@ -879,6 +879,7 @@ class PddCollectService:
                 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,)
@@ -1350,7 +1351,12 @@ class PddCollectService:
             ),
             None,
         )
-        node = self._find_option_node(root, target) if latest_visible else 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", ""))
@@ -1375,6 +1381,9 @@ class PddCollectService:
         while True:
             self._check_cancelled()
             latest_xml = self._dump_hierarchy(device)
+            if self._is_big_image_viewer(latest_xml):
+                self._recover_spec_panel_from_big_image(device)
+                return ColorPriceSample(None, None, None)
             latest_root = _parse_xml(latest_xml)
             snapshot = parse_spec_panel(latest_xml)
             latest_node = self._find_option_node(latest_root, target)
@@ -1428,6 +1437,140 @@ class PddCollectService:
             else:
                 selection_sleeps += 1
 
+    def _clear_default_size_selection(self, device: Any) -> None:
+        """逐屏寻找默认尺码；只在唯一选中项明确时点击一次取消。"""
+
+        previous_signature: Optional[
+            tuple[tuple[str, tuple[str, ...]], ...]
+        ] = None
+        stable_edge_reads = 0
+        for swipe_count in range(self._max_spec_swipes + 1):
+            self._check_cancelled()
+            xml_data = self._dump_hierarchy(device)
+            root = _parse_xml(xml_data)
+            snapshot = parse_spec_panel(xml_data)
+            size_dimension = next(
+                (item for item in snapshot.dimensions if item.key == "size"),
+                None,
+            )
+            if size_dimension is not None:
+                selected_values = []
+                for value in size_dimension.values:
+                    state_node = self._find_option_node(root, value.text)
+                    if state_node is not None and self._node_is_selected(state_node):
+                        selected_values.append((value.text, state_node))
+
+                if len(selected_values) == 1:
+                    target, state_node = selected_values[0]
+                    click_node = self._find_safe_option_click_node(state_node, target)
+                    if click_node is not None:
+                        bounds = _parse_bounds(click_node.get("bounds", ""))
+                        assert bounds is not None
+                        device.click(
+                            (bounds[0] + bounds[2]) // 2,
+                            (bounds[1] + bounds[3]) // 2,
+                        )
+                        self._sleep(self._color_poll_interval)
+                        latest_root = _parse_xml(self._dump_hierarchy(device))
+                        latest_node = self._find_option_node(latest_root, target)
+                        if latest_node is None or not self._node_is_selected(
+                            latest_node
+                        ):
+                            break
+                    # 页面不支持取消时也不能重复点击，否则可能重新选中或误操作。
+                    break
+                if not selected_values:
+                    break
+
+            signature = tuple(
+                (item.key, tuple(value.text for value in item.values))
+                for item in snapshot.dimensions
+            )
+            stable_edge_reads = (
+                stable_edge_reads + 1 if signature == previous_signature else 0
+            )
+            previous_signature = signature
+            if stable_edge_reads >= 2 or swipe_count >= self._max_spec_swipes:
+                break
+            region = self._vertical_region(root)
+            if region is None:
+                break
+            self._swipe_region(device, region, horizontal=False, reverse=False)
+            self._sleep(0.35)
+
+        # 查找尺码时页面可能已经滚到底部，逐色采价前必须回到顶部。
+        self._move_spec_panel_to_top(device)
+
+    def _move_spec_panel_to_top(self, device: Any) -> None:
+        """把规格面板纵向恢复到顶部；连续两次不变化即停止。"""
+
+        previous_signature: Optional[tuple[tuple[str, Bounds], ...]] = None
+        stable_edge_reads = 0
+        for _ in range(self._max_spec_swipes):
+            self._check_cancelled()
+            xml_data = self._dump_hierarchy(device)
+            root = _parse_xml(xml_data)
+            signature = tuple(
+                (
+                    _preferred_node_label(node),
+                    bounds,
+                )
+                for node in root.iter("node")
+                if (bounds := _parse_bounds(node.get("bounds", ""))) is not None
+                and (
+                    _is_dimension_heading(_preferred_node_label(node))
+                    or node.get("clickable") == "true"
+                )
+            )
+            stable_edge_reads = (
+                stable_edge_reads + 1 if signature == previous_signature else 0
+            )
+            previous_signature = signature
+            if stable_edge_reads >= 2:
+                return
+            region = self._vertical_region(root)
+            if region is None:
+                return
+            self._swipe_region(device, region, horizontal=False, reverse=True)
+            self._sleep(0.35)
+
+    @staticmethod
+    def _is_big_image_viewer(xml_data: str | bytes) -> bool:
+        """用 ViewPager 和页码共同识别误入的商品大图页。"""
+
+        root = _parse_xml(xml_data)
+        has_pager = any(
+            (node.get("class") or "").endswith("ViewPager")
+            for node in root.iter("node")
+        )
+        has_page_number = any(
+            re.fullmatch(r"\d+\s*/\s*\d+", _preferred_node_label(node))
+            for node in root.iter("node")
+        )
+        has_dimension = any(
+            _is_dimension_heading(_preferred_node_label(node))
+            for node in root.iter("node")
+        )
+        return has_pager and has_page_number and not has_dimension
+
+    def _recover_spec_panel_from_big_image(self, device: Any) -> None:
+        """误入大图后只返回一次，并确认规格面板已经恢复。"""
+
+        device.press("back")
+        for _ in range(3):
+            self._check_cancelled()
+            self._sleep(self._color_poll_interval)
+            xml_data = self._dump_hierarchy(device)
+            try:
+                if _is_spec_panel_open(xml_data):
+                    return
+            except PddCollectError:
+                break
+        raise PddCollectError(
+            "PDD_PAGE_SPEC_PANEL_LOST",
+            "点击颜色后进入大图，返回一次仍未恢复规格面板",
+        )
+
     def _color_view_signature(
         self, xml_data: str | bytes
     ) -> tuple[tuple[str, Bounds], ...]:
@@ -1677,6 +1820,36 @@ class PddCollectService:
                 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}