Coverage for src/crawler/abstract_crawlers/base_crawler.py: 63%

1import logging 

2import time 

3from collections.abc import Iterable 

4from datetime import datetime, timedelta 

5from email.policy import EmailPolicy 

6from typing import TYPE_CHECKING, Literal 

7 

8import aiohttp 

9import regex 

10import requests 

11from bs4 import BeautifulSoup 

12from django.conf import settings 

13from django.contrib.auth.models import User 

14from django.db.utils import IntegrityError 

15from django.utils import timezone 

16from langcodes import standardize_tag 

17from lingua import LanguageDetector, LanguageDetectorBuilder 

18from opentelemetry import trace 

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

20 build_jats_data_from_html_field, 

21) 

22from ptf.cmds.xml.jats.builder.references import ( 

23 get_article_title_xml, 

24 get_author_xml, 

25 get_fpage_xml, 

26 get_lpage_xml, 

27 get_source_xml, 

28 get_year_xml, 

29) 

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

31from ptf.model_data import ( 

32 ArticleData, 

33 ContributorDict, 

34 IssueData, 

35 ResourceData, 

36 TitleDict, 

37 create_abstract, 

38 create_contributor, 

39 create_extid, 

40 create_issuedata, 

41 create_publisherdata, 

42 create_subj, 

43 create_titledata, 

44) 

45from ptf.model_data_converter import update_data_for_jats 

46from ptf.models import ExtLink 

47from ptf_external.session import get_session 

48from pylatexenc.latex2text import LatexNodes2Text 

49from pysolr import SolrError 

50from requests_cache import CachedSession 

51 

52from crawler.cmds.xml_cmds import addOrUpdateGDMLIssueXmlCmd 

53from crawler.models import Source 

54from crawler.models.extlink_checked import ExtlinkChecked 

55from crawler.types import CitationLiteral 

56from crawler.utils import ( 

57 add_pdf_link_to_xarticle, 

58 add_source_link_to_xarticle, 

59 cleanup_str, 

60 get_all_cols, 

61 get_or_create_collection, 

62) 

63 

64if TYPE_CHECKING: 

65 from collections.abc import Callable 

66 

67 from bs4 import Tag 

68 

69 

70class CrawlerTitleDict(TitleDict): 

71 title_tex: str | None 

72 

73 

74class BaseCollectionCrawler: 

75 """ 

76 Base collection for the crawlers. 

77 To create a crawler: 

78 1) derive a class from BaseCollectionCrawler and name it XXXCrawler 

79 2) override the functions parse_collection_content, parse_issue_content and parse_article_content 

80 3) update factory.py so that crawler_factory can return your new crawler 

81 """ 

82 

83 logger = logging.getLogger(__name__) 

84 tracer = trace.get_tracer(__name__) 

85 

86 source_name = "" 

87 source_domain = "" 

88 source_website = "" 

89 

90 issue_href = "" 

91 

92 collection = None 

93 source = None 

94 user = None 

95 session: requests.Session | CachedSession 

96 async_session: aiohttp.ClientSession 

97 is_checkable = True 

98 verify = True 

99 headers = {} 

100 

101 requests_interval = getattr(settings, "REQUESTS_INTERVAL", 90) 

102 "seconds to wait between two http requests" 

103 requests_timeout = 60 

104 "seconds to wait before aborting the connection (if no bytes are recieved)" 

105 

106 latext_parser = LatexNodes2Text() 

107 

108 # Override the values in your concrete crawler if the formulas in text (titles, abstracts) 

109 # do not use the "$" to surround tex formulas 

110 delimiter_inline_formula = "$" 

111 delimiter_disp_formula = "$" 

112 

113 # HACK : Workaround for tests (monkeypatching) 

114 # We store the class here, so we can monkeypatch it when running tests 

115 # subCrawlers = { 

116 # LofplCrawler: None 

117 # } 

118 subCrawlers: dict[type["BaseCollectionCrawler"], "BaseCollectionCrawler | None"] = {} 

119 

120 _language_detector: LanguageDetector | None = None 

121 _language_detector_builder = LanguageDetectorBuilder.from_all_languages() 

122 

123 match_headers = False 

124 "Whereas to include headers in requests cache key" 

125 orcid_re = r"https\:\/\/orcid\.org\/(?P<orcid>\d{4}-\d{4}-\d{4}-\d{4})" 

126 

127 ignore_missing_pdf = True 

128 "Set this to False on a Crawler-basis to allow inserting articles without PDFs" 

129 pid_year_restrictions: dict[str, int] = {} 

130 "pid -> excluded years count" 

131 

132 pause_function: "Callable[[int], None]" 

133 "Overridable the pause function (used in celery tasks to speedup aborting)" 

134 

135 @classmethod 

136 def get_view_id(cls): 

137 return cls.source_domain 

138 

139 @property 

140 def language_detector(self): 

141 """Crawler Instance singleton for language builder. 

142 Late init of LanguageDetector to save on memory""" 

143 if not self._language_detector: 

144 self._language_detector = self._language_detector_builder.build() 

145 return self._language_detector 

146 

147 def __init__( 

148 self, 

149 *args, 

150 username: str, 

151 collection_id: str, 

152 dry: bool = False, 

153 publisher: str = "", 

154 collection_url: str | None = None, 

155 backend=None, 

156 pause_function=staticmethod(time.sleep), # noqa: B008 

157 ): 

158 if not collection_url: 158 ↛ 159line 158 didn't jump to line 159 because the condition on line 158 was never true

159 all_cols = get_all_cols() 

160 col = all_cols[collection_id] 

161 

162 collection_url = col["sources"].get(self.source_domain, None) 

163 if collection_url is None: 

164 raise ValueError( 

165 f"Source {self.source_domain} not found for collection {collection_id}" 

166 ) 

167 self.collection_url = collection_url 

168 for CrawlerClass in self.subCrawlers: 168 ↛ 169line 168 didn't jump to line 169 because the loop on line 168 never started

169 self.subCrawlers[CrawlerClass] = CrawlerClass( 

170 *args, 

171 username=username, 

172 collection_id=collection_id, 

173 dry=dry, 

174 publisher=publisher, 

175 collection_url=collection_url, 

176 ) 

177 self.logger = logging.getLogger(__name__ + "." + self.source_domain) 

178 # self.logger = logging.getLogger(__name__) 

179 

180 self.username = username 

181 

182 self.collection_id = collection_id 

183 

184 self.dry = dry 

185 self.publisher = publisher 

186 

187 # Classproperty : We sometimes want to use the session without initializing the class (rot monitoring) 

188 BaseCollectionCrawler.session = requests.Session() 

189 

190 self.pause_function = pause_function 

191 

192 # Skipped when running tests 

193 self.initialize() 

194 

195 self.backend = backend 

196 

197 def initialize(self): 

198 """ 

199 Acts as a "second" init function to skip model accesses during test data generation 

200 """ 

201 self.collection = get_or_create_collection(self.collection_id) 

202 self.source = self.get_or_create_source() 

203 self.user = User.objects.get(username=self.username) 

204 BaseCollectionCrawler.session = get_session() 

205 BaseCollectionCrawler.session.verify = self.verify 

206 self.session.pause_function = self.pause_function 

207 self.session.delay = self.requests_interval 

208 

209 @classmethod 

210 def can_crawl(cls, pid: str) -> bool: 

211 return True 

212 

213 def parse_collection_content(self, content: str) -> list[IssueData]: 

214 """ 

215 Parse the HTML content with BeautifulSoup 

216 returns a list of xissue. 

217 Override this function in a derived class 

218 """ 

219 return [] 

220 

221 def parse_issue_content(self, content: str, xissue: IssueData): 

222 """ 

223 Parse the HTML content with BeautifulSoup 

224 Fills the xissue.articles 

225 Override this function in a derived class. 

226 

227 CAV : You are supposed to create articles there. Please assign a PID to each article. 

228 The PID can be `a + article_index`, like this : `a0` `a21` 

229 """ 

230 

231 def parse_article_content( 

232 self, content: str, xissue: IssueData, xarticle: ArticleData, url: str 

233 ) -> ArticleData | None: 

234 """ 

235 Parse the HTML content with BeautifulSoup 

236 returns the xarticle. 

237 Override this function in a derived class. 

238 The xissue is passed to the function in case the article page has issue information (ex: publisher) 

239 The article url is also passed as a parameter 

240 

241 CAV : You are supposed to assign articles pid again here 

242 """ 

243 return xarticle 

244 

245 @tracer.start_as_current_span("crawl_collection") 

246 def crawl_collection(self): 

247 # TODO: Comments, filter 

248 """ 

249 Crawl an entire collection. ptf.models.Container objects are created. 

250 - get the HTML content of the collection_url 

251 - parse the HTML content with beautifulsoup to extract the list of issues 

252 - merge the xissues (some Source can have multiple pages for 1 volume/issue. We create only 1 container) 

253 - crawl each issue if col_only is False 

254 - Returns the list of merged issues. 

255 It is an OrderedDict {pid: {"issues": xissues}} 

256 The key is the pid of the merged issues. 

257 Ex: The source may have Ex: Volume 6 (2000) and Volume 6 (1999) 

258 the pid is then made with 1999-2000__6_ 

259 """ 

260 

261 if self.source is None: 

262 raise RuntimeError("ERROR: the source is not set") 

263 

264 content = self.download_file(self.collection_url) 

265 if content: 

266 xissues = self.parse_collection_content(content) 

267 else: 

268 # download_file returns None (404) 

269 return None 

270 

271 """ 

272 Some collections split the same volumes in different pages 

273 Ex: Volume 6 (2000) and Volume 6 (1999) 

274 We merge the 2 xissues with the same volume number => Volume 6 (1999-2000) 

275 """ 

276 # merged_xissues = self.merge_xissues(xissues) 

277 

278 xissues_dict = {str(i.pid): i for i in xissues} 

279 

280 return xissues_dict 

281 

282 def start_process_issue(self, xissue: IssueData): 

283 # Some source, like EuDML do not have a separate HTML pages for an issue's table of content. 

284 # The list of articles directly come from the collection HTML page: the xissue has no url attribute 

285 issue_url = xissue.url 

286 if issue_url is not None: 

287 if issue_url.endswith(".pdf"): 

288 add_pdf_link_to_xarticle(xissue, issue_url) 

289 xissue.url = None 

290 else: 

291 content = self.download_file(issue_url) 

292 with self.tracer.start_as_current_span("parse_issue_content"): 

293 self.parse_issue_content(content, xissue) 

294 

295 @tracer.start_as_current_span("crawl_issue") 

296 def crawl_issue(self, xissue: IssueData): 

297 """ 

298 Crawl 1 wag page of an issue. 

299 - get the HTML content of the issue 

300 - parse the HTML content with beautifulsoup to extract the list of articles and/or the issue metadata 

301 - crawl each article 

302 """ 

303 

304 self.start_process_issue(xissue) 

305 

306 xarticles = xissue.articles 

307 

308 parsed_xarticles = [] 

309 

310 for xarticle in xarticles: 

311 parsed_xarticle = self.crawl_article(xarticle, xissue) 

312 if parsed_xarticle is not None: 

313 parsed_xarticles.append(parsed_xarticle) 

314 

315 xissue.articles = parsed_xarticles 

316 

317 issue_has_pdf = self.article_has_pdf(xissue) 

318 

319 if self.ignore_missing_pdf: 

320 xissue.articles = [a for a in xissue.articles if self.article_has_pdf(a)] 

321 if self.dry: 

322 return 

323 if len(xissue.articles) == 0 and not issue_has_pdf: 

324 return 

325 self.process_resource_metadata(xissue, resource_type="issue") 

326 

327 self.add_xissue_into_database(xissue) 

328 

329 @staticmethod 

330 def article_has_source(art: ArticleData | IssueData): 

331 return ( 

332 next( 

333 (e_link for e_link in art.ext_links if e_link["rel"] == "source"), 

334 None, 

335 ) 

336 is not None 

337 ) 

338 

339 @staticmethod 

340 def article_has_pdf(art: ArticleData | IssueData): 

341 return ( 

342 next( 

343 (link for link in art.ext_links if link["rel"] in ["article-pdf", "article-html"]), 

344 None, 

345 ) 

346 is not None 

347 ) 

348 

349 def crawl_article(self, xarticle: ArticleData, xissue: IssueData): 

350 # ARTICLE URL as en ExtLink (to display the link in the article page) 

351 if xarticle.url is None: 

352 if not self.article_has_source(xarticle): 352 ↛ 358line 352 didn't jump to line 358 because the condition on line 352 was always true

353 if xissue.url: 

354 article_source = xissue.url 

355 else: 

356 article_source = self.collection_url 

357 add_source_link_to_xarticle(xarticle, article_source, self.source_domain) 

358 return self.process_article_metadata(xarticle) 

359 

360 if self.parse_article_content.__func__ != BaseCollectionCrawler.parse_article_content: 

361 content = self.download_file(xarticle.url) 

362 xarticle.pid = f"{xissue.pid}_{xarticle.pid}" 

363 

364 try: 

365 with self.tracer.start_as_current_span("parse_article_content"): 

366 parsed_xarticle = self.parse_article_content( 

367 content, xissue, xarticle, xarticle.url 

368 ) 

369 except ValueError as e: 

370 self.logger.warning(e) 

371 self.logger.warning("Retrying in 5 mins while invalidating cache") 

372 self.pause_function(5 * 60) 

373 content = self.download_file(xarticle.url, headers={"Cache-Control": "no-cache"}) 

374 with self.tracer.start_as_current_span("parse_article_content"): 

375 parsed_xarticle = self.parse_article_content( 

376 content, xissue, xarticle, xarticle.url 

377 ) 

378 

379 if not parsed_xarticle: 379 ↛ 380line 379 didn't jump to line 380 because the condition on line 379 was never true

380 return None 

381 

382 xarticle = parsed_xarticle 

383 

384 if xarticle.doi: 

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

386 

387 if not self.article_has_source(xarticle) and xarticle.url: 

388 add_source_link_to_xarticle(xarticle, xarticle.url, self.source_domain) 

389 

390 # The article title may have formulas surrounded with '$' 

391 return self.process_article_metadata(xarticle) 

392 

393 def process_resource_metadata(self, xresource: ResourceData, resource_type="article"): 

394 tag = "article-title" if resource_type == "article" else "issue-title" 

395 

396 # Process title tex 

397 ckeditor_data = build_jats_data_from_html_field( 

398 xresource.title_tex, 

399 tag=tag, 

400 text_lang=xresource.lang, 

401 delimiter_inline=self.delimiter_inline_formula, 

402 delimiter_disp=self.delimiter_disp_formula, 

403 ) 

404 

405 xresource.title_html = ckeditor_data["value_html"] 

406 # xresource.title_tex = ckeditor_data["value_tex"] 

407 xresource.title_xml = ckeditor_data["value_xml"] 

408 

409 abstracts_to_parse = [ 

410 xabstract for xabstract in xresource.abstracts if xabstract["tag"] == "abstract" 

411 ] 

412 # abstract may have formulas surrounded with '$' 

413 if len(abstracts_to_parse) > 0: 

414 for xabstract in abstracts_to_parse: 

415 ckeditor_data = build_jats_data_from_html_field( 

416 xabstract["value_tex"], 

417 tag="abstract", 

418 text_lang=xabstract["lang"], 

419 resource_lang=xresource.lang, 

420 field_type="abstract", 

421 delimiter_inline=self.delimiter_inline_formula, 

422 delimiter_disp=self.delimiter_disp_formula, 

423 ) 

424 

425 xabstract["value_html"] = ckeditor_data["value_html"] 

426 # xabstract["value_tex"] = ckeditor_data["value_tex"] 

427 xabstract["value_xml"] = ckeditor_data["value_xml"] 

428 

429 # Remove useless titles (duplicates) 

430 titles = [] 

431 for trans_title in xresource.titles: 

432 if trans_title["title_html"] != xresource.title_html: 432 ↛ 431line 432 didn't jump to line 431 because the condition on line 432 was always true

433 # trans_title["title_html"] = self.latex_converter.latex_to_text( 

434 # trans_title["title_html"] 

435 # ) 

436 titles.append(trans_title) 

437 

438 xresource.titles = titles 

439 

440 # remove duplicated keywords and add "msc" type to msc keywords : 

441 xresource_keywords = [] 

442 for resource_keyword_dict in xresource.kwds: 

443 resource_keyword = resource_keyword_dict["value"] 

444 if resource_keyword == "" or resource_keyword in [ 444 ↛ 447line 444 didn't jump to line 447 because the condition on line 444 was never true

445 xresource_keyword["value"] for xresource_keyword in xresource_keywords 

446 ]: 

447 continue 

448 

449 if any(char.isdigit() for char in resource_keyword): 

450 resource_keyword_dict["type"] = "msc" 

451 

452 xresource_keywords.append(resource_keyword_dict) 

453 xresource.kwds = xresource_keywords 

454 

455 return xresource 

456 

457 def process_article_metadata(self, xarticle: ArticleData): 

458 self.process_resource_metadata(xarticle) 

459 for bibitem in xarticle.bibitems: 

460 bibitem.type = "unknown" 

461 update_data_for_jats(xarticle, with_label=False) 

462 

463 return xarticle 

464 

465 def download_file(self, url: str, headers=None): 

466 """ 

467 Downloads a page and returns its content (decoded string). 

468 """ 

469 

470 if headers is None: 

471 headers = {} 

472 for attempt in range(3): 

473 session: CachedSession = get_session() # type:ignore 

474 response = session.get( 

475 url, 

476 headers=headers, 

477 ) 

478 

479 content = self.decode_response(response) 

480 if content == "" or not content: 

481 self.logger.debug("Got empty content while fetching ! ") 

482 # 15 mins, 30 mins, 45 mins 

483 delay_minutes = attempt * 15 

484 self.logger.debug( 

485 f"Retrying in {delay_minutes}mins ({(datetime.now() + timedelta(minutes=delay_minutes)).time()})", 

486 extra={"url": url}, 

487 ) 

488 self.pause_function(delay_minutes * 60) 

489 continue 

490 return content 

491 raise ValueError(f"Could not decode content at {url}") 

492 

493 def decode_response(self, response: requests.Response, encoding: str | None = None): 

494 """Override this if the content-type headers from the sources are advertising something else than the actual content 

495 SASA needs this""" 

496 # Force 

497 if encoding: 

498 response.encoding = encoding 

499 return response.text 

500 

501 # Attempt to get encoding using HTTP headers 

502 content_type_tag = response.headers.get("Content-Type", None) 

503 

504 if content_type_tag: 504 ↛ 511line 504 didn't jump to line 511 because the condition on line 504 was always true

505 charset = self.parse_content_type_charset(content_type_tag) 

506 if charset: 506 ↛ 507line 506 didn't jump to line 507 because the condition on line 506 was never true

507 response.encoding = charset 

508 return response.text 

509 

510 # Attempt to get encoding using HTML meta charset tag 

511 soup = BeautifulSoup(response.text, "html5lib") 

512 charset = soup.select_one("meta[charset]") 

513 if charset: 

514 htmlencoding = charset.get("charset") 

515 if isinstance(htmlencoding, str): 515 ↛ 520line 515 didn't jump to line 520 because the condition on line 515 was always true

516 response.encoding = htmlencoding 

517 return response.text 

518 

519 # Attempt to get encoding using HTML meta content type tag 

520 content_type_tag = soup.select_one( 

521 'meta[http-equiv="Content-Type"],meta[http-equiv="content-type"]' 

522 ) 

523 if content_type_tag: 

524 content_type = content_type_tag.get("content") 

525 if isinstance(content_type, str): 525 ↛ 531line 525 didn't jump to line 531 because the condition on line 525 was always true

526 charset = self.parse_content_type_charset(content_type) 

527 if charset: 527 ↛ 531line 527 didn't jump to line 531 because the condition on line 527 was always true

528 response.encoding = charset 

529 return response.text 

530 

531 return response.text 

532 

533 @staticmethod 

534 def parse_content_type_charset(content_type: str): 

535 header = EmailPolicy.header_factory("content-type", content_type) 

536 if "charset" in header.params: 

537 return header.params.get("charset") 

538 

539 @tracer.start_as_current_span("add_xissue_to_database") 

540 def add_xissue_into_database(self, xissue: IssueData) -> IssueData: 

541 xissue.journal = self.collection 

542 xissue.source = self.source_domain 

543 

544 if xissue.fyear == 0: 

545 raise ValueError("Failsafe : Cannot insert issue without a year") 

546 

547 xpub = create_publisherdata() 

548 xpub.name = self.publisher 

549 xissue.publisher = xpub 

550 xissue.last_modified_iso_8601_date_str = timezone.now().isoformat() 

551 

552 attempt = 1 

553 success = False 

554 

555 while not success and attempt < 4: 

556 try: 

557 params = {"xissue": xissue, "use_body": False} 

558 cmd = addOrUpdateGDMLIssueXmlCmd(params) 

559 cmd.do() 

560 success = True 

561 self.logger.debug(f"Issue {xissue.pid} inserted in database") 

562 return xissue 

563 except SolrError: 

564 self.logger.warning( 

565 f"Encoutered SolrError while inserting issue {xissue.pid} in database" 

566 ) 

567 attempt += 1 

568 self.logger.debug(f"Attempt {attempt}. sleeping 10 seconds.") 

569 self.pause_function(10) 

570 except Exception as e: 

571 self.logger.error( 

572 f"Got exception while attempting to insert {xissue.pid} in database : {e}" 

573 ) 

574 raise 

575 

576 if success is False: 

577 raise ConnectionRefusedError("Cannot connect to SolR") 

578 

579 assert False, "Unreachable" 

580 

581 def get_metadata_using_citation_meta( 

582 self, 

583 xarticle: ArticleData, 

584 xissue: IssueData, 

585 soup: BeautifulSoup, 

586 what: list[CitationLiteral] | None = None, 

587 ): 

588 """ 

589 :param xarticle: the xarticle that will collect the metadata 

590 :param xissue: the xissue that will collect the publisher 

591 :param soup: the BeautifulSoup object of tha article page 

592 :param what: list of citation_ items to collect. 

593 :return: None. The given article is modified 

594 """ 

595 

596 if what is None: 596 ↛ 597line 596 didn't jump to line 597 because the condition on line 596 was never true

597 what = [] 

598 if "title" in what: 

599 # TITLE 

600 citation_title_node = soup.select_one("meta[name='citation_title']") 

601 if citation_title_node: 601 ↛ 606line 601 didn't jump to line 606 because the condition on line 601 was always true

602 title = citation_title_node.get("content") 

603 if isinstance(title, str): 603 ↛ 606line 603 didn't jump to line 606 because the condition on line 603 was always true

604 xarticle.title_tex = title 

605 

606 if "author" in what: 606 ↛ 635line 606 didn't jump to line 635 because the condition on line 606 was always true

607 # AUTHORS 

608 citation_author_nodes = soup.select("meta[name^='citation_author']") 

609 current_author: ContributorDict | None = None 

610 for citation_author_node in citation_author_nodes: 

611 if citation_author_node.get("name") == "citation_author": 

612 text_author = citation_author_node.get("content") 

613 if not isinstance(text_author, str): 613 ↛ 614line 613 didn't jump to line 614 because the condition on line 613 was never true

614 raise ValueError("Cannot parse author") 

615 if text_author == "": 615 ↛ 616line 615 didn't jump to line 616 because the condition on line 615 was never true

616 current_author = None 

617 continue 

618 current_author = create_contributor(role="author", string_name=text_author) 

619 xarticle.contributors.append(current_author) 

620 continue 

621 if current_author is None: 621 ↛ 622line 621 didn't jump to line 622 because the condition on line 621 was never true

622 self.logger.warning("Couldn't parse citation author") 

623 continue 

624 if citation_author_node.get("name") == "citation_author_institution": 

625 text_institution = citation_author_node.get("content") 

626 if not isinstance(text_institution, str): 626 ↛ 627line 626 didn't jump to line 627 because the condition on line 626 was never true

627 continue 

628 current_author["addresses"].append(text_institution) 

629 if citation_author_node.get("name") == "citation_author_ocrid": 629 ↛ 630line 629 didn't jump to line 630 because the condition on line 629 was never true

630 text_orcid = citation_author_node.get("content") 

631 if not isinstance(text_orcid, str): 

632 continue 

633 current_author["orcid"] = text_orcid 

634 

635 if "pdf" in what: 

636 # PDF 

637 citation_pdf_node = soup.select_one('meta[name="citation_pdf_url"]') 

638 if citation_pdf_node: 

639 pdf_url = citation_pdf_node.get("content") 

640 if isinstance(pdf_url, str): 640 ↛ 643line 640 didn't jump to line 643 because the condition on line 640 was always true

641 add_pdf_link_to_xarticle(xarticle, pdf_url) 

642 

643 if "lang" in what: 

644 # LANG 

645 citation_lang_node = soup.select_one("meta[name='citation_language']") 

646 if citation_lang_node: 646 ↛ 652line 646 didn't jump to line 652 because the condition on line 646 was always true

647 # TODO: check other language code 

648 content_text = citation_lang_node.get("content") 

649 if isinstance(content_text, str): 649 ↛ 652line 649 didn't jump to line 652 because the condition on line 649 was always true

650 xarticle.lang = standardize_tag(content_text) 

651 

652 if "abstract" in what: 

653 # ABSTRACT 

654 abstract_node = soup.select_one("meta[name='citation_abstract']") 

655 if abstract_node is not None: 

656 abstract = abstract_node.get("content") 

657 if not isinstance(abstract, str): 657 ↛ 658line 657 didn't jump to line 658 because the condition on line 657 was never true

658 raise ValueError("Couldn't parse abstract from meta") 

659 abstract = BeautifulSoup(abstract, "html.parser").text 

660 lang = abstract_node.get("lang") 

661 if not isinstance(lang, str): 

662 lang = self.detect_language(abstract, xarticle) 

663 xarticle.abstracts.append(create_abstract(lang=lang, value_tex=abstract)) 

664 

665 if "page" in what: 

666 # PAGES 

667 citation_fpage_node = soup.select_one("meta[name='citation_firstpage']") 

668 if citation_fpage_node: 

669 page = citation_fpage_node.get("content") 

670 if isinstance(page, str): 670 ↛ 675line 670 didn't jump to line 675 because the condition on line 670 was always true

671 page = page.split("(")[0] 

672 if len(page) < 32: 672 ↛ 675line 672 didn't jump to line 675 because the condition on line 672 was always true

673 xarticle.fpage = page 

674 

675 citation_lpage_node = soup.select_one("meta[name='citation_lastpage']") 

676 if citation_lpage_node: 

677 page = citation_lpage_node.get("content") 

678 if isinstance(page, str): 678 ↛ 683line 678 didn't jump to line 683 because the condition on line 678 was always true

679 page = page.split("(")[0] 

680 if len(page) < 32: 680 ↛ 683line 680 didn't jump to line 683 because the condition on line 680 was always true

681 xarticle.lpage = page 

682 

683 if "doi" in what: 

684 # DOI 

685 citation_doi_node = soup.select_one("meta[name='citation_doi']") 

686 if citation_doi_node: 686 ↛ 695line 686 didn't jump to line 695 because the condition on line 686 was always true

687 doi = citation_doi_node.get("content") 

688 if isinstance(doi, str): 688 ↛ 695line 688 didn't jump to line 695 because the condition on line 688 was always true

689 doi = doi.strip() 

690 pos = doi.find("10.") 

691 if pos > 0: 691 ↛ 692line 691 didn't jump to line 692 because the condition on line 691 was never true

692 doi = doi[pos:] 

693 xarticle.doi = doi 

694 

695 if "mr" in what: 

696 # MR 

697 citation_mr_node = soup.select_one("meta[name='citation_mr']") 

698 if citation_mr_node: 698 ↛ 707line 698 didn't jump to line 707 because the condition on line 698 was always true

699 mr = citation_mr_node.get("content") 

700 if isinstance(mr, str): 700 ↛ 707line 700 didn't jump to line 707 because the condition on line 700 was always true

701 mr = mr.strip() 

702 if mr.find("MR") == 0: 702 ↛ 707line 702 didn't jump to line 707 because the condition on line 702 was always true

703 mr = mr[2:] 

704 extid = create_extid("mr-item-id", mr) 

705 xarticle.extids.append(extid) 

706 

707 if "zbl" in what: 

708 # ZBL 

709 citation_zbl_node = soup.select_one("meta[name='citation_zbl']") 

710 if citation_zbl_node: 710 ↛ 719line 710 didn't jump to line 719 because the condition on line 710 was always true

711 zbl = citation_zbl_node.get("content") 

712 if isinstance(zbl, str): 712 ↛ 719line 712 didn't jump to line 719 because the condition on line 712 was always true

713 zbl = zbl.strip() 

714 if zbl.find("Zbl") == 0: 714 ↛ 719line 714 didn't jump to line 719 because the condition on line 714 was always true

715 zbl = zbl[3:].strip() 

716 extid = create_extid("zbl-item-id", zbl) 

717 xarticle.extids.append(extid) 

718 

719 if "publisher" in what: 

720 # PUBLISHER 

721 citation_publisher_node = soup.select_one("meta[name='citation_publisher']") 

722 if citation_publisher_node: 722 ↛ 731line 722 didn't jump to line 731 because the condition on line 722 was always true

723 pub = citation_publisher_node.get("content") 

724 if isinstance(pub, str): 724 ↛ 731line 724 didn't jump to line 731 because the condition on line 724 was always true

725 pub = pub.strip() 

726 if pub != "": 726 ↛ 731line 726 didn't jump to line 731 because the condition on line 726 was always true

727 xpub = create_publisherdata() 

728 xpub.name = pub 

729 xissue.publisher = xpub 

730 

731 if "keywords" in what: 

732 # KEYWORDS 

733 citation_kwd_nodes = soup.select("meta[name='citation_keywords']") 

734 for kwd_node in citation_kwd_nodes: 

735 kwds = kwd_node.get("content") 

736 if isinstance(kwds, str): 736 ↛ 734line 736 didn't jump to line 734 because the condition on line 736 was always true

737 kwds = kwds.split(",") 

738 for kwd in kwds: 

739 if kwd == "": 739 ↛ 740line 739 didn't jump to line 740 because the condition on line 739 was never true

740 continue 

741 kwd = kwd.strip() 

742 xarticle.kwds.append({"type": "", "lang": xarticle.lang, "value": kwd}) 

743 

744 if "references" in what: 

745 citation_references = soup.select("meta[name='citation_reference']") 

746 for index, tag in enumerate(citation_references): 

747 content = tag.get("content") 

748 if not isinstance(content, str): 748 ↛ 749line 748 didn't jump to line 749 because the condition on line 748 was never true

749 raise ValueError("Cannot parse citation_reference meta") 

750 label = str(index + 1) 

751 if regex.match(r"^\[\d+\].*", content): 751 ↛ 752line 751 didn't jump to line 752 because the condition on line 751 was never true

752 label = None 

753 xarticle.bibitems.append(self.__parse_meta_citation_reference(content, label)) 

754 

755 def get_metadata_using_dcterms( 

756 self, 

757 xarticle: ArticleData, 

758 soup: "Tag", 

759 what: "Iterable[Literal['abstract', 'keywords', 'date_published', 'article_type']]", 

760 ): 

761 if "abstract" in what: 761 ↛ 769line 761 didn't jump to line 769 because the condition on line 761 was always true

762 abstract_tag = soup.select_one("meta[name='DCTERMS.abstract']") 

763 if abstract_tag: 763 ↛ 769line 763 didn't jump to line 769 because the condition on line 763 was always true

764 abstract_text = self.get_str_attr(abstract_tag, "content") 

765 xarticle.abstracts.append( 

766 create_abstract(lang="en", value_tex=cleanup_str(abstract_text)) 

767 ) 

768 

769 if "keywords" in what: 769 ↛ 778line 769 didn't jump to line 778 because the condition on line 769 was always true

770 keyword_tags = soup.select("meta[name='DC.subject']") 

771 for tag in keyword_tags: 

772 kwd_text = tag.get("content") 

773 if not isinstance(kwd_text, str) or len(kwd_text) == 0: 773 ↛ 774line 773 didn't jump to line 774 because the condition on line 773 was never true

774 continue 

775 kwd = create_subj(value=kwd_text) 

776 xarticle.kwds.append(kwd) 

777 

778 if "date_published" in what: 778 ↛ 779line 778 didn't jump to line 779 because the condition on line 778 was never true

779 published_tag = soup.select_one("meta[name='DC.Date.created']") 

780 if published_tag: 

781 published_text = self.get_str_attr(published_tag, "content") 

782 xarticle.date_published = published_text 

783 

784 if "article_type" in what: 784 ↛ 785line 784 didn't jump to line 785 because the condition on line 784 was never true

785 type_tag = soup.select_one("meta[name='DC.Type.articleType']") 

786 if type_tag: 

787 type_text = self.get_str_attr(type_tag, "content") 

788 xarticle.atype = type_text 

789 

790 def create_xissue( 

791 self, 

792 url: str | None, 

793 year: int, 

794 volume_number: str | None, 

795 issue_number: str | None = None, 

796 vseries: str | None = None, 

797 lyear: int | None = None, 

798 ): 

799 if url is not None and url.endswith("/"): 799 ↛ 800line 799 didn't jump to line 800 because the condition on line 799 was never true

800 url = url[:-1] 

801 xissue = create_issuedata() 

802 xissue.url = url 

803 

804 year_str = year 

805 xissue.fyear = year 

806 if lyear: 806 ↛ 807line 806 didn't jump to line 807 because the condition on line 806 was never true

807 year_str = f"{year}_{lyear}" 

808 xissue.lyear = lyear 

809 xissue.pid = self.get_issue_pid( 

810 self.collection_id, year_str, volume_number, issue_number, vseries 

811 ) 

812 

813 if volume_number is not None: 

814 xissue.volume = regex.sub(r"[^\w-]+", "_", volume_number) 

815 

816 if issue_number is not None: 

817 xissue.number = issue_number.replace(",", "-") 

818 

819 if vseries is not None: 819 ↛ 820line 819 didn't jump to line 820 because the condition on line 819 was never true

820 xissue.vseries = vseries 

821 return xissue 

822 

823 def detect_language(self, text: str, article: ArticleData | None = None): 

824 if article and article.lang is not None and article.lang != "und": 

825 return article.lang 

826 

827 language = self.language_detector.detect_language_of(text) 

828 

829 if not language: 829 ↛ 830line 829 didn't jump to line 830 because the condition on line 829 was never true

830 return "und" 

831 return language.iso_code_639_1.name.lower() 

832 

833 def get_str_attr(self, tag: "Tag", attr: str): 

834 """Equivalent of `tag.get(attr)`, but ensures the return value is a string""" 

835 node_attr = tag.get(attr)