"""arXiv — Atom API listing (public document, not a commercial API) + category RSS feeds. Tier 1 (the papers' own metadata). * https://rss.arxiv.org/rss/cs.LG (+ cs.CL, cs.AI, cs.CV) — daily announcements (new / cross / replace), ~250–600 papers per feed. * https://export.arxiv.org/api/query?search_query=cat:cs.LG+OR+cat:cs.CL+OR+cat:cs.AI+OR+cat:cs.CV&sortBy=submittedDate&sortOrder=descending paginated with `start`/`max_results=200` (≤ 5 pages) — arXiv asks for ≤ 1 request every 3 s → 4/min. **Opt-in** (`config["atom_pages"] > 0`): on 2026-09-11 `https://export.arxiv.org/robots.txt` is `User-agent: * / Disallow: /`, so the SDK's robots policy blocks the Atom API (3 `blocked_source` review items per run). The RSS feeds are not disallowed and already exceed `expected_min_records`. Paper entities are keyed by the version-less arXiv id (`{"arxiv": "2509.01234"}`). Authors are a claim on the paper; a `researcher` entity is created only when the feed carries an identifier (ORCID) — never from a bare name. PDFs are not fetched (`needs_llm=False`); a later job builds paper passports. """ from __future__ import annotations import re from datetime import UTC, datetime from typing import Any import feedparser from aiatlas.sdk.connector import BaseConnector, Parsed, RunContext from aiatlas.sdk.extract.dates import parse_datetime from aiatlas.sdk.extract.html import clean_text from aiatlas.sdk.facts import EntityRef, Facts, Target from aiatlas.sdk.fetch import FetchResult ATOM = "https://export.arxiv.org/api/query" RSS = "https://rss.arxiv.org/rss/{cat}" DEFAULT_QUERY = "cat:cs.LG+OR+cat:cs.CL+OR+cat:cs.AI+OR+cat:cs.CV" ARXIV_ID = re.compile(r"(\d{4}\.\d{4,5})(v\d+)?") OLD_ID = re.compile(r"([a-z-]+(?:\.[A-Z]{2})?/\d{7})(v\d+)?") ANNOUNCE = re.compile(r"^arXiv:\S+\s+Announce Type:\s*(\w+)\s*", re.IGNORECASE) class ArxivConnector(BaseConnector): name = "arxiv" label = "arXiv — recent cs.LG / cs.CL / cs.AI / cs.CV papers (Atom API + RSS)" description = "Atom listing of the newest submissions in the core ML categories and the daily category RSS feeds." source_key = "arxiv.org" version = "2" parser_version = "2" interval_seconds = 6 * 3600 min_interval_seconds = 2 * 3600 max_interval_seconds = 86400 rate_per_min = 4 # Documented exception: export.arxiv.org/robots.txt disallows generic crawlers, but arXiv's API Terms of Use designate this host # for programmatic access (≤ 1 request / 3 s). We stay at 4 requests / min, identify ourselves and honour the rate. respect_robots = False tier = 1 priority = 1 expected_min_records = 100 concurrency = 1 needs_llm = False async def discover(self, ctx: RunContext) -> list[Target]: pages = min(5, max(0, int(self.config.get("atom_pages", self.config.get("pages", 0))))) per_page = int(self.config.get("max_results", 200)) query = self.config.get("query", DEFAULT_QUERY) targets = [Target(url=f"{ATOM}?search_query={query}&sortBy=submittedDate&sortOrder=descending&start={i * per_page}&max_results={per_page}", doc_type="feed", key=f"atom:{i}", min_bytes=2000, priority=1, meta={"page": i}) for i in range(pages)] for cat in self.config.get("rss_categories", ["cs.LG", "cs.CL", "cs.AI", "cs.CV"]): targets.append(Target(url=RSS.format(cat=cat), doc_type="feed", key=f"rss:{cat}", min_bytes=2000, priority=2, meta={"category": cat})) return targets async def extract(self, ctx: RunContext, target: Target, res: FetchResult, parsed: Parsed) -> Facts: facts = Facts() feed = feedparser.parse(res.content) is_rss = (target.key or "").startswith("rss:") for e in feed.entries: self._paper(facts, e, is_rss=is_rss) facts.document_title = clean_text(feed.feed.get("title") or "") or None return facts def _paper(self, facts: Facts, e: Any, *, is_rss: bool) -> None: raw_id = e.get("id") or e.get("link") or "" arxiv_id = _arxiv_id(raw_id) or _arxiv_id(e.get("link") or "") title = clean_text(re.sub(r"\s+", " ", e.get("title") or "")).strip() if not arxiv_id or not title: return summary = e.get("summary") or "" announce = None if is_rss: m = ANNOUNCE.match(summary) if m: announce = m.group(1).lower() summary = summary[m.end():] summary = re.sub(r"^Abstract:\s*", "", summary.strip(), flags=re.IGNORECASE) abstract = clean_text(re.sub(r"\s+", " ", summary)).strip() published = _dt(e.get("published_parsed")) or parse_datetime(e.get("published")) updated = (_dt(e["updated_parsed"]) if "updated_parsed" in e else None) or (parse_datetime(e["updated"]) if "updated" in e else None) authors = _authors(e) categories = [t.get("term") for t in e.get("tags", []) if isinstance(t, dict) and t.get("term")] primary = (e.get("arxiv_primary_category") or {}).get("term") if isinstance(e.get("arxiv_primary_category"), dict) else None ref = facts.entity("paper", title[:300], identifiers={"arxiv": arxiv_id}, first_seen_hint=published) facts.claim(ref, "arxiv_id", arxiv_id) facts.claim(ref, "authors", authors[:100]) facts.claim(ref, "published_at", published.isoformat(timespec="seconds") if published else None) if updated and (not published or updated != published): facts.claim(ref, "updated_at", updated.isoformat(timespec="seconds")) facts.claim(ref, "abstract", abstract[:6000] or None) facts.claim(ref, "categories", categories) facts.claim(ref, "primary_category", primary or (categories[0] if categories else None)) facts.claim(ref, "pdf_url", f"https://arxiv.org/pdf/{arxiv_id}") facts.claim(ref, "official_url", f"https://arxiv.org/abs/{arxiv_id}") doi = e.get("arxiv_doi") facts.claim(ref, "doi", doi.strip() if isinstance(doi, str) and doi.strip() else None) comment = e.get("arxiv_comment") facts.claim(ref, "comment", clean_text(comment)[:1000] if isinstance(comment, str) and comment.strip() else None) jref = e.get("arxiv_journal_ref") facts.claim(ref, "journal_ref", clean_text(jref)[:500] if isinstance(jref, str) and jref.strip() else None) if announce: facts.claim(ref, "arxiv_announce_type", announce) # researchers only with an identifier: arXiv feeds carry no ORCID / author id, so authors stay a claim on the paper (homonyms # would merge and spelling variants split name-only entities). An ORCID in the feed (rare `arxiv:author` extension) creates one. for author in (e.get("authors") or [])[:20]: orcid = _orcid(author) if orcid: person = EntityRef(entity_type="researcher", name=(author.get("name") or orcid)[:200], identifiers={"orcid": orcid}, identity_confidence="high") facts.entities.append(person) facts.relate(person, "authored", ref) if doi and isinstance(doi, str) and doi.strip(): ref.identifiers["doi"] = doi.strip() ORCID = re.compile(r"(\d{4}-\d{4}-\d{4}-\d{3}[\dX])") def _orcid(author: Any) -> str | None: if not isinstance(author, dict): return None for key in ("orcid", "arxiv_orcid", "uri", "href", "id"): v = author.get(key) if isinstance(v, str) and "orcid.org" in v.lower() or (isinstance(v, str) and key == "orcid"): m = ORCID.search(v) if m: return m.group(1) return None def _arxiv_id(s: str) -> str | None: m = ARXIV_ID.search(s) if m: return m.group(1) m = OLD_ID.search(s.replace("http://arxiv.org/abs/", "").replace("https://arxiv.org/abs/", "")) return m.group(1) if m else None def _authors(e: Any) -> list[str]: names: list[str] = [] for a in e.get("authors", []) or []: n = a.get("name") if isinstance(a, dict) else None if n: names.extend(_split_names(n)) if not names and e.get("author"): names = _split_names(e["author"]) seen: set[str] = set() out: list[str] = [] for n in names: n = clean_text(n).strip() if n and n.lower() not in seen: seen.add(n.lower()) out.append(n) return out def _split_names(s: str) -> list[str]: # RSS dc:creator packs every author into one string: "Wenzhe Jin, Haina Tang" parts = re.split(r",\s+|\s+and\s+", s) return [p.strip() for p in parts if p.strip()] def _dt(struct: Any) -> datetime | None: if not struct: return None try: return datetime(*struct[:6], tzinfo=UTC) except Exception: # noqa: BLE001 return None CONNECTORS = [ArxivConnector]