anti-plagiarism/scripts/parsers/cyberleninka.py

"""Парсер КиберЛенинки.

КиберЛенинка — крупнейший русскоязычный открытый репозиторий научных статей.
Сайт: https://cyberleninka.ru

Используем парсинг HTML страниц со строгим rate limiting (1 req/sec),
а не недокументированное API.
"""

import contextlib
import logging
import re
import time
from typing import Any

import httpx
from base import BaseParser
from bs4 import BeautifulSoup

logger = logging.getLogger(__name__)

BASE_URL = "https://cyberleninka.ru"
SEARCH_URL = f"{BASE_URL}/api/search"
ARTICLE_URL = f"{BASE_URL}/article"
RATE_LIMIT_DELAY = 1.0  # секунд между запросами


class CyberLeninkaParser(BaseParser):
    """Парсер КиберЛенинки через HTML + API поиска."""

    source_name = "cyberleninka"

    def __init__(self) -> None:
        super().__init__()
        self.client = httpx.Client(
            headers={
                "User-Agent": "AcademicHelper/1.0 (noreply@jze9.ru)",
                "Accept-Language": "ru-RU,ru;q=0.9",
            },
            timeout=30.0,
            follow_redirects=True,
        )

    def fetch(
        self,
        query: str = "",
        limit: int = 500,
        subject: str | None = None,
    ) -> list[dict[str, Any]]:
        """
        Получить статьи из КиберЛенинки.

        Args:
            query: Поисковый запрос
            limit: Максимальное количество статей
            subject: Предметная область (опционально)

        Returns:
            Список сырых словарей статей
        """
        results = []
        page = 0

        while len(results) < limit:
            try:
                params: dict[str, Any] = {
                    "q": query,
                    "size": min(10, limit - len(results)),
                    "from": page * 10,
                }

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

                data = response.json()
                items = data.get("articles") or []

                if not items:
                    break

                results.extend(items)
                page += 1

                if len(items) < 10:
                    break

                time.sleep(RATE_LIMIT_DELAY)

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

        return results[:limit]

    def transform(self, raw: dict[str, Any]) -> dict[str, Any]:
        """
        Преобразовать статью КиберЛенинки в унифицированный формат.

        Структура ответа API поиска КиберЛенинки:
        - id: числовой ID
        - name: название
        - authors: строка с авторами
        - journal: название журнала
        - year: год
        - annotation: аннотация
        - link: путь к статье
        """
        raw_id = raw.get("id") or raw.get("link", "")
        ext_id = str(raw_id)

        if not ext_id:
            return {}

        # Авторы (строка вида "Иванов И.И., Петров П.П.")
        authors_str = raw.get("authors", "") or ""
        authors = _parse_cyberleninka_authors(authors_str)

        # Год
        year_raw = raw.get("year")
        year = None
        if year_raw:
            with contextlib.suppress(ValueError, TypeError):
                year = int(str(year_raw)[:4])

        # URL
        link = raw.get("link", "")
        url = f"{BASE_URL}{link}" if link.startswith("/") else link or None

        return {
            "source": self.source_name,
            "ext_id": ext_id,
            "doi": raw.get("doi") or None,
            "title": (raw.get("name") or "").strip() or None,
            "authors": authors,
            "year": year,
            "lang": "ru",  # КиберЛенинка — только русскоязычные
            "journal": raw.get("journal") or None,
            "volume": raw.get("volume") or None,
            "issue": raw.get("number") or None,
            "pages": raw.get("pages") or None,
            "abstract": (raw.get("annotation") or "").strip() or None,
            "url": url,
            "full_text": None,
        }

    def fetch_article_details(self, url: str) -> dict[str, Any]:
        """
        Получить детали статьи из HTML страницы.

        Извлекает метатеги citation_* по Dublin Core / Google Scholar стандарту.

        Args:
            url: URL страницы статьи

        Returns:
            Словарь с дополнительными метаданными
        """
        try:
            time.sleep(RATE_LIMIT_DELAY)
            response = self.client.get(url)
            response.raise_for_status()

            soup = BeautifulSoup(response.text, "html.parser")
            meta = {}

            # Извлечь citation_* мета теги
            for tag in soup.find_all("meta"):
                name = tag.get("name", "")
                content = tag.get("content", "")
                if name.startswith("citation_") and content:
                    key = name[9:]  # Убрать "citation_"
                    meta[key] = content

            return {
                "doi": meta.get("doi"),
                "title": meta.get("title"),
                "abstract": meta.get("abstract"),
                "year": meta.get("publication_date", "")[:4] or None,
                "journal": meta.get("journal_title"),
                "volume": meta.get("volume"),
                "issue": meta.get("issue"),
                "pages": f"{meta.get('firstpage', '')}-{meta.get('lastpage', '')}".strip("-") or None,
            }
        except Exception as e:
            logger.warning(f"Ошибка получения деталей статьи {url}: {e}")
            return {}


def _parse_cyberleninka_authors(authors_str: str) -> list[dict[str, str]]:
    """
    Разбить строку авторов КиберЛенинки на список.

    Форматы: "Иванов И.И.", "Иванов И.И., Петров П.П."
    """
    if not authors_str.strip():
        return []

    results = []
    # Разделитель — запятая, но не внутри инициалов
    parts = re.split(r",\s+(?=[А-ЯA-Z])", authors_str.strip())

    for part in parts:
        part = part.strip().rstrip(",")
        words = part.split()

        if not words:
            continue

        # Ищем паттерн: Фамилия Инициалы (последнее слово — инициалы)
        if len(words) >= 2:
            # Проверить, последнее ли слово — инициалы (содержит точки)
            if "." in words[-1]:
                last_name = " ".join(words[:-1])
                initials = words[-1]
            else:
                last_name = words[0]
                initials = "".join(w[0].upper() + "." for w in words[1:] if w)
        else:
            last_name = words[0]
            initials = ""

        results.append({
            "last_name": last_name,
            "first_name": "",
            "initials": initials,
        })

    return results