diff --git a/client/src/pdd_collect_service.py b/client/src/pdd_collect_service.py index cdf1a9f..119b215 100644 --- a/client/src/pdd_collect_service.py +++ b/client/src/pdd_collect_service.py @@ -718,6 +718,11 @@ def parse_spec_panel(xml_data: str | bytes) -> SpecSnapshot: 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: @@ -1697,15 +1702,19 @@ class PddCollectService: 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) @@ -1717,9 +1726,13 @@ class PddCollectService: ) if dimension.key == "size": size_name = dimension.name + size_found = True for value in dimension.values: sizes[value.text] = sizes.get(value.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 @@ -1741,14 +1754,219 @@ class PddCollectService: ) self._sleep(0.35) - if not sizes: + if not size_found or xml_data is None: return None + + xml_data, start_confirmed = self._move_second_dimension_to_start( + device, xml_data + ) + 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: + raise PddCollectError( + "PDD_DATA_SPEC_INCOMPLETE", + f"滚动过程中丢失第二规格:{size_name}", + ) + size_name = size_dimension.name + for value in size_dimension.values: + sizes[value.text] = sizes.get(value.text, False) or value.available + + signature = self._second_dimension_signature(xml_data) + 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 _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) + 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 + ) -> 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: + node = self._find_option_node(root, value.text) + if node is None: + continue + bounds = _parse_bounds(node.get("bounds", "")) + if bounds is not None: + result.append((value.text, bounds)) + return tuple(result) + + 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, ...], diff --git a/client/test/test_pdd_collect_service.py b/client/test/test_pdd_collect_service.py index 52667e7..2a409d2 100644 --- a/client/test/test_pdd_collect_service.py +++ b/client/test/test_pdd_collect_service.py @@ -460,6 +460,55 @@ class DefaultPackageDevice(DefaultSizeDevice): return super()._spec_xml().replace('text="尺码"', 'text="套餐"') +class HorizontalPackageDevice: + """模拟独立横向套餐列表,每屏只有三个且相邻页面有重叠。""" + + packages = tuple(f"套餐{i}" for i in range(1, 8)) + + def __init__(self, expected_count=7): + self.page = 1 + self.expected_count = expected_count + self.swipes = [] + self.clicks = [] + + def dump_hierarchy(self): + start = min(self.page * 2, len(self.packages) - 3) + visible = self.packages[start : start + 3] + options = [] + for index, text in enumerate(visible): + left = 36 + index * 320 + options.append( + f'' + ) + return f""" + + + + + + + + {''.join(options)} + + + + """ + + def swipe(self, x1, y1, x2, y2, duration=0.35): + self.swipes.append((x1, y1, x2, y2, duration)) + if abs(x2 - x1) <= abs(y2 - y1): + return + if x1 > x2: + self.page = min(self.page + 1, 2) + else: + self.page = max(self.page - 1, 0) + + class FakeClock: """测试用时钟:sleep 只推进虚拟时间,不真的等待。""" @@ -1171,6 +1220,44 @@ class PddCollectParserTest(unittest.TestCase): self.assertEqual(data["dimensions"][1]["key"], "size") self.assertEqual(data["dimensions"][1]["name"], "套餐") + def test_horizontal_packages_are_scanned_from_left_to_right_without_clicks(self): + device = HorizontalPackageDevice() + 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, "套餐(7)") + self.assertEqual( + [value.text for value in result.values], + list(device.packages), + ) + self.assertEqual(device.clicks, []) + self.assertTrue( + all(abs(x2 - x1) > abs(y2 - y1) for x1, y1, x2, y2, _ in device.swipes) + ) + + def test_package_heading_count_rejects_partial_collection(self): + device = HorizontalPackageDevice(expected_count=8) + service = PddCollectService( + PddDeviceService(lambda _serial: device), + "USB-001", + "client-001", + sleeper=lambda _seconds: None, + ) + + with self.assertRaises(PddCollectError) as raised: + service._collect_size_dimension(device) + + self.assertEqual(raised.exception.code, "PDD_DATA_SPEC_INCOMPLETE") + self.assertEqual(raised.exception.diagnostics["expected_count"], 8) + self.assertEqual(raised.exception.diagnostics["collected_count"], 7) + 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)