Coverage for src/qdrant_loader_mcp_server/mcp/handlers/search/organize.py: 100%

1from __future__ import annotations 

2 

3import re 

4 

5from ....search.components.models.hybrid import HybridSearchResult 

6 

7 

8def organize_by_hierarchy( 

9 results: list[HybridSearchResult], 

10) -> dict[str, list[HybridSearchResult]]: 

11 hierarchy_groups: dict[str, list[HybridSearchResult]] = {} 

12 for result in results: 

13 file_path_val = getattr(result, "file_path", None) 

14 breadcrumb_text_val = getattr(result, "breadcrumb_text", None) 

15 source_title_val = getattr(result, "source_title", None) 

16 

17 if getattr(result, "source_type", None) == "localfile" and file_path_val: 

18 path_parts = [p for p in re.split(r"[\\/]", str(file_path_val)) if p] 

19 root_title = path_parts[0] if path_parts else "Root" 

20 elif breadcrumb_text_val: 

21 breadcrumb_parts = str(breadcrumb_text_val).split(" > ") 

22 root_title = ( 

23 breadcrumb_parts[0] 

24 if breadcrumb_parts 

25 else (source_title_val or "Root") 

26 ) 

27 else: 

28 root_title = source_title_val or "Root" 

29 

30 if root_title not in hierarchy_groups: 

31 hierarchy_groups[root_title] = [] 

32 hierarchy_groups[root_title].append(result) 

33 

34 for group in hierarchy_groups.values(): 

35 

36 def sort_key(x): 

37 x_file_path = getattr(x, "file_path", None) 

38 x_source_title = getattr(x, "source_title", None) or "" 

39 if getattr(x, "source_type", None) == "localfile" and x_file_path: 

40 folder_depth = ( 

41 len([p for p in re.split(r"[\\/]", str(x_file_path)) if p]) - 1 

42 ) 

43 return (folder_depth, x_source_title) 

44 else: 

45 return (getattr(x, "depth", 0) or 0, x_source_title) 

46 

47 group.sort(key=sort_key) 

48 

49 return hierarchy_groups