Coverage for src / crawler / zbmath.py: 0%

1# import shutil 

2# import tempfile 

3# import urllib 

4# import xml.etree.ElementTree as ET 

5 

6 

7import logging 

8import time 

9from typing import TYPE_CHECKING 

10 

11import langcodes 

12from matching.matching import get_article_params, zbmath_matching_query 

13from ptf import model_data 

14from ptf.cmds.xml.ckeditor.utils import ( 

15 build_jats_data_from_html_field, 

16 get_html_and_xml_from_text_with_formulas, 

17) 

18from ptf.model_data_converter import get_contrib_xml 

19from requests import HTTPError, ReadTimeout 

20 

21from crawler.utils import add_pdf_link_to_xarticle, get_session 

22 

23if TYPE_CHECKING: 

24 from ptf.models import Article 

25 

26 

27last_zbmath_request = 0 

28request_delay = 5 

29_logger = logging.getLogger(__name__) 

30ZBMATH_URL = "https://zbmath.org" 

31ZBMATH_API_URL = "https://api.zbmath.org/v1/document/{query}" 

32ZBMATH_PDF_URL = "https://zbmath.org/pdf/{zbmathid}.pdf" 

33 

34 

35def parse_zbmath_article(zbmath_article): 

36 xarticle = model_data.create_articledata() 

37 for json_author in zbmath_article["contributors"]["authors"]: 

38 author = model_data.create_contributor(role="author", string_name=json_author["name"]) 

39 author["contrib_xml"] = get_contrib_xml(author) 

40 xarticle.contributors.append(author) 

41 

42 # Lang 

43 languages = zbmath_article["language"]["languages"] 

44 if len(languages) > 0: 

45 xarticle.lang = zbmath_article["language"]["languages"][0] 

46 

47 # Title 

48 xarticle.title_tex = zbmath_article["title"]["title"] 

49 

50 ckeditor_data = build_jats_data_from_html_field( 

51 xarticle.title_tex, 

52 tag="article-title", 

53 text_lang=xarticle.lang, 

54 delimiter_inline="$", 

55 delimiter_disp="$", 

56 ) 

57 xarticle.title_html = ckeditor_data["value_html"] 

58 xarticle.title_xml = ckeditor_data["value_xml"] 

59 

60 # Abstract 

61 zbl_abstract = next( 

62 ( 

63 c 

64 for c in zbmath_article["editorial_contributions"] 

65 if c["contribution_type"] in "summary" 

66 ), 

67 None, 

68 ) 

69 if zbl_abstract: 

70 abstract_data = model_data.create_abstract(value_tex=zbl_abstract["text"]) 

71 if zbl_abstract["language"]: 

72 lang = langcodes.Language.find(zbl_abstract["language"]).language 

73 if lang and xarticle.lang and lang != xarticle.lang: 

74 abstract_data["lang"] = lang 

75 

76 abstract_data["value_html"], abstract_data["value_xml"] = ( 

77 get_html_and_xml_from_text_with_formulas( 

78 abstract_data["value_tex"], delimiter_inline="\\(", delimiter_disp="\\[" 

79 ) 

80 ) 

81 xarticle.abstracts.append(abstract_data) 

82 

83 # Keywords 

84 for kwd in zbmath_article["keywords"]: 

85 if not kwd: 

86 continue 

87 xarticle.kwds.append(model_data.create_subj(value=kwd)) 

88 # MSC 

89 for msc in zbmath_article["msc"]: 

90 if msc["scheme"] != "msc2020": 

91 continue 

92 xarticle.kwds.append(model_data.create_subj(value=msc["code"])) 

93 

94 # Pages 

95 pages: str = zbmath_article["source"]["pages"] 

96 if pages: 

97 pages.split("-") 

98 if len(pages) > 0: 

99 xarticle.fpage = pages[0] 

100 if len(pages) > 1: 

101 xarticle.lpage = pages[1] 

102 

103 # extids 

104 zbmath_id = zbmath_article["identifier"] 

105 if zbmath_id is None: 

106 zbmath_id = zbmath_article["id"] 

107 xarticle.extids.append(("zbl-item-id", zbmath_id.strip())) 

108 

109 for link in zbmath_article["links"]: 

110 match link["type"]: 

111 case "doi": 

112 xarticle.doi = link["identifier"].strip() 

113 xarticle.pid = xarticle.doi.replace("/", "_").replace(".", "_").replace("-", "_") 

114 xarticle.extids.append(("doi", link["identifier"].strip())) 

115 case "eudml": 

116 xarticle.extids.append(("eudml-item-id", link["identifier"].strip())) 

117 case "arxiv": 

118 xarticle.extids.append(("arxiv", link["identifier"].strip())) 

119 case "emis_ft": 

120 add_pdf_link_to_xarticle(xarticle, link["url"].strip()) 

121 return xarticle 

122 

123 

124def zbmath_request_article(zblid): 

125 session = get_session() 

126 response = session.get( 

127 ZBMATH_API_URL.format(query=zblid), 

128 headers={**session.headers, "Content-Type": "text/json"}, 

129 ) 

130 try: 

131 response.raise_for_status() 

132 except HTTPError: 

133 return 

134 

135 response = response.json() 

136 return parse_zbmath_article(response["result"]) 

137 

138 

139def zbmath_request_article_by_doi(doi: str): 

140 session = get_session() 

141 search = f"_structured_search?results_per_page=1&DOI={doi}" 

142 response = session.get( 

143 ZBMATH_API_URL.format(query=search), 

144 headers={**session.headers, "Content-Type": "text/json"}, 

145 ) 

146 if response.status_code == 404: 

147 logging.debug(f"ZBMATH API {doi} not found") 

148 return 

149 

150 response.raise_for_status() 

151 

152 response = response.json() 

153 if response["status"]["nr_total_results"] > 1: 

154 logging.error(f"ZBMATH API found multiple candidates for doi {doi}") 

155 return 

156 # raise ValueError(f"ZBMATH API found multiple candidates for doi {doi}") 

157 if response["status"]["nr_total_results"] == 0: 

158 logging.debug(f"ZBMATH API {doi} not found") 

159 return 

160 return parse_zbmath_article(response["result"][0]) 

161 

162 

163def zbmath_request_article_by_extid(extid: str): 

164 pass 

165 

166 

167def _zbmath_query_retry(params: dict, timeout: int): 

168 global last_zbmath_request 

169 RETRIES = 3 

170 for i in range(RETRIES): 

171 try: 

172 delta = (last_zbmath_request + request_delay) - time.time() 

173 if delta > 0: 

174 _logger.info(f"Waiting {round(delta)}s before making another request") 

175 time.sleep(delta) 

176 

177 response = zbmath_matching_query(params, timeout, get_session()) 

178 if not getattr(response, "from_cache", False): 

179 last_zbmath_request = time.time() 

180 return response 

181 except ReadTimeout as e: 

182 if i >= RETRIES - 1: 

183 raise e 

184 _logger.warning("Encountered ReadTimeout while fetching Zbmath. retrying in 60 secs") 

185 time.sleep(60) 

186 

187 assert False, "unreachable" 

188 

189 

190def match_zbl_article(article: "Article"): 

191 """Finds article using matching score""" 

192 # query zbl 

193 params = get_article_params(article) 

194 response = _zbmath_query_retry(params, timeout=30) 

195 results = response.json() 

196 

197 try: 

198 item = results["results"][0][0] 

199 except (KeyError, IndexError): 

200 _logger.debug(f"Couldn't get any matching results for resource {article.pid}. Skipping") 

201 return 

202 return item 

203 

204 

205def zbmath_get_pdfurl(zbmathid: str): 

206 return ZBMATH_PDF_URL.format(zbmathid=zbmathid)