fix(core): ходить в CORE через sing-box — с российских адресов ключ не работает

Заливка встала после двух порций: прогоны стали заканчиваться за две минуты
с нулём документов. Выглядело как поломка сети, и на ложные следы ушло время —
MTU в норме (1472 байта проходят), IPv6 ни при чём, Cloudflare из того же
контейнера качается на 2.2 МБ/с, ключ и квота целы (с домашней машины тот же
запрос отвечает за 6с, лимит нетронут).

Разница оказалась в адресе. С прода:
  напрямую      — код 000, обрыв на 25с (соединение есть, тело не приходит)
  через sing-box — код 200 за 1.7с, 207 КБ
Анонимные короткие запросы проходят и напрямую — поэтому блокировка и
маскировалась под сетевой сбой.

CORE ведёт себя как OpenRouter, ради которого singbox-proxy и заводили,
поэтому решение то же: httpx получает proxy из CORE_PROXY_URL (умолчание —
socks5://singbox-proxy:1080, пусто = напрямую для локальных прогонов).
В образ индексатора добавлен socksio: без него httpx не умеет SOCKS5.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

docker-compose.prod.yml

@@ -152,6 +152,10 @@ services:
     depends_on:
       elasticsearch:
         condition: service_healthy
+      # CORE не отвечает на ключ с российских адресов — парсер ходит через
+      # sing-box (см. scripts/parsers/core.py)
+      singbox-proxy:
+        condition: service_started
 
   worker-notifier:
     build:

scripts/deploy.sh

@@ -43,6 +43,7 @@ fi
 
 REBUILD=""
 FRONTEND_CHANGED=0
+PARSERS_CHANGED=0
 if [ "$FULL" = 1 ]; then
   REBUILD="$ALL_BACKEND"
   FRONTEND_CHANGED=1
@@ -51,6 +52,12 @@ else
     echo "$CHANGED" | grep -qE "^services/$svc/" && REBUILD="$REBUILD $svc"
   done
   echo "$CHANGED" | grep -qE '^services/frontend/' && FRONTEND_CHANGED=1
+  # Парсеры не вшиты в образ, а смонтированы в worker-indexer (см. compose),
+  # поэтому пересобирать нечего. Но процесс воркера держит модуль парсера в
+  # памяти с прошлого прогона: без перезапуска правка молча не применяется —
+  # заливка продолжает ходить по старому коду, и это не видно ниоткуда, кроме
+  # тела запросов в логах. Ловилось на CORE 16.09.2026.
+  echo "$CHANGED" | grep -qE '^scripts/parsers/' && PARSERS_CHANGED=1
 fi
 REBUILD="$(echo "$REBUILD" | xargs || true)"
 
@@ -111,6 +118,13 @@ echo "==> [4/6] Применение секретов к неизменивши
 # shellcheck disable=SC2086
 $COMPOSE up -d $ALL_BACKEND
 
+# Правка парсера доезжает монтированием, но модуль уже загружен в память
+# воркера — перезапускаем, если образ и так не пересобирался выше.
+if [ "$PARSERS_CHANGED" = 1 ] && ! echo " $REBUILD " | grep -q " worker-indexer "; then
+  echo "==> [4.5/6] Парсеры менялись — перезапуск worker-indexer"
+  $COMPOSE restart worker-indexer
+fi
+
 echo "==> [5/6] Миграции БД (с ретраем)"
 for i in $(seq 1 10); do
   $COMPOSE exec -T api alembic upgrade head && break

scripts/ops/keep_ingesting.py

@@ -115,12 +115,25 @@ def main() -> None:
 
         started = []
         for source_id, source_type, last_run_id in sources:
-            # Источник исчерпан, если прошлый прогон не добавил ни одной статьи
-            # и не был отменён: дальше в дампе/бакете для нас ничего нет
+            # Источник исчерпан, если прогоны перестали добавлять статьи: дальше
+            # в дампе/бакете/выдаче для нас ничего нет.
+            #
+            # Но ОДИН пустой прогон — ещё не приговор: ровно так же выглядят
+            # обрыв связи с API и прогон на устаревшем коде парсера. На CORE
+            # 16.09.2026 из-за этого источник был помечен исчерпанным после
+            # единственной неудачи и больше не запускался — молча, без ошибки.
+            # Поэтому ждём два пустых прогона подряд: разовый сбой переживём,
+            # а реально кончившийся источник остановится всего на один прогон
+            # позже.
             if last_run_id and not args.keep_going_empty:
-                prev = s.get(ParseRun, last_run_id)
-                if prev and prev.status == "done" and prev.added == 0 and prev.fetched == 0:
-                    print(f"{source_type}: источник исчерпан, пропускаем")
+                last_two = s.execute(text("""
+                    SELECT status, added, fetched FROM parse_runs
+                    WHERE source_id = :sid ORDER BY id DESC LIMIT 2
+                """), {"sid": source_id}).all()
+                if len(last_two) == 2 and all(
+                    r.status == "done" and r.added == 0 and r.fetched == 0 for r in last_two
+                ):
+                    print(f"{source_type}: источник исчерпан (два пустых прогона), пропускаем")
                     continue
 
             src = s.get(ParseSource, source_id)

scripts/parsers/core.py

@@ -25,10 +25,20 @@
 обогащать корпус нечем.
 
 Грабли API, все проверены живьём:
+- **с российских адресов ключ не работает**: прямой запрос с прода виснет без
+  ошибки (соединение есть, тело ответа не приходит), через `singbox-proxy` тот
+  же запрос отвечает за 1.7с. Анонимные короткие запросы проходят и напрямую,
+  из-за чего поломка выглядит как сетевая. Адрес прокси — `CORE_PROXY_URL`;
 - у `/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 агрегирует репозитории и отдаёт одну статью несколько раз под разными
@@ -41,9 +51,9 @@
 import hashlib
 import logging
 import os
+import re
 import time
 from collections.abc import Iterator
-from datetime import UTC, datetime
 from typing import Any
 
 import httpx
@@ -55,8 +65,13 @@ 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
+# CORE отдаёт данные по ключу только за пределами РФ: с прода прямой запрос
+# молча виснет (TCP есть, тело ответа не приходит), а тот же запрос через
+# sing-box возвращается за 1.7с. Анонимные мелкие запросы проходят и напрямую,
+# поэтому со стороны это выглядит как поломка сети, а не блокировка. Тот же
+# приём, что для OpenRouter в worker-gpu. Пусто = ходить напрямую.
+DEFAULT_PROXY = "socks5://singbox-proxy:1080"
 
 
 class COREParser(BaseParser):
@@ -65,15 +80,21 @@ class COREParser(BaseParser):
     source_name = "core"
     bulk = True
 
-    def __init__(self, api_key: str | None = None) -> None:
+    def __init__(self, api_key: str | None = None, proxy_url: str | None = None) -> None:
         super().__init__()
         self.api_key = api_key or os.environ.get("CORE_API_KEY", "")
+        # Вне compose-сети (локальный прогон, тесты) имени singbox-proxy нет —
+        # там прокси отключают, выставив CORE_PROXY_URL пустым
+        self.proxy_url = self._proxy(proxy_url)
         if not self.api_key:
             logger.warning("CORE: ключ не задан (CORE_API_KEY) — API ответит 401")
-        # Страница весит ~5 МБ, 12с как у PMC тут мало: обычный ответ ~7с.
-        # 60с — с запасом на джиттер, но без риска съесть весь бюджет прогона
+        # Страница весит ~5 МБ, обычный ответ ~7с. 30с — запас на джиттер в
+        # десять раз, но без риска съесть бюджет прогона: три попытки по 60с
+        # отъедали 180с из 1500с на одной залипшей странице (поймано 16.09.2026,
+        # та же грабля, что чинили в pmc_bulk)
         self.client = httpx.Client(
-            timeout=60,
+            timeout=30,
+            proxy=self.proxy_url or None,
             headers={
                 "Authorization": f"Bearer {self.api_key}",
                 "User-Agent": "AcademicHelper/1.0 (noreply@jze9.ru)",
@@ -81,6 +102,13 @@ class COREParser(BaseParser):
         )
         self.last_token: str | None = None
 
+    @staticmethod
+    def _proxy(proxy_url: str | None) -> str:
+        """Адрес прокси: явный аргумент, иначе CORE_PROXY_URL, иначе умолчание."""
+        if proxy_url is not None:
+            return proxy_url
+        return os.environ.get("CORE_PROXY_URL", DEFAULT_PROXY)
+
     def fetch(  # type: ignore[override]
         self,
         limit: int = 1000,
@@ -91,45 +119,47 @@ class COREParser(BaseParser):
         progress_cb: ProgressCallback | None = None,
         **_ignored: Any,
     ) -> Iterator[dict[str, Any]]:
-        """Статьи с полным текстом, год за годом от свежих к старым.
+        """Статьи с полным текстом, архив за архивом.
 
         Args:
             limit: сколько статей с текстом отдать за прогон
-            query: поисковый запрос CORE; пустой — всё за год
-            year_from: нижняя граница годов (включительно)
-            year_to: верхняя граница; по умолчанию текущий год
-            resume_token: позиция вида "2019:4200" — год и смещение в нём
+            query: список архивов — номера через запятую/пробел или выражение
+                `(repositories.id:N OR ...)`. Пустой — обычный поиск по словам
+            year_from: отсекать статьи старше этого года (необязательно)
+            year_to: отсекать статьи новее этого года (необязательно)
+            resume_token: позиция вида "1298: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):]
+        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 year in years:
+        for part in parts:
             offset = start_offset
-            start_offset = 0      # смещение относится только к году из токена
+            start_offset = 0      # смещение относится только к архиву из токена
 
             while given < limit and offset < MAX_OFFSET:
-                results = self._page(query, year, offset)
+                results = self._page(part, query, offset)
                 if results is None:      # API не ответил — прогон закончен
                     return
                 if not results:
-                    logger.info("CORE: год %d исчерпан на смещении %d", year, offset)
+                    logger.info("CORE: архив %s исчерпан на смещении %d", part, offset)
                     break
 
-                token = f"{year}:{offset}"
+                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 тыс.
                     # символов у разных статей практически исключено
@@ -149,30 +179,45 @@ class COREParser(BaseParser):
                         break
 
                 offset += PAGE
-                self.last_token = f"{year}:{offset}"
+                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 пройдены, отдано %d", bottom, top, given)
+        logger.info("CORE: пройдено архивов: %d, отдано %d", len(parts), given)
 
     @staticmethod
-    def _parse_token(token: str | None) -> tuple[int | None, int]:
-        """Разобрать позицию "год:смещение"; мусор — начать сначала."""
+    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
-        year, _, offset = token.partition(":")
+        part, _, offset = token.rpartition(":")
         try:
-            return int(year), int(offset)
+            return part, 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}"
+    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):
@@ -188,12 +233,12 @@ class COREParser(BaseParser):
                 return resp.json().get("results") or []
             except Exception as e:
                 logger.warning(
-                    "CORE: страница %d:%d не удалась (%d/%d): %s",
-                    year, offset, attempt + 1, RETRIES, e,
+                    "CORE: страница %s:%d не удалась (%d/%d): %s",
+                    part, offset, attempt + 1, RETRIES, e,
                 )
                 time.sleep(2 * (attempt + 1))
 
-        logger.error("CORE: страница %d:%d не далась за %d попыток", year, offset, RETRIES)
+        logger.error("CORE: страница %s:%d не далась за %d попыток", part, offset, RETRIES)
         return None
 
     def transform(self, raw: dict[str, Any]) -> dict[str, Any]:

scripts/parsers/tests/test_core.py

@@ -1,11 +1,15 @@
 """Юнит-тесты парсера CORE — чистая логика, без сети.
 
-Стерегут то, ради чего парсер и написан иначе, чем остальные: отбор статей с
-полным текстом, отсев дублей CORE (одна статья под разными id) и позицию
-продолжения "год:смещение" — из-за неё выборка режется по годам.
+Стерегут то, ради чего парсер написан иначе остальных: отбор статей с полным
+текстом, отсев дублей CORE (одна статья под разными id), нарезку выборки по
+архивам-поставщикам и позицию продолжения «архив:смещение».
+
+Нарезка по архивам — не украшение: связка условий через `AND` в API CORE не
+работает (запрос по архивам вместе с годом отдаёт годы вперемешку), поэтому в
+запрос идёт ровно одно условие, а годы отсекаются уже на нашей стороне.
 """
 
-from core import MIN_CHARS, COREParser
+from core import DEFAULT_PROXY, MIN_CHARS, COREParser
 
 LONG = "слово " * 400        # заведомо длиннее MIN_CHARS
 
@@ -20,49 +24,65 @@ SAMPLE = {
     "downloadUrl": "https://core.ac.uk/download/1.pdf",
     "abstract": "Аннотация",
     "fullText": LONG,
-    "resume_token": "2019:100",
+    "resume_token": "1298:100",
 }
 
 
+def test_repos_from_or_expression():
+    q = "(repositories.id:1298 OR repositories.id:21908 OR repositories.id:949)"
+    assert COREParser._repos(q) == ["1298", "21908", "949"]
+
+
+def test_repos_from_plain_list():
+    assert COREParser._repos("1298, 21908 949") == ["1298", "21908", "949"]
+
+
+def test_repos_without_numbers_falls_back_to_word_search():
+    # Номеров нет — единственная часть «q»: обычный поиск по словам
+    assert COREParser._repos("нейронные сети") == ["q"]
+    assert COREParser._repos("") == ["q"]
+
+
 def test_parse_token():
-    assert COREParser._parse_token("2019:4200") == (2019, 4200)
+    assert COREParser._parse_token("1298:4200") == ("1298", 4200)
+    assert COREParser._parse_token("q:300") == ("q", 300)
     assert COREParser._parse_token(None) == (None, 0)
     assert COREParser._parse_token("мусор") == (None, 0)
-    assert COREParser._parse_token("год:смещение") == (None, 0)
+    assert COREParser._parse_token("архив:смещение") == (None, 0)
 
 
 def test_transform_maps_fields():
-    t = COREParser().transform(SAMPLE)
+    t = COREParser(proxy_url="").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["resume_token"] == "1298: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)) == {}
+    assert COREParser(proxy_url="").transform(dict(SAMPLE, fullText="коротко")) == {}
+    assert COREParser(proxy_url="").transform(dict(SAMPLE, fullText=None)) == {}
+    assert COREParser(proxy_url="").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
+    assert COREParser(proxy_url="").transform(dict(SAMPLE, language={"code": "zz"}))["lang"] is None
+    assert COREParser(proxy_url="").transform(dict(SAMPLE, language=None))["lang"] is None
 
 
 def _parser_with_pages(pages):
     """Парсер, у которого выдача подменена заранее заготовленными страницами."""
-    p = COREParser(api_key="test")
+    p = COREParser(api_key="test", proxy_url="")
     calls = []
 
-    def fake_page(query, year, offset):
-        calls.append((year, offset))
+    def fake_page(part, query, offset):
+        calls.append((part, offset))
         return pages.pop(0) if pages else []
 
     p._page = fake_page       # type: ignore[method-assign]
@@ -79,7 +99,7 @@ def test_fetch_drops_core_duplicates():
         {"id": 4, "fullText": "коротко"},
     ]
     p = _parser_with_pages([page])
-    got = list(p.fetch(limit=10, year_from=2019, year_to=2019))
+    got = list(p.fetch(limit=10, query="repositories.id:1298"))
     assert [w["id"] for w in got] == [1, 3]
 
 
@@ -88,22 +108,49 @@ 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"]
+    got = list(p.fetch(limit=10, query="repositories.id:1298"))
+    assert [w["resume_token"] for w in got] == ["1298:0", "1298:100"]
 
 
-def test_fetch_walks_years_down_and_resumes():
+def test_fetch_walks_archives_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)]
+    list(p.fetch(limit=10, query="(repositories.id:1298 OR repositories.id:949)",
+                 resume_token="1298:300"))
+    # Начали с архива из токена и его смещения, пустой архив — переход к следующему
+    assert p.calls == [("1298", 300), ("949", 0)]
 
 
-def test_fetch_ignores_token_of_unknown_year():
+def test_fetch_ignores_token_of_unknown_archive():
     p = _parser_with_pages([[]])
-    list(p.fetch(limit=10, year_from=2019, year_to=2019, resume_token="1800:500"))
-    assert p.calls == [(2019, 0)]
+    list(p.fetch(limit=10, query="repositories.id:1298", resume_token="99999:500"))
+    assert p.calls == [("1298", 0)]
+
+
+def test_fetch_filters_years_on_our_side():
+    # Годы API не фильтрует, поэтому отсекаем сами — и только если заданы
+    page = [
+        {"id": 1, "fullText": LONG, "yearPublished": 2004},
+        {"id": 2, "fullText": LONG + "два", "yearPublished": 2015},
+        {"id": 3, "fullText": LONG + "три", "yearPublished": 2030},
+    ]
+    p = _parser_with_pages([list(page)])
+    got = list(p.fetch(limit=10, query="repositories.id:1298", year_from=2010, year_to=2026))
+    assert [w["id"] for w in got] == [2]
+
+    p2 = _parser_with_pages([list(page)])
+    got2 = list(p2.fetch(limit=10, query="repositories.id:1298"))
+    assert [w["id"] for w in got2] == [1, 2, 3]
 
 
 def test_min_chars_threshold_is_meaningful():
     assert MIN_CHARS >= 1000
+
+
+def test_proxy_choice(monkeypatch):
+    # CORE не отвечает на ключ с российских адресов — по умолчанию идём через
+    # sing-box, но вне compose-сети прокси отключают пустым значением
+    monkeypatch.delenv("CORE_PROXY_URL", raising=False)
+    assert COREParser._proxy(None) == DEFAULT_PROXY
+    monkeypatch.setenv("CORE_PROXY_URL", "socks5://иной:1080")
+    assert COREParser._proxy(None) == "socks5://иной:1080"
+    assert COREParser._proxy("") == ""          # явное «без прокси» сильнее окружения

services/worker-indexer/requirements.txt

@@ -12,3 +12,4 @@ langdetect==1.0.9
 beautifulsoup4==4.12.3  # парсер CyberLeninka: детали статьи (fetch_article_details)
 pydantic-settings==2.2.1
 httpx==0.27.0
+socksio==1.0.0  # SOCKS5-прокси для httpx (CORE блокирует запросы из РФ, см. scripts/parsers/core.py)