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

1"""Campus calendar queries backed by the shared dataset snapshot.""" 

2 

3import hashlib 

4import json 

5from datetime import date 

6from typing import Optional 

7 

8from data_api.data.manager import nthudata 

9from data_api.data.nthudata import NTHUDataManager 

10from data_api.utils.calendars import filter_calendar_events 

11 

12JSON_PATH = "calendars.json" 

13LIBRARY_JSON_PATH = "libraries/calendars.json" 

14LIBRARY_PREFIX = "library-" 

15 

16 

17class CalendarsService: 

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

19 self.manager = manager 

20 

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

22 """Get calendar metadata without events.""" 

23 campus_version, campus = await self._get_data(JSON_PATH) 

24 library_version, libraries = await self._get_data(LIBRARY_JSON_PATH) 

25 # A list spans two snapshots; neither source's version alone identifies it. 

26 commit_hash = None 

27 if campus_version is not None and library_version is not None: 

28 commit_hash = hashlib.sha256( 

29 json.dumps([campus_version, library_version]).encode() 

30 ).hexdigest() 

31 calendars = campus + [self._library_calendar(calendar) for calendar in libraries] 

32 return commit_hash, [ 

33 {key: value for key, value in calendar.items() if key != "events"} 

34 for calendar in calendars 

35 ] 

36 

37 async def _get_data(self, path: str) -> tuple[Optional[str], list[dict]]: 

38 return await self.manager.get(path) 

39 

40 @staticmethod 

41 def _library_calendar(calendar: dict) -> dict: 

42 return { 

43 **calendar, 

44 "id": f"{LIBRARY_PREFIX}{calendar['id']}", 

45 "category": "library", 

46 "source": "NTHU Library", 

47 } 

48 

49 async def get_calendar(self, calendar_id: str) -> tuple[Optional[str], Optional[dict]]: 

50 """Get one calendar, including its events.""" 

51 is_library = calendar_id.startswith(LIBRARY_PREFIX) 

52 path = LIBRARY_JSON_PATH if is_library else JSON_PATH 

53 source_id = calendar_id.removeprefix(LIBRARY_PREFIX) if is_library else calendar_id 

54 commit_hash, calendars = await self._get_data(path) 

55 for calendar in calendars: 

56 if calendar["id"] == source_id: 

57 return commit_hash, self._library_calendar(calendar) if is_library else calendar 

58 return commit_hash, None 

59 

60 async def search_calendar_events( 

61 self, 

62 calendar_id: str, 

63 start: Optional[date] = None, 

64 end: Optional[date] = None, 

65 keyword: Optional[str] = None, 

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

67 """Filter by overlapping local dates and keyword, then sort by start.""" 

68 commit_hash, calendar = await self.get_calendar(calendar_id) 

69 if calendar is None: 

70 return commit_hash, None 

71 events = filter_calendar_events(calendar["events"], start, end, keyword) 

72 return commit_hash, sorted(events, key=lambda event: (event["start"], event["id"])) 

73 

74 async def get_calendar_event( 

75 self, calendar_id: str, event_id: str 

76 ) -> tuple[Optional[str], Optional[dict]]: 

77 """Get a single event within its calendar.""" 

78 commit_hash, calendar = await self.get_calendar(calendar_id) 

79 if calendar is not None: 

80 for event in calendar["events"]: 

81 if event["id"] == event_id: 

82 return commit_hash, event 

83 return commit_hash, None 

84 

85 

86calendars_service = CalendarsService()