diff --git a/client/src/pdd_collect_service.py b/client/src/pdd_collect_service.py index f408ee4..6db10e9 100644 --- a/client/src/pdd_collect_service.py +++ b/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} diff --git a/client/test/test_pdd_collect_service.py b/client/test/test_pdd_collect_service.py index c7f03a9..e172345 100644 --- a/client/test/test_pdd_collect_service.py +++ b/client/test/test_pdd_collect_service.py @@ -300,6 +300,95 @@ class DelayedHorizontalSwipeDevice(SnakeColorDevice): self.sizes_visible = True +class ImageColorDevice(SnakeColorDevice): + """图片型颜色卡片：点图片会进入大图，点底部文字才选中颜色。""" + + def __init__(self, home_xml): + super().__init__(home_xml) + self.big_image_open = False + + def dump_hierarchy(self): + if self.big_image_open: + return """ + + + + + """ + return super().dump_hierarchy() + + def click(self, x, y): + if self.panel_open and 740 <= y < 820: + self.clicks.append((x, y)) + self.big_image_open = True + return + super().click(x, y) + + def press(self, key): + if key == "back": + self.big_image_open = False + + def _spec_xml(self): + xml_data = super()._spec_xml() + old = ( + '' + ) + new = ( + '' + '' + '' + '' + '' + ) + return xml_data.replace(old, new) + + +class DefaultSizeDevice(SnakeColorDevice): + """尺码在下方且默认选中，取消后才能完整遍历颜色。""" + + def __init__(self, home_xml, *, can_clear=True): + super().__init__(home_xml) + self.size_selected = True + self.can_clear = can_clear + self.size_clicks = 0 + + def click(self, x, y): + if self.panel_open and self.sizes_visible and 1210 <= y <= 1300: + self.clicks.append((x, y)) + self.size_clicks += 1 + if self.can_clear: + self.size_selected = False + return + super().click(x, y) + + def swipe(self, x1, y1, x2, y2, duration=0.35): + self.swipes.append(((x1, y1, x2, y2), {"duration": duration})) + self.swipe_panel_states.append(self.panel_open) + if abs(x2 - x1) > abs(y2 - y1): + self.page = 1 if x1 > x2 else 0 + else: + self.sizes_visible = y1 > y2 + + def _spec_xml(self): + xml_data = super()._spec_xml() + if self.sizes_visible: + selected = str(self.size_selected).lower() + xml_data = xml_data.replace( + 'content-desc="M"\n clickable="true"', + f'content-desc="M"\n clickable="true" selected="{selected}"', + ) + return xml_data + + class FakeClock: """测试用时钟：sleep 只推进虚拟时间，不真的等待。""" @@ -822,6 +911,86 @@ class PddCollectParserTest(unittest.TestCase): for item in data["skus"] )) + def test_image_color_card_clicks_caption_instead_of_opening_big_image(self): + device = ImageColorDevice(self.home_xml) + device.panel_open = True + device.page = 0 + service = PddCollectService( + PddDeviceService(lambda _serial: device), + "USB-001", + "client-001", + sleeper=lambda _seconds: None, + ) + + sample = service._click_and_sample_color(device, "A色") + + self.assertEqual(1000, sample.price_cent) + self.assertFalse(device.big_image_open) + self.assertGreaterEqual(device.clicks[0][1], 820) + + def test_big_image_recovery_presses_back_only_once(self): + device = ImageColorDevice(self.home_xml) + device.panel_open = True + device.big_image_open = True + press_calls = [] + original_press = device.press + + def record_press(key): + press_calls.append(key) + original_press(key) + + device.press = record_press + service = PddCollectService( + PddDeviceService(lambda _serial: device), + "USB-001", + "client-001", + sleeper=lambda _seconds: None, + ) + + service._recover_spec_panel_from_big_image(device) + + self.assertEqual(["back"], press_calls) + self.assertFalse(device.big_image_open) + + def test_default_size_is_cleared_once_before_colors_and_sizes_are_collected(self): + device = DefaultSizeDevice(self.home_xml) + service = PddCollectService( + PddDeviceService(lambda _serial: device), + "USB-001", + "client-001", + sleeper=lambda _seconds: None, + ) + + data = service.collect( + FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123") + ).to_pdd_data() + + self.assertEqual(1, device.size_clicks) + self.assertFalse(device.size_selected) + self.assertEqual( + ["A色", "B色", "C色", "F色", "E色", "D色"], + device.clicked_colors, + ) + self.assertEqual( + ["M", "L"], + [item["text"] for item in data["dimensions"][1]["values"]], + ) + + def test_size_that_cannot_be_cleared_is_not_clicked_repeatedly(self): + device = DefaultSizeDevice(self.home_xml, can_clear=False) + device.panel_open = True + service = PddCollectService( + PddDeviceService(lambda _serial: device), + "USB-001", + "client-001", + sleeper=lambda _seconds: None, + ) + + service._clear_default_size_selection(device) + + self.assertEqual(1, device.size_clicks) + self.assertTrue(device.size_selected) + 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)