"""Парсер КиберЛенинки. КиберЛенинка — крупнейший русскоязычный открытый репозиторий научных статей. Сайт: https://cyberleninka.ru Поиск — через JSON-API `POST /api/search` (`{"mode":"articles", "q", "size", "from"}`) со строгим rate limiting (1 req/sec). Ответ подсвечивает совпадения тегами … в name/annotation — чистим при трансформации. """ import contextlib import html import logging import re import time from typing import Any import httpx from base import BaseParser 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: # /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 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-теги подсветки (…), декодировать сущности ("), обрезать.""" 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