"""Парсер CORE (core.ac.uk) — полные тексты открытых репозиториев. Зачем нужен: полный текст есть у малой доли корпуса, и по-русски дыра самая большая — CyberLeninka отдаёт только HTML-страницу статьи, тела не даёт. CORE агрегирует открытые репозитории (в том числе вузовские русско- и белорусскоязычные) и кладёт готовый текст прямо в выдачу поиска — его не надо ни скачивать отдельно, ни извлекать из PDF. Замер на живом API 16.09.2026: страница в 100 записей приходит за ~7с и содержит 47 статей с текстом длиннее 1500 символов, медиана — 12.7 тыс. символов. То есть один запрос ≈ 47 документов корпуса. Чем сузить выборку до русских текстов: в общем потоке CORE кириллицы нет вовсе (проверено: 0 из 48 полных текстов за 2019). Язык у работ размечен плохо — `language` сплошь None или "zz", фильтровать по нему нельзя. Зато работает отбор по архиву-поставщику: запрос вида `(repositories.id:1298 OR repositories.id:21908 OR ...)` по вузовским репозиториям России и Беларуси даёт 79 полных текстов из 100 записей, 78 из них — кириллица. Сам список архивов живёт в поле `query` источника, а не в коде: его правят из админки, не пересобирая образ. Массовый парсер (`bulk = True`): отдаёт статьи генератором и пишется пачками через COPY, как Википедия и PMC (см. bulk_writer.py). Статьи без текста пропускаются молча — они уже есть у нас как метаданные из OpenAlex, а обогащать корпус нечем. Грабли API, все проверены живьём: - у `/v3/search/works` обязателен слэш на конце: иначе 301, а при редиректе теряется заголовок Authorization и запрос уходит анонимным; - `offset` упирается в 100 000 (под капотом Azure Search, глубже — 400), поэтому выборка режется на части — по одному архиву-поставщику на запрос; - **условия через `AND` не связываются**: `(repositories.id:...) AND yearPublished:2026` отдаёт 2.3 млн работ вперемешку по годам, то есть условия объединяются по «или» и год работает лишь как подсказка ранжированию. По отдельности каждое условие фильтрует честно (`yearPublished:2026` — ровно 2026, `repositories.id:1298` — ровно этот архив). Поэтому в запрос идёт РОВНО ОДНО условие: номер архива. Годы, если заданы, отсекаются уже на нашей стороне; - `fullText` — не фильтруемое поле, `_exists_:fullText` отвечает 500. Отбирать статьи с текстом приходится на своей стороне, оплачивая трафиком; - CORE агрегирует репозитории и отдаёт одну статью несколько раз под разными id (проверено: 90810208 и 354263199 — один текст). Дедупликация корпуса идёт только по ext_id, поэтому такие пары легли бы отдельными документами и потом ловились бы как заимствование друг у друга. Отсеиваем по хешу текста в пределах прогона: копии приходят в соседних строках выдачи. """ import hashlib import logging import os import re import time from collections.abc import Iterator from typing import Any import httpx from base import BaseParser, ProgressCallback logger = logging.getLogger(__name__) API_URL = "https://api.core.ac.uk/v3/search/works/" # слэш обязателен, см. докстринг PAGE = 100 # максимум записей за запрос MAX_OFFSET = 100_000 # потолок глубины у Azure Search MIN_CHARS = 1500 # короче — обрывок или аннотация, а не статья RETRIES = 3 class COREParser(BaseParser): """Полные тексты из CORE. Отдаёт статьи потоком, режет выборку по годам.""" source_name = "core" bulk = True def __init__(self, api_key: str | None = None) -> None: super().__init__() self.api_key = api_key or os.environ.get("CORE_API_KEY", "") if not self.api_key: logger.warning("CORE: ключ не задан (CORE_API_KEY) — API ответит 401") # Страница весит ~5 МБ, обычный ответ ~7с. 30с — запас на джиттер в # десять раз, но без риска съесть бюджет прогона: три попытки по 60с # отъедали 180с из 1500с на одной залипшей странице (поймано 16.09.2026, # та же грабля, что чинили в pmc_bulk) self.client = httpx.Client( timeout=30, headers={ "Authorization": f"Bearer {self.api_key}", "User-Agent": "AcademicHelper/1.0 (noreply@jze9.ru)", }, ) self.last_token: str | None = None def fetch( # type: ignore[override] self, limit: int = 1000, query: str = "", year_from: int | None = None, year_to: int | None = None, resume_token: str | None = None, progress_cb: ProgressCallback | None = None, **_ignored: Any, ) -> Iterator[dict[str, Any]]: """Статьи с полным текстом, архив за архивом. Args: limit: сколько статей с текстом отдать за прогон query: список архивов — номера через запятую/пробел или выражение `(repositories.id:N OR ...)`. Пустой — обычный поиск по словам year_from: отсекать статьи старше этого года (необязательно) year_to: отсекать статьи новее этого года (необязательно) resume_token: позиция вида "1298:4200" — архив и смещение в нём progress_cb: см. base.ProgressCallback """ parts = self._repos(query) start_part, start_offset = self._parse_token(resume_token) if start_part in parts: parts = parts[parts.index(start_part):] else: start_offset = 0 given = 0 seen: set[str] = set() for part in parts: offset = start_offset start_offset = 0 # смещение относится только к архиву из токена while given < limit and offset < MAX_OFFSET: results = self._page(part, query, offset) if results is None: # API не ответил — прогон закончен return if not results: logger.info("CORE: архив %s исчерпан на смещении %d", part, offset) break token = f"{part}:{offset}" for work in results: text = work.get("fullText") or "" if len(text) < MIN_CHARS: continue year = work.get("yearPublished") if year and ((year_from and year < year_from) or (year_to and year > year_to)): continue # Дубли CORE: один текст под разными id (см. докстринг). # Хеша начала текста хватает — совпадение первых 5 тыс. # символов у разных статей практически исключено digest = hashlib.sha1(text[:5000].encode()).hexdigest() if digest in seen: continue seen.add(digest) given += 1 # Токен указывает на страницу, из которой пришла статья, а # не на следующую: пачка может прерваться на её середине, и # тогда следующий прогон перечитает страницу целиком. # Лишний запрос дешевле потерянных статей, дубли отсекает # ON CONFLICT на ext_id work["resume_token"] = token yield work if given >= limit: break offset += PAGE self.last_token = f"{part}:{offset}" if progress_cb and not progress_cb(given): logger.info("CORE: выборка остановлена по запросу (%d)", given) return logger.info("CORE: пройдено архивов: %d, отдано %d", len(parts), given) @staticmethod def _repos(query: str) -> list[str]: """Номера архивов из настройки источника. Принимает и выражение `(repositories.id:1298 OR ...)`, и простой список «1298, 21908». Если номеров нет вовсе — единственная часть с именем `q`: тогда парсер просто ищет по словам запроса. """ ids = re.findall(r"repositories\.id:(\d+)", query or "") if not ids: ids = re.findall(r"\b\d{2,7}\b", query or "") return ids or ["q"] @staticmethod def _parse_token(token: str | None) -> tuple[str | None, int]: """Разобрать позицию "архив:смещение"; мусор — начать сначала.""" if not token or ":" not in token: return None, 0 part, _, offset = token.rpartition(":") try: return part, int(offset) except ValueError: logger.warning("CORE: непонятная позиция %r, начинаем сначала", token) return None, 0 def _page(self, part: str, query: str, offset: int) -> list[dict[str, Any]] | None: """Одна страница выдачи; None — API не отвечает, прогон пора кончать. В запросе РОВНО одно условие: связка через AND в этом API не работает (см. докстринг модуля), поэтому годы отсекаются уже после выборки. """ q = f"repositories.id:{part}" if part != "q" else (query or "*") params = {"q": q, "limit": PAGE, "offset": offset} for attempt in range(RETRIES): try: resp = self.client.get(API_URL, params=params) if resp.status_code == 429: # Лимит тарифа: подождать и повторить, а не ронять прогон wait = int(resp.headers.get("retry-after", 30)) logger.warning("CORE: лимит запросов, ждём %dс", wait) time.sleep(min(wait, 60)) continue resp.raise_for_status() return resp.json().get("results") or [] except Exception as e: logger.warning( "CORE: страница %s:%d не удалась (%d/%d): %s", part, offset, attempt + 1, RETRIES, e, ) time.sleep(2 * (attempt + 1)) logger.error("CORE: страница %s:%d не далась за %d попыток", part, offset, RETRIES) return None def transform(self, raw: dict[str, Any]) -> dict[str, Any]: """Привести статью CORE к унифицированному формату корпуса.""" ext = raw.get("id") text = raw.get("fullText") or "" if not ext or len(text) < MIN_CHARS: return {} lang = raw.get("language") or {} code = lang.get("code") if isinstance(lang, dict) else lang journals = raw.get("journals") or [] journal = journals[0].get("title") if journals and isinstance(journals[0], dict) else None return { "source": self.source_name, "ext_id": f"core:{ext}", "title": (raw.get("title") or f"CORE {ext}")[:1000], "authors": self.normalize_authors(self._authors(raw)), "doi": raw.get("doi"), "year": raw.get("yearPublished"), # zz у CORE значит «язык не определён» — лучше пусто, чем мусор "lang": code if code and code != "zz" else None, "journal": journal[:300] if journal else None, "url": raw.get("downloadUrl") or f"https://core.ac.uk/works/{ext}", "abstract": (raw.get("abstract") or text[:2000])[:2000], "text": text, "resume_token": raw.get("resume_token"), } @staticmethod def _authors(raw: dict[str, Any]) -> list[dict[str, str]]: """Авторы CORE приходят одной строкой "Фамилия, Имя Отчество".""" out = [] for author in raw.get("authors") or []: name = author.get("name") if isinstance(author, dict) else author if not name: continue last, _, first = str(name).partition(",") out.append({"last_name": last.strip(), "first_name": first.strip()}) return out