"""Run bounded, recorded pages for the 2000–2016 Andreev backfill. Uses PatchScholar's native fetch and review-pack operations. Serial ingestion only; reruns resume successful pages. This is a collection-specific run recipe. """ import argparse import csv import json import re import sqlite3 import subprocess from datetime import datetime, timezone from pathlib import Path ROOT = Path(__file__).resolve().parents[1] SPEC = ROOT / "searches/2026-09-11-backfill" DATA = ROOT / "data/batches/2026-09-11-backfill" FAMILIES = { "arxiv-core": "arxiv-decade-000", "arxiv-related": "arxiv-related-000", "openalex-core": "openalex-decade-000", "openalex-abstract": "openalex-abstract-000", "openalex-abstract-cursor": "openalex-abstract-000", "openalex-ysr": "openalex-ysr-000", "openalex-splitting": "openalex-splitting-000", "openalex-gatemon": "openalex-gatemon-000", } ADDITIONAL = {"openalex-josephson-mesoscopic": "mesoscopic Josephson", "openalex-josephson-phase": "Josephson current phase","openalex-seeds": "","openalex-josephson-subgap": "Josephson subgap", "openalex-josephson-states": "Josephson bound states", "openalex-proximity": "superconducting proximity effect", "openalex-shiba": "Shiba"} def db(): return sqlite3.connect(f"file:{ROOT / 'state/patchscholar.db'}?mode=ro", uri=True) def stamp(): return datetime.now(timezone.utc).isoformat() def run(family, pages, start="2000-01-01", end="2016-09-08", batch="2026-09-11-backfill"): SPEC = ROOT / "searches" / batch DATA = ROOT / "data/batches" / batch SPEC.mkdir(parents=True, exist_ok=True) DATA.mkdir(parents=True, exist_ok=True) provider = family.split("-")[0] if family == "openalex-seeds": dois = ["10.1103/PhysRevLett.67.3836", "10.1103/PhysRevLett.71.1625", "10.1103/PhysRevB.46.12841", "10.1103/RevModPhys.69.731", "10.1103/PhysRevLett.74.1657", "10.1103/PhysRevLett.83.2652"] initial = {"filter": "doi:" + "|".join("https://doi.org/" + d.lower() for d in dois) + f",from_publication_date:{start},to_publication_date:{end}", "per-page": 200, "cursor": "*"} elif family in ADDITIONAL: initial = {"filter": f"title_and_abstract.search:{ADDITIONAL[family]},from_publication_date:{start},to_publication_date:{end}", "per-page": 200, "cursor": "*"} else: initial = json.loads((ROOT / "searches/2026-09-09-completion" / (FAMILIES[family]+".json")).read_text()) initial = json.loads(json.dumps(initial).replace("201609090000 TO 202609090000", start.replace("-", "") + "0000 TO " + end.replace("-", "") + "2359").replace("from_publication_date:2016-09-09,to_publication_date:2026-09-09", f"from_publication_date:{start},to_publication_date:{end}")) if family.endswith("-cursor"): initial.pop("page", None) initial["cursor"] = "*" current = initial for page in range(100): key = f"{family}-{page:03d}" result_path = DATA / f"{key}.result.json" if result_path.exists(): previous = json.loads(result_path.read_text()) if previous["pagination"]["exhausted"]: print(f"{family}: exhausted", flush=True) return current = previous["pagination"]["next_spec"] continue if not pages: return pages -= 1 path = SPEC / f"{key}.json" path.write_text(json.dumps(current, indent=2)+"\n") with db() as connection: before = {r[0] for r in connection.execute("SELECT id FROM canonical_work")} started = stamp() result = subprocess.run(["patchscholar", "fetch", "raw", provider, "search" if provider == "arxiv" else "works", "--spec-file", str(path)], cwd=ROOT, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) finished = stamp() ledger_path = ROOT / "searches/ledger.csv" with ledger_path.open() as stream: headers = next(csv.reader(stream)) entry = dict.fromkeys(headers, "") entry.update(search_id=f"{batch}-{key}-{started[:19]}", provider=provider, spec_file=str(path.relative_to(ROOT)), date_from=start, date_to=end, started_at=started, finished_at=finished) if result.returncode: diagnostics = [label for pattern, label in [(r"429|rate.limit", "rate limited"), (r"502|503|504", "provider gateway unavailable"), (r"timed? ?out|Timeout", "timeout"), (r"SSL|certificate", "TLS error"), (r"resolve|NameResolution", "DNS error"), (r"403|Forbidden", "access denied")] if re.search(pattern, result.stderr, re.I)] print(key + ": safe failure classification: " + (", ".join(diagnostics) or "unclassified provider failure"), flush=True) entry.update(status="failed", retrieved_count=0, new_candidates=0, continuation=json.dumps(current), notes=f"Provider command failed with exit {result.returncode}; no completion claimed. Provider stderr suppressed to avoid credential-bearing URLs.") else: payload = json.loads(result.stdout) result_path.write_text(json.dumps(payload, indent=2)+"\n") with db() as connection: after = {r[0] for r in connection.execute("SELECT id FROM canonical_work")} touched = [r[0] for r in connection.execute("SELECT DISTINCT w.canonical_work_id FROM work w JOIN source_record s ON s.id=w.raw_source_record_id WHERE s.payload_path=?", (payload["payload_path"],))] new_ids = sorted(after-before) (DATA / f"{key}.ids.json").write_text(json.dumps({"new":new_ids,"touched":touched})) entry.update(status="exhausted" if payload["pagination"]["exhausted"] else "partial", retrieved_count=payload["fetched_records"], new_candidates=len(new_ids), continuation=json.dumps(payload["pagination"]), notes=f"{result_path.relative_to(ROOT)}; new_candidates counts new canonical records; attachment and screening tracked separately. Existing reviews preserved.") print(f"{key}: retrieved={payload['fetched_records']} new={len(new_ids)} total={payload['pagination'].get('provider_total_results')} exhausted={payload['pagination']['exhausted']}", flush=True) with ledger_path.open("a", newline="") as stream: csv.DictWriter(stream, fieldnames=headers).writerow(entry) if result.returncode: raise SystemExit(f"Stopped {key} after recorded provider failure") if payload["pagination"]["exhausted"]: return current = payload["pagination"]["next_spec"] if not current: raise SystemExit("No continuation despite non-exhaustion") raise SystemExit("Safety bound of 100 pages reached") if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("family", choices=[*FAMILIES, *ADDITIONAL]) parser.add_argument("--pages", type=int, default=1) parser.add_argument("--start", default="2000-01-01") parser.add_argument("--end", default="2016-09-08") parser.add_argument("--batch", default="2026-09-11-backfill") args = parser.parse_args() run(args.family, args.pages, args.start, args.end, args.batch)