From 3d4fcecbb22483434c87a8ec04e0a2bf1f118c56 Mon Sep 17 00:00:00 2001 From: jze9 Date: Wed, 16 Sep 2026 17:05:34 +0500 Subject: [PATCH] =?UTF-8?q?feat(core):=20=D0=BF=D0=BE=D0=B4=D0=BA=D0=BB?= =?UTF-8?q?=D1=8E=D1=87=D0=B8=D1=82=D1=8C=20CORE=20=D0=BA=D0=B0=D0=BA=20?= =?UTF-8?q?=D0=B8=D1=81=D1=82=D0=BE=D1=87=D0=BD=D0=B8=D0=BA=20=D0=BF=D0=BE?= =?UTF-8?q?=D0=BB=D0=BD=D1=8B=D1=85=20=D1=82=D0=B5=D0=BA=D1=81=D1=82=D0=BE?= =?UTF-8?q?=D0=B2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Оба массовых источника исчерпаны — wikipedia_ru и pmc_bulk отдают ноль, сторож честно пишет «источник исчерпан, пропускаем», и корпус стоит. Брать новые статьи неоткуда, а главная дыра прежняя: полный текст есть у 11% корпуса, по-русски — почти нигде (CyberLeninka отдаёт только HTML страницы, тела статьи там нет вовсе). CORE кладёт готовый текст прямо в выдачу поиска: его не надо ни качать отдельно, ни извлекать из PDF. Парсер массовый, как Википедия и PMC — отдаёт статьи генератором, пишется пачками через COPY, помнит позицию. Что выяснено живьём и учтено в коде: - у /v3/search/works обязателен слэш на конце, иначе 301, а на редиректе теряется заголовок Authorization и запрос уходит анонимным; - offset упирается в 100 000 (под капотом Azure Search), поэтому выборка режется по годам, а позиция продолжения — строка «год:смещение»; - fullText не фильтруемое поле, _exists_:fullText отвечает 500 — статьи с текстом приходится отбирать на своей стороне; - CORE отдаёт одну статью под разными id из разных репозиториев, а корпус дедуплицируется только по ext_id: такие пары легли бы отдельными документами и ловились бы как заимствование друг у друга. Отсеиваем по хешу начала текста в пределах прогона. Позиция указывает на страницу, из которой пришла статья, а не на следующую: пачка может прерваться на середине страницы, и тогда прогон перечитает её целиком. Лишний запрос дешевле потерянных статей, дубли отсекает ON CONFLICT. Замер на живом API: страница в 100 записей приходит за ~7с. В общем потоке CORE кириллицы нет вовсе (0 из 48 полных текстов за 2019), язык размечен негодно — сплошь None и «zz». Зато работает отбор по архиву: запрос по 41 вузовскому репозиторию России и Беларуси даёт 79 полных текстов из 100 записей, 78 из них кириллические. Список архивов живёт в поле query источника, а не в коде — правится из админки без пересборки. Co-Authored-By: Claude Opus 5 --- scripts/ops/keep_ingesting.py | 2 +- scripts/parsers/core.py | 237 +++++++++++++++++++++ scripts/parsers/tests/test_core.py | 109 ++++++++++ services/worker-indexer/app/tasks/index.py | 11 + 4 files changed, 358 insertions(+), 1 deletion(-) create mode 100644 scripts/parsers/core.py create mode 100644 scripts/parsers/tests/test_core.py diff --git a/scripts/ops/keep_ingesting.py b/scripts/ops/keep_ingesting.py index f47f44a..bdc8fcd 100755 --- a/scripts/ops/keep_ingesting.py +++ b/scripts/ops/keep_ingesting.py @@ -31,7 +31,7 @@ heartbeat — без этой чистки сторож видит «занят import argparse -BULK_TYPES = ("wikipedia_ru", "pmc_bulk") +BULK_TYPES = ("wikipedia_ru", "pmc_bulk", "core") STALE_MINUTES = 10 diff --git a/scripts/parsers/core.py b/scripts/parsers/core.py new file mode 100644 index 0000000..162f20d --- /dev/null +++ b/scripts/parsers/core.py @@ -0,0 +1,237 @@ +"""Парсер 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), + поэтому выборка режется по годам: в каждом году свой отсчёт; +- `fullText` — не фильтруемое поле, `_exists_:fullText` отвечает 500. + Отбирать статьи с текстом приходится на своей стороне, оплачивая трафиком; +- CORE агрегирует репозитории и отдаёт одну статью несколько раз под разными + id (проверено: 90810208 и 354263199 — один текст). Дедупликация корпуса + идёт только по ext_id, поэтому такие пары легли бы отдельными документами + и потом ловились бы как заимствование друг у друга. Отсеиваем по хешу + текста в пределах прогона: копии приходят в соседних строках выдачи. +""" + +import hashlib +import logging +import os +import time +from collections.abc import Iterator +from datetime import UTC, datetime +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 # короче — обрывок или аннотация, а не статья +DEFAULT_YEAR_FROM = 2010 +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 МБ, 12с как у PMC тут мало: обычный ответ ~7с. + # 60с — с запасом на джиттер, но без риска съесть весь бюджет прогона + self.client = httpx.Client( + timeout=60, + 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: поисковый запрос CORE; пустой — всё за год + year_from: нижняя граница годов (включительно) + year_to: верхняя граница; по умолчанию текущий год + resume_token: позиция вида "2019:4200" — год и смещение в нём + progress_cb: см. base.ProgressCallback + """ + top = year_to or datetime.now(UTC).year + bottom = year_from or DEFAULT_YEAR_FROM + years = list(range(top, bottom - 1, -1)) + + start_year, start_offset = self._parse_token(resume_token) + if start_year in years: + years = years[years.index(start_year):] + else: + start_offset = 0 + + given = 0 + seen: set[str] = set() + for year in years: + offset = start_offset + start_offset = 0 # смещение относится только к году из токена + + while given < limit and offset < MAX_OFFSET: + results = self._page(query, year, offset) + if results is None: # API не ответил — прогон закончен + return + if not results: + logger.info("CORE: год %d исчерпан на смещении %d", year, offset) + break + + token = f"{year}:{offset}" + for work in results: + text = work.get("fullText") or "" + if len(text) < MIN_CHARS: + 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"{year}:{offset}" + if progress_cb and not progress_cb(given): + logger.info("CORE: выборка остановлена по запросу (%d)", given) + return + + logger.info("CORE: годы %d-%d пройдены, отдано %d", bottom, top, given) + + @staticmethod + def _parse_token(token: str | None) -> tuple[int | None, int]: + """Разобрать позицию "год:смещение"; мусор — начать сначала.""" + if not token or ":" not in token: + return None, 0 + year, _, offset = token.partition(":") + try: + return int(year), int(offset) + except ValueError: + logger.warning("CORE: непонятная позиция %r, начинаем сначала", token) + return None, 0 + + def _page(self, query: str, year: int, offset: int) -> list[dict[str, Any]] | None: + """Одна страница выдачи; None — API не отвечает, прогон пора кончать.""" + q = f"yearPublished:{year}" + if query: + q = f"({query}) AND {q}" + 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: страница %d:%d не удалась (%d/%d): %s", + year, offset, attempt + 1, RETRIES, e, + ) + time.sleep(2 * (attempt + 1)) + + logger.error("CORE: страница %d:%d не далась за %d попыток", year, 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 diff --git a/scripts/parsers/tests/test_core.py b/scripts/parsers/tests/test_core.py new file mode 100644 index 0000000..4b7f80f --- /dev/null +++ b/scripts/parsers/tests/test_core.py @@ -0,0 +1,109 @@ +"""Юнит-тесты парсера CORE — чистая логика, без сети. + +Стерегут то, ради чего парсер и написан иначе, чем остальные: отбор статей с +полным текстом, отсев дублей CORE (одна статья под разными id) и позицию +продолжения "год:смещение" — из-за неё выборка режется по годам. +""" + +from core import MIN_CHARS, COREParser + +LONG = "слово " * 400 # заведомо длиннее MIN_CHARS + +SAMPLE = { + "id": 123456, + "title": "Нейронные сети в медицине", + "authors": [{"name": "Кузьмин, Ярослав Вадимович"}, {"name": "Савенок А."}], + "doi": "10.1234/x", + "yearPublished": 2019, + "language": {"code": "ru"}, + "journals": [{"title": "Вестник"}], + "downloadUrl": "https://core.ac.uk/download/1.pdf", + "abstract": "Аннотация", + "fullText": LONG, + "resume_token": "2019:100", +} + + +def test_parse_token(): + assert COREParser._parse_token("2019:4200") == (2019, 4200) + assert COREParser._parse_token(None) == (None, 0) + assert COREParser._parse_token("мусор") == (None, 0) + assert COREParser._parse_token("год:смещение") == (None, 0) + + +def test_transform_maps_fields(): + t = COREParser().transform(SAMPLE) + assert t["ext_id"] == "core:123456" + assert t["source"] == "core" + assert t["year"] == 2019 and t["lang"] == "ru" + assert t["journal"] == "Вестник" + assert t["text"] == LONG + assert t["resume_token"] == "2019:100" + # Автор приходит одной строкой "Фамилия, Имя Отчество" + assert t["authors"][0]["last_name"] == "Кузьмин" + assert t["authors"][0]["initials"] == "Я.В." + + +def test_transform_skips_short_text(): + assert COREParser().transform(dict(SAMPLE, fullText="коротко")) == {} + assert COREParser().transform(dict(SAMPLE, fullText=None)) == {} + assert COREParser().transform(dict(SAMPLE, id=None)) == {} + + +def test_transform_drops_undefined_language(): + # zz у CORE значит «язык не определён» — в корпус такое класть незачем + assert COREParser().transform(dict(SAMPLE, language={"code": "zz"}))["lang"] is None + assert COREParser().transform(dict(SAMPLE, language=None))["lang"] is None + + +def _parser_with_pages(pages): + """Парсер, у которого выдача подменена заранее заготовленными страницами.""" + p = COREParser(api_key="test") + calls = [] + + def fake_page(query, year, offset): + calls.append((year, offset)) + return pages.pop(0) if pages else [] + + p._page = fake_page # type: ignore[method-assign] + p.calls = calls # type: ignore[attr-defined] + return p + + +def test_fetch_drops_core_duplicates(): + # Одна и та же статья под разными id — CORE так отдаёт всегда + page = [ + {"id": 1, "fullText": LONG}, + {"id": 2, "fullText": LONG}, + {"id": 3, "fullText": LONG + "иное"}, + {"id": 4, "fullText": "коротко"}, + ] + p = _parser_with_pages([page]) + got = list(p.fetch(limit=10, year_from=2019, year_to=2019)) + assert [w["id"] for w in got] == [1, 3] + + +def test_fetch_position_points_at_own_page(): + # Токен обязан указывать на страницу, откуда пришла статья: пачка может + # прерваться на середине, и следующий прогон перечитает её целиком + pages = [[{"id": 1, "fullText": LONG}], [{"id": 2, "fullText": LONG + "два"}]] + p = _parser_with_pages(pages) + got = list(p.fetch(limit=10, year_from=2019, year_to=2019)) + assert [w["resume_token"] for w in got] == ["2019:0", "2019:100"] + + +def test_fetch_walks_years_down_and_resumes(): + p = _parser_with_pages([[], []]) + list(p.fetch(limit=10, year_from=2018, year_to=2019, resume_token="2019:300")) + # Начали с года из токена и его смещения, пустой год — переход к следующему + assert p.calls == [(2019, 300), (2018, 0)] + + +def test_fetch_ignores_token_of_unknown_year(): + p = _parser_with_pages([[]]) + list(p.fetch(limit=10, year_from=2019, year_to=2019, resume_token="1800:500")) + assert p.calls == [(2019, 0)] + + +def test_min_chars_threshold_is_meaningful(): + assert MIN_CHARS >= 1000 diff --git a/services/worker-indexer/app/tasks/index.py b/services/worker-indexer/app/tasks/index.py index da96cba..d0da0f0 100644 --- a/services/worker-indexer/app/tasks/index.py +++ b/services/worker-indexer/app/tasks/index.py @@ -589,6 +589,17 @@ def _parser_for(source_type: str, cfg: dict[str, Any]) -> tuple[Any, dict[str, A "dump_path": query or None, # query = путь к дампу, если задан "start_offset": int(cfg.get("resume_token") or 0), } + elif source_type == "core": + # Массовый источник: позиция — "год:смещение". Выборка режется по + # годам, потому что offset у CORE упирается в 100 000 (см. core.py) + from core import COREParser as P + kwargs = { + "limit": limit, + "query": query, + "year_from": cfg.get("year_from"), + "year_to": cfg.get("year_to"), + "resume_token": cfg.get("resume_token"), + } elif source_type == "pmc_bulk": from pmc_bulk import PMCBulkParser as P kwargs = {