<a href="http://export.arxiv.org/api/query?search_query=...&start=0&max_results=2000&sortBy=lastUpdatedDate">http://export.arxiv.org/api/query?search_query=...&start=0&max_results=2000&sortBy=lastUpdatedDate</a>

<arxiv:id>2402.12345v1</arxiv:id>
<arxiv:authors>
  <arxiv:author>
    <arxiv:name>John Doe</arxiv:name>
    <arxiv:affiliation>MIT</arxiv:affiliation>
  </arxiv:author>
</arxiv:authors>
<arxiv:title>Advances in Neural Network Training</arxiv:title>
<arxiv:summary>Abstract text...</arxiv:summary>
<arxiv:published>2024-02-15T18:30:00Z</arxiv:published>
<arxiv:updated>2024-02-20T10:15:00Z</arxiv:updated>
<arxiv:primary_category>cs.LG</arxiv:primary_category>
<arxiv:categories>
  <arxiv:category>cs.LG</arxiv:category>
  <arxiv:category>math.OC</arxiv:category>
</arxiv:categories>
<arxiv:doi>10.48550/arXiv.2402.12345</arxiv:doi>
<arxiv:journal_reference>ICML 2024</arxiv:journal_reference>
<arxiv:comment>20 pages, 5 figures</arxiv:comment>

pip install requests feedparser xmltodict

#!/usr/bin/env python3
"""
arXiv OAI-PMH Metadata Harvester POC
Fetches papers from cs.LG + math.OC, saves to JSON
"""
import json
import time
from datetime import datetime, timedelta
import requests
from xml.etree import ElementTree as ET

# Configuration
ARXIV_OAI_URL = "http://export.arxiv.org/oai2"
CATEGORIES = ["cs.LG", "math.OC"]  # Primary categories
OUTPUT_FILE = "arxiv_papers.json"
PAPERS_TO_FETCH = 100
REQUEST_DELAY = 3  # seconds between requests (conservative)

def fetch_oai_records(category, start_date=None, max_papers=100):
    """
    Fetch metadata from arXiv via OAI-PMH ListRecords verb.
    Returns list of dicts with paper metadata.
    """
    papers = []
    params = {
        "verb": "ListRecords",
        "metadataPrefix": "oai_dc",  # Dublin Core (simple, widely supported)
        "set": f"cs.archive:{category}",  # Category filter
    }

    if start_date:
        params["from"] = start_date  # ISO 8601: YYYY-MM-DD

    token_url = ARXIV_OAI_URL

    while len(papers) < max_papers:
        print(f"[{category}] Fetching batch (total: {len(papers)})...")

        try:
            resp = requests.get(ARXIV_OAI_URL, params=params, timeout=10)
            resp.raise_for_status()
        except Exception as e:
            print(f"Error fetching: {e}")
            break

        # Parse XML response
        root = ET.fromstring(resp.content)
        ns = {
            "oai": "http://www.openarchives.org/OAI/2.0/",
            "dc": "http://purl.org/dc/elements/1.1/",
            "arxiv": "http://arxiv.org/OAI/arXivRaw/"
        }

        # Extract records
        for record in root.findall(".//oai:record", ns):
            if len(papers) >= max_papers:
                break

            header = record.find(".//oai:header", ns)
            metadata = record.find(".//oai:metadata", ns)

            if metadata is None or header.find(".//oai:status", ns) is not None:
                # Skip deleted records
                continue

            # Extract fields (DC format)
            arxiv_elem = metadata.find("arxiv:arXivRaw", ns)
            if arxiv_elem is not None:
                # arXiv extended format
                paper = {
                    "arxiv_id": header.find("oai:identifier", ns).text.replace("oai:arXiv:", ""),
                    "title": (arxiv_elem.find("title").text or "").strip(),
                    "authors": [
                        a.text for a in arxiv_elem.findall("authors/author")
                    ],
                    "summary": (arxiv_elem.find("summary").text or "").strip(),
                    "published": arxiv_elem.find("datestamp").text,
                    "categories": arxiv_elem.find("categories").text.split(),
                    "doi": arxiv_elem.find("doi").text if arxiv_elem.find("doi") is not None else None,
                }
            else:
                # Fallback to Dublin Core (basic)
                dc = metadata.find(".//dc:creator", ns)
                paper = {
                    "arxiv_id": header.find("oai:identifier", ns).text.replace("oai:arXiv:", ""),
                    "title": (metadata.find(".//dc:title", ns).text or "").strip(),
                    "authors": [a.text for a in metadata.findall(".//dc:creator", ns)],
                    "summary": (metadata.find(".//dc:description", ns).text or "").strip(),
                    "published": header.find("oai:datestamp", ns).text,
                    "categories": [metadata.find(".//dc:subject", ns).text] if metadata.find(".//dc:subject", ns) is not None else [],
                    "doi": None,
                }

            papers.append(paper)

        # Check for resumption token (pagination)
        resumption = root.find(".//oai:resumptionToken", ns)
        if resumption is None or resumption.text is None or len(papers) >= max_papers:
            break

        # Continue with next batch
        params = {"verb": "ListRecords", "resumptionToken": resumption.text}
        time.sleep(REQUEST_DELAY)

    return papers

def fetch_via_search_api(query, max_papers=100):
    """
    Alternative: Fetch via arXiv Search API.
    More flexible for complex queries.
    """
    papers = []
    api_url = "http://export.arxiv.org/api/query"

    params = {
        "search_query": query,
        "start": 0,
        "max_results": min(2000, max_papers),
        "sortBy": "lastUpdatedDate",
        "sortOrder": "descending"
    }

    print(f"Fetching via API: {query}")

    try:
        resp = requests.get(api_url, params=params, timeout=10)
        resp.raise_for_status()
    except Exception as e:
        print(f"Error: {e}")
        return papers

    root = ET.fromstring(resp.content)
    ns = {"atom": "http://www.w3.org/2005/Atom", "arxiv": "http://arxiv.org/schemas/atom"}

    for entry in root.findall("atom:entry", ns):
        if len(papers) >= max_papers:
            break

        paper = {
            "arxiv_id": entry.find("atom:id", ns).text.split("/abs/")[-1],
            "title": entry.find("atom:title", ns).text.strip(),
            "authors": [a.find("atom:name", ns).text for a in entry.findall("atom:author", ns)],
            "summary": entry.find("atom:summary", ns).text.strip(),
            "published": entry.find("atom:published", ns).text,
            "categories": [
                c.get("term") for c in entry.findall("atom:category", ns)
            ],
            "doi": entry.find("arxiv:doi", ns).text if entry.find("arxiv:doi", ns) is not None else None,
        }
        papers.append(paper)

    return papers

def main():
    """Main: Fetch papers from multiple categories, save as JSON."""
    all_papers = []
    papers_per_cat = PAPERS_TO_FETCH // len(CATEGORIES)

    print(f"Fetching {PAPERS_TO_FETCH} papers from {CATEGORIES}")
    print(f"({papers_per_cat} per category)\n")

    # Method 1: OAI-PMH (Recommended)
    print("<mark>= Method 1: OAI-PMH (Recommended) </mark>=")
    for cat in CATEGORIES:
        cat_papers = fetch_oai_records(cat, max_papers=papers_per_cat)
        all_papers.extend(cat_papers)
        print(f"  {cat}: {len(cat_papers)} papers\n")
        time.sleep(REQUEST_DELAY)

    # Method 2: Search API (Alternative)
    # query = "cat:cs.LG AND cat:math.OC"
    # api_papers = fetch_via_search_api(query, max_papers=50)
    # all_papers.extend(api_papers)

    # Save to JSON
    with open(OUTPUT_FILE, "w") as f:
        json.dump(all_papers, f, indent=2, default=str)

    print(f"\nSaved {len(all_papers)} papers to {OUTPUT_FILE}")
    print(f"Sample paper:\n{json.dumps(all_papers[0] if all_papers else {}, indent=2)}")

if __name__ == "__main__":
    main()

[
  {
    "arxiv_id": "2402.12345",
    "title": "Efficient Gradient Estimation in Neural Networks",
    "authors": ["Alice Smith", "Bob Jones"],
    "summary": "We propose a novel method for...",
    "published": "2024-02-15T18:30:00Z",
    "categories": ["cs.LG", "math.OC"],
    "doi": "10.48550/arXiv.2402.12345"
  },
  ...
]

python arxiv_poc.py
# Output: arxiv_papers.json with 100 papers + metadata

Daily Ingest Pipeline:
┌─────────────────────────┐
│  OAI-PMH Full Harvest   │  (Tier 1: cs.LG, math.OC, stat.ML)
│   Nightly @ 02:00 MSK   │  2-3 hours, ~100-200 new papers
└────────┬────────────────┘
         │
         ▼
┌─────────────────────────┐
│   Parse & Deduplicate   │
│   (Check arxiv_id)      │
└────────┬────────────────┘
         │
         ▼
┌─────────────────────────┐
│  Insert to базаатей    │
│  (SQLite / PostgreSQL)  │
└─────────────────────────┘

Hourly Incremental Updates:
┌─────────────────────────┐
│  Search API Query       │  (Targeted: "new papers in cat:cs.LG")
│   Hourly @ :00 MSK      │  15-30 min, ~8-16 new papers
└────────┬────────────────┘
         │
         ▼
┌─────────────────────────┐
│   Merge Results         │
│   (Avoid duplicates)    │
└─────────────────────────┘

## arXiv Bulk Metadata Import

Use OAI-PMH to keep a real-time database of neural network optimization papers.

### One-Time Setup

1. Install dependencies:
   ```bash
   pip install requests feedparser xmltodict
   ```

2. Create database schema (SQLite):
   ```sql
   CREATE TABLE arxiv_papers (
     arxiv_id TEXT PRIMARY KEY,
     title TEXT,
     authors TEXT,  -- JSON array
     abstract TEXT,
     published TEXT,
     updated TEXT,
     categories TEXT,  -- JSON array
     doi TEXT,
     journal_reference TEXT,
     fetched_at TIMESTAMP
   );
   CREATE INDEX idx_arxiv_cat ON arxiv_papers(categories);
   CREATE INDEX idx_arxiv_published ON arxiv_papers(published);
   ```

3. Run initial harvest:
   ```bash
   python scripts/arxiv_harvest.py --mode=full --categories=cs.LG,math.OC,stat.ML
   # Takes ~2 hours for ~100k papers
   ```

### Automated Daily Updates

Add to cron (02:00 MSK every night):
```bash
0 2 * * * python /root/scripts/arxiv_harvest.py --mode=incremental

import sqlite3
conn = sqlite3.connect('arxiv_papers.db')
# Get recent papers in optimization
papers = conn.execute(
    "SELECT title, authors, published FROM arxiv_papers "
    "WHERE categories LIKE '%math.OC%' "
    "ORDER BY published DESC LIMIT 10"
).fetchall()