Coverage for src/data_api/domain/departments/services.py: 100%

1""" 

2Departments domain service. 

3 

4Handles department and personnel directory data. 

5""" 

6 

7from typing import Optional 

8 

9from thefuzz import fuzz 

10 

11from data_api.data.manager import nthudata 

12from data_api.data.nthudata import NTHUDataManager 

13 

14JSON_PATH = "directory.json" 

15FUZZY_SEARCH_THRESHOLD_DEPT = 80 

16FUZZY_SEARCH_THRESHOLD_PERSON = 80 

17FUZZY_SEARCH_THRESHOLD_PERSON_TITLE = 90 

18 

19 

20def _department_match_score(query: str, department: dict, fuzzy: bool) -> int: 

21 if not fuzzy: 

22 return 100 if query == department["name"] else 0 

23 score = fuzz.partial_ratio(query, department["name"]) 

24 return score if score >= FUZZY_SEARCH_THRESHOLD_DEPT else 0 

25 

26 

27def _person_match_score(query: str, person: dict, fuzzy: bool) -> int: 

28 if not fuzzy: 

29 return 100 if query == person["name"] or query == person.get("title") else 0 

30 name_score = fuzz.partial_ratio(query, person["name"]) 

31 if name_score >= FUZZY_SEARCH_THRESHOLD_PERSON: 

32 return name_score 

33 title_score = fuzz.partial_ratio(query, person.get("title") or "") 

34 return title_score if title_score >= FUZZY_SEARCH_THRESHOLD_PERSON_TITLE else 0 

35 

36 

37class DepartmentsService: 

38 """Service for department directory operations.""" 

39 

40 def __init__(self, manager: NTHUDataManager = nthudata): 

41 self.manager = manager 

42 

43 async def get_all_departments(self) -> tuple[Optional[str], list[dict]]: 

44 """Get all departments.""" 

45 return await self.manager.get(JSON_PATH) 

46 

47 async def get_directory( 

48 self, query: str | None = None, fuzzy: bool = True 

49 ) -> tuple[Optional[str], list[dict]]: 

50 """Return matching departments, retaining only matching people when needed.""" 

51 commit_hash, directory_data = await self.get_all_departments() 

52 if not query: 

53 return commit_hash, directory_data 

54 

55 results = [] 

56 for department in directory_data: 

57 if department_score := _department_match_score(query, department, fuzzy): 

58 results.append((department_score, department)) 

59 continue 

60 

61 details = department.get("details", {}) 

62 matching_people = [ 

63 (score, person) 

64 for person in details.get("people", []) 

65 if (score := _person_match_score(query, person, fuzzy)) 

66 ] 

67 if matching_people: 

68 results.append( 

69 ( 

70 max(score for score, _ in matching_people), 

71 { 

72 **department, 

73 "details": { 

74 **details, 

75 "people": [person for _, person in matching_people], 

76 }, 

77 }, 

78 ) 

79 ) 

80 

81 results.sort(key=lambda item: item[0], reverse=True) 

82 return commit_hash, [department for _, department in results] 

83 

84 async def search_departments_and_people( 

85 self, query: str, fuzzy: bool = True 

86 ) -> tuple[Optional[str], dict[str, list]]: 

87 """Search departments and people using fuzzy or exact matching.""" 

88 commit_hash, directory_data = await self.manager.get(JSON_PATH) 

89 if not query: 

90 return commit_hash, {"departments": [], "people": []} 

91 

92 dept_results = [] 

93 person_results = [] 

94 

95 for department in directory_data: 

96 # Search departments 

97 if dept_similarity := _department_match_score(query, department, fuzzy): 

98 dept_results.append((dept_similarity, department)) 

99 

100 # Search people in this department 

101 people = department.get("details", {}).get("people", []) 

102 for person in people: 

103 if score := _person_match_score(query, person, fuzzy): 

104 person_results.append((score, person)) 

105 

106 dept_results.sort(key=lambda x: x[0], reverse=True) 

107 person_results.sort(key=lambda x: x[0], reverse=True) 

108 

109 return commit_hash, { 

110 "departments": [dept for _, dept in dept_results], 

111 "people": [person for _, person in person_results], 

112 } 

113 

114 

115# Global service instance 

116departments_service = DepartmentsService()