Coverage for src / crawler / by_source / episciences_crawler.py: 13%

154 statements  

« prev     ^ index     » next       coverage.py v7.13.1, created at 2026-07-15 13:33 +0000

1import json 

2import math 

3import os 

4import re 

5from urllib.parse import parse_qs, urljoin, urlparse 

6 

7from bs4 import BeautifulSoup 

8from lxml import etree 

9from ptf.cmds.xml.jats.jats_parser import JatsArticle 

10from ptf.model_data import ( 

11 create_articledata, 

12 create_issuedata, 

13) 

14from pylatexenc.latex2text import LatexNodes2Text 

15 

16from crawler.abstract_crawlers.matching_crawler import MatchingCrawler 

17from crawler.crawler_utils import get_issue_pid 

18from crawler.utils import regex_to_dict 

19 

20# TODO : Find another system to save the title information with 

21# Maybe a toggle attribute on collection ? 

22 

23HAS_TITLES = {"COMPOSIT": False, "ARIMA": True} 

24 

25 

26# We could improve our data further by augmenting the articles using arxiv 

27# (references) 

28class EpisciencesCrawler(MatchingCrawler): 

29 source_name = "Episciences" 

30 source_domain = "EPISCIENCES" 

31 source_website = "https://www.episciences.org/" 

32 

33 episciences_id_re = r"https://api\.episciences\.org/api/volumes\?rvcode=(?P<episciences_id>\w+)&pagination=false" 

34 

35 headers = {"accept_encoding": "utf-8", "accept": "application/ld+json"} 

36 latex_converter: LatexNodes2Text 

37 

38 def __init__(self, *args, **kwargs): 

39 super().__init__(*args, **kwargs) 

40 self.latex_converter = LatexNodes2Text(math_mode="verbatim") 

41 dict = regex_to_dict(self.episciences_id_re, self.collection_url) 

42 self.episciences_id = dict["episciences_id"] 

43 

44 def parse_collection_content(self, content): 

45 data = json.loads(content) 

46 xissues = [] 

47 

48 # Would a dedicated class be preferable ? Or maybe a smarter crawler overall 

49 if self.collection_id == "DMTCS": 

50 for issue in data: 

51 xissues.append(self.prefetch_dmcts_issue(issue)) 

52 else: 

53 for issue in data: 

54 url = f"https://api.episciences.org/api/volumes/{issue['vid']}?rvcode={self.episciences_id}&pagination=false" 

55 data = self.download_file(url) 

56 issue_content = json.loads(data) 

57 xissues.append(self.prefetch_episciences_issue(issue_content)) 

58 

59 xissues.sort(key=lambda x: x.volume) 

60 issues_by_volume = {} 

61 for issue in xissues: 

62 if issue.volume not in issues_by_volume: 

63 issues_by_volume[issue.volume] = [] 

64 issues_by_volume[issue.volume].append(issue) 

65 

66 for volume_issues in issues_by_volume.values(): 

67 year_iterable = [] 

68 try: 

69 year_iterable = [int(i.fyear) for i in volume_issues] 

70 except ValueError: 

71 pass 

72 firstyear = min(year_iterable) 

73 lastyear = max(year_iterable) 

74 if firstyear != lastyear: 

75 for i in volume_issues: 

76 i.fyear = firstyear 

77 i.lyear = lastyear 

78 

79 return xissues 

80 

81 def prefetch_dmcts_issue(self, issue: dict): 

82 if "vol_year" in issue: 

83 year = str(issue["vol_year"]) 

84 else: 

85 mid_art = issue["papers"][math.floor(len(issue["papers"]) / 2)] 

86 mid_art_url = urljoin(self.collection_url, mid_art["@id"]) 

87 mid_art_content = self.download_file(mid_art_url) 

88 mid_art_data = json.loads(mid_art_content) 

89 year = mid_art_data["document"]["journal"]["journal_issue"]["publication_date"]["year"] 

90 issue_title = issue["titles"]["en"] 

91 

92 # parsed_url = urlparse(self.collection_url) 

93 # issue_view_url = f"https://{parse_qs(parsed_url.query)['rvcode'][0]}.episciences.org/volume/view/id/{issue['vid']}" 

94 

95 if "DMTCS Proceedings" in issue_title: 

96 xissue = create_issuedata() 

97 xissue.url = None 

98 xissue.fyear = int(year) 

99 xissue.pid = get_issue_pid( 

100 self.collection_id, xissue.fyear, "special_" + str(issue["vid"]) 

101 ) 

102 # HACK : handle this elsewhere ? transform title_tex into title_xml 

103 # Is title_xml here even valid ? 

104 xissue.title_tex = issue_title 

105 xissue.title_html = issue_title 

106 xissue.title_xml = issue_title 

107 xissue.volume = issue_title 

108 

109 else: 

110 # Vol. 1 

111 # Vol. 3 no. 2 

112 # Vol. 17 no.2 

113 # vol. 24, no 2 

114 # Vol. 18 no. 2, Permutation Patterns 2015 

115 # Vol. 19 no. 4, FCT '15 

116 # vol. 27:2 

117 # vol. 25:3 special issue ICGT'22 

118 # vol. 26:1, Permutation Patterns 2023 

119 title_dict = regex_to_dict( 

120 r"[Vv]ol. (?P<volume>\d+)(?:(?:(?:,? no\.? ?)|(?:\:))?(?P<number>\d+))?(?:,? (?P<title>.+))?", 

121 issue_title, 

122 error_msg="Couldn't parse issue title", 

123 ) 

124 xissue = self.create_xissue( 

125 None, year, title_dict["volume"], title_dict.get("number", None) 

126 ) 

127 if title_dict["title"] is not None: 

128 # HACK : handle this elsewhere ? transform title_tex into title_xml 

129 # Is title_xml here even valid ? 

130 xissue.title_tex = title_dict["title"] 

131 xissue.title_html = title_dict["title"] 

132 xissue.title_xml = title_dict["title"] 

133 if issue["papers"] is not None: 

134 for index, paper in enumerate(issue["papers"]): 

135 xarticle = create_articledata() 

136 xarticle.url = urljoin( 

137 self.collection_url, f"/api/papers/export/{paper['@id']}/zbjats" 

138 ) 

139 xarticle.pid = f"a{index}" 

140 xissue.articles.append(xarticle) 

141 return xissue 

142 

143 def prefetch_episciences_issue(self, issue: dict): 

144 """ 

145 Episciences doesn't provides issue years. 

146 We have to parse the year from the first article of the issue (publication date). 

147 """ 

148 year = None 

149 if "vol_year" in issue and issue["vol_year"] is not None: 

150 year = int(issue["vol_year"]) 

151 elif "year" in issue and issue["year"] is not None: 

152 year = int(issue["year"]) 

153 else: 

154 match = re.search(r"\d{4}", issue["titles"]["en"]) 

155 if match: 

156 year = int(match.group(0)) 

157 

158 # Impossible to find issue year from metadata nor title 

159 # Fallback : fetch an article in the middle opf the issue and get its publication year 

160 mid_art = issue["papers"][math.floor(len(issue["papers"]) / 2)] 

161 mid_art_url = urljoin(self.collection_url, mid_art["@id"]) 

162 mid_art_content = self.download_file(mid_art_url) 

163 mid_art_data = json.loads(mid_art_content) 

164 year = int( 

165 mid_art_data["document"]["journal"]["journal_issue"]["publication_date"]["year"] 

166 ) 

167 

168 # parsed_url = urlparse(self.collection_url) 

169 # issue_view_url = f"https://{parse_qs(parsed_url.query)['rvcode'][0]}..org/volume/view/id/{issue['vid']}" 

170 

171 issue_title = issue["titles"]["en"] 

172 xissue = self.create_xissue(None, year, None, None) 

173 xissue.lang = "en" 

174 try: 

175 title_dict = regex_to_dict( 

176 r"Volume (?P<number>\d+)(?:, Issue (?P<issue>\d+))?(?:, (?P<title>[\w ]+))?", 

177 issue_title, 

178 error_msg="Couldn't parse issue title", 

179 ) 

180 

181 xissue.volume = title_dict["number"] 

182 if title_dict["issue"]: 

183 xissue.number = title_dict["issue"] 

184 except ValueError: 

185 self.logger.error( 

186 f"Could not parse volume number from title : {issue_title}. Fallbacking to issue poosition." 

187 ) 

188 xissue.volume = f"S{issue['position']}" 

189 xissue.title_tex = issue_title 

190 # if title_dict["title"]: 

191 # xissue.title_tex = title_dict["title"] 

192 

193 if self.collection_id in HAS_TITLES: 

194 if HAS_TITLES[self.collection_id]: 

195 xissue.title_tex = issue_title 

196 if "fr" in issue["titles"]: 

197 title_trans = issue["titles"]["fr"] 

198 xissue.titles.append( 

199 self.create_trans_title( 

200 xresource_lang=xissue.lang, 

201 resource_type="issue", 

202 title_str=title_trans, 

203 lang="fr", 

204 ) 

205 ) 

206 xissue.pid = get_issue_pid(self.collection_id, year, xissue.volume, xissue.number) 

207 

208 url_parsed = urlparse(self.collection_url) 

209 domain = f"{url_parsed.scheme}://{url_parsed.hostname}" 

210 query_string = parse_qs(url_parsed.query) 

211 for index, paper in enumerate(issue["papers"]): 

212 xarticle = create_articledata() 

213 paper_id = paper["@id"].removeprefix("/api/papers/") 

214 xarticle.url = urljoin(domain, f"api/papers/export/{paper_id}/zbjats") 

215 xarticle.pid = f"a{index}" 

216 xissue.articles.append(xarticle) 

217 

218 xissue.url = ( 

219 f"https://{query_string['rvcode'][0]}.episciences.org/volume/view/id/{issue['vid']}" 

220 ) 

221 return xissue 

222 

223 def parse_issue_content(self, content, xissue): 

224 soup = BeautifulSoup(content, "html5lib") 

225 for a in xissue.articles: 

226 if not a.url: 

227 raise ValueError("Could not find article position : no article url") 

228 id_search = re.match( 

229 r"https://api.episciences.org/api/papers/export/(\d+)/zbjats", a.url 

230 ) 

231 if not id_search: 

232 raise ValueError("Could not find article position : article url failed to parse") 

233 html_tag = soup.select_one(f"#papers > h3 > a[href='/{id_search.group(1)}']") 

234 if not html_tag: 

235 raise ValueError( 

236 "Could not find article position : article not found in issue html" 

237 ) 

238 a.position = int(next(html_tag.previous_siblings).strip().removesuffix(".")) 

239 

240 xissue.articles.sort(key=lambda a: a.position) 

241 return super().parse_issue_content(content, xissue) 

242 

243 def parse_article_content(self, content, xissue, xarticle, url): 

244 parser = etree.XMLParser( 

245 huge_tree=True, recover=True, remove_blank_text=False, remove_comments=True 

246 ) 

247 dom = etree.fromstring(content.encode("utf-8"), parser) 

248 # Fix for invalid self-uri tag 

249 xslt = etree.parse( 

250 os.path.dirname(os.path.realpath(__file__)) + "/episciences-zbjats.xsl", parser 

251 ) 

252 transform = etree.XSLT(xslt) 

253 tree = transform(dom) 

254 

255 parsed_xarticle = JatsArticle(tree=tree.getroot()) 

256 parsed_xarticle.pid = xarticle.pid 

257 return parsed_xarticle