Coverage for src/crawler/by_source/episciences_crawlers/episciences_crawler.py: 17%
164 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-09-23 14:47 +0000
1import json
2import logging
3import math
4import os
5import re
6from urllib.parse import parse_qs, urljoin, urlparse
8from bs4 import BeautifulSoup
9from lxml import etree
10from ptf.cmds.xml.jats.jats_parser import JatsArticle
11from ptf.model_data import (
12 create_articledata,
13)
14from pylatexenc.latex2text import LatexNodes2Text
16from crawler.abstract_crawlers.matching_crawler import MatchingCrawler
17from crawler.crawler_utils import get_issue_pid
18from crawler.utils import regex_to_dict
20# TODO : Find another system to save the title information with
21# Maybe a toggle attribute on collection ?
23HAS_TITLES = {"COMPOSIT": False, "ARIMA": True}
25logger = logging.getLogger(__name__)
28# We could improve our data further by augmenting the articles using arxiv
29# (references)
30class EpisciencesCrawler(MatchingCrawler):
31 source_name = "Episciences"
32 source_domain = "EPISCIENCES"
33 view_id = "EPISCIENCES"
34 source_website = "https://www.episciences.org/"
36 episciences_id_re = r"https://api\.episciences\.org/api/volumes\?rvcode=(?P<episciences_id>\w+)&pagination=false"
37 volume_title_re = r"Volume (?P<number>\d+)(?:, Issue (?P<issue>\d+))?(?:, (?P<title>[\w ]+))?"
38 article_volume_match_re = r"(?:volume|vol\.?)\s*(\d+)"
39 headers = {"accept_encoding": "utf-8", "accept": "application/ld+json"}
40 latex_converter: LatexNodes2Text
42 @classmethod
43 def get_view_id(cls):
44 return "EPISCIENCES"
46 def __init__(self, *args, **kwargs):
47 super().__init__(*args, **kwargs)
48 self.latex_converter = LatexNodes2Text(math_mode="verbatim")
49 dict = regex_to_dict(self.episciences_id_re, self.collection_url)
50 self.episciences_id = dict["episciences_id"]
52 def parse_collection_content(self, content):
53 data = json.loads(content)
54 xissues = []
55 for issue in data:
56 # if "special_issue" in issue.get("vol_type"):
57 # continue
58 url = f"https://api.episciences.org/api/volumes/{issue['vid']}?rvcode={self.episciences_id}&pagination=false"
59 issue_content = json.loads(self.download_file(url))
60 # Would a dedicated class be preferable ? Or maybe a smarter crawler overall
61 xissues.append(self.prefetch_episciences_issue(issue_content))
63 xissues.sort(key=lambda x: x.volume)
64 issues_by_volume = {}
65 for issue in xissues:
66 if issue.volume not in issues_by_volume:
67 issues_by_volume[issue.volume] = []
68 issues_by_volume[issue.volume].append(issue)
70 for volume_issues in issues_by_volume.values():
71 year_iterable = []
72 try:
73 year_iterable = [int(i.fyear) for i in volume_issues]
74 except ValueError:
75 pass
76 firstyear = min(year_iterable)
77 for i in volume_issues:
78 i.fyear = firstyear
80 return xissues
82 def process_issue_title(self, xissue: dict, issue_title: str):
83 title_dict = regex_to_dict(
84 self.volume_title_re,
85 issue_title,
86 error_msg="Couldn't parse issue title",
87 )
89 xissue.volume = title_dict["number"]
90 if title_dict["issue"]:
91 xissue.number = title_dict["issue"]
92 return xissue
94 def prefetch_episciences_issue(self, issue: dict):
95 """
96 Episciences doesn't provides issue years.
97 We have to parse the year from the first article of the issue (publication date).
98 """
99 year = None
100 if "vol_year" in issue and issue["vol_year"] is not None:
101 year = int(issue["vol_year"][0:4])
102 elif "year" in issue and issue["year"] is not None:
103 year = int(issue["year"][0:4])
104 else:
105 match = re.search(r"\d{4}", issue["titles"]["en"])
106 if match:
107 year = int(match.group(0))
109 if not year:
110 # Impossible to find issue year from metadata nor title
111 # Fallback : fetch an article in the middle opf the issue and get its publication year
112 mid_art = issue["papers"][math.floor(len(issue["papers"]) / 2)]
113 mid_art_url = urljoin(self.collection_url, mid_art["@id"])
114 mid_art_content = self.download_file(mid_art_url)
115 mid_art_data = json.loads(mid_art_content)
116 year = int(
117 mid_art_data["document"]["journal"]["journal_issue"]["publication_date"]["year"]
118 )
120 # parsed_url = urlparse(self.collection_url)
121 # issue_view_url = f"https://{parse_qs(parsed_url.query)['rvcode'][0]}..org/volume/view/id/{issue['vid']}"
123 issue_title = issue["titles"]["en"]
124 xissue = self.create_xissue(None, year, None, None)
125 xissue.lang = "en"
126 try:
127 xissue = self.process_issue_title(xissue, issue_title)
128 except ValueError:
129 self.logger.error(
130 f"Could not parse volume number from title : {issue_title}. Fallbacking to issue poosition."
131 )
132 xissue.volume = issue_title
133 xissue.title_tex = issue_title
134 # if title_dict["title"]:
135 # xissue.title_tex = title_dict["title"]
137 if self.collection_id in HAS_TITLES:
138 if HAS_TITLES[self.collection_id]:
139 xissue.title_tex = issue_title
140 if "fr" in issue["titles"]:
141 title_trans = issue["titles"]["fr"]
142 xissue.titles.append(
143 self.create_trans_title(
144 xresource_lang=xissue.lang,
145 resource_type="issue",
146 title_str=title_trans,
147 lang="fr",
148 )
149 )
150 xissue.pid = get_issue_pid(self.collection_id, year, xissue.volume, xissue.number)
152 url_parsed = urlparse(self.collection_url)
153 query_string = parse_qs(url_parsed.query)
154 self.prefetch_issue_content(xissue, issue_data=issue)
155 xissue.url = (
156 f"https://{query_string['rvcode'][0]}.episciences.org/volume/view/id/{issue['vid']}"
157 )
158 return xissue
160 def prefetch_issue_content(self, xissue, issue_data):
161 url_parsed = urlparse(self.collection_url)
162 domain = f"{url_parsed.scheme}://{url_parsed.hostname}"
163 for index, paper in enumerate(issue_data["papers"]):
164 xarticle = create_articledata()
165 paper_id = paper["@id"].removeprefix("/api/papers/")
166 xarticle.url = urljoin(domain, f"api/papers/export/{paper_id}/zbjats")
167 xarticle.pid = f"a{index}"
168 xissue.articles.append(xarticle)
170 def parse_issue_content(self, content, xissue):
171 soup = BeautifulSoup(content, "html5lib")
172 for a in xissue.articles:
173 if not a.url:
174 raise ValueError("Could not find article position : no article url")
175 id_search = re.match(
176 r"https://api.episciences.org/api/papers/export/(\d+)/zbjats", a.url
177 )
178 if not id_search:
179 raise ValueError("Could not find article position : article url failed to parse")
180 html_tag = soup.select_one(f"#papers > h3 > a[href='/{id_search.group(1)}']")
181 if not html_tag:
182 raise ValueError(
183 "Could not find article position : article not found in issue html"
184 )
185 a.position = int(next(html_tag.previous_siblings).strip().removesuffix("."))
187 xissue.articles.sort(key=lambda a: a.position)
188 return super().parse_issue_content(content, xissue)
190 def get_article_volume(self, tree, xissue):
191 try:
192 volume = tree.xpath("//volume")[0].text
193 match = re.compile(self.article_volume_match_re, re.IGNORECASE).search(volume)
195 if match:
196 volume = match.group(1)
198 if xissue.number in [None, ""]: # define xissue number when aggregated
199 match = re.search(r"Issues ([\d\-–,\s]+):", tree.xpath("//volume")[0].text)
200 if match:
201 xissue.number = re.sub(r"[,\s–—]+", "-", match.group(1).strip()).strip("-")
202 return volume
203 except IndexError:
204 # no article volume found for HRJ Volume 41
205 return xissue.volume
207 def make_issue_date(self, xarticle, xissue, parsed_xarticle):
208 """
209 For Episcience, Issues date metadata are not relevant, an Issue fdate is retrieved from it's first articles publication date.
210 """
211 article_date_published = parsed_xarticle.date_published
212 if xarticle.pid == xissue.articles[0].pid:
213 xissue.fyear = article_date_published
214 elif article_date_published < xissue.fyear:
215 xissue.fyear = article_date_published
217 def parse_article_content(self, content, xissue, xarticle, url):
218 # we use the zbjats format because it gives the most complete response
219 parser = etree.XMLParser(
220 huge_tree=True, recover=True, remove_blank_text=False, remove_comments=True
221 )
222 dom = etree.fromstring(content.encode("utf-8"), parser)
223 # Fix for invalid self-uri tag
224 xslt = etree.parse(
225 os.path.dirname(os.path.realpath(__file__)) + "/episciences-zbjats.xsl", parser
226 )
227 transform = etree.XSLT(xslt)
228 tree = transform(dom)
230 parsed_xarticle = JatsArticle(tree=tree.getroot())
231 parsed_xarticle.pid = xarticle.pid
232 parsed_xarticle.title_tex = self.latex_converter.latex_to_text(parsed_xarticle.title_tex)
234 abstract_langs = []
235 for abstract in parsed_xarticle.abstracts:
236 if abstract["lang"] in abstract_langs:
237 parsed_xarticle.abstracts.remove(abstract)
238 else:
239 abstract["value_tex"] = self.latex_converter.latex_to_text(abstract["value_tex"])
240 abstract_langs.append(abstract["lang"])
242 volume = self.get_article_volume(tree, xissue)
244 if not parsed_xarticle.date_published:
245 parsed_xarticle.date_published = int(
246 parsed_xarticle.date_published_iso_8601_date_str[0:4]
247 )
249 self.make_issue_date(xarticle, xissue, parsed_xarticle)
251 if volume not in [xissue.title_tex, xissue.volume]:
252 logger.warning(
253 "Article belongs to another issue, but is also part of this special issue. Skipping."
254 )
255 return None
257 return parsed_xarticle