Coverage for src / crawler / by_source / ams_crawler.py: 15%
139 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-07-15 13:33 +0000
1import html
2import json
3import os
4from urllib.parse import urljoin
5from uuid import uuid4
7from bs4 import BeautifulSoup, Tag
8from opentelemetry import trace
9from ptf.cmds.xml.ckeditor.ckeditor_parser import CkeditorParser
10from ptf.cmds.xml.ckeditor.utils import get_abstract_xml
11from ptf.model_data import create_abstract, create_articledata, create_contributor, create_subj
12from ptf.utils import execute_cmd
14from crawler.abstract_crawlers.threaded_crawler import ThreadedCrawler
15from crawler.cmds.mixed_citation import ExtLinkXml, MixedCitation
16from crawler.tests.data_generation.decorators import skip_generation
17from crawler.utils import add_pdf_link_to_xarticle, cleanup_str
20class AmsCrawler(ThreadedCrawler):
21 source_name = "American Mathematical Society"
22 source_domain = "AMS"
23 source_website = "https://www.ams.org/"
24 tracer = trace.get_tracer(__name__)
26 @classmethod
27 def get_view_id(cls):
28 return "AMS"
30 @skip_generation
31 def parse_collection_content(self, content):
32 xissues = []
33 soup = BeautifulSoup(content, "html.parser")
34 issues_data_tag = soup.select_one(
35 ".container main[role='main'] script[type='text/javascript']:not([src])"
36 )
37 data = json.loads(self.get_col_issues(issues_data_tag.text))
38 issues = data["issues"]
39 self.group_by_year = data["group_by_year"] == "Y"
40 self.ams_code = data["ams_code"].lower()
41 for i in issues:
42 number = i.get("IssueNumber", None)
43 if number:
44 number = str(number)
45 if self.group_by_year:
46 number = None
47 # For AMS, xissue.url is NOT a real URL, but the AMS issue ID
48 # Issue data is fetched from an API and thus every issue url is the same
49 xissues.append(
50 self.create_xissue(
51 str(i["IssueId"]),
52 int(i["Year"]),
53 str(i["Volume"]),
54 number,
55 )
56 )
58 if self.group_by_year:
59 # We take only the first issue advertised by the website
60 # All ignored issues will be present inside the API on the next step anyways
61 years = {}
62 for i in xissues:
63 if i.fyear not in years:
64 years[i.fyear] = []
65 years[i.fyear].append(i)
67 xissues = [y[0] for y in years.values()]
68 return xissues
70 def get_col_issues(self, input: str):
71 """
72 AMS Issues are listed inside an inline js script
73 We have to spawn a nodejs subprocess to convert javascript into json"""
75 filename = "/tmp/crawler/puppeteer/" + str(uuid4())
76 filename_out = filename + "-out"
77 os.makedirs(os.path.dirname(filename), exist_ok=True)
78 with open(filename, "w") as file:
79 file.write(input)
81 content = None
82 attempt = 0
83 while not content and attempt < 3:
84 attempt += 1
85 cmd = f"{os.path.dirname(os.path.realpath(__file__))}/ams_crawler_col.js -f {filename} -o {filename_out}"
86 execute_cmd(cmd)
88 if os.path.isfile(filename_out):
89 with open(filename_out) as file_:
90 content = file_.read()
92 os.remove(filename)
93 os.remove(filename_out)
95 if not content:
96 raise ValueError("Couldn't parse collection content")
97 return content
99 def download_issue_summary(self, issue_id):
100 response = self.session.post(
101 "https://pubs.ams.org/product/GetJournalIssueDetail",
102 data={"productCode": self.ams_code, "issueId": issue_id},
103 headers={
104 "User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:149.0) Gecko/20100101 Firefox/149.0"
105 },
106 )
107 return response.text
109 def start_process_issue(self, xissue):
110 issue_url = xissue.url
111 if not issue_url:
112 raise ValueError("Issue does not have an URL")
113 content = self.download_issue_summary(issue_url)
114 # API response is somehow a list of issues
115 # Currently CAMS somehow puts every article inside a different issue in the list...
116 articles = []
118 if self.group_by_year:
119 for issue in json.loads(content):
120 articles.extend(issue["Articles"])
121 else:
122 issue_json = next(i for i in json.loads(content) if str(i["IssueId"]) == xissue.url)
123 articles = issue_json["Articles"]
125 with self.tracer.start_as_current_span("parse_issue_content"):
126 self.parse_ams_issue_content(articles, xissue)
128 def parse_ams_issue_content(self, articles: list[dict], xissue):
129 for index, article_dict in enumerate(articles):
130 xarticle = create_articledata()
131 xarticle.title_tex = article_dict["Title"]
132 # ...
133 # https://pubs.ams.org/mcom/2000-69-231/S0025-5718-00-01249-7
134 if article_dict["DOI"] != "DOI_PREFIX_HERE_S0025-5718-00-01249-7":
135 xarticle.doi = article_dict["DOI"]
137 xarticle.pid = f"a_{index}"
138 xarticle.fpage = str(article_dict["StartPage"])
139 xarticle.lpage = str(article_dict["EndPage"])
140 xarticle.date_published = article_dict["PostDate"]
142 if article_dict["DocumentType"] == "BOOKREV":
143 if xarticle.title_tex == "":
144 book_title = article_dict["BookReviews"][0]["Title"]
145 xarticle.title_tex = "Book review: " + book_title
146 if article_dict["PrimaryMsc"] is not None:
147 for msc in article_dict["PrimaryMsc"].split(", "):
148 xarticle.kwds.append(create_subj(type="msc", value=cleanup_str(msc)))
150 ckeditor_data = CkeditorParser(
151 html_value=article_dict["Abstract"],
152 mml_formulas="",
153 )
154 abstract = create_abstract(
155 lang="en",
156 value_xml=get_abstract_xml(ckeditor_data.value_xml, lang="en"),
157 value_tex=ckeditor_data.value_tex,
158 value_html=ckeditor_data.value_html,
159 )
160 xarticle.abstracts.append(abstract)
162 # TODO : EnhancedReferences
163 # TODO : UnenhancedReferences
164 # TODO : BibliographicInfo
166 add_pdf_link_to_xarticle(
167 xarticle,
168 urljoin("https://www.ams.org/journals/", self.ams_code + article_dict["PdfUrl"]),
169 )
170 xarticle.url = urljoin(
171 self.collection_url,
172 self.ams_code + "/" + article_dict["IssueDirectory"] + "/" + article_dict["PII"],
173 )
174 if article_dict["MRNumber"]:
175 xarticle.extids.append(("mr-item-id", article_dict["MRNumber"]))
177 for author in article_dict["Authors"]:
178 # TODO : AMS Provides Firstname/MiddleName/LastName but we do not have Middlename fields
179 # How should we proceed about that ?
180 xarticle.contributors.append(
181 create_contributor(
182 role="author",
183 string_name=html.unescape(author["FullName"]),
184 email=html.unescape(author["Email"] or ""),
185 addresses=[html.unescape(author["Affiliation"] or "")],
186 )
187 )
189 soup = BeautifulSoup(article_dict["EnhancedReferences"], "html5lib")
190 refs = soup.select("ul > li")
191 for ref in refs:
192 xarticle.bibitems.append(self.parse_ref(ref))
194 xissue.articles.append(xarticle)
196 def parse_ref(self, ref: "Tag"):
197 citation_builder = MixedCitation()
198 for el in ref.children:
199 if isinstance(el, str):
200 if el in [", DOI ", " DOI ", "DOI"]:
201 continue
202 citation_builder.elements.append(el)
203 continue
204 if isinstance(el, Tag):
205 if el.name == "a":
206 if el.text.startswith("10."):
207 extlink = ExtLinkXml(urljoin("https://doi.org/", el.text))
208 citation_builder.elements.append(extlink)
209 el.decompose()
210 continue
212 href = el.get("href")
213 if not isinstance(href, str):
214 continue
215 if href.startswith("https://mathscinet.ams.org/mathscinet-getitem"):
216 extlink = ExtLinkXml(href)
217 citation_builder.elements.append(extlink)
218 el.decompose()
219 continue
220 citation_builder.elements.append(el.get_text())
221 return citation_builder.get_jats_ref()
223 # def parse_ref(self, ref: "Tag"):
224 # citation_builder = MixedCitation()
225 # # Everything behind the title should be authors
226 # title_element = ref.select_one("em")
227 # if title_element:
228 # authors = list(title_element.previous_siblings)
229 # # if len(authors) != 1:
230 # # self.logger.error("Could not correctly parse reference. Fallback to text")
231 # # citation_builder.elements.append(ref.get_text())
232 # # return citation_builder.get_jats_ref()
233 # # Temporary fix : structured bibitems parsing is sometimes incorrect.
234 # # Better have no data than incorrect data (?)
235 # for el in authors:
236 # citation_builder.elements.append(el.get_text())
237 # for el in authors:
238 # el.extract()
239 # # authors_el = GenericRefElement()
240 # # authors_el.name = "person-group"
241 # # citation_builder.elements.append(authors_el)
242 # # authors_text = authors[0].text
243 # # if authors_text.endswith(", "):
244 # # authors_text = authors_text.removesuffix(", ")
245 # # authors_el.elements.append(authors_text)
246 # # citation_builder.elements.append(", ")
247 # # else:
248 # # authors_el.elements.append(authors_text)
250 # article_title = MixedCitation()
251 # article_title.name = "article-title"
252 # citation_builder.elements.append(article_title)
253 # article_title.elements.append(title_element.text)
254 # title_element.decompose()
256 # # everything before a tag is text
257 # first_link = ref.select_one("a")
258 # if first_link:
259 # texts = list(first_link.previous_siblings)
260 # if len(texts) == 0:
261 # raise ValueError("first_link previous_siblings is empty")
262 # for el in reversed(texts):
263 # citation_builder.elements.append(el.get_text().removesuffix(", Preprint, arXiv:"))
264 # el.extract()
266 # for link in ref.select("a"):
267 # url = link.get("href")
268 # if not isinstance(url, str):
269 # raise ValueError("Citation extlink does not have a valid url")
270 # reflink = ExtLinkXml(url)
271 # citation_builder.elements.append(reflink)
272 # else:
273 # citation_builder.elements.append(ref.get_text())
275 # return citation_builder.get_jats_ref()