Coverage for src/crawler/by_source/dml_e_crawler.py: 28%

1from urllib.parse import urljoin 

2 

3import regex 

4from bs4 import BeautifulSoup, Tag 

5from ptf.model_data import ( 

6 ArticleData, 

7 IssueData, 

8 create_articledata, 

9 create_contributor, 

10 create_extid, 

11 create_extlink, 

12) 

13 

14from crawler.abstract_crawlers.matching_crawler import MatchingCrawler 

15from crawler.crawler_utils import article_has_source 

16from crawler.models import ExtlinkChecked 

17from crawler.utils import add_pdf_link_to_xarticle 

18 

19 

20class Dml_eCrawler(MatchingCrawler): 

21 """ 

22 DML_E is quite peculiar : 

23 There is no issue page, and articles are separated into "years" instead of volumes/issues. 

24 volume/issue number is stored inside each article page. 

25 In order to being able to parse volume and issue numbers, we must parse the articles before creating volumes and issues. 

26 """ 

27 

28 source_domain = "DML_E" 

29 source_name = "Proyecto DML-E: Biblioteca Digital de Matemáticas " 

30 source_website = "http://dmle.icmat.es/" 

31 

32 # 1987, 1: 1-17 

33 # 1999,19: 1-11 

34 # 2008, 53-62, 

35 # 1963 (1-2): 

36 # 2000, 51 (1): 49-58, 13 Ref. 

37 # 2006, 57 (Extra): 327-342, 10 Ref. 

38 issue_regex = r"\d+,? ?(?:(?P<volume>\d+),? ?)?(?:\((?P<number>[\d\w\-]+)\))?(?:[:,])? ?(?:(?P<page_start>\d+)-(?P<page_end>\d+))?" 

39 

40 def parse_collection_content(self, content): 

41 xissues = [] 

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

43 pagination_elements = soup.select("div.prevnext a") 

44 for page in pagination_elements: 

45 href = page.get("href") 

46 if not isinstance(href, str): 46 ↛ 47line 46 didn't jump to line 47 because the condition on line 46 was never true

47 continue 

48 href = urljoin(self.collection_url, href) 

49 content = self.download_file(href) 

50 xissues = [*xissues, *self.parse_collection_page(content, href)] 

51 

52 return xissues 

53 

54 def parse_collection_page(self, content: str, url: str): 

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

56 xissues = [] 

57 current_year = False 

58 issues_tags = soup.select("a[name], ul.art_info") 

59 for issue_tag in issues_tags: 

60 if issue_tag.name == "a": 

61 current_year = issue_tag.get("name") 

62 if not isinstance(current_year, str): 62 ↛ 63line 62 didn't jump to line 63 because the condition on line 62 was never true

63 raise ValueError("Issue year cannot be parsed") 

64 continue 

65 

66 if not current_year: 66 ↛ 67line 66 didn't jump to line 67 because the condition on line 66 was never true

67 raise ValueError("Issue year not found") 

68 issue = self.create_xissue(url, int(current_year), current_year) 

69 self.parse_issue_tag(issue_tag, issue) 

70 xissues.append(issue) 

71 return xissues 

72 

73 # def parse_issue_content(self, content, xissue): 

74 # pass 

75 

76 def parse_issue_tag(self, tag: Tag, xissue: IssueData): 

77 if not xissue.url: 77 ↛ 78line 77 didn't jump to line 78 because the condition on line 77 was never true

78 raise ValueError("xissue must have an URL") 

79 article_tags = tag.select("li") 

80 for index, art_tag in enumerate(article_tags): 

81 href_tag = art_tag.select_one("a[href]") 

82 if not href_tag: 82 ↛ 83line 82 didn't jump to line 83 because the condition on line 82 was never true

83 raise ValueError("Cannot parse article") 

84 url = href_tag.get("href") 

85 if not isinstance(url, str): 85 ↛ 86line 85 didn't jump to line 86 because the condition on line 85 was never true

86 raise ValueError("Cannot parse Article URL") 

87 url = urljoin(xissue.url, url) 

88 

89 title = href_tag.text 

90 

91 article = create_articledata() 

92 article.title_tex = title 

93 article.url = url 

94 article.pid = "a" + str(index) 

95 xissue.articles.append(article) 

96 

97 def parse_dml_e_article_content(self, content, xissue, xarticle: ArticleData, url, pid): 

98 xarticle.pid = pid 

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

100 table_lines = soup.select("div#centro table tr") 

101 issue_volume: str | None = None 

102 issue_number: str | None = None 

103 for line in table_lines: 

104 header_tag = line.select_one("th") 

105 value_tag = line.select_one("td") 

106 if not value_tag: 

107 raise ValueError("Cannot parse article") 

108 

109 # PDF 

110 if not header_tag: 

111 href_tag = line.select_one("a") 

112 if not href_tag: 

113 raise ValueError("Cannot parse article pdf link") 

114 href = href_tag.get("href") 

115 if not isinstance(href, str): 

116 raise ValueError("Cannot parse article pdf link") 

117 add_pdf_link_to_xarticle(xarticle, self.source_website + href) 

118 continue 

119 

120 # Title 

121 if header_tag.text == "Título español": 

122 xarticle.title_tex = value_tag.text 

123 continue 

124 if header_tag.text == "Título original": 

125 xarticle.title_tex = value_tag.text 

126 continue 

127 if header_tag.text == "Título inglés": 

128 xarticle.title_tex = value_tag.text 

129 continue 

130 

131 # Author 

132 if header_tag.text == "Autor/es": 

133 authors_tags = value_tag.select("a") 

134 for a in authors_tags: 

135 author = create_contributor() 

136 author["role"] = "author" 

137 author["string_name"] = a.text 

138 xarticle.contributors.append(author) 

139 continue 

140 # Page 

141 if header_tag.text == "Publicación": 

142 volume_re = list(regex.finditer(self.issue_regex, value_tag.text)) 

143 if len(volume_re) != 0: 

144 # raise ValueError("Cannot parse Article page") 

145 volume_data = volume_re[0].groupdict() 

146 

147 if volume_data["page_start"] and volume_data["page_end"]: 

148 xarticle.page_range = ( 

149 volume_data["page_start"] + "-" + volume_data["page_end"] 

150 ) 

151 if "volume" in volume_data: 

152 issue_volume = volume_data["volume"] 

153 if "number" in volume_data: 

154 issue_number = volume_data["number"] 

155 else: 

156 raise ValueError("issue volume or number not found") 

157 

158 # LANG 

159 if header_tag.text == "Idioma": 

160 languages = {"Inglés": "en", "Español": "es", "Francés": "fr"} 

161 if value_tag.text in languages: 

162 xarticle.lang = languages[value_tag.text] 

163 

164 if header_tag.text == "Código MathReviews" and value_tag.text.startswith("MR"): 

165 extid = create_extid("mr-item-id", value_tag.text) 

166 xarticle.extids.append(extid) 

167 

168 if header_tag.text == "Código Z-Math" and value_tag.text.startswith("Zbl "): 

169 # Space in zblid... http://dmle.icmat.es/revistas/detalle.php?numero=95 

170 zblid = value_tag.text.removeprefix("Zbl ").strip() 

171 # zblid does not seems to exist http://dmle.icmat.es/revistas/detalle.php?numero=4004 

172 if not zblid.startswith("pre"): 

173 extid = create_extid("zbl-item-id", zblid) 

174 xarticle.extids.append(extid) 

175 

176 return xarticle, issue_volume, issue_number 

177 

178 def crawl_issue(self, xissue: IssueData): 

179 if hasattr(xissue, "url") and xissue.url: 

180 content = self.download_file(xissue.url) 

181 self.parse_issue_content(content, xissue) 

182 

183 dml_e_issues: dict[str, IssueData] = {} 

184 

185 xarticles = xissue.articles 

186 

187 for xarticle in xarticles: 

188 parsed_xarticle, xissue_vol, xissue_number = self.crawl_dml_e_article(xarticle, xissue) 

189 if parsed_xarticle is None: 

190 continue 

191 if xissue_vol or xissue_number: 

192 issue_tag = (xissue_vol or "") + "_" + (xissue_number or "") 

193 else: 

194 issue_tag = xissue.fyear 

195 if not issue_tag: 

196 raise ValueError("issue_tag is None") 

197 if issue_tag not in dml_e_issues: 

198 dml_e_issues[issue_tag] = self.create_xissue( 

199 xissue.url, xissue.fyear, xissue_vol, xissue_number or None 

200 ) 

201 dml_e_issues[issue_tag].articles.append(parsed_xarticle) 

202 

203 for value in dml_e_issues.values(): 

204 if self.ignore_missing_pdf: 

205 value.articles = [a for a in value.articles if self.article_has_pdf(a)] 

206 if self.dry: 

207 return 

208 issue_has_pdf = self.article_has_pdf(value) 

209 if len(value.articles) == 0 and not issue_has_pdf: 

210 continue 

211 for index, article in enumerate(value.articles): 

212 article.pid = f"{value.pid}_a{index}" 

213 self.process_resource_metadata(value, resource_type="issue") 

214 self.add_xissue_into_database(value) 

215 

216 def crawl_dml_e_article(self, xarticle: ArticleData, xissue: IssueData): 

217 parsed_xarticle = xarticle 

218 if not hasattr(xarticle, "url") or not xarticle.url: 

219 raise ValueError("article does not have an url") 

220 # self.progress_bar.text(f"{xarticle.pid} - {xarticle.url}") 

221 

222 content = self.download_file(xarticle.url) 

223 pid = f"{xissue.pid}_{xarticle.pid}" 

224 

225 parsed_xarticle, xissue_vol, xissue_number = self.parse_dml_e_article_content( 

226 content, xissue, xarticle, xarticle.url, pid 

227 ) 

228 

229 if not article_has_source(parsed_xarticle) and parsed_xarticle.url: 

230 ext_link = create_extlink() 

231 ext_link["rel"] = "source" 

232 ext_link["location"] = parsed_xarticle.url 

233 ext_link["metadata"] = self.source_domain 

234 parsed_xarticle.ext_links.append(ext_link) 

235 

236 # The article title may have formulas surrounded with '$' 

237 return self.process_article_metadata(parsed_xarticle), xissue_vol, xissue_number 

238 

239 @classmethod 

240 def check_pdf_link_validity(cls, url, verify=True): 

241 # we overwrite this base_crawler method to manage the links to pdf that are not article pdf. 

242 # Avoid downloading the whole PDF 

243 # CHUNK_SIZE = 100 # number of characters fetched 

244 # If the url contains Movingwall it does not lead to the article 

245 # TODO this should be in the harvest tasks 

246 if "Movingwall" in url: 

247 print("The url does not link to the PDF article because of o moving wall") 

248 return ( 

249 False, 

250 None, 

251 { 

252 "status": ExtlinkChecked.Status.ERROR, 

253 "message": "The url does not link to the PDF article because of a moving wall", 

254 }, 

255 ) 

256 return super().check_pdf_link_validity(url)