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

1""" 

2Announcements domain service. 

3 

4Simple data fetching and filtering service for announcements. 

5""" 

6 

7import re 

8from typing import Optional 

9 

10from thefuzz import fuzz 

11 

12from data_api.data.manager import nthudata 

13from data_api.data.nthudata import NTHUDataManager 

14from data_api.utils.search import fuzzy_matches 

15 

16# Constants 

17ANNOUNCEMENTS_JSON = "announcements.json" 

18ANNOUNCEMENTS_LIST_JSON = "announcements_list.json" 

19FUZZY_SEARCH_THRESHOLD = 80 

20 

21 

22def _matches_source_url(link: str, url: str) -> bool: 

23 def split_url(value: str) -> tuple[bool, str, str]: 

24 scheme = re.match(r"https?://", value, re.IGNORECASE) 

25 without_scheme = value[scheme.end() :] if scheme else value 

26 parts = re.split(r"([/?#])", without_scheme, maxsplit=1) 

27 return bool(scheme), parts[0], "".join(parts[1:]) 

28 

29 _, link_host, link_tail = split_url(link) 

30 has_scheme, query_host, query_tail = split_url(url) 

31 normalized_link = link_host.casefold() + link_tail 

32 if has_scheme or ( 

33 query_host.casefold() in link_host.casefold() and (not query_tail or "." in query_host) 

34 ): 

35 url = query_host.casefold() + query_tail 

36 return url in normalized_link 

37 

38 

39class AnnouncementsService: 

40 """Service for fetching and filtering announcements.""" 

41 

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

43 self.manager = manager 

44 

45 async def get_announcements( 

46 self, 

47 department: Optional[str] = None, 

48 title: Optional[str] = None, 

49 language: Optional[str] = None, 

50 url: Optional[str] = None, 

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

52 """ 

53 Get announcements with optional filtering. 

54 

55 Returns: 

56 tuple: (commit_hash, filtered_announcements) 

57 """ 

58 commit_hash, announcements_data = await self.manager.get(ANNOUNCEMENTS_JSON) 

59 

60 if url: 

61 announcements_data = [ 

62 announcement 

63 for announcement in announcements_data 

64 if _matches_source_url(announcement["link"], url) 

65 ] 

66 if department: 

67 announcements_data = [ 

68 announcement 

69 for announcement in announcements_data 

70 if announcement["department"] == department 

71 ] 

72 if title: 

73 announcements_data = [ 

74 announcement 

75 for announcement in announcements_data 

76 if title in announcement["title"] 

77 ] 

78 if language: 

79 announcements_data = [ 

80 announcement 

81 for announcement in announcements_data 

82 if announcement.get("language") == language 

83 ] 

84 

85 return commit_hash, announcements_data 

86 

87 async def get_announcements_list( 

88 self, department: Optional[str] = None 

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

90 """Get announcements list (without article content).""" 

91 commit_hash, announcements_list = await self.manager.get(ANNOUNCEMENTS_LIST_JSON) 

92 

93 if department: 

94 announcements_list = [ 

95 announcement 

96 for announcement in announcements_list 

97 if announcement["department"] == department 

98 ] 

99 

100 return commit_hash, announcements_list 

101 

102 @staticmethod 

103 def _matches_fuzzy_source( 

104 source: dict, 

105 department: Optional[str], 

106 language: Optional[str], 

107 url: Optional[str], 

108 ) -> bool: 

109 if url and not _matches_source_url(source["link"], url): 

110 return False 

111 if language and source.get("language") != language: 

112 return False 

113 if department: 

114 dept_name = source.get("department", "") 

115 if fuzz.partial_ratio(department, dept_name) < FUZZY_SEARCH_THRESHOLD: 

116 return False 

117 return True 

118 

119 async def fuzzy_search_announcements( 

120 self, 

121 department: Optional[str] = None, 

122 title: Optional[str] = None, 

123 language: Optional[str] = None, 

124 url: Optional[str] = None, 

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

126 """ 

127 Fuzzy search within the nested structure. 

128 Returns the original structure but with non-matching articles removed. 

129 """ 

130 # 1. 取得原始資料 

131 commit_hash, raw_data = await self.manager.get(ANNOUNCEMENTS_JSON) 

132 

133 filtered_results = [] 

134 

135 # 2. 遍歷每一個處室/來源 

136 for source in raw_data: 

137 if not self._matches_fuzzy_source(source, department, language, url): 

138 continue 

139 

140 # 取出該處室的所有文章 

141 original_articles = source.get("articles", []) 

142 

143 matched_articles = list(original_articles) 

144 if title: 

145 matched_articles = fuzzy_matches( 

146 original_articles, title, "title", FUZZY_SEARCH_THRESHOLD 

147 ) 

148 if not matched_articles: 

149 continue 

150 

151 # 3. 重組資料結構 

152 # 複製一份處室資訊 (避免修改到原始快取)，並替換 articles 

153 new_source = source.copy() 

154 new_source["articles"] = matched_articles 

155 

156 filtered_results.append(new_source) 

157 

158 return commit_hash, filtered_results 

159 

160 

161# Global service instance 

162announcements_service = AnnouncementsService()