Radar académico: ETL sobre una API pública de papers

Mantenerse al día con la investigación académica es un problema de volumen. En áreas como machine learning o econometría, arXiv publica cientos de papers por semana. Leer todos es imposible; no leer ninguno es peligroso. Lo que necesitaba era un filtro: un sistema que me dijera qué se está publicando en las áreas que me interesan, quiénes son los autores más activos, y qué keywords están ganando tracción.

arXiv tiene una API pública. No requiere API key, no tiene tier de pago, y devuelve metadata completa en formato Atom XML. Es el punto de partida perfecto.

El diseño del schema: fact table pattern

El modelo de datos tiene que responder una pregunta temporal: "¿Qué papers aparecieron hoy para esta búsqueda, y cómo se compara con la semana pasada?". Eso requiere una fact table que registre cada aparición de un paper en un snapshot:

class ArxivQueryResult(models.Model):
    query      = models.ForeignKey(ArxivQuery, on_delete=models.CASCADE)
    paper      = models.ForeignKey(ArxivPaper, on_delete=models.CASCADE)
    rank       = models.PositiveIntegerField()
    scraped_at = models.DateTimeField()

    class Meta:
        unique_together = ("query", "paper", "scraped_at")
        indexes = [models.Index(fields=["query", "scraped_at"])]

El paper es una entidad deduplicada por arxiv_id — si el mismo paper aparece en dos queries distintas, solo se guarda una vez. Los autores y categorías son relaciones many-to-many:

class ArxivPaper(models.Model):
    arxiv_id     = models.CharField(max_length=50, unique=True)
    title        = models.TextField()
    abstract     = models.TextField()
    pdf_url      = models.URLField()
    published_at = models.DateTimeField()
    authors      = models.ManyToManyField(ArxivAuthor)
    categories   = models.ManyToManyField(ArxivCategory)

Las queries configurables (ArxivQuery) permiten activar y desactivar búsquedas sin tocar código. Cada query tiene un max_results ajustable y un flag is_active.

El ETL: respetando los límites

arXiv pide explícitamente 3 segundos entre requests. Respetar eso es importante — no solo por cortesía, sino porque si te bloquean, no hay plan B gratuito.

def run_full_arxiv_snapshot():
    queries = ArxivQuery.objects.filter(is_active=True)
    scraped_at = timezone.now()

    for query in queries:
        papers = _fetch_arxiv_papers(query.query_text, query.max_results)

        for rank, item in enumerate(papers, 1):
            paper = _upsert_paper(item)
            ArxivQueryResult.objects.get_or_create(
                query=query, paper=paper, scraped_at=scraped_at,
                defaults={"rank": rank}
            )

        time.sleep(3)  # Cortesía con arXiv

El _upsert_paper usa update_or_create con arxiv_id como clave natural. Si un paper ya existe, actualiza su abstract y fecha de modificación. Los autores y categorías se sincronizan con set() para evitar duplicados en las relaciones M2M.

El parser XML es stdlib puro — xml.etree.ElementTree, sin dependencias externas:

def _fetch_arxiv_papers(query_text, max_results):
    url = "https://export.arxiv.org/api/query"
    params = {
        "search_query": f"all:{query_text}",
        "start": 0,
        "max_results": max_results,
        "sortBy": "submittedDate",
        "sortOrder": "descending"
    }
    response = requests.get(url, params=params, timeout=30)
    root = ET.fromstring(response.content)
    # Parsear entries del feed Atom...

Los cuatro endpoints

La API expone cuatro vistas, todas protegidas con IsEmployeeUser:

1. POST /api/arxiv/snapshot/ — Dispara el ETL manualmente. Útil para actualizar sin esperar al próximo ciclo.

2. GET /api/arxiv/queries/ — Lista todas las queries configuradas (activas e inactivas). El frontend las usa para poblar un dropdown.

3. GET /api/arxiv/papers/?query_id=N — Papers del snapshot más reciente para una query. Incluye título, abstract truncado, autores, categorías y link al PDF.

4. GET /api/arxiv/stats/?query_id=N — Analíticas computadas on-the-fly:

def get(self, request):
    # Top 15 autores más frecuentes
    top_authors = Counter(author_names).most_common(15)

    # Top 30 keywords (stopwords EN+ES filtradas)
    words = [w for w in title_words if w.lower() not in STOPWORDS]
    top_keywords = Counter(words).most_common(30)

    # Papers por categoría arXiv
    papers_by_category = Counter(category_codes)

    # Evolución temporal: papers por snapshot
    papers_over_time = results.values("scraped_at").annotate(count=Count("id"))

Las stats se calculan en memoria con Counter, no con queries SQL complejas. Para el volumen de datos que manejo (cientos de papers, no millones), es más rápido y más legible.

Tests: 23 en el backend

No quería que un cambio futuro rompiera la deduplicación o los permisos. Los tests cubren:

• Permisos: anónimos y usuarios regulares no pueden acceder (403/401)
• Queries: devuelve activas e inactivas, formato correcto
• Papers: filtra por snapshot más reciente, autores como lista de strings, orden por rank
• Stats: autores rankeados por frecuencia, stopwords excluidas de keywords, query_id requerido
• Snapshot: mock del ETL, manejo de errores 500

El mock del ETL es importante — no quiero que los tests llamen a arXiv de verdad:

@patch("arxiv.views.run_full_arxiv_snapshot")
def test_snapshot_success(self, mock_etl):
    mock_etl.return_value = timezone.now()
    response = self.client.post("/api/arxiv/snapshot/")
    self.assertEqual(response.status_code, 200)

En la próxima entrega: el frontend con cuatro paneles de visualización y la integración en el hub de analytics.