Coverage for src/crawler/by_source/elibm_crawler.py: 52%
247 statements
« prev ^ index » next coverage.py v7.6.4, created at 2025-01-15 14:09 +0000
1from collections import OrderedDict
3from bs4 import BeautifulSoup
4from ptf.model_data import (
5 IssueData,
6 create_abstract,
7 create_articledata,
8 create_contributor,
9 create_issuedata,
10 create_subj,
11)
13from crawler.base_crawler import BaseCollectionCrawler
14from crawler.utils import add_pdf_link_to_xarticle
17class ElibmCrawler(BaseCollectionCrawler):
18 source_name = "Electronic Library of Mathematics"
19 source_domain = "ELIBM"
20 source_website = "https://www.elibm.org"
22 def __init__(self, *args, **kwargs):
23 super().__init__(*args, **kwargs)
24 if self.collection_id == "DOCMA":
25 self.delimiter_inline_formula = "\\("
26 self.delimiter_disp_formula = "\\["
28 def parse_collection_content(self, content):
29 """
30 Parse the HTML page of Annals of Math and returns a list of xissue.
31 Each xissue has its pid/volume/number/year metadata + its url
33 self.periode is set at the end based on the xissue years of the HTML page
34 """
35 soup = BeautifulSoup(content, "html.parser")
36 xissues = []
38 # Extract the list of issues
39 link_nodes = soup.find_all("a")
41 # eLibM puts special issue titles as volume number
42 # to create a issue pid, we use S1, S2...
43 last_special_issue_number = 0
45 for link_node in link_nodes:
46 url = link_node.get("href")
47 text = link_node.get_text()
48 if url.startswith("/issue"):
49 xissue, last_special_issue_number = self.create_elibm_xissue(
50 url, text, last_special_issue_number
51 )
53 # eLibM lists the special issues at the end.
54 # set the periode_begin if we find a special issue
55 if last_special_issue_number == 1: 55 ↛ 56line 55 didn't jump to line 56 because the condition on line 55 was never true
56 self.periode_begin = self.get_first_year(xissues[-1].year)
58 if xissue: 58 ↛ 45line 58 didn't jump to line 45 because the condition on line 58 was always true
59 xissues.append(xissue)
61 self.periode_end = self.get_first_year(xissues[0].year)
63 if last_special_issue_number == 0: 63 ↛ 66line 63 didn't jump to line 66 because the condition on line 63 was always true
64 self.periode_begin = self.get_first_year(xissues[-1].year)
66 self.periode = self.get_or_create_periode()
68 return xissues
70 def get_first_year(self, year):
71 if "/" in year: 71 ↛ 72line 71 didn't jump to line 72 because the condition on line 71 was never true
72 year = year.split("/")[0]
74 return year
76 def create_elibm_xissue(self, url, text, last_special_issue_number):
77 if "(" not in text or ")" not in text: 77 ↛ 78line 77 didn't jump to line 78 because the condition on line 77 was never true
78 return None, None
80 parts = text.split("(")
82 year = parts[1].split(")")[0]
83 year = year.replace("/", "-")
85 # volume might not be an integer. eLibM puts special issue titles as volume number.
86 volume = parts[0].strip()
88 number = ""
89 if "No. " in volume:
90 parts = volume.split("No. ")
91 volume = parts[0].strip()
92 number = parts[1].strip()
94 try:
95 volume_for_pid = int(volume)
96 except ValueError:
97 last_special_issue_number += 1
98 volume_for_pid = f"S{last_special_issue_number}"
100 xissue = create_issuedata()
101 xissue.pid = f"{self.collection_id}_{year}__{volume_for_pid}_{number}"
102 xissue.year = year
103 xissue.volume = volume
104 xissue.number = number
105 xissue.url = self.source_website + url
107 return xissue, last_special_issue_number
109 def parse_issue_content(self, content, xissue):
110 soup = BeautifulSoup(content, "html.parser")
111 article_nodes = soup.find_all("div", {"class": "title"})
113 for index_article, article_node in enumerate(article_nodes):
114 article_link_node = article_node.find("a")
115 if article_link_node: 115 ↛ 113line 115 didn't jump to line 113 because the condition on line 115 was always true
116 url = article_link_node.get("href")
117 xarticle = create_articledata()
118 xarticle.pid = "a" + str(index_article)
119 xarticle.url = self.source_website + url
121 # eLibM lists the articles in the reverse order, except for one special issue
122 if xissue.volume == "Mahler Selecta": 122 ↛ 123line 122 didn't jump to line 123 because the condition on line 122 was never true
123 xissue.articles.append(xarticle)
124 else:
125 xissue.articles.insert(0, xarticle)
127 # if the issue has only 1 article, eLibM skip the issue page and directly display the article page
128 if len(xissue.articles) == 0:
129 title_node = soup.find("h2", {"class": "document_title"})
130 if title_node is not None: 130 ↛ exitline 130 didn't return from function 'parse_issue_content' because the condition on line 130 was always true
131 xarticle = create_articledata()
132 xarticle.pid = "a0"
133 xarticle.url = xissue.url
135 xissue.articles.append(xarticle)
137 def parse_article_content(self, content, xissue, xarticle, url, pid):
138 """
139 Parse the content with Beautifulsoup and returns an ArticleData
140 """
141 xarticle = create_articledata()
142 xarticle.pid = pid
143 xarticle.lang = "en"
145 soup = BeautifulSoup(content, "html.parser")
147 # TITLE
148 title_node = soup.find("h2", {"class": "document_title"})
149 if title_node: 149 ↛ 153line 149 didn't jump to line 153 because the condition on line 149 was always true
150 xarticle.title_tex = title_node.get_text()
152 # AUTHORS
153 citation_author_node = soup.find("h3", {"class": "document_author"})
154 if citation_author_node: 154 ↛ 173line 154 didn't jump to line 173 because the condition on line 154 was always true
155 text = citation_author_node.get_text()
156 if text: 156 ↛ 173line 156 didn't jump to line 173 because the condition on line 156 was always true
157 parts = text.split(";")
158 for part in parts:
159 text_author = part.strip()
161 role = "author"
162 if "(ed.)" in text_author:
163 role = "editor"
164 text_author = text_author.split("(ed.)")[0].strip()
166 author = create_contributor()
167 author["role"] = role
168 author["string_name"] = text_author
170 xarticle.contributors.append(author)
172 # PDF
173 link_nodes = soup.find_all("a")
174 for link_node in link_nodes:
175 url = link_node.get("href")
176 if url.startswith("/ft/"):
177 pdf_url = self.source_website + url
178 add_pdf_link_to_xarticle(xarticle, pdf_url)
180 panel_nodes = soup.find_all("h3", {"class": "panel-title"})
181 for panel_node in panel_nodes:
182 text = panel_node.get_text()
183 content_node = panel_node.parent.parent.find("div", {"class": "panel-body"})
185 if text == "Summary":
186 # ABSTRACT
187 abstract = content_node.get_text()
188 xabstract = create_abstract(tag="abstract", value_tex=abstract, lang=xarticle.lang)
189 xarticle.abstracts.append(xabstract)
191 elif text == "Mathematics Subject Classification":
192 # MSC
193 subjs = content_node.get_text().split(", ")
194 for subj in subjs:
195 subject = create_subj()
196 subject["value"] = subj
197 subject["type"] = "msc"
198 subject["lang"] = "en"
199 xarticle.kwds.append(subject)
201 elif text == "Keywords/Phrases":
202 # Keywords
203 subjs = content_node.get_text().split(", ")
204 for subj in subjs:
205 subject = create_subj()
206 subject["value"] = subj
207 subject["lang"] = "en"
208 xarticle.kwds.append(subject)
210 # PAGES
211 citation_node = soup.find("h5", {"class": "document_source"})
212 if citation_node: 212 ↛ 231line 212 didn't jump to line 231 because the condition on line 212 was always true
213 text = citation_node.get_text()
214 year = f"({xissue.year})"
215 if year in text: 215 ↛ 231line 215 didn't jump to line 231 because the condition on line 215 was always true
216 text = text.split(year)[0]
218 if "p." in text:
219 text = text.split("p.")[0].split(",")[-1].strip()
220 xarticle.size = text
222 elif "-" in text:
223 parts = text.split("-")
224 first_page = parts[-2].split(" ")[-1]
225 last_page = parts[-1].split(",")[0].split(" ")[0]
227 xarticle.fpage = first_page
228 xarticle.lpage = last_page
230 # DOI
231 doi_node = citation_node.next_sibling
232 if doi_node.name == "div": 232 ↛ 233line 232 didn't jump to line 233 because the condition on line 232 was never true
233 text = doi_node.get_text()
234 if text.startswith("DOI: "):
235 doi = text[5:]
237 xarticle.doi = doi
238 xarticle.pid = doi.replace("/", "_").replace(".", "_").replace("-", "_")
240 return xarticle
242 def crawl_collection(self):
243 if self.source is None:
244 raise RuntimeError("ERROR: the source is not set")
246 content = self.download_file(self.collection_url)
247 xissues = self.parse_collection_content(content)
249 """
250 Some collections split the same volumes in different pages
251 Ex: Volume 6 (2000) and Volume 6 (1999)
252 We merge the 2 xissues with the same volume number => Volume 6 (1999-2000)
253 """
254 xissues_dict = self.merge_xissues(xissues)
256 filtered_xissues = xissues_dict
257 # Filter the issues to crawl if start_pid was set in the constructor
258 if self.start_pid is not None:
259 filtered_xissues = {}
260 start = False
261 for pid in xissues_dict:
262 if pid == self.start_pid:
263 start = True
264 if start:
265 filtered_xissues[pid] = xissues_dict[pid]
267 return filtered_xissues
269 def merge_xissues(self, xissues: list[IssueData]):
270 """
271 Some collections split the same volumes in different pages
272 Ex: Volume 6 (2000) and Volume 6 (1999)
273 We merge the 2 xissues with the same volume number => Volume 6 (1999-2000)
274 """
276 merged_xissues = OrderedDict()
278 for xissue in xissues:
279 xissues_with_same_volume = [
280 item
281 for item in xissues
282 if xissue.volume == item.volume
283 and xissue.number == item.number
284 and xissue.vseries == item.vseries
285 and (item.volume or item.number)
286 ]
288 if len(xissues_with_same_volume) < 2:
289 if xissue.pid is None:
290 raise ValueError("Issue does not have a PID")
291 merged_xissues[xissue.pid] = {"issues": [xissue]}
292 first_issue = xissue
293 year = xissue.year
294 else:
295 first_issue = xissues_with_same_volume[0]
296 volume = xissues_with_same_volume[0].volume
297 number = xissues_with_same_volume[0].number
298 vseries = xissues_with_same_volume[0].vseries
300 # Compute the year based on all issues with the same volume/number
301 begin = end = year = xissues_with_same_volume[0].year
302 if not year:
303 raise ValueError("year is not defined")
305 if "-" in year:
306 parts = year.split("-")
307 begin = parts[0]
308 end = parts[1]
310 for xissue_with_same_volume in xissues_with_same_volume[1:]:
311 new_begin = new_end = xissue_with_same_volume.year
313 if not xissue_with_same_volume.year:
314 raise ValueError("xissue year is not defined")
316 if "-" in xissue_with_same_volume.year:
317 parts = year.split("-")
318 new_begin = parts[0]
319 new_end = parts[1]
321 if begin is None or end is None or new_begin is None or new_end is None:
322 continue
323 begin_int = int(begin)
324 end_int = int(end)
325 new_begin_int = int(new_begin)
326 new_end_int = int(new_end)
328 if new_begin_int < begin_int:
329 begin = new_begin
330 if new_end_int > end_int:
331 end = new_end
333 if begin != end:
334 year = f"{begin}-{end}"
335 else:
336 year = begin
338 # We can now set the real pid
339 pid = f"{self.collection_id}_{year}_{vseries}_{volume}_{number}"
340 for issue in xissues_with_same_volume:
341 issue.pid = pid
343 if pid not in merged_xissues:
344 merged_xissues[pid] = {
345 "issues": xissues_with_same_volume,
346 }
348 # We can set the year only for the first xissue because it is the one used to collect
349 # all the articles.
350 # See crawl_issue with merged_xissue = self.crawl_one_issue_url(xissues_to_crawl[0])
351 # But we need to use a separate variable (merged_year) because parse_article_content may rely on the year
352 first_issue.merged_year = year
354 return merged_xissues
356 def crawl_issue(self, merged_xissues: dict[str, list[IssueData]]):
357 """
358 Wrapper around crawl_elibm_issue, to handle issues declared in multiple web pages.
359 """
361 xissues_to_crawl = merged_xissues["issues"]
363 merged_xissue = xissues_to_crawl[0]
364 self.crawl_elibm_issue(merged_xissue)
366 if len(xissues_to_crawl) > 1:
367 for raw_xissue in xissues_to_crawl[1:]:
368 self.crawl_elibm_issue(raw_xissue)
370 merged_xissue.articles = raw_xissue.articles + merged_xissue.articles
372 # Updates the article pid
373 for article_index, xarticle in enumerate(merged_xissue):
374 if raw_xissue.pid in xarticle.pid:
375 xarticle.pid = f"{raw_xissue.pid}_a{str(article_index)}"
377 # Now that the issue pages have been downloaded/read, we can set the merged pid
378 # The merged_year was set in self.merge_xissues
379 # merged_xissue.pid
380 merged_xissue.year = merged_xissue.merged_year
382 if not self.test_mode and len(merged_xissue.articles) > 0:
383 self.add_xissue_into_database(merged_xissue)
385 def crawl_elibm_issue(self, xissue: IssueData):
386 """
387 Crawl 1 wag page of an issue.
388 - get the HTML content of the issue
389 - parse the HTML content with beautifulsoup to extract the list of articles and/or the issue metadata
390 - crawl each article
391 """
393 # Some source, like EuDML do not have a separate HTML pages for an issue's table of content.
394 # The list of articles directly come from the collection HTML page: the xissue has no url attribute
395 if hasattr(xissue, "url") and xissue.url:
396 content = self.download_file(xissue.url)
397 self.parse_issue_content(content, xissue)
399 xarticles = xissue.articles
401 parsed_xarticles = []
403 for xarticle in xarticles:
404 parsed_xarticle = self.crawl_article(xarticle, xissue)
405 if parsed_xarticle is not None:
406 parsed_xarticles.append(parsed_xarticle)
408 xissue.articles = parsed_xarticles