anti-plagiarism/scripts/parsers/cyberleninka.py

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

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

Поиск — через JSON-API `POST /api/search` (`{"mode":"articles", "q", "size", "from"}`)
со строгим rate limiting (1 req/sec). Ответ подсвечивает совпадения тегами <b>…</b>
в name/annotation — чистим при трансформации.
"""

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

import httpx
from base import BaseParser, ProgressCallback

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,
        progress_cb: ProgressCallback | None = None,
    ) -> list[dict[str, Any]]:
        """
        Получить статьи из КиберЛенинки.

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

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

        while len(results) < limit:
            try:
                # /api/search КиберЛенинки — это POST с JSON-телом (GET даёт 405).
                # mode=articles обязателен, иначе поиск не по статьям.
                payload: dict[str, Any] = {
                    "mode": "articles",
                    "q": query,
                    "size": min(10, limit - len(results)),
                    "from": page * 10,
                }

                response = self.client.post(SEARCH_URL, json=payload)
                response.raise_for_status()

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

                if not items:
                    break

                results.extend(items)
                page += 1

                if progress_cb and not progress_cb(min(len(results), limit)):
                    logger.info(
                        f"КиберЛенинка: выборка остановлена по запросу (получено {len(results)})"
                    )
                    break

                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 {}

        # Авторы: API отдаёт список строк "Фамилия И. О." (иногда — одну строку)
        raw_authors = raw.get("authors") or []
        authors = (
            _parse_cyberleninka_authors(raw_authors)
            if isinstance(raw_authors, str)
            else _authors_from_list(raw_authors)
        )

        # Год
        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

        # OCR-фрагменты текста статьи — приходят прямо в ответе поиска (список
        # кусков, обычно начало статьи + фрагмент с ключевыми словами), без
        # доп. запроса. Не полный текст всей статьи, но заметно богаче аннотации —
        # используем как full_text для более точных Winnowing-отпечатков (L1).
        ocr = raw.get("ocr")
        full_text = _clean(" ".join(ocr)) if isinstance(ocr, list) and ocr else None

        return {
            "source": self.source_name,
            "ext_id": ext_id,
            "doi": raw.get("doi") or None,
            "title": _clean(raw.get("name")) 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": _clean(raw.get("annotation")) or None,
            "url": url,
            "full_text": full_text,
        }

    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()

            # Ленивый импорт: заливке (fetch/transform) bs4 не нужен, только деталям статьи
            from bs4 import BeautifulSoup

            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 _clean(text: str | None) -> str:
    """Убрать HTML-теги подсветки (<b>…</b>), декодировать сущности (&quot;), обрезать."""
    if not text:
        return ""
    return html.unescape(re.sub(r"<[^>]+>", "", str(text))).strip()


def _authors_from_list(items: list) -> list[dict[str, str]]:
    """Авторы из списка строк 'Фамилия И. О.' → [{last_name, initials}, ...]."""
    authors: list[dict[str, str]] = []
    for item in items:
        name = _clean(str(item))
        words = name.split()
        if not words:
            continue
        initials = " ".join(words[1:]) if len(words) >= 2 else ""
        authors.append({"last_name": words[0], "initials": initials})
    return authors


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