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

1""" 

2Libraries domain service. 

3 

4Handles library data fetching and search, plus the library RSS feeds 

5crawled by NTHU-Data-Scraper. 

6""" 

7 

8import re 

9import ssl 

10from datetime import datetime, timedelta 

11from typing import Optional 

12 

13import httpx 

14import truststore 

15from bs4 import BeautifulSoup 

16from pydantic import TypeAdapter 

17 

18from data_api.core.exceptions import ( 

19 UpstreamException, 

20 UpstreamResponseException, 

21) 

22from data_api.core.upstream import live_client, upstream_errors 

23from data_api.data.manager import nthudata 

24from data_api.data.nthudata import NTHUDataManager 

25from data_api.models.libraries import LibraryLostAndFound, LibrarySpace 

26 

27RSS_JSON_PATH = "libraries/rss.json" 

28MISSING_LOST_AND_FOUND_TABLE = "Lost-and-found results table is missing" 

29DEFAULT_HEADERS = { 

30 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3" 

31} 

32SPACE_ADAPTER = TypeAdapter(list[LibrarySpace]) 

33ctx = truststore.SSLContext(ssl.PROTOCOL_TLS_CLIENT) 

34 

35 

36def parse_lost_items(html: str) -> list[dict[str, str]]: 

37 """Parse complete results, distinguishing an empty page from a broken one.""" 

38 soup = BeautifulSoup(html, "html.parser") 

39 table = soup.find("table") 

40 if table is None: 

41 content = soup.find(id="content") 

42 if content is None: 

43 raise UpstreamResponseException(MISSING_LOST_AND_FOUND_TABLE) 

44 heading = content.find("h1") 

45 if heading is None: 

46 raise UpstreamResponseException(MISSING_LOST_AND_FOUND_TABLE) 

47 if "Lost and Found System" in heading.get_text() and re.search( 

48 r"\u76ee\u524d\u7121\u8cc7\u6599\s*!!", content.get_text() 

49 ): 

50 return [] 

51 raise UpstreamResponseException(MISSING_LOST_AND_FOUND_TABLE) 

52 

53 rows = table.find_all("tr") 

54 if not rows: 

55 raise UpstreamResponseException("Lost-and-found table has no header") 

56 titles = [cell.get_text(strip=True) for cell in rows[0].find_all(["td", "th"])] 

57 column_names = set(titles) 

58 if column_names != set(LibraryLostAndFound.model_fields) or len(titles) != len(column_names): 

59 raise UpstreamResponseException("Lost-and-found table has invalid columns") 

60 

61 items = [] 

62 for row in rows[1:]: 

63 cells = [ 

64 re.sub(r"\s+", " ", cell.get_text().strip()) for cell in row.find_all(["td", "th"]) 

65 ] 

66 if len(cells) != len(titles): 

67 raise UpstreamResponseException("Lost-and-found row has invalid cells") 

68 items.append(dict(zip(titles, cells))) 

69 return items 

70 

71 

72class LibrariesService: 

73 """Service for library data operations.""" 

74 

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

76 self.manager = manager 

77 self.client: httpx.AsyncClient | None = None 

78 

79 async def get_space_availability(self) -> list[dict]: 

80 """Get validated live library space data.""" 

81 with upstream_errors(): 

82 async with live_client(self.client, verify=ctx) as client: 

83 response = await client.get( 

84 "https://libsms.lib.nthu.edu.tw/RWDAPI_New/GetDevUseStatus.aspx", 

85 headers=DEFAULT_HEADERS, 

86 ) 

87 response.raise_for_status() 

88 data = response.json() 

89 

90 if not isinstance(data, dict) or not isinstance(data.get("resmsg"), str): 

91 raise UpstreamResponseException("Library space response has no result message") 

92 if data["resmsg"] != "\u6210\u529f": 

93 raise UpstreamException("Library space service reported a failure") 

94 rows = data.get("rows") 

95 if not isinstance(rows, list): 

96 raise UpstreamResponseException("Library space response has no rows list") 

97 SPACE_ADAPTER.validate_python(rows) 

98 return rows 

99 

100 async def get_lost_and_found_items(self) -> list[dict[str, str]]: 

101 """Get all live lost items from the last six months.""" 

102 date_end = datetime.now() 

103 date_start = date_end - timedelta(days=6 * 30) 

104 post_data = { 

105 "place": "0", 

106 "date_start": date_start.strftime("%Y-%m-%d"), 

107 "date_end": date_end.strftime("%Y-%m-%d"), 

108 "catalog": "ALL", 

109 "keyword": "", 

110 "SUMIT": "\u9001\u51fa", 

111 } 

112 with upstream_errors(): 

113 async with live_client(self.client, verify=ctx) as client: 

114 response = await client.post( 

115 "https://adage.lib.nthu.edu.tw/find/search_it.php", 

116 data=post_data, 

117 headers=DEFAULT_HEADERS, 

118 ) 

119 response.raise_for_status() 

120 return parse_lost_items(response.text) 

121 

122 async def get_rss_items(self, rss_type: str) -> tuple[Optional[str], Optional[list[dict]]]: 

123 """Get the items of one library RSS feed.""" 

124 commit_hash, rss_data = await self.manager.get(RSS_JSON_PATH) 

125 return commit_hash, rss_data.get(rss_type) 

126 

127 

128# Global service instance 

129libraries_service = LibrariesService()