anti-plagiarism/scripts/parsers/arxiv.py

"""Парсер arXiv API.

arXiv — открытый препринт-сервер в физике, математике, CS и биологии.
API: https://info.arxiv.org/help/api/index.html

Использует Atom XML API.
"""

import time
import logging
import xml.etree.ElementTree as ET
from typing import Any

import httpx

from base import BaseParser

logger = logging.getLogger(__name__)

ARXIV_API = "https://export.arxiv.org/api/query"
RATE_LIMIT_DELAY = 3.0  # arXiv требует не более 1 запроса/3сек

# Namespace XML
NS = {
    "atom": "http://www.w3.org/2005/Atom",
    "arxiv": "http://arxiv.org/schemas/atom",
    "dc": "http://purl.org/dc/elements/1.1/",
    "opensearch": "http://a9.com/-/spec/opensearch/1.1/",
}


class ArxivParser(BaseParser):
    """Парсер arXiv API."""

    source_name = "arxiv"

    def __init__(self) -> None:
        super().__init__()
        self.client = httpx.Client(
            headers={"User-Agent": "AcademicHelper/1.0 (noreply@jze9.ru)"},
            timeout=30.0,
        )

    def fetch(
        self,
        query: str = "",
        limit: int = 500,
        categories: list[str] | None = None,
        year_from: int | None = None,
    ) -> list[dict[str, Any]]:
        """
        Получить препринты из arXiv.

        Args:
            query: Поисковый запрос
            limit: Максимальное количество документов
            categories: Список категорий arXiv (cs.AI, math.ST и т.д.)
            year_from: Год публикации от

        Returns:
            Список сырых словарей
        """
        results = []
        start = 0
        max_results = min(100, limit)

        # Формируем запрос
        search_query = ""
        if query:
            search_query = f"all:{query}"
        if categories:
            cat_query = " OR ".join(f"cat:{c}" for c in categories)
            search_query = f"({search_query}) AND ({cat_query})" if search_query else f"({cat_query})"

        if not search_query:
            search_query = "all:neural network"  # Default query

        while len(results) < limit:
            try:
                params = {
                    "search_query": search_query,
                    "start": start,
                    "max_results": min(max_results, limit - len(results)),
                    "sortBy": "submittedDate",
                    "sortOrder": "descending",
                }

                response = self.client.get(ARXIV_API, params=params)
                response.raise_for_status()

                entries = self._parse_xml(response.text)
                if not entries:
                    break

                results.extend(entries)
                start += len(entries)

                if len(entries) < max_results:
                    break

                time.sleep(RATE_LIMIT_DELAY)

            except httpx.HTTPStatusError as e:
                logger.error(f"arXiv HTTP ошибка: {e.response.status_code}")
                break
            except Exception as e:
                logger.error(f"Ошибка запроса arXiv: {e}")
                break

        return results[:limit]

    def _parse_xml(self, xml_text: str) -> list[dict[str, Any]]:
        """Парсить Atom XML ответ arXiv."""
        try:
            root = ET.fromstring(xml_text)
        except ET.ParseError as e:
            logger.error(f"Ошибка парсинга XML arXiv: {e}")
            return []

        entries = []
        for entry in root.findall("atom:entry", NS):
            entries.append(self._parse_entry(entry))

        return entries

    def _parse_entry(self, entry: ET.Element) -> dict[str, Any]:
        """Парсить один entry из Atom XML."""
        def text(path: str) -> str | None:
            elem = entry.find(path, NS)
            return elem.text.strip() if elem is not None and elem.text else None

        arxiv_id = text("atom:id") or ""
        # Нормализовать: убрать версию
        if "abs/" in arxiv_id:
            arxiv_id = arxiv_id.split("abs/")[-1].split("v")[0]

        # Авторы
        authors = []
        for author_elem in entry.findall("atom:author", NS):
            name = text("atom:name") if author_elem.find("atom:name", NS) is not None else None
            if name:
                parts = name.strip().split()
                if len(parts) >= 2:
                    authors.append({
                        "last_name": parts[-1],
                        "first_name": " ".join(parts[:-1]),
                    })
                elif parts:
                    authors.append({"last_name": parts[0], "first_name": ""})

        # Год из published
        published = text("atom:published") or ""
        year = int(published[:4]) if published[:4].isdigit() else None

        # DOI
        doi = None
        for link in entry.findall("atom:link", NS):
            if link.get("title") == "doi":
                doi = link.get("href", "").replace("http://dx.doi.org/", "")
                break

        # URL — предпочитаем прямую ссылку на PDF (её качает enrich_full_text),
        # иначе HTML-страницу аннотации
        url = None
        for link in entry.findall("atom:link", NS):
            if link.get("title") == "pdf" or link.get("type") == "application/pdf":
                url = link.get("href")
                break
        if not url:
            for link in entry.findall("atom:link", NS):
                if link.get("type") == "text/html":
                    url = link.get("href")
                    break

        # Категории
        categories = [
            cat.get("term", "")
            for cat in entry.findall("atom:category", NS)
        ]

        return {
            "id": arxiv_id,
            "title": text("atom:title") or "",
            "abstract": text("atom:summary") or "",
            "authors": authors,
            "year": year,
            "published": published,
            "doi": doi,
            "url": url,
            "categories": categories,
            "journal": "arXiv",
        }

    def transform(self, raw: dict[str, Any]) -> dict[str, Any]:
        """Преобразовать документ arXiv в унифицированный формат."""
        ext_id = raw.get("id", "")
        if not ext_id:
            return {}

        # Нормализовать авторов
        authors = self.normalize_authors(raw.get("authors", []))

        # Определить язык (arXiv — преимущественно английский)
        lang = "en"

        # Категории как JSON
        categories = raw.get("categories", [])

        return {
            "source": self.source_name,
            "ext_id": f"arxiv:{ext_id}",
            "doi": raw.get("doi") or None,
            "title": (raw.get("title") or "").replace("\n", " ").strip() or None,
            "authors": authors,
            "year": raw.get("year"),
            "lang": lang,
            "journal": "arXiv",
            "volume": None,
            "issue": None,
            "pages": None,
            "abstract": (raw.get("abstract") or "").replace("\n", " ").strip() or None,
            "url": raw.get("url"),
            "full_text": None,
        }