6099ef621f
Прод-воркер падал на index.run_parser с 'No module named bs4': парсер импортил beautifulsoup4 на верхнем уровне, а в образе worker-indexer его нет. Но bs4 нужен только для fetch_article_details (детали статьи), а заливке (fetch+transform) — нет. - импорт bs4 сделан ленивым (внутри fetch_article_details) → заливка работает даже без bs4 в образе; - beautifulsoup4 добавлен в worker-indexer/requirements.txt (для деталей статьи). Это была вторая причина, почему CyberLeninka никогда не наполняла базу (первая — GET вместо POST, 405). Проверено вживую: fetch без bs4 в пути работает. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
263 lines
9.1 KiB
Python
263 lines
9.1 KiB
Python
"""Парсер КиберЛенинки.
|
||
|
||
КиберЛенинка — крупнейший русскоязычный открытый репозиторий научных статей.
|
||
Сайт: 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
|
||
|
||
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
|
||
|
||
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": 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()
|
||
|
||
# Ленивый импорт: заливке (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>), декодировать сущности ("), обрезать."""
|
||
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
|