#!/usr/bin/env python3 """ openalex_citation_crawler.py BFS citation graph crawler using the OpenAlex API — completely free, no key, 50 req/sec in the polite pool (just pass your email as mailto= param). Also provides a Barabási-Albert synthetic graph for instant offline testing when no network is available (or when you just want a known structure). Feeds into pagerank_eigenvalue_survey.run_survey() for spectral bin analysis. Usage: # Real citation graph (online): python openalex_citation_crawler.py --seed "PageRank Brin Page 1998" --hops 2 # Synthetic scale-free graph (offline, instant): python openalex_citation_crawler.py --synthetic --nodes 300 # Save and survey: python openalex_citation_crawler.py --seed "HITS Kleinberg 1999" --hops 2 --out hits_graph.json Requires: requests, networkx, scipy, numpy uv pip install requests networkx scipy numpy """ import argparse import json import sys import time from collections import deque from dataclasses import dataclass, field, asdict from pathlib import Path from typing import List, Dict, Optional, Set import requests import networkx as nx import numpy as np sys.path.insert(0, str(Path(__file__).parent)) try: from pagerank_eigenvalue_survey import run_survey HAS_SURVEY = True except ImportError: HAS_SURVEY = False print("[warn] pagerank_eigenvalue_survey not found — will print basic PageRank only") # ── OpenAlex API ────────────────────────────────────────────────────────────── OA_BASE = "https://api.openalex.org" # Using the project email puts requests into the polite pool (50 req/sec). OA_EMAIL = "allaunjsilverfox@gmail.com" @dataclass class OAPaper: work_id: str title: str year: Optional[int] cite_count: int doi: Optional[str] references: List[str] = field(default_factory=list) # OpenAlex work IDs citations: List[str] = field(default_factory=list) # OpenAlex work IDs class OAClient: def __init__(self, rate_delay: float = 0.05): # 20 req/sec, well under 50 self._last = 0.0 self._delay = rate_delay self._stats = {"requests": 0, "errors": 0} def _get(self, path: str, params: Dict = None) -> Optional[dict]: elapsed = time.monotonic() - self._last if elapsed < self._delay: time.sleep(self._delay - elapsed) self._last = time.monotonic() self._stats["requests"] += 1 p = {"mailto": OA_EMAIL, **(params or {})} try: resp = requests.get(f"{OA_BASE}{path}", params=p, timeout=15) if resp.status_code == 429: print(" [rate-limit] waiting 30s") time.sleep(30) return self._get(path, params) resp.raise_for_status() return resp.json() except Exception as exc: self._stats["errors"] += 1 print(f" [oa error] {path}: {exc}") return None def search(self, query: str, limit: int = 5) -> List[dict]: data = self._get("/works", { "search": query, "per-page": limit, "select": "id,title,publication_year,cited_by_count,doi", }) return data.get("results", []) if data else [] def paper(self, work_id: str) -> Optional[OAPaper]: # work_id may be full URL or short form like W2741809807 short_id = work_id.replace("https://openalex.org/", "") data = self._get(f"/works/{short_id}", { "select": "id,title,publication_year,cited_by_count,doi,referenced_works" }) if not data or "id" not in data: return None return OAPaper( work_id=data["id"], title=data.get("title") or "", year=data.get("publication_year"), cite_count=data.get("cited_by_count", 0), doi=data.get("doi"), references=[r for r in data.get("referenced_works", []) if r], ) def citing(self, work_id: str, limit: int = 50) -> List[str]: """IDs of papers that cite work_id.""" short_id = work_id.replace("https://openalex.org/", "") data = self._get("/works", { "filter": f"cites:{short_id}", "per-page": limit, "select": "id", "sort": "cited_by_count:desc", # take the most-cited citers first }) if not data: return [] return [r["id"] for r in data.get("results", []) if r.get("id")] def stats(self) -> str: return f"requests={self._stats['requests']} errors={self._stats['errors']}" # ── BFS ─────────────────────────────────────────────────────────────────────── def crawl_openalex(seed_id: str, hops: int, max_refs: int, max_cites: int, max_papers: int) -> Dict[str, OAPaper]: client = OAClient() visited: Set[str] = set() papers: Dict[str, OAPaper] = {} queue: deque = deque([(seed_id, 0)]) print(f"\nOpenAlex BFS: seed={seed_id} hops={hops} max={max_papers}") while queue and len(papers) < max_papers: wid, depth = queue.popleft() if wid in visited: continue visited.add(wid) paper = client.paper(wid) if not paper: continue papers[wid] = paper print(f" [{depth}/{hops}] {len(papers):>4} cite={paper.cite_count:>6}" f" {paper.title[:65]}") if depth < hops: # References are already in the paper response ref_ids = paper.references[:max_refs] cite_ids = client.citing(wid, limit=max_cites) paper.citations = cite_ids for nid in ref_ids + cite_ids: if nid and nid not in visited: queue.append((nid, depth + 1)) print(f"\nDone: {len(papers)} papers | {client.stats()}") return papers # ── Synthetic scale-free graph ──────────────────────────────────────────────── def build_synthetic_webgraph(n: int = 300, m: int = 3, spam_nodes: int = 15, seed: int = 42) -> nx.DiGraph: """ Barabási-Albert preferential attachment graph → directed citation graph. Properties matching the real web: - Power-law degree distribution (few hubs, many leaves) - Small-world connectivity - Natural community structure emerges at n≥200 Spam injection: `spam_nodes` low-degree nodes → hub node 0 (artificial authority boost, should trigger bad-link gate on merger). """ rng = np.random.default_rng(seed) ug = nx.barabasi_albert_graph(n, m, seed=seed) g = ug.to_directed() # Prune ~30% of back-edges to make it more citation-like (mostly one-way) edges_to_remove = [(u, v) for u, v in list(g.edges()) if u > v and rng.random() < 0.7] g.remove_edges_from(edges_to_remove) # Add spam link farm → node 0 (highest PageRank target) for i in range(n, n + spam_nodes): g.add_node(i, spam=True) g.add_edge(i, 0) if i > n: g.add_edge(i, i - 1) # farm cross-links print(f"Synthetic graph: {g.number_of_nodes()} nodes, " f"{g.number_of_edges()} edges (BA n={n} m={m} + {spam_nodes} spam)") return g # ── Build NetworkX graph from OA papers ────────────────────────────────────── def oa_to_networkx(papers: Dict[str, OAPaper]) -> nx.DiGraph: g = nx.DiGraph() id_to_int = {wid: i for i, wid in enumerate(papers)} for wid, p in papers.items(): nid = id_to_int[wid] g.add_node(nid, work_id=wid, title=p.title[:80], year=p.year or 0, cite_count=p.cite_count) for wid, p in papers.items(): src = id_to_int[wid] for ref_id in p.references: if ref_id in id_to_int: g.add_edge(src, id_to_int[ref_id]) # src cites ref print(f"Graph: {g.number_of_nodes()} nodes, {g.number_of_edges()} edges") return g # ── Lightweight survey fallback ─────────────────────────────────────────────── def basic_survey(g: nx.DiGraph): """Minimal survey when pagerank_eigenvalue_survey is not importable.""" pr = nx.pagerank(g, alpha=0.85) try: hubs, auths = nx.hits(g, max_iter=300) except Exception: hubs = auths = {n: 1 / len(g) for n in g} top = sorted(pr.items(), key=lambda x: -x[1])[:15] print("\n[PageRank top 15]") for nid, score in top: title = g.nodes[nid].get("title", f"node-{nid}") year = g.nodes[nid].get("year", "") print(f" {score:.5f} [{year}] {title}") top_auth = sorted(auths.items(), key=lambda x: -x[1])[:10] print("\n[HITS authority top 10]") for nid, score in top_auth: title = g.nodes[nid].get("title", f"node-{nid}") print(f" {score:.5f} {title}") # ── Main ────────────────────────────────────────────────────────────────────── def main(): ap = argparse.ArgumentParser( description="OpenAlex BFS citation crawler + spectral survey (no API key needed)") mode = ap.add_mutually_exclusive_group() mode.add_argument("--seed", default="PageRank Brin Page 1998", help="Title/author query (default: PageRank paper)") mode.add_argument("--seed-id", help="OpenAlex work ID e.g. W2741809807") mode.add_argument("--synthetic", action="store_true", help="Use synthetic Barabási-Albert graph (offline, instant)") ap.add_argument("--nodes", type=int, default=300, help="Synthetic graph node count") ap.add_argument("--hops", type=int, default=2) ap.add_argument("--max-refs", type=int, default=40) ap.add_argument("--max-cites", type=int, default=30) ap.add_argument("--max-papers", type=int, default=300) ap.add_argument("--out", default=None, help="Save paper JSON here") ap.add_argument("--no-survey", action="store_true") args = ap.parse_args() if args.synthetic: g = build_synthetic_webgraph(n=args.nodes) else: client = OAClient() if args.seed_id: seed_id = args.seed_id if not seed_id.startswith("https://"): seed_id = f"https://openalex.org/{seed_id}" else: print(f"Searching OpenAlex: {args.seed}") results = client.search(args.seed, limit=3) if not results: print("ERROR: no results") sys.exit(1) best = results[0] seed_id = best["id"] print(f" seed: [{best.get('publication_year','')}] {best.get('title','?')}") print(f" id: {seed_id}") papers = crawl_openalex( seed_id, args.hops, args.max_refs, args.max_cites, args.max_papers) if args.out: Path(args.out).write_text( json.dumps({wid: asdict(p) for wid, p in papers.items()}, indent=2)) print(f"Saved → {args.out}") g = oa_to_networkx(papers) if not args.no_survey: if HAS_SURVEY: run_survey(g) else: basic_survey(g) if __name__ == "__main__": main()