"""Парсер КиберЛенинки. КиберЛенинка — крупнейший русскоязычный открытый репозиторий научных статей. Сайт: https://cyberleninka.ru Используем парсинг HTML страниц со строгим rate limiting (1 req/sec), а не недокументированное API. """ import re import time import logging from typing import Any import httpx from bs4 import BeautifulSoup 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: 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: try: year = int(str(year_raw)[:4]) except (ValueError, TypeError): pass # 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