1#!/usr/bin/env python32"""WP44 — harvest arXiv metadata (set=cs) through OAI-PMH (metadataPrefix=arXiv).34Harvests by monthly datestamp windows from --from to --until; each window's raw5XML pages are stored verbatim (immutable raw snapshot) under <out>/raw/<window>/6page_NNNN.xml and parsed into <out>/arxiv_cs.parquet at the end (``--parse-only``7re-parses). Politeness: >= 3 s between requests, honours Retry-After on 503.8Note: OAI ``from``/``until`` filter on the record *datestamp* (last update), so a9window contains every record updated in that window; ``created`` gives the first10submission date. Duplicates across windows are resolved at parse time (latest11datestamp wins).12"""13import argparse, glob, json, os, re, time, datetime as dt, calendar14import requests15from lxml import etree16import pyarrow as pa, pyarrow.parquet as pq1718BASE = "https://oaipmh.arxiv.org/oai"19NS = {"oai": "http://www.openarchives.org/OAI/2.0/", "ax": "http://arxiv.org/OAI/arXiv/"}20UA = "WP44-UQO-systematic-mapping/0.1 (academic research; contact simon-pierre.boucher@uqo.ca)"21SCHEMA = pa.schema([22 ("arxiv_id", pa.string()), ("datestamp", pa.string()), ("created", pa.string()), ("updated", pa.string()),23 ("title", pa.string()), ("abstract", pa.string()), ("authors", pa.list_(pa.string())), ("categories", pa.list_(pa.string())),24 ("primary_category", pa.string()), ("comments", pa.string()), ("journal_ref", pa.string()), ("doi", pa.string()),25 ("license", pa.string()), ("report_no", pa.string()), ("msc_class", pa.string()), ("acm_class", pa.string()), ("set_specs", pa.list_(pa.string())),26])272829def month_windows(start, end):30 y, m = start.year, start.month31 while (y, m) <= (end.year, end.month):32 last = calendar.monthrange(y, m)[1]33 a = dt.date(y, m, 1); b = min(dt.date(y, m, last), end)34 yield a, b35 m += 136 if m == 13:37 y, m = y + 1, 1383940def fetch(session, params, log):41 for attempt in range(12):42 try:43 r = session.get(BASE, params=params, timeout=180, headers={"User-Agent": UA})44 except requests.RequestException as e:45 log(f" network error {e}; sleep 30"); time.sleep(30); continue46 if r.status_code == 503:47 wait = int(r.headers.get("Retry-After", "30")); log(f" 503 retry-after {wait}"); time.sleep(wait + 1); continue48 if r.status_code != 200:49 log(f" HTTP {r.status_code}; sleep 30"); time.sleep(30); continue50 return r.text51 raise RuntimeError("too many failures")525354def harvest(a, log):55 s = requests.Session()56 state_path = os.path.join(a.out, "harvest_state.json")57 state = json.load(open(state_path)) if os.path.exists(state_path) else {"windows": {}}58 for w0, w1 in month_windows(dt.date.fromisoformat(a.frm), dt.date.fromisoformat(a.until)):59 key = f"{w0}_{w1}"60 if state["windows"].get(key, {}).get("status") == "done":61 continue62 d = os.path.join(a.out, "raw", key); os.makedirs(d, exist_ok=True)63 params = {"verb": "ListRecords", "metadataPrefix": "arXiv", "set": a.set, "from": str(w0), "until": str(w1)}64 page = 0; nrec = 0; t0 = time.time()65 while True:66 xml = fetch(s, params, log)67 with open(os.path.join(d, f"page_{page:04d}.xml"), "w") as fh:68 fh.write(xml)69 root = etree.fromstring(xml.encode())70 err = root.find("oai:error", NS)71 if err is not None:72 if err.get("code") == "noRecordsMatch":73 break74 raise RuntimeError(f"OAI error {err.get('code')}: {err.text}")75 nrec += len(root.findall(".//oai:record", NS))76 tok = root.find(".//oai:resumptionToken", NS)77 page += 178 if tok is None or not (tok.text or "").strip():79 break80 params = {"verb": "ListRecords", "resumptionToken": tok.text.strip()}81 time.sleep(a.delay)82 state["windows"][key] = {"status": "done", "pages": page, "records": nrec, "seconds": round(time.time() - t0),83 "when": dt.datetime.now(dt.UTC).isoformat()}84 json.dump(state, open(state_path, "w"), indent=1)85 log(f"window {key}: {page} pages, {nrec} records, {time.time()-t0:.0f}s")86 time.sleep(a.delay)878889def parse(a, log):90 recs = {}91 files = sorted(glob.glob(os.path.join(a.out, "raw", "*", "page_*.xml")))92 for f in files:93 root = etree.parse(f).getroot()94 for rec in root.findall(".//oai:record", NS):95 h = rec.find("oai:header", NS)96 if h.get("status") == "deleted":97 continue98 m = rec.find("oai:metadata/ax:arXiv", NS)99 if m is None:100 continue101 g = lambda t: (m.findtext(f"ax:{t}", namespaces=NS) or "").strip() or None102 aid = g("id"); ds = h.findtext("oai:datestamp", namespaces=NS)103 if aid in recs and recs[aid]["datestamp"] >= ds:104 continue105 cats = (g("categories") or "").split()106 authors = []107 for au in m.findall("ax:authors/ax:author", NS):108 kn = au.findtext("ax:keyname", namespaces=NS) or ""; fn = au.findtext("ax:forenames", namespaces=NS) or ""109 authors.append((fn + " " + kn).strip())110 recs[aid] = dict(arxiv_id=aid, datestamp=ds, created=g("created"), updated=g("updated"),111 title=re.sub(r"\s+", " ", g("title") or ""), abstract=re.sub(r"\s+", " ", g("abstract") or ""),112 authors=authors, categories=cats, primary_category=cats[0] if cats else None,113 comments=g("comments"), journal_ref=g("journal-ref"), doi=(g("doi") or "").lower() or None, license=g("license"),114 report_no=g("report-no"), msc_class=g("msc-class"), acm_class=g("acm-class"),115 set_specs=[x.text for x in h.findall("oai:setSpec", NS)])116 cols = {n: [r[n] for r in recs.values()] for n in SCHEMA.names}117 pq.write_table(pa.Table.from_pydict(cols, schema=SCHEMA), os.path.join(a.out, "arxiv_cs.parquet"), compression="zstd")118 rep = {"files": len(files), "records": len(recs), "parsed_at": dt.datetime.now(dt.UTC).isoformat(), "set": a.set, "from": a.frm, "until": a.until}119 json.dump(rep, open(os.path.join(a.out, "arxiv_parse_report.json"), "w"), indent=1); log(json.dumps(rep))120121122def main():123 ap = argparse.ArgumentParser(); ap.add_argument("--out", required=True); ap.add_argument("--set", default="cs")124 ap.add_argument("--from", dest="frm", default="2018-01-01"); ap.add_argument("--until", default="2026-11-02")125 ap.add_argument("--delay", type=float, default=3.5); ap.add_argument("--parse-only", action="store_true")126 a = ap.parse_args(); os.makedirs(a.out, exist_ok=True)127 log = lambda s: print(f"[{dt.datetime.now():%H:%M:%S}] {s}", flush=True)128 if not a.parse_only:129 harvest(a, log)130 parse(a, log)131132133if __name__ == "__main__":134 main()135