Coverage for src/crawler/by_source/impan_crawler.py: 83%

1import time 

2from urllib.parse import urljoin 

3 

4import lingua 

5from bs4 import BeautifulSoup 

6from lingua import LanguageDetectorBuilder 

7from ptf.model_data import ( 

8 IssueData, 

9 create_abstract, 

10 create_articledata, 

11 create_issuedata, 

12 create_subj, 

13) 

14 

15from crawler.abstract_crawlers.matching_crawler import MatchingCrawler 

16from crawler.types import CitationLiteral 

17from crawler.utils import cleanup_str, regex_to_dict 

18 

19 

20class ImpanCrawler(MatchingCrawler): 

21 source_name = "Institute of Mathematics Polish Academy of Sciences" 

22 source_domain = "IMPAN" 

23 source_website = "https://www.impan.pl/" 

24 

25 requests_timeout = 120 

26 _language_detector_builder = LanguageDetectorBuilder.from_languages( 

27 lingua.Language.ENGLISH, 

28 lingua.Language.FRENCH, 

29 lingua.Language.POLISH, 

30 lingua.Language.RUSSIAN, 

31 lingua.Language.GERMAN, 

32 ) 

33 

34 def parse_collection_content(self, content): 

35 """ 

36 Discrete Analysis. 

37 We ignore the journal web page and query Crossref to get the list of articles. 

38 We query crossref for each article to get the list of xissues based on the publication date. 

39 Each xissue has its year + list of articles with their URLs 

40 """ 

41 

42 soup = BeautifulSoup(content, "html.parser") 

43 xissues_dict: dict[str, IssueData] = {} 

44 

45 # Extract the list of issues 

46 volume_nodes = soup.select("div.year") 

47 

48 for volume_node in volume_nodes: 

49 year = volume_node.get_text() 

50 

51 issues_nodes = volume_node.parent 

52 if issues_nodes is None: 52 ↛ 53line 52 didn't jump to line 53 because the condition on line 52 was never true

53 continue 

54 issues_nodes = issues_nodes.select("div.issues") 

55 

56 for issue_node in issues_nodes: 

57 issues_link_node = issue_node.select("a") 

58 for issue_link_node in issues_link_node: 

59 href = self.get_str_attr(issue_link_node, "href") 

60 url = urljoin(self.source_website, href) 

61 

62 xissue = self.create_impan_xissue(url, int(year)) 

63 # Prevent duplicate issues 

64 # NOTE : is this needed ? 

65 pid = xissue.pid 

66 if pid is None: 66 ↛ 67line 66 didn't jump to line 67 because the condition on line 66 was never true

67 continue 

68 if pid in xissues_dict: 

69 self.logger.debug("Duplicate issue in connection", extra={"pid": pid}) 

70 continue 

71 xissues_dict[pid] = xissue 

72 

73 return list(xissues_dict.values()) 

74 

75 def create_impan_xissue(self, url: str, year: int): 

76 url = url.removesuffix("/") 

77 parts = url.split("/") 

78 issue_number = parts[-1].replace(",", "-") 

79 volume_number = parts[-2] 

80 

81 xissue = create_issuedata() 

82 if volume_number == "all": 

83 xissue.pid = f"{self.collection_id}_{year}__{issue_number}" 

84 xissue.volume = issue_number 

85 else: 

86 xissue.pid = f"{self.collection_id}_{year}__{volume_number}_{issue_number}" 

87 xissue.volume = volume_number 

88 xissue.number = issue_number.replace(",", "-") 

89 

90 xissue.fyear = year 

91 

92 xissue.url = url 

93 

94 return xissue 

95 

96 def parse_issue_content(self, content, xissue: IssueData, retries=0): 

97 soup = BeautifulSoup(content, "html.parser") 

98 article_nodes = soup.select("div.info") 

99 

100 # Custom retry system for impan 

101 if len(article_nodes) == 0 and xissue.url: 101 ↛ 102line 101 didn't jump to line 102 because the condition on line 101 was never true

102 if retries > 3: 

103 self.logger.info( 

104 "Maximum number of retires reached. This issue seems to be empty", 

105 {"pid": xissue.pid, "url": xissue.url}, 

106 ) 

107 return 

108 

109 self.logger.debug("Couldn't find articles... Retrying", {"pid": xissue.pid}) 

110 time.sleep(60) 

111 return self.parse_issue_content( 

112 self.download_file(xissue.url), 

113 xissue=xissue, 

114 retries=retries + 1, 

115 ) 

116 

117 for index_article, article_node in enumerate(article_nodes): 

118 xarticle = create_articledata() 

119 xarticle.pid = "a" + str(index_article) 

120 

121 article_link_node = article_node.select_one("a") 

122 if article_link_node is None: 122 ↛ 123line 122 didn't jump to line 123 because the condition on line 122 was never true

123 raise ValueError(f"[{self.source_domain}] {xarticle.pid} : Issue link is None") 

124 

125 href = self.get_str_attr(article_link_node, "href") 

126 

127 xissue_url = xissue.url 

128 if xissue_url is None: 128 ↛ 129line 128 didn't jump to line 129 because the condition on line 128 was never true

129 raise ValueError(f"[{self.source_domain}] {xarticle.pid} : Issue url is None") 

130 xarticle.url = urljoin(xissue_url, href) 

131 

132 xissue.articles.append(xarticle) 

133 

134 def parse_article_content(self, content, xissue, xarticle, url): 

135 """ 

136 Parse the content with Beautifulsoup and returns an ArticleData 

137 """ 

138 

139 # We only parse the arXiv id in the Discrete Analysis article page 

140 soup = BeautifulSoup(content, "html.parser") 

141 

142 title_info = soup.select_one("div.info") 

143 if title_info is None: 143 ↛ 144line 143 didn't jump to line 144 because the condition on line 143 was never true

144 self.logger.error( 

145 f"[{self.source_domain}] {self.collection_id} {xarticle.pid} : Title not found" 

146 ) 

147 return None 

148 title_node = title_info.select_one("a") 

149 if title_node is None: 149 ↛ 151line 149 didn't jump to line 151 because the condition on line 149 was always true

150 title_node = soup.select_one("h2.product-title") 

151 if title_node is None: 151 ↛ 152line 151 didn't jump to line 152 because the condition on line 151 was never true

152 self.logger.error( 

153 f"[{self.source_domain}] {self.collection_id} {xarticle.pid} : Title not found" 

154 ) 

155 return None 

156 

157 title_tex = title_node.get_text() 

158 xarticle.title_tex = title_tex 

159 xarticle.lang = self.detect_language(xarticle.title_tex) 

160 

161 what: list[CitationLiteral] = [ 

162 "author", 

163 "page", 

164 "doi", 

165 "publisher", 

166 "page", 

167 ] 

168 

169 # If download button has the "buy" class, skip adding the pdf. 

170 if not soup.select_one("a.button.download.noborder.buy"): 170 ↛ 173line 170 didn't jump to line 173 because the condition on line 170 was always true

171 what.append("pdf") 

172 

173 self.get_metadata_using_citation_meta(xarticle, xissue, soup, what) 

174 

175 # abstract 

176 abstract_mml_node = soup.select_one("div.details.abstract p") 

177 if abstract_mml_node is None: 

178 self.logger.debug("Abstract not found", extra={"pid": xarticle.pid}) 

179 else: 

180 abstract_tex = abstract_mml_node.get_text() 

181 

182 xarticle.abstracts.append(create_abstract(value_tex=abstract_tex, lang=xarticle.lang)) 

183 

184 # MSC 

185 msc_tag = soup.select_one(".source.msc") 

186 if msc_tag: 186 ↛ 205line 186 didn't jump to line 205 because the condition on line 186 was always true

187 msc_text = cleanup_str(msc_tag.text) 

188 # weird but sometimes the DOI is in the MSC div... 

189 if not msc_text.startswith("DOI: "): 

190 # MSC: Primary 93B25; Secondary 93B11, 93B15, 93B17. 

191 # MSC: 35J30, 35J45, 47A53 

192 regex_dict = regex_to_dict( 

193 r"MSC: (?:Primary )?(?P<primary>[\w, ]+)(?:; Secondary (?P<secondary>[\w, ]+).)?", 

194 msc_text, 

195 ) 

196 

197 msc_list = regex_dict["primary"].split(", ") 

198 

199 if regex_dict["secondary"] is not None: 

200 msc_list.extend(regex_dict["secondary"].split(", ")) 

201 

202 for msc in msc_list: 

203 xarticle.kwds.append(create_subj(type="msc", value=cleanup_str(msc))) 

204 

205 return xarticle