feat(core): подключить CORE как источник полных текстов

Оба массовых источника исчерпаны — 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 <noreply@anthropic.com>

scripts/ops/keep_ingesting.py

@@ -15,11 +15,36 @@
 Останавливается сам, когда источник исчерпан: парсер перестаёт отдавать статьи,
 прогон закрывается с нулём добавленных, и скрипт больше его не трогает
 (--stop-when-empty, по умолчанию включено).
+
+Перед подсчётом «занято ли» снимает зависшие прогоны: если worker-indexer упал
+или перезапустился, прогон навсегда остаётся в статусе running с замолчавшим
+heartbeat — без этой чистки сторож видит «занято» бесконечно и не запускает
+вообще ничего, для всех источников сразу (порог тот же, что и в панели
+отладки — services/api/app/api/admin.py, stale_cutoff).
+
+Молчащий heartbeat сам по себе ещё не значит «мёртв»: пачка COPY на большой
+базе идёт десятки минут без единого тика. Поэтому прогон снимается, только
+если его таска нет среди выполняющихся на воркерах queue.index; не ответил
+хоть один такой воркер — не снимается ничего. Иначе сторож снимал живой
+медленный прогон и тут же запускал его дубль по тем же статьям.
 """
 
 import argparse
 
-BULK_TYPES = ("wikipedia_ru", "pmc_bulk")
+BULK_TYPES = ("wikipedia_ru", "pmc_bulk", "core")
+STALE_MINUTES = 10
+
+
+def live_task_ids(celery_app) -> set[str] | None:
+    """ID тасков, выполняющихся на воркерах queue.index; None — кто-то из них не ответил."""
+    queues = celery_app.control.inspect(timeout=10).active_queues() or {}
+    workers = [w for w, qs in queues.items() if any(q["name"] == "queue.index" for q in qs)]
+    if not workers:
+        return None
+    active = celery_app.control.inspect(destination=workers, timeout=10).active()
+    if not active or set(active) != set(workers):
+        return None
+    return {t["id"] for tasks in active.values() for t in tasks}
 
 
 def main() -> None:
@@ -38,7 +63,44 @@ def main() -> None:
 
     types = tuple(t.strip() for t in args.types.split(",") if t.strip())
 
+    live = live_task_ids(celery_app)
+
     with db_session() as s:
+        if live is None:
+            print("воркеры queue.index не ответили — живость прогонов не проверить, не снимаю")
+            reaped = []
+        else:
+            reaped = s.execute(text(f"""
+                UPDATE parse_runs SET status='error', stage='finished',
+                       error='зависший прогон снят автоматически: таск не выполняется, heartbeat молчал дольше {STALE_MINUTES} минут',
+                       finished_at=now()
+                WHERE status='running'
+                  AND (heartbeat_at IS NULL OR heartbeat_at < now() - interval '{STALE_MINUTES} minutes')
+                  AND (celery_task_id IS NULL OR NOT (celery_task_id = ANY(CAST(:live AS text[]))))
+                RETURNING id, source_id
+            """), {"live": sorted(live)}).all()
+        # queued без heartbeat — тот же зомби другого вида: send_task не дошёл
+        # или упал между двумя commit при постановке в очередь (celery_task_id
+        # так и остался пустым), воркер о такой задаче никогда не узнает
+        reaped += s.execute(text(f"""
+            UPDATE parse_runs SET status='error', stage='finished',
+                   error='зависший прогон снят автоматически: висел в очереди дольше {STALE_MINUTES} минут',
+                   finished_at=now()
+            WHERE status='queued'
+              AND celery_task_id IS NULL
+              AND started_at < now() - interval '{STALE_MINUTES} minutes'
+            RETURNING id, source_id
+        """)).all()
+        if reaped:
+            for source_id in {row.source_id for row in reaped}:
+                s.execute(text("""
+                    UPDATE parse_sources SET last_status='error',
+                           last_error='зависший прогон снят автоматически'
+                    WHERE id=:sid AND last_status='running'
+                """), {"sid": source_id})
+            s.commit()
+            print(f"снято зависших прогонов: {len(reaped)} ({', '.join(str(r.id) for r in reaped)})")
+
         active = s.execute(text(
             "SELECT count(*) FROM parse_runs WHERE status IN ('queued','running')"
         )).scalar_one()

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

scripts/parsers/pmc_bulk.py

@@ -14,7 +14,7 @@
 import logging
 import re
 from collections.abc import Iterator
-from concurrent.futures import ThreadPoolExecutor
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from typing import Any
 from urllib.parse import quote
 
@@ -39,8 +39,11 @@ class PMCBulkParser(BaseParser):
     def __init__(self, workers: int = 12) -> None:
         super().__init__()
         self.workers = workers
+        # 60с на файл при бюджете задачи в 1500с — один залипший запрос съедал
+        # почти весь бюджет. Обычный GET сюда укладывается в доли секунды,
+        # 12с — с большим запасом на джиттер, но без риска съесть весь прогон
         self.client = httpx.Client(
-            timeout=60,
+            timeout=12,
             headers={"User-Agent": "AcademicHelper/1.0 (noreply@jze9.ru)"},
         )
         # Позиция листинга: бакет отдаётся страницами, и продолжать прогон
@@ -82,8 +85,13 @@ class PMCBulkParser(BaseParser):
                 logger.info("PMC bulk: бакет закончился")
                 return
 
+            # as_completed вместо map(): map() отдаёт результаты строго по
+            # порядку отправки, поэтому один залипший запрос блокирует все
+            # уже готовые — даже если остальные 11 потоков давно отработали
             with ThreadPoolExecutor(max_workers=self.workers) as pool:
-                for raw in pool.map(self._fetch_article, ids):
+                futures = {pool.submit(self._fetch_article, i): i for i in ids}
+                for future in as_completed(futures):
+                    raw = future.result()
                     if raw is None:
                         continue
                     given += 1

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

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 = {