Coverage for src / crawler / by_source / cup_crawler.py: 10%
189 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-07-15 13:33 +0000
1import logging
2import re
3from urllib.parse import urljoin
5from bs4 import BeautifulSoup, Tag
6from ptf.cmds.xml.xml_utils import escape
7from ptf.model_data import create_abstract, create_articledata, create_contributor
9from crawler.abstract_crawlers.matching_crawler import MatchingCrawler
10from crawler.cmds.mixed_citation import (
11 ExtLinkXml,
12 GenericRefElement,
13 MixedCitation,
14)
15from crawler.utils import cleanup_str, regex_to_dict
17logger = logging.getLogger(__name__)
20class CupCrawler(MatchingCrawler):
21 source_name = "Cambridge University Press"
22 source_domain = "CUP"
23 source_website = "https://www.cambridge.org/core/"
25 issue_re = r"Issue (?P<issue>\S+)"
26 issue_error_re = r"Volume (?P<issue_nb>\d+)"
27 volume_re = r"Volume (?P<volume>\d+)"
28 archive_volume_re = r"Vol (?P<volume>\d+)"
29 archive_year_re = r"Archive content \n\n\n (?P<year>\S+)"
31 pid_year_restrictions = {
32 "GLMJ": 6,
33 "CJM": 6,
34 "CMB": 6,
35 }
37 def parse_collection_content(self, content):
38 xissues = []
39 soup = BeautifulSoup(content, "html.parser")
41 volumes_tag = soup.select(
42 "div.journal-all-issues > ul > li > div.content > ul.accordion > li.accordion-navigation"
43 )
44 for volume_tag in volumes_tag:
45 issue_defaut_nb = "1"
46 volume = volume_tag.select_one("a")
47 if volume is None:
48 raise ValueError("Couldn't parse volume tag")
50 try:
51 volume_group = regex_to_dict(
52 self.volume_re, volume.text, error_msg="Couldn't parse volume number"
53 )
54 except ValueError:
55 try:
56 volume_group = regex_to_dict(
57 self.archive_volume_re,
58 volume.text,
59 error_msg="Couldn't parse volume number",
60 )
61 except ValueError:
62 raise ValueError(f"Couldn't parse volume number from text: {volume.text}")
64 issues_tag = volume_tag.select("div > ul > li > ul > li > a")
66 ## If no issue listed : we consider the volume has only one issue
67 if not issues_tag:
68 issue_href = volume.get("href")
69 year_span = volume.select_one("span.date")
70 if not year_span:
71 raise ValueError("Couldn't parse year for volume with no issue")
72 year = year_span.text.split(" ")[-1]
73 xissues.append(
74 self.create_xissue(
75 urljoin(self.source_website, issue_href),
76 int(year),
77 volume_group.get("volume"),
78 "1",
79 )
80 )
81 continue
83 # Get all the volume listed issues
84 for issue_tag in issues_tag:
85 issue_nb, issue_href, issue_year, issue_defaut_nb = self.get_issue_data(
86 issue_tag, issue_defaut_nb
87 )
88 # # Cambridge has declared articles younger than 5 not as open access
89 # if issue_year < current_year:
90 xissues.append(
91 self.create_xissue(
92 urljoin(self.source_website, issue_href),
93 issue_year,
94 volume_group.get("volume"),
95 issue_nb,
96 )
97 )
98 return xissues
100 def get_issue_data(self, issue_tag, default_issue_nb):
101 """
102 Get issue number in classic case but also in the special case of volume 27 with no issue number (defaults to issue 1)
103 """
104 year_span = issue_tag.select_one("span.date")
105 if not year_span:
106 raise ValueError("Couldn't parse year for issue")
107 year = int(year_span.text.split(" ")[-1])
109 issue_href = issue_tag.get("href")
110 if not isinstance(issue_href, str):
111 raise ValueError("Couldn't parse issue href")
113 try:
114 issue = regex_to_dict(
115 self.issue_re, issue_tag.text, error_msg="Couldn't parse issue number"
116 )
117 except ValueError:
118 try:
119 issue = regex_to_dict(
120 self.issue_error_re, issue_tag.text, error_msg="Couldn't parse issue number"
121 )
122 except ValueError:
123 raise ValueError(f"Couldn't parse issue number from text: {issue_tag.text}")
125 issue_nb = issue.get("issue")
126 return issue_nb, issue_href, year, default_issue_nb
128 def parse_issue_content(self, content, xissue):
129 soup = BeautifulSoup(content, "html.parser")
130 articles = soup.select("div.representation")
131 article_number = 0
132 for article in articles:
133 if (
134 article.select_one(".access-modal > .status.open-access > .icon.open-access")
135 is None
136 ):
137 logger.debug("Article is not open access. skipping.")
138 continue
139 xarticle = create_articledata()
140 article_href = article.select_one("a.part-link").get("href")
141 if not isinstance(article_href, str):
142 raise ValueError("Couldn't parse article href")
143 xarticle.url = urljoin(self.source_website, article_href)
144 xarticle.pid = "a" + str(article_number)
145 xissue.articles.append(xarticle)
146 article_number += 1
148 has_pagination = soup.select_one("ul.pagination a:-soup-contains-own('Next »')")
149 if has_pagination:
150 pagination_link = has_pagination.get("href")
151 if isinstance(pagination_link, str):
152 page_url = urljoin(xissue.url, pagination_link)
153 content = self.download_file(page_url)
155 self.parse_issue_content(content, xissue)
157 def parse_article_content(self, content, xissue, xarticle, url):
158 soup = BeautifulSoup(content, "html.parser")
160 self.get_metadata_using_citation_meta(
161 xarticle,
162 xissue,
163 soup,
164 [
165 "pdf",
166 "page",
167 "doi",
168 "publisher",
169 "keywords",
170 "references",
171 ],
172 )
174 ## Title
175 title_tag = soup.select_one("hgroup > h1")
176 if title_tag is None:
177 raise ValueError(f"Couldn't parse article title for article with url: {xarticle.url}")
178 xarticle.title_tex = cleanup_str(title_tag.text)
180 ## Abstract
181 abstract_tag = soup.select_one("div.abstract")
183 if abstract_tag:
184 abstract = cleanup_str(abstract_tag.text)
185 xarticle.abstracts.append(create_abstract(value_tex=abstract, lang=xarticle.lang))
186 else:
187 logger.info(f"No abstract found for article with url: {xarticle.url}")
189 ## keywords
190 keywords_tag = soup.select_one("div.keywords")
191 keywords = keywords_tag.select("span") if keywords_tag else []
192 for keyword in keywords:
193 xarticle.kwds.append(
194 {"type": "", "lang": xarticle.lang, "value": cleanup_str(keyword.text)}
195 )
197 ## Contributors name doi email
198 self.parse_cup_contributors(soup, xarticle)
200 references_list = soup.select_one("#references-list")
201 if references_list:
202 xarticle.bibitems = self.parse_cambridge_references(references_list)
203 return xarticle
205 def parse_cup_contributors(self, soup, xarticle):
206 # Fetch ORCIDs [Name, ORCID]
207 contributors = soup.select_one("div.contributors-details")
208 if not contributors:
209 raise ValueError("Couldn't parse contributors")
211 orcid_by_name = {}
212 for orcid_link in contributors.find_all("a", {"data-test-orcid": True}):
213 name = orcid_link["data-test-orcid"]
214 href = orcid_link.get("href", "")
215 orcid_id = href.rstrip("/").split("/")[-1] if href else None
216 orcid_by_name[name] = orcid_id
218 # Fetch Emails [Name, Email]
219 email_by_name = {}
220 for corresp in contributors.find_all(class_="corresp"):
221 mailto = corresp.find("a", href=re.compile(r"^mailto:"))
222 if mailto:
223 email = mailto["href"].replace("mailto:", "")
224 # Le nom du correspondant est souvent juste avant dans le texte
225 # On cherche dans les blocs .author le lien corresp
226 email_by_name["__corresp__"] = email # sera affiné ci-dessous
228 # Fetch Authors
229 for author_block in contributors.find_all(attrs={"data-test-author": True}):
230 string_name = author_block["data-test-author"]
232 # Split name into first and last name
233 parts = string_name.strip().split()
234 if len(parts) >= 2:
235 first_name = " ".join(parts[:-1])
236 last_name = parts[-1]
237 else:
238 first_name = ""
239 last_name = string_name
241 # ORCID
242 orcid = orcid_by_name.get(string_name)
244 # Email
245 email = ""
246 mailto_tag = author_block.find("a", href=re.compile(r"^mailto:"))
247 if mailto_tag:
248 email = mailto_tag["href"].replace("mailto:", "")
250 xarticle.contributors.append(
251 create_contributor(
252 role="author",
253 string_name=string_name,
254 first_name=first_name,
255 last_name=last_name,
256 orcid=orcid,
257 email=email,
258 )
259 )
260 return xarticle
262 def parse_cambridge_references(self, soup: Tag):
263 bibitems = []
264 for item in soup.select(".circle-list__item"):
265 citation_builder = MixedCitation()
266 label_tag = item.select_one(".circle-list__item__number")
267 if label_tag:
268 citation_builder.label = escape(cleanup_str(label_tag.text))
269 citation_content = item.select_one(".circle-list__item__grouped__content")
270 if citation_content:
271 self.parse_cambridge_ref_nodes(citation_content, citation_builder)
273 # Group all StringNames into one PersonGroup object
274 persongroup_builder = GenericRefElement()
275 persongroup_builder.name = "person-group"
276 # Index of StringNames objects
277 i = [
278 index
279 for index, element in enumerate(citation_builder.elements)
280 if isinstance(element, GenericRefElement) and element.name == "string-name"
281 ]
282 if len(i) > 0:
283 persongroup_builder.elements = citation_builder.elements[i[0] : i[-1] + 1]
284 del citation_builder.elements[i[0] : i[-1] + 1]
285 citation_builder.elements.insert(i[0], persongroup_builder)
287 bibitems.append(citation_builder.get_jats_ref())
288 return bibitems
290 def parse_cambridge_ref_nodes(
291 self,
292 current_tag: Tag,
293 current_builder: GenericRefElement,
294 ):
295 "recursive function that parses references tags"
296 for element in current_tag.children:
297 if isinstance(element, str):
298 current_builder.elements.append(escape(element))
299 continue
300 if isinstance(element, Tag):
301 tag_class = element.get("class")
302 if isinstance(tag_class, list):
303 if len(tag_class) > 0:
304 tag_class = tag_class[0]
305 else:
306 tag_class = None
308 if not tag_class:
309 continue
310 if tag_class in ("mathjax-tex-wrapper", "aop-lazy-load-image"):
311 continue
312 if element.name == "a":
313 href = element.get("href")
314 if isinstance(href, str):
315 current_builder.elements.append(" ")
316 current_builder.elements.append(
317 ExtLinkXml(escape(href), escape(element.text))
318 )
319 continue
321 if tag_class in [
322 "surname",
323 "given-names",
324 "string-name",
325 "person-group",
326 "publisher-name",
327 "source",
328 "volume",
329 "year",
330 "fpage",
331 "lpage",
332 "article-title",
333 "issue",
334 "chapter-title",
335 "inline-formula",
336 "collab",
337 "alternatives",
338 "italic",
339 "publisher-loc",
340 "roman",
341 "edition",
342 "suffix",
343 ]:
344 refnode_builder = GenericRefElement()
345 refnode_builder.name = tag_class
346 current_builder.elements.append(refnode_builder)
347 self.parse_cambridge_ref_nodes(element, refnode_builder)
348 continue
350 self.logger.warning(f"Couldn't insert tag into mixed citation : {tag_class}")
351 current_builder.elements.append(escape(element.text))