Coverage for src/crawler/by_source/msp_crawler.py: 11%

1import os 

2from urllib.parse import urljoin, urlparse 

3 

4from bs4 import BeautifulSoup, Tag 

5from ptf.cmds.xml.ckeditor.ckeditor_parser import CkeditorParser 

6from ptf.cmds.xml.jats.builder.issue import get_abstract_xml 

7from ptf.cmds.xml.jats.builder.references import get_article_title_xml, get_ext_link_xml 

8from ptf.cmds.xml.jats.jats_parser import parse_mixed_citation_into_ref 

9from ptf.model_data import ( 

10 ArticleData, 

11 create_abstract, 

12 create_articledata, 

13 create_subj, 

14) 

15 

16from crawler.abstract_crawlers.threaded_crawler import ThreadedCrawler 

17from crawler.utils import cleanup_str, regex_to_dict 

18 

19 

20class MspCrawler(ThreadedCrawler): 

21 source_name = "Mathematical Sciences Publishers" 

22 source_domain = "MSP" 

23 source_website = "https://msp.org/" 

24 

25 issue_re = r"\/\w+\/(?P<year>\d+)\/(?P<volume>\d+)(?:\-(?P<number>\d+))?" 

26 

27 def parse_collection_content(self, content): 

28 xissues = [] 

29 soup = BeautifulSoup(content, "html.parser") 

30 issues = soup.select("td.issues-area a.about[href]") 

31 for issue in issues: 

32 issue_href = issue.get("href") 

33 if not isinstance(issue_href, str): 

34 raise TypeError("Couldn't parse issue href") 

35 

36 issue_dict = regex_to_dict( 

37 self.issue_re, issue_href, error_msg="Couldn't parse issue data" 

38 ) 

39 

40 xissues.append( 

41 self.create_xissue( 

42 urljoin(self.source_website, issue_href), 

43 int(issue_dict["year"]), 

44 issue_dict["volume"], 

45 issue_dict.get("number", None), 

46 ) 

47 ) 

48 return xissues 

49 

50 def parse_issue_content(self, content, xissue): 

51 soup = BeautifulSoup(content, "html.parser") 

52 if xissue.url is None: 

53 raise ValueError("Cannot parse article : issue url is None") 

54 

55 incomplete = soup.select_one(".incomplete") 

56 if ( 

57 incomplete 

58 and cleanup_str(incomplete.text) != "Publication of this issue is now complete." 

59 ): 

60 self.logger.debug( 

61 "Ignoring : Issue is not available due to S2O policy", 

62 extra={"pid": xissue.pid}, 

63 ) 

64 return 

65 

66 issue_doi_tag = soup.select_one("div.issue-doi a") 

67 if issue_doi_tag: 

68 xissue.doi = cleanup_str(issue_doi_tag.text) 

69 

70 articles = soup.select("#toc-area .title") 

71 for index, article_tag in enumerate(articles): 

72 xarticle = create_articledata() 

73 article_href = article_tag.get("href") 

74 if not isinstance(article_href, str): 

75 raise TypeError("Couldn't parse article url") 

76 xarticle.url = urljoin(xissue.url, article_href) 

77 xarticle.pid = "a" + str(index) 

78 xissue.articles.append(xarticle) 

79 

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

81 soup = BeautifulSoup(content, "html.parser") 

82 

83 # Warn : meta doi is sometimes incorrect 

84 self.get_metadata_using_citation_meta( 

85 xarticle, xissue, soup, ["title", "author", "page", "pdf", "publisher"] 

86 ) 

87 

88 if not self.is_article_openaccess(xarticle): 

89 return 

90 

91 doi_tag = soup.select_one(".paper-doi > a") 

92 if doi_tag: 

93 xarticle.doi = doi_tag.text 

94 

95 article_data: dict[str, Tag] = {} 

96 article_sections = soup.select("#content-area > .article") 

97 for section in article_sections: 

98 if section.select_one(".copyright-license"): 

99 continue 

100 

101 tabs = section.select("tr") 

102 section_title_tag = tabs[0].select_one("h5") 

103 if not section_title_tag: 

104 self.logger.debug(f"Skipping {tabs[0].name} section at {xarticle.pid}") 

105 continue 

106 section_title = section_title_tag.text 

107 section_title_tag.decompose() 

108 del section_title_tag 

109 

110 section_content = tabs[0] 

111 if len(tabs) > 1: 

112 section_content = tabs[1] 

113 section_tag = section_content.select_one("tr > td.article-area") 

114 if section_tag: 

115 article_data[section_title] = section_tag 

116 del article_sections 

117 

118 if "Keywords" in article_data and article_data["Keywords"] != "": 

119 for kwd in article_data["Keywords"].text.split(", "): 

120 xarticle.kwds.append(create_subj(lang="en", type="kwd", value=kwd)) 

121 

122 if ( 

123 "Mathematical Subject Classification 2010" in article_data 

124 and article_data["Mathematical Subject Classification 2010"] != "" 

125 ): 

126 msc_long_text = ( 

127 cleanup_str(article_data["Mathematical Subject Classification 2010"].text) 

128 .replace("Primary: ", "") 

129 .replace(" Secondary: ", ", ") 

130 ) 

131 for kwd in msc_long_text.split(", "): 

132 xarticle.kwds.append(create_subj(lang="en", type="msc", value=kwd)) 

133 

134 if "Abstract" in article_data and article_data["Abstract"] != "": 

135 for img in article_data["Abstract"].select("img"): 

136 img.decompose() 

137 abstract_str = "".join(str(e) for e in article_data["Abstract"].select("p")) 

138 test = CkeditorParser( 

139 html_value=abstract_str, 

140 mml_formulas="", 

141 ) 

142 

143 abstract = create_abstract( 

144 lang="en", 

145 value_xml=get_abstract_xml(test.value_xml, lang="en"), 

146 value_tex=test.value_tex, 

147 value_html=test.value_html, 

148 ) 

149 

150 xarticle.abstracts.append(abstract) 

151 

152 self.parse_msp_references(xarticle) 

153 return xarticle 

154 

155 def is_article_openaccess(self, xarticle: ArticleData): 

156 stream = next( 

157 (stream for stream in xarticle.streams if stream["rel"] == "full-text"), None 

158 ) 

159 if stream is None: 

160 self.logger.error(f"Could not find pdf for article {xarticle.url}") 

161 return False 

162 pdf_url = stream["location"] 

163 

164 isok, *_ = self.check_pdf_link_validity(url=pdf_url, verify=True) 

165 

166 return isok 

167 

168 def parse_msp_references(self, xarticle: ArticleData): 

169 url = urlparse(xarticle.url) 

170 dirname = os.path.dirname(url.path) 

171 filename = os.path.basename(url.path) 

172 url = url._replace(path=urljoin(str(dirname) + "/", str(filename).replace("p", "b"))) 

173 

174 content = self.download_file(str(url.geturl())) 

175 soup = BeautifulSoup(content, "html.parser") 

176 references = soup.select("#content-area table.article:last-of-type tr") 

177 

178 # TODO : extensive parsing (authors, title etc...) 

179 # Currently, only the text is inserted 

180 for ref_node in references: 

181 td = ref_node.select("td") 

182 value_xml = self.parse_single_ref(td[1]) 

183 label = cleanup_str(td[0].text) 

184 ref = parse_mixed_citation_into_ref(value_xml, label=label) 

185 xarticle.bibitems.append(ref) 

186 

187 def parse_single_ref(self, tag: Tag): 

188 xml_list = [] 

189 ext_links = [] 

190 authors_closed = False 

191 

192 for element in tag.contents: 

193 if isinstance(element, str): 

194 xml_list.append(element) 

195 continue 

196 if isinstance(element, Tag): 

197 if element.name == "b" and not authors_closed: 

198 xml_list.append(f"<string-name>{element.text}</string-name>") 

199 elif element.name == "i" and not authors_closed: 

200 temp_element = xml_list.pop() 

201 xml_list = [ 

202 f'<person-group person-group-type="author">{cleanup_str("".join(xml_list))}</person-group>', 

203 temp_element, 

204 ] 

205 xml_list.append(get_article_title_xml(element.text)) 

206 del temp_element 

207 

208 link = element.select_one("a") 

209 if link: 

210 link_href = link.get("href") 

211 if isinstance(link_href, str) and link_href.startswith("https://doi.org/"): 

212 link_href = link_href.removeprefix("https://doi.org/") 

213 ext_links.append(get_ext_link_xml(link_href, link_href, "doi")) 

214 

215 authors_closed = True 

216 elif element.name == "a": 

217 pass 

218 continue 

219 

220 return cleanup_str("".join(xml_list) + "".join(ext_links))