Repository Public
1"""arXiv adapter: OAI-PMH (``arXivRaw``) metadata-first discovery → Creative Commons licence filter →2pre-2018 check on the v1 submission date → LaTeX e-print download → ``<stem>.meta.json`` stub.34Endpoint (verified live 2026-11-09): ``https://oaipmh.arxiv.org/oai`` (Identify → baseURL, granularity5YYYY-MM-DD, earliestDatestamp 2005-09-16); legacy fallback ``http://export.arxiv.org/oai2``.67OAI quirks that matter here:8* ``from``/``until`` select on the record *datestamp* (last metadata change), NOT on the submission date.9 A paper first submitted in 2012 and revised in 2020 carries a 2020 datestamp. A complete enumeration of10 papers with v1 in [F, U] therefore needs the datestamp window [F, today]; the v1 date is filtered locally.11* A set (e.g. ``q-fin``) includes cross-listed papers whose primary category is elsewhere.12* ``<license>`` is the licence currently attached to the record; it is absent for many old submissions13 (→ UNKNOWN, excluded). The default arXiv licence normalises to ARXIV-NONEXCLUSIVE and is excluded.14* The e-print served for an unversioned id is the LATEST version; we request the latest version submitted15 before 2018-01-01 (``/src/<id>v<k>``) so that the text itself is pre-2018.16"""17from __future__ import annotations1819import datetime as _dt20import email.utils21import gzip22import io23import json24import random25import re26import tarfile27import time28from collections.abc import Iterator29from dataclasses import dataclass, field30from pathlib import Path3132from lxml import etree3334from htr.data.taxonomy import discipline_from_arxiv35from htr.licenses.policy import LicenseRecord, normalize_license36from htr.utils.hashing import sha256_bytes, short_id37from htr.utils.http import PoliteClient38from htr.utils.io import append_jsonl, atomic_write_bytes, ensure_dir, write_json39from htr.utils.provenance import utc_now4041OAI_ENDPOINTS = ("https://oaipmh.arxiv.org/oai", "http://export.arxiv.org/oai2")42SRC_BASE = "https://export.arxiv.org/src" # /e-print/<id> answers 301 → /src/<id>; we call /src directly (1 request)43EPRINT_BASE = "https://export.arxiv.org/e-print"44NS = {"oai": "http://www.openarchives.org/OAI/2.0/", "raw": "http://arxiv.org/OAI/arXivRaw/"}45PRE2018 = _dt.date(2018, 1, 1)46CC_CODES_DEFAULT = ("CC0", "CC-BY", "CC-BY-SA", "CC-BY-NC", "CC-BY-NC-SA")47PUB_DATE_EVIDENCE = "arxiv v1 submitted (OAI arXivRaw)"484950# --------------------------------------------------------------------------------------------- records51@dataclass52class ArxivRecord:53 arxiv_id: str54 datestamp: str | None55 setspecs: list[str]56 title: str | None57 authors_raw: str | None58 categories: list[str]59 license_uri: str | None60 versions: list[tuple[str, str]] # (label "v1", ISO date)61 abstract: str | None62 doi: str | None = None63 journal_ref: str | None = None64 comments: str | None = None65 submitter: str | None = None6667 @property68 def v1_date(self) -> str | None:69 return self.versions[0][1] if self.versions else None7071 @property72 def license_code(self) -> str:73 return normalize_license(None, self.license_uri) if self.license_uri else "UNKNOWN"7475 @property76 def authors(self) -> list[str]:77 return split_authors(self.authors_raw or "")7879 @property80 def author_group(self) -> str | None:81 return author_group(self.authors)8283 def text_version(self, cutoff: _dt.date = PRE2018) -> tuple[str, str] | None:84 """Latest version submitted strictly before ``cutoff`` (the text we download)."""85 ok = [v for v in self.versions if _dt.date.fromisoformat(v[1]) < cutoff]86 return ok[-1] if ok else None8788 def to_meta(self) -> dict:89 return {90 "id": self.arxiv_id, "datestamp": self.datestamp, "setspecs": self.setspecs, "title": self.title,91 "authors": self.authors_raw, "categories": " ".join(self.categories), "license": self.license_uri,92 "versions": [{"version": v, "date": d} for v, d in self.versions], "abstract": self.abstract,93 "doi": self.doi, "journal_ref": self.journal_ref, "comments": self.comments, "submitter": self.submitter,94 }959697def _clean(s: str | None) -> str | None:98 return re.sub(r"\s+", " ", s).strip() if s else None99100101def split_authors(raw: str) -> list[str]:102 """Split an arXiv authors string ("A. Smith, B. Jones (MIT) and C. Doe") into names."""103 s = raw104 for _ in range(3): # affiliations, possibly nested105 s = re.sub(r"\([^()]*\)", " ", s)106 s = re.sub(r"\\[`'^\"~=.uvHc]\s*\{?([A-Za-z])\}?", r"\1", s) # strip TeX accents107 s = s.replace("{", "").replace("}", "")108 parts = re.split(r",|;|\band\b|&", re.sub(r"\s+", " ", s))109 out = []110 for p in parts:111 p = p.strip(" .")112 if p and not re.fullmatch(r"(et al|Jr|Sr|II|III)", p, re.IGNORECASE):113 out.append(p)114 return out115116117def author_group(authors: list[str]) -> str | None:118 """Same convention as the JATS parser: normalised first-author surname + '_' + first initial."""119 if not authors:120 return None121 toks = authors[0].split()122 if not toks:123 return None124 return re.sub(r"[^a-z]", "", toks[-1].lower()) + ("_" + toks[0][0].lower() if len(toks) > 1 else "")125126127def _parse_date(s: str) -> str | None:128 try:129 return email.utils.parsedate_to_datetime(s.strip()).date().isoformat()130 except (TypeError, ValueError, IndexError):131 return None132133134def _rec_text(md: etree._Element, tag: str) -> str | None:135 el = md.find(f"raw:{tag}", NS)136 return _clean(el.text) if el is not None and el.text else None137138139def parse_record(rec: etree._Element) -> ArxivRecord | None:140 header = rec.find("oai:header", NS)141 if header is None or header.get("status") == "deleted":142 return None143 md = rec.find("oai:metadata/raw:arXivRaw", NS)144 if md is None:145 return None146 versions = []147 for v in md.findall("raw:version", NS):148 d = _parse_date(v.findtext("raw:date", default="", namespaces=NS))149 if d:150 versions.append((v.get("version") or f"v{len(versions) + 1}", d))151 versions.sort(key=lambda x: int(re.sub(r"\D", "", x[0]) or 0))152 return ArxivRecord(153 arxiv_id=_rec_text(md, "id") or "",154 datestamp=header.findtext("oai:datestamp", namespaces=NS),155 setspecs=[s.text for s in header.findall("oai:setSpec", NS) if s.text],156 title=_rec_text(md, "title"),157 authors_raw=_rec_text(md, "authors"),158 categories=(_rec_text(md, "categories") or "").split(),159 license_uri=_rec_text(md, "license"),160 versions=versions,161 abstract=_rec_text(md, "abstract"),162 doi=_rec_text(md, "doi"),163 journal_ref=_rec_text(md, "journal-ref"),164 comments=_rec_text(md, "comments"),165 submitter=_rec_text(md, "submitter"),166 )167168169# ------------------------------------------------------------------------------------------ OAI client170class OaiClient:171 def __init__(self, client: PoliteClient | None = None, endpoint: str | None = None):172 self.http = client or PoliteClient(timeout=240)173 self.endpoint = endpoint174 self.pages = 0175176 def resolve_endpoint(self) -> str:177 if self.endpoint:178 return self.endpoint179 for ep in OAI_ENDPOINTS:180 try:181 r = self.http.get(ep, params={"verb": "Identify"})182 if r.status_code == 200 and b"<Identify>" in r.content:183 self.endpoint = ep184 return ep185 except Exception: # noqa: BLE001 - try the fallback186 continue187 raise RuntimeError(f"no arXiv OAI-PMH endpoint reachable among {OAI_ENDPOINTS}")188189 def _page(self, params: dict) -> etree._Element:190 """One OAI page; PoliteClient already retries 503 (Retry-After / exponential); we add an outer backoff."""191 last: Exception | None = None192 for attempt in range(4):193 try:194 r = self.http.get(self.resolve_endpoint(), params=params)195 r.raise_for_status()196 self.pages += 1197 return etree.fromstring(r.content, etree.XMLParser(huge_tree=True, resolve_entities=False, no_network=True))198 except Exception as e: # noqa: BLE001199 last = e200 time.sleep(30 * 2**attempt)201 raise RuntimeError(f"OAI page failed after retries: {last}")202203 def list_records(self, set_spec: str, date_from: str | None, date_until: str | None) -> Iterator[ArxivRecord]:204 params: dict = {"verb": "ListRecords", "metadataPrefix": "arXivRaw", "set": set_spec}205 if date_from:206 params["from"] = date_from207 if date_until:208 params["until"] = date_until209 while True:210 root = self._page(params)211 err = root.find("oai:error", NS)212 if err is not None:213 if err.get("code") == "noRecordsMatch":214 return215 raise RuntimeError(f"OAI error {err.get('code')}: {err.text}")216 for rec in root.iterfind("oai:ListRecords/oai:record", NS):217 parsed = parse_record(rec)218 if parsed is not None and parsed.arxiv_id:219 yield parsed220 tok = root.find("oai:ListRecords/oai:resumptionToken", NS)221 if tok is None or not (tok.text or "").strip():222 return223 params = {"verb": "ListRecords", "resumptionToken": tok.text.strip()}224225226# ----------------------------------------------------------------------------------------- e-prints227def classify_payload(data: bytes) -> tuple[str, str | None]:228 """Return (kind, suffix): kind ∈ {latex, pdf, html, unknown}; suffix ∈ {.tar.gz, .gz, .tar, .tex}."""229 head = data.lstrip()[:8]230 if head.startswith(b"%PDF"):231 return "pdf", None232 if data[:2] == b"\x1f\x8b":233 try:234 inner = gzip.GzipFile(fileobj=io.BytesIO(data)).read(1 << 20)235 except OSError:236 return "unknown", None237 if inner.lstrip()[:5] == b"%PDF-":238 return "pdf", None239 if len(inner) > 262 and inner[257:262] == b"ustar":240 return "latex", ".tar.gz"241 return "latex", ".gz"242 if len(data) > 262 and data[257:262] == b"ustar":243 return "latex", ".tar"244 if head.lower().startswith((b"<!doc", b"<html")):245 return "html", None246 if b"\\documentclass" in data[:200000] or b"\\documentstyle" in data[:200000] or b"\\begin{document}" in data:247 return "latex", ".tex"248 return "unknown", None249250251def is_tar_gz(data: bytes) -> bool:252 try:253 with tarfile.open(fileobj=io.BytesIO(data), mode="r:gz"):254 return True255 except (tarfile.TarError, OSError, EOFError):256 return False257258259def stem_of(arxiv_id: str) -> str:260 return arxiv_id.replace("/", "_")261262263@dataclass264class Counters:265 records_seen: int = 0266 duplicates: int = 0267 license_excluded: int = 0268 no_versions: int = 0269 before_window: int = 0270 not_pre2018: int = 0271 eligible: int = 0272 pdf_only: int = 0273 fetched: int = 0274 errors: int = 0275 by_license: dict = field(default_factory=dict)276 by_set: dict = field(default_factory=dict)277 by_v1_year_eligible: dict = field(default_factory=dict)278 by_license_eligible: dict = field(default_factory=dict)279 windows_harvested: list = field(default_factory=list)280 truncated_at: dict | None = None281282 def inc(self, name: str, key: str) -> None:283 d = getattr(self, name)284 d[key] = d.get(key, 0) + 1285286287def check_record(rec: ArxivRecord, allowed: set[str], v1_from: _dt.date, v1_until: _dt.date, st: Counters) -> bool:288 """Licence + v1-date filters; updates counters; True when the record is eligible."""289 code = rec.license_code290 st.inc("by_license", code)291 if code not in allowed:292 st.license_excluded += 1293 return False294 if not rec.v1_date:295 st.no_versions += 1296 return False297 v1 = _dt.date.fromisoformat(rec.v1_date)298 if v1 >= PRE2018 or v1 > v1_until:299 st.not_pre2018 += 1300 return False301 if v1 < v1_from:302 st.before_window += 1303 return False304 st.eligible += 1305 st.inc("by_v1_year_eligible", rec.v1_date[:4])306 st.inc("by_license_eligible", code)307 return True308309310def build_stub(rec: ArxivRecord, version: tuple[str, str], data: bytes, raw_path: Path, url: str) -> dict:311 code = rec.license_code312 lic = LicenseRecord(code=code, uri=rec.license_uri, evidence="oai arXivRaw license", raw=rec.license_uri)313 notes = [f"text version fetched: {version[0]} ({version[1]})"]314 last = rec.versions[-1]315 if last[0] != version[0]:316 notes.append(f"later versions exist (latest {last[0]} {last[1]}); not used")317 notes.append("licence is the record's current arXiv licence; per-version licence history is not exposed by OAI")318 return {319 "doc_id": short_id("doc", "arxiv", rec.arxiv_id),320 "source": "arxiv",321 "source_ids": {k: v for k, v in (("arxiv_id", rec.arxiv_id), ("doi", rec.doi)) if v},322 "title": rec.title,323 "authors": rec.authors,324 "author_group": rec.author_group,325 "source_url": url,326 "license": lic.to_dict(),327 "acquired_at": utc_now(),328 "raw_path": str(raw_path),329 "raw_sha256": sha256_bytes(data),330 "raw_bytes": len(data),331 "raw_format": "latex",332 "pub_date": rec.v1_date,333 "pub_year": int(rec.v1_date[:4]) if rec.v1_date else None,334 "pub_date_evidence": PUB_DATE_EVIDENCE,335 "subject_area": discipline_from_arxiv(rec.categories),336 "subject_raw": rec.categories,337 "doc_type": "preprint",338 "venue": rec.journal_ref,339 "publisher": "arXiv",340 "provider_meta": {**rec.to_meta(), "version_fetched": version[0], "version_date": version[1], "eprint_url": f"{EPRINT_BASE}/{rec.arxiv_id}{version[0]}"},341 "notes": notes,342 }343344345def fetch_eprint(http: PoliteClient, rec: ArxivRecord, out_dir: Path, st: Counters, log) -> bool:346 version = rec.text_version()347 if version is None:348 st.not_pre2018 += 1349 return False350 url = f"{SRC_BASE}/{rec.arxiv_id}{version[0]}"351 r = http.get(url)352 r.raise_for_status()353 data = r.content354 kind, suffix = classify_payload(data)355 if kind != "latex":356 if kind == "pdf":357 st.pdf_only += 1358 else:359 st.errors += 1360 if log is not None:361 log.write(json.dumps({"arxiv_id": rec.arxiv_id, "skipped": kind, "url": url, "note": "no LaTeX source (PDF-only submission)" if kind == "pdf" else "unrecognised payload", "content_type": r.headers.get("content-type")}) + "\n")362 return False363 if suffix == ".tar.gz" and not is_tar_gz(data):364 suffix = ".gz"365 raw_path = out_dir / f"{stem_of(rec.arxiv_id)}{suffix}"366 atomic_write_bytes(raw_path, data)367 write_json(out_dir / f"{stem_of(rec.arxiv_id)}.meta.json", build_stub(rec, version, data, raw_path, url))368 st.fetched += 1369 return True370371372# ------------------------------------------------------------------------------------------ driver373# Archives created recently: every record in the set (incl. cross-lists) was stamped after creation, so earlier374# datestamp windows are provably empty — and the OAI server answers empty windows very slowly (90–400 s).375SET_CREATED = {"econ": "2017-09-01", "eess": "2017-09-01"}376377378def year_windows(date_from: str, date_until: str, set_spec: str | None = None) -> list[tuple[str, str]]:379 a, b = _dt.date.fromisoformat(date_from), _dt.date.fromisoformat(date_until)380 created = SET_CREATED.get((set_spec or "").split(":")[0])381 if created:382 a = max(a, _dt.date.fromisoformat(created))383 out = []384 for y in range(a.year, b.year + 1):385 lo, hi = max(a, _dt.date(y, 1, 1)), min(b, _dt.date(y, 12, 31))386 out.append((lo.isoformat(), hi.isoformat()))387 return out388389390def _harvest(oai: OaiClient, sets: list[str], windows_by_set: dict[str, list[tuple[str, str]]], allowed: set[str], v1_from: _dt.date, v1_until: _dt.date, st: Counters, target_per_set: int | None, on_eligible=None, progress=None, deadline: float | None = None) -> list[ArxivRecord]:391 seen: set[str] = set()392 pool: list[ArxivRecord] = []393 for s in sets:394 n_set = 0395 for lo, hi in windows_by_set[s]:396 if target_per_set and n_set >= target_per_set:397 break398 for rec in oai.list_records(s, lo, hi):399 if deadline and time.time() > deadline:400 st.truncated_at = {"set": s, "window": [lo, hi], "last_datestamp": rec.datestamp, "sets_not_started": sets[sets.index(s) + 1:]}401 return pool402 st.records_seen += 1403 if rec.arxiv_id in seen:404 st.duplicates += 1405 continue406 seen.add(rec.arxiv_id)407 if check_record(rec, allowed, v1_from, v1_until, st):408 st.inc("by_set", s)409 n_set += 1410 pool.append(rec)411 if on_eligible:412 on_eligible(rec, s)413 if progress and st.records_seen % 1000 == 0:414 progress(s, rec.datestamp, st)415 st.windows_harvested.append({"set": s, "from": lo, "until": hi, "pages_total": oai.pages, "records_seen_total": st.records_seen})416 if progress:417 progress(s, hi, st)418 return pool419420421def discover_and_fetch_arxiv(422 out_dir: Path,423 sets: list[str],424 date_from: str,425 date_until: str,426 allowed_codes: set[str] | None = None,427 max_docs: int | None = 50,428 seed: int = 1,429 client: PoliteClient | None = None,430 log_path: Path | None = None,431 metadata_only: bool = False,432 metadata_out: Path | None = None,433 datestamp_until: str | None = None,434 pool_factor: int = 4,435 progress=None,436 max_seconds: float | None = None,437) -> dict:438 """Metadata-first discovery + acquisition of CC-licensed pre-2018 arXiv LaTeX sources.439440 ``date_from``/``date_until`` bound the **v1 submission date** (capped at 2017-12-31). In fetch mode the OAI441 datestamp windows are the same years visited in seeded random order until ``pool_factor × max_docs``442 eligible records are found (a sample; papers revised after ``date_until`` are not reachable). In443 ``metadata_only`` mode the datestamp window is [date_from, datestamp_until or today] — the complete444 enumeration — and eligible records are appended to ``metadata_out`` (JSONL) without downloading.445 """446 allowed = set(allowed_codes or CC_CODES_DEFAULT)447 http = client or PoliteClient(timeout=240)448 oai = OaiClient(http)449 out_dir = ensure_dir(out_dir)450 st = Counters()451 v1_from, v1_until = _dt.date.fromisoformat(date_from), min(_dt.date.fromisoformat(date_until), PRE2018 - _dt.timedelta(days=1))452 t0 = time.time()453 deadline = t0 + max_seconds if max_seconds else None454 if metadata_only:455 ds_until = datestamp_until or _dt.date.today().isoformat()456 windows = {s: year_windows(date_from, ds_until, s) for s in sets}457 mpath = metadata_out or (out_dir / "_eligible_metadata.jsonl")458 mpath.unlink(missing_ok=True)459460 def keep(rec: ArxivRecord, s: str) -> None:461 append_jsonl(mpath, {"set": s, "license_code": rec.license_code, "v1_date": rec.v1_date, **rec.to_meta()})462463 _harvest(oai, sets, windows, allowed, v1_from, v1_until, st, None, keep, progress, deadline)464 out = {**st.__dict__, "metadata_out": str(mpath), "datestamp_window": [date_from, ds_until]}465 else:466 rng = random.Random(seed)467 windows = {}468 for s in sets:469 w = year_windows(date_from, v1_until.isoformat(), s)470 rng.shuffle(w)471 windows[s] = w472 target = None if not max_docs else max(1, -(-max_docs * pool_factor // len(sets)))473 pool = _harvest(oai, sets, windows, allowed, v1_from, v1_until, st, target, None, progress)474 rng.shuffle(pool)475 log = open(log_path, "a", encoding="utf-8") if log_path else None476 try:477 for rec in pool:478 if max_docs and st.fetched >= max_docs:479 break480 try:481 fetch_eprint(http, rec, out_dir, st, log)482 except Exception as e: # noqa: BLE001483 st.errors += 1484 if log:485 log.write(json.dumps({"arxiv_id": rec.arxiv_id, "error": repr(e)[:300]}) + "\n")486 finally:487 if log:488 log.close()489 out = {**st.__dict__, "candidate_pool": len(pool)}490 out["oai_endpoint"] = oai.endpoint491 out["oai_pages"] = oai.pages492 out["http"] = http.stats493 out["elapsed_s"] = round(time.time() - t0, 1)494 return out495