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

1""" 

2Dining domain service. 

3 

4Handles business logic for dining data fetching and filtering. 

5""" 

6 

7from datetime import datetime 

8from typing import Optional 

9from zoneinfo import ZoneInfo 

10 

11from thefuzz import fuzz 

12 

13from data_api.data.manager import nthudata 

14from data_api.data.nthudata import NTHUDataManager 

15from data_api.domain.dining import enums 

16from data_api.utils.search import fuzzy_matches 

17 

18JSON_PATH = "dining.json" 

19FUZZY_SEARCH_THRESHOLD = 60 

20 

21 

22def is_restaurant_open(restaurant: dict, day: str) -> bool: 

23 """ 

24 Check if restaurant is open on specified day. 

25 

26 Args: 

27 restaurant: Restaurant data dict 

28 day: Day of week in English lowercase 

29 

30 Returns: 

31 True if restaurant may be open, False if definitely closed 

32 """ 

33 note = restaurant.get("note", "").lower() 

34 for keyword in enums.DiningScheduleKeyword.BREAK_KEYWORDS: 

35 for day_zh in enums.DiningScheduleKeyword.DAY_EN_TO_ZH.get(day, []): 

36 if keyword in note and day_zh in note: 

37 return False 

38 return True 

39 

40 

41def _filter_by_schedule(dining_data: list[dict], schedule: Optional[str]) -> list[dict]: 

42 if schedule is None: 

43 return dining_data 

44 

45 day = schedule 

46 if schedule == "today": 

47 weekday = datetime.now(ZoneInfo("Asia/Taipei")).weekday() 

48 day = {5: "saturday", 6: "sunday"}.get(weekday, "weekday") 

49 

50 filtered = [] 

51 for building in dining_data: 

52 restaurants = [ 

53 restaurant 

54 for restaurant in building["restaurants"] 

55 if is_restaurant_open(restaurant, day) 

56 ] 

57 if restaurants: 

58 filtered.append({**building, "restaurants": restaurants}) 

59 return filtered 

60 

61 

62class DiningService: 

63 """Service for dining data operations.""" 

64 

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

66 self.manager = manager 

67 

68 async def get_dining_data( 

69 self, 

70 building_name: Optional[str] = None, 

71 restaurant_name: Optional[str] = None, 

72 schedule: Optional[str] = None, 

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

74 """Get grouped dining data with building, name, and opening-day filters.""" 

75 commit_hash, dining_data = await self.manager.get(JSON_PATH) 

76 

77 filtered = [] 

78 

79 for b in dining_data: 

80 # 篩建築物 

81 if building_name and b["building"] != building_name: 

82 continue 

83 # 篩餐廳 

84 restaurants = b["restaurants"] 

85 if restaurant_name: 

86 restaurants = [r for r in restaurants if restaurant_name in r["name"]] 

87 

88 # 沒餐廳就不要放進去，免得空殼 building 出現 

89 if restaurants: 

90 filtered.append( 

91 { 

92 **b, 

93 "restaurants": restaurants, 

94 } 

95 ) 

96 

97 return commit_hash, _filter_by_schedule(filtered, schedule) 

98 

99 async def get_open_restaurants( 

100 self, 

101 schedule: str, 

102 building_name: Optional[str] = None, 

103 restaurant_name: Optional[str] = None, 

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

105 """Get possibly open restaurants after fuzzy building and name filtering.""" 

106 commit_hash, dining_data = await self.fuzzy_search_dining_data( 

107 building_name=building_name, restaurant_name=restaurant_name, schedule=schedule 

108 ) 

109 

110 return commit_hash, [ 

111 restaurant for building in dining_data for restaurant in building["restaurants"] 

112 ] 

113 

114 async def fuzzy_search_dining_data( 

115 self, 

116 building_name: Optional[str] = None, 

117 restaurant_name: Optional[str] = None, 

118 schedule: Optional[str] = None, 

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

120 """ 

121 Fuzzy search dining data while maintaining the nested structure. 

122 Returns: list[DiningBuilding] 

123 """ 

124 commit_hash, raw_data = await self.manager.get(JSON_PATH) 

125 filtered_results = [] 

126 

127 # 2. 遍歷每一棟建築 

128 for building in raw_data: 

129 # --- Level 1: 建築名稱篩選 --- 

130 # 如果有給 building_name，就先檢查建築名稱是否符合 

131 if building_name: 

132 b_score = fuzz.partial_ratio(building_name, building.get("building", "")) 

133 if b_score < FUZZY_SEARCH_THRESHOLD: 

134 # 建築名稱不符合，直接跳過整棟 

135 continue 

136 

137 # 取得該建築內的所有餐廳 

138 original_restaurants = building.get("restaurants", []) 

139 matched_restaurants = [] 

140 

141 # --- Level 2: 餐廳名稱篩選 --- 

142 if restaurant_name: 

143 # 如果有搜餐廳名，則過濾內部的餐廳 

144 matched_restaurants = fuzzy_matches( 

145 original_restaurants, restaurant_name, "name", FUZZY_SEARCH_THRESHOLD 

146 ) 

147 

148 # 如果這棟樓裡面，沒有任何一家餐廳符合搜尋，這棟樓就不用回傳了 

149 # (除非使用者只搜了建築名，沒搜餐廳名，那下面 else 會處理) 

150 if not matched_restaurants: 

151 continue 

152 

153 else: 

154 # 如果沒有搜餐廳名 (只搜建築)，則保留該建築內所有餐廳 

155 matched_restaurants = original_restaurants 

156 

157 # 3. 重組資料結構 

158 # 複製一份建築資料，避免改到快取 

159 new_building = building.copy() 

160 new_building["restaurants"] = matched_restaurants 

161 

162 filtered_results.append(new_building) 

163 

164 return commit_hash, _filter_by_schedule(filtered_results, schedule) 

165 

166 

167# Global service instance 

168dining_service = DiningService()