Coverage for src/crawler/crawler_utils.py: 16%
211 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-29 13:39 +0000
1# This file contains utils functions related to ArticleData or IssueData parsing and population
2# Some of the functions present here were initially present in base_crawler but then moved here.
5import logging
6from collections.abc import Callable
7from email.policy import EmailPolicy
9import regex
10from bs4 import BeautifulSoup
11from langcodes import standardize_tag
12from ptf.cmds.xml.jats.builder.references import (
13 get_article_title_xml,
14 get_author_xml,
15 get_fpage_xml,
16 get_lpage_xml,
17 get_source_xml,
18 get_year_xml,
19)
20from ptf.cmds.xml.jats.jats_parser import parse_mixed_citation_into_ref
21from ptf.model_data import (
22 ArticleData,
23 ContributorDict,
24 IssueData,
25 create_abstract,
26 create_contributor,
27 create_extid,
28 create_issuedata,
29 create_publisherdata,
30)
32from crawler.types import CitationLiteral
33from crawler.utils import add_pdf_link_to_xarticle, cleanup_str
35references_mapping = {
36 "citation_title": get_article_title_xml,
37 "citation_journal_title": get_source_xml,
38 "citation_publication_date": get_year_xml,
39 "citation_firstpage": get_fpage_xml,
40 "citation_lastpage": get_lpage_xml,
41}
43logger = logging.getLogger(__name__)
46def parse_content_type_charset(content_type: str):
47 header = EmailPolicy.header_factory("content-type", content_type)
48 if "charset" in header.params:
49 return header.params.get("charset")
52def parse_meta_citation_reference(content: str, label=None):
53 categories = content.split(";")
55 if len(categories) == 1:
56 return parse_mixed_citation_into_ref(content, label=label)
58 citation_data = [c.split("=") for c in categories if "=" in c]
59 del categories
61 xml_string = ""
62 authors_parsed = False
63 authors_strings = []
64 for data in citation_data:
65 key = data[0].strip()
66 citation_content = data[1]
67 if key == "citation_author":
68 authors_strings.append(get_author_xml(template_str=citation_content))
69 continue
70 elif not authors_parsed:
71 xml_string += ", ".join(authors_strings)
72 authors_parsed = True
74 if key in references_mapping:
75 xml_string += " " + references_mapping[key](citation_content)
77 return parse_mixed_citation_into_ref(xml_string, label=label)
80def set_pages(article: ArticleData, pages: str, separator: str = "-"):
81 pages_split = pages.split(separator)
82 if len(pages_split) == 0: 82 ↛ 83line 82 didn't jump to line 83 because the condition on line 82 was never true
83 article.page_range = pages
84 if len(pages_split) > 0 and pages[0].isnumeric(): 84 ↛ exitline 84 didn't return from function 'set_pages' because the condition on line 84 was always true
85 article.fpage = pages_split[0]
86 if (
87 len(pages_split) > 1
88 and pages_split[0] != pages_split[1]
89 and pages_split[1].isnumeric()
90 ):
91 article.lpage = pages_split[1]
94def get_issue_pid(
95 collection_id: str,
96 year: int,
97 volume_number: str | None = None,
98 issue_number: str | None = None,
99 series: str | None = None,
100):
101 # Replace any non-word character with an underscore
102 pid = f"{collection_id}_{year}"
103 if series is not None: 103 ↛ 104line 103 didn't jump to line 104 because the condition on line 103 was never true
104 pid += f"_{series}"
105 if volume_number is not None: 105 ↛ 107line 105 didn't jump to line 107 because the condition on line 105 was always true
106 pid += f"_{volume_number}"
107 if issue_number is not None: 107 ↛ 109line 107 didn't jump to line 109 because the condition on line 107 was always true
108 pid += f"_{issue_number}"
109 pid = regex.sub(r"[^a-zA-Z0-9-]+", "_", cleanup_str(pid))
110 return pid
113def create_xissue(
114 collection_id: str,
115 url: str | None,
116 year: int,
117 volume_number: str | None,
118 issue_number: str | None = "1",
119 vseries: str | None = None,
120):
121 if url is not None and url.endswith("/"): 121 ↛ 122line 121 didn't jump to line 122 because the condition on line 121 was never true
122 url = url[:-1]
123 xissue = create_issuedata()
124 xissue.url = url
126 xissue.pid = get_issue_pid(collection_id, year, volume_number, issue_number, vseries)
128 xissue.fyear = year
130 if volume_number is not None: 130 ↛ 133line 130 didn't jump to line 133 because the condition on line 130 was always true
131 xissue.volume = regex.sub(r"[^a-zA-Z0-9-]+", "_", volume_number)
133 if issue_number is not None: 133 ↛ 136line 133 didn't jump to line 136 because the condition on line 133 was always true
134 xissue.number = issue_number.replace(",", "-")
136 if vseries is not None: 136 ↛ 137line 136 didn't jump to line 137 because the condition on line 136 was never true
137 xissue.vseries = vseries
138 return xissue
141def get_metadata_using_citation_meta(
142 xarticle: ArticleData,
143 xissue: IssueData,
144 soup: BeautifulSoup,
145 what: list[CitationLiteral] | None = None,
146 detect_language_fct: Callable[[str, ArticleData], str] | None = None,
147):
148 """
149 :param xarticle: the xarticle that will collect the metadata
150 :param xissue: the xissue that will collect the publisher
151 :param soup: the BeautifulSoup object of tha article page
152 :param what: list of citation_ items to collect.
153 :return: None. The given article is modified
154 """
156 if what is None:
157 what = []
158 if "title" in what:
159 # TITLE
160 citation_title_node = soup.select_one("meta[name='citation_title']")
161 if citation_title_node:
162 title = citation_title_node.get("content")
163 if isinstance(title, str):
164 xarticle.title_tex = title
166 if "author" in what:
167 # AUTHORS
168 citation_author_nodes = soup.select("meta[name^='citation_author']")
169 current_author: ContributorDict | None = None
170 for citation_author_node in citation_author_nodes:
171 if citation_author_node.get("name") == "citation_author":
172 text_author = citation_author_node.get("content")
173 if not isinstance(text_author, str):
174 raise ValueError("Cannot parse author")
175 if text_author == "":
176 current_author = None
177 continue
178 current_author = create_contributor(role="author", string_name=text_author)
179 xarticle.contributors.append(current_author)
180 continue
181 if current_author is None:
182 logger.warning("Couldn't parse citation author")
183 continue
184 if citation_author_node.get("name") == "citation_author_institution":
185 text_institution = citation_author_node.get("content")
186 if not isinstance(text_institution, str):
187 continue
188 current_author["addresses"].append(text_institution)
189 if citation_author_node.get("name") == "citation_author_ocrid":
190 text_orcid = citation_author_node.get("content")
191 if not isinstance(text_orcid, str):
192 continue
193 current_author["orcid"] = text_orcid
195 if "pdf" in what:
196 # PDF
197 citation_pdf_node = soup.select_one('meta[name="citation_pdf_url"]')
198 if citation_pdf_node:
199 pdf_url = citation_pdf_node.get("content")
200 if isinstance(pdf_url, str):
201 add_pdf_link_to_xarticle(xarticle, pdf_url)
203 if "lang" in what:
204 # LANG
205 citation_lang_node = soup.select_one("meta[name='citation_language']")
206 if citation_lang_node:
207 # TODO: check other language code
208 content_text = citation_lang_node.get("content")
209 if isinstance(content_text, str):
210 xarticle.lang = standardize_tag(content_text)
212 if "abstract" in what:
213 # ABSTRACT
214 abstract_node = soup.select_one("meta[name='citation_abstract']")
215 if abstract_node is not None:
216 abstract = abstract_node.get("content")
217 if not isinstance(abstract, str):
218 raise ValueError("Couldn't parse abstract from meta")
219 abstract = BeautifulSoup(abstract, "html.parser").text
220 lang = abstract_node.get("lang")
221 if not isinstance(lang, str):
222 if not detect_language_fct:
223 return
224 lang = detect_language_fct(abstract, xarticle)
225 xarticle.abstracts.append(create_abstract(lang=lang, value_tex=abstract))
227 if "page" in what:
228 # PAGES
229 citation_fpage_node = soup.select_one("meta[name='citation_firstpage']")
230 if citation_fpage_node:
231 page = citation_fpage_node.get("content")
232 if isinstance(page, str):
233 page = page.split("(")[0]
234 if len(page) < 32:
235 xarticle.fpage = page
237 citation_lpage_node = soup.select_one("meta[name='citation_lastpage']")
238 if citation_lpage_node:
239 page = citation_lpage_node.get("content")
240 if isinstance(page, str):
241 page = page.split("(")[0]
242 if len(page) < 32:
243 xarticle.lpage = page
245 if "doi" in what:
246 # DOI
247 citation_doi_node = soup.select_one("meta[name='citation_doi']")
248 if citation_doi_node:
249 doi = citation_doi_node.get("content")
250 if isinstance(doi, str):
251 doi = doi.strip()
252 pos = doi.find("10.")
253 if pos > 0:
254 doi = doi[pos:]
255 xarticle.doi = doi
257 if "mr" in what:
258 # MR
259 citation_mr_node = soup.select_one("meta[name='citation_mr']")
260 if citation_mr_node:
261 mr = citation_mr_node.get("content")
262 if isinstance(mr, str):
263 mr = mr.strip()
264 if mr.find("MR") == 0:
265 mr = mr[2:]
266 extid = create_extid("mr-item-id", mr)
267 xarticle.extids.append(extid)
269 if "zbl" in what:
270 # ZBL
271 citation_zbl_node = soup.select_one("meta[name='citation_zbl']")
272 if citation_zbl_node:
273 zbl = citation_zbl_node.get("content")
274 if isinstance(zbl, str):
275 zbl = zbl.strip()
276 if zbl.find("Zbl") == 0:
277 zbl = zbl[3:].strip()
278 extid = create_extid("zbl-item-id", zbl)
279 xarticle.extids.append(extid)
281 if "publisher" in what:
282 # PUBLISHER
283 citation_publisher_node = soup.select_one("meta[name='citation_publisher']")
284 if citation_publisher_node:
285 pub = citation_publisher_node.get("content")
286 if isinstance(pub, str):
287 pub = pub.strip()
288 if pub != "":
289 xpub = create_publisherdata()
290 xpub.name = pub
291 xissue.publisher = xpub
293 if "keywords" in what:
294 # KEYWORDS
295 citation_kwd_nodes = soup.select("meta[name='citation_keywords']")
296 for kwd_node in citation_kwd_nodes:
297 kwds = kwd_node.get("content")
298 if isinstance(kwds, str):
299 kwds = kwds.split(",")
300 for kwd in kwds:
301 if kwd == "":
302 continue
303 kwd = kwd.strip()
304 xarticle.kwds.append({"type": "", "lang": xarticle.lang, "value": kwd})
306 if "references" in what:
307 citation_references = soup.select("meta[name='citation_reference']")
308 for index, tag in enumerate(citation_references):
309 content = tag.get("content")
310 if not isinstance(content, str):
311 raise TypeError("Cannot parse citation_reference meta")
312 label = str(index + 1)
313 if regex.match(r"^\[\d+\].*", content):
314 label = None
315 xarticle.bibitems.append(parse_meta_citation_reference(content, label))
318def article_has_pdf(art: ArticleData | IssueData):
319 return next((link for link in art.ext_links if link["rel"] == "article-pdf"), None) is not None
322def article_has_source(art: ArticleData | IssueData):
323 return (
324 next(
325 (e_link for e_link in art.ext_links if e_link["rel"] == "source"),
326 None,
327 )
328 is not None
329 )