feat: collect all quartile categories and classify VAK journals

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## 0.18.0
+
+- Добавлено поле `vak_categories` с результатом проверки журнала по предоставленным спискам К1/К2, источником и способом сопоставления. Совпадение по названию отмечается отдельно от подтверждённого ISSN; текущий снимок не объявляется исторической категорией.
+- Инструмент обогащения сохраняет все доступные квартильные категории всех систем, включая дополнительные сведения для публикаций с уже известным квартилем.
+
 ## 0.17.0
 
 - Добавлено отдельное поле публикации `quartiles` с квартилями Q1–Q4, системой, годом метрики и источником. Данные ВШЭ переносятся без удаления исходных значений; новые сведения сопоставляются по DOI и ISSN.

README.md

@@ -155,7 +155,7 @@ node --check app/static/admin.js
 
 Тесты с данными используют временную SQLite; отдельные API smoke-тесты запускают приложение с его текущей конфигурацией. Браузерная проверка: `pip install playwright`, `python -m playwright install chromium`, затем `python tests/browser_admin.py`. Она проверяет реальный рендеринг страниц, клавиатуру, диалог, фильтры и восстановление прогресса после ошибки. Скриншоты сохраняются во временную папку; путь выводится в конце.
 
-Версия сервиса: `0.17.0`. Админка всегда показывает версии backend и frontend в footer.
+Версия сервиса: `0.18.0`. Админка всегда показывает версии backend и frontend в footer.
 
 ## Разовый импорт публикаций
 
@@ -202,7 +202,7 @@ JSON-справочника записываются с LF для воспрои
 
 ### Квартили публикаций и схема БД
 
-Полный состав полей `employee_publications`, структура отдельного поля `quartiles`, ограничения сопоставления и команды обогащения описаны в [docs/publication-database.md](docs/publication-database.md). Для существующей БД добавлено nullable JSONB-поле миграцией `010_publication_quartiles.sql`; данные переносятся отдельным инструментом, а не при старте приложения.
+Полный состав полей `employee_publications`, структура отдельного поля `quartiles`, ограничения сопоставления и команды обогащения описаны в [docs/publication-database.md](docs/publication-database.md). Для существующей БД добавлены nullable JSONB-поля миграциями `010_publication_quartiles.sql` и `011_publication_vak_categories.sql`; данные переносятся отдельным инструментом, а не при старте приложения.
 
 Квартили Q1–Q4 из данных ВШЭ сохраняются с исходной системой; неизвестный год метрики остаётся неизвестным. Для остальных публикаций журнал устанавливается по DOI, рейтинг — по ISSN. Исторические SJR доступны через открытые данные университета Эстремадуры за 2009–2025; CiteScore/JCR читаются из проверенных каталогов. При отсутствии метрики за год публикации допускается явно разрешённый последний доступный год, с сохранением его года и источника. Отсутствие DOI, неоднозначность и недоступная метрика фиксируются без присвоения вымышленного квартиля. Парсер сохраняет новое поле при обновлении записи.
 

app/db.py

@@ -39,11 +39,12 @@ def _ensure_runtime_schema() -> None:
         table_names = set(inspector.get_table_names())
     if "employee_publications" in table_names:
         publication_columns = {column["name"] for column in inspector.get_columns("employee_publications")}
-        if "quartiles" not in publication_columns:
-            with engine.begin() as connection:
-                connection.execute(text("ALTER TABLE employee_publications ADD COLUMN " +
-                                        ("IF NOT EXISTS " if engine.dialect.name == "postgresql" else "") +
-                                        "quartiles " + ("JSONB" if engine.dialect.name == "postgresql" else "JSON")))
+        for column in ("quartiles", "vak_categories"):
+            if column not in publication_columns:
+                with engine.begin() as connection:
+                    connection.execute(text("ALTER TABLE employee_publications ADD COLUMN " +
+                                            ("IF NOT EXISTS " if engine.dialect.name == "postgresql" else "") +
+                                            column + " " + ("JSONB" if engine.dialect.name == "postgresql" else "JSON")))
     if "employees" in table_names and "employee_news_links" not in table_names:
         models.EmployeeNewsLink.__table__.create(bind=engine, checkfirst=True)
         inspector = inspect(engine)

app/models.py

@@ -109,6 +109,7 @@ class EmployeePublication(Base):
     document_url: Mapped[str | None] = mapped_column(Text)
     citation_text: Mapped[str | None] = mapped_column(Text)
     quartiles: Mapped[dict | None] = mapped_column(json_type)
+    vak_categories: Mapped[dict | None] = mapped_column(json_type)
     annotation: Mapped[dict | None] = mapped_column(json_type)
     description: Mapped[dict | None] = mapped_column(json_type)
     authors: Mapped[list | None] = mapped_column(json_type)

app/services/council.py

@@ -11,13 +11,12 @@ from pathlib import Path
 from urllib.parse import quote, urlsplit
 from zoneinfo import ZoneInfo
 
-from bs4 import BeautifulSoup
 from sqlalchemy import select
 from sqlalchemy.orm import Session, selectinload
 
 from app.models import Employee
 from app.services.academic_degrees import academic_degrees
-from app.services.council_source import _doi, _journal_names, _name, _quartiles, load_reference
+from app.services.council_source import _doi, _journal_names, _name, _quartiles, load_reference, _description_journals
 from app.services.council_catalog import catalog, publication_catalog
 from app.services.publication_quartiles import labels as quartile_labels
 from app.services.council_assessment import (
@@ -97,19 +96,6 @@ def _k2_journals() -> dict:
     return result
 
 
-def _description_journals(description: dict) -> list[str]:
-    names = []
-    for field in ("short", "shortLeft"):
-        value = description.get(field)
-        values = value.values() if isinstance(value, dict) else [value]
-        for text in values:
-            if isinstance(text, str):
-                text = BeautifulSoup(text, "html.parser").get_text(" ", strip=True)
-                # HSE short descriptions start with the journal, followed by the year.
-                names.append(re.split(r"\b(?:19|20)\d{2}\b", text, maxsplit=1)[0].strip(" .,:;"))
-    return names
-
-
 @lru_cache(maxsize=1)
 def _articles_by_author() -> dict[str, list[dict]]:
     return load_reference(EXPORT)

app/services/council_source.py

@@ -38,6 +38,21 @@ def _journal_names(data: dict) -> list[str]:
     return names
 
 
+def _description_journals(description: dict) -> list[str]:
+    from bs4 import BeautifulSoup
+
+    names = []
+    for field in ("short", "shortLeft"):
+        value = description.get(field)
+        values = value.values() if isinstance(value, dict) else [value]
+        for text in values:
+            if isinstance(text, str):
+                text = BeautifulSoup(text, "html.parser").get_text(" ", strip=True)
+                # HSE short descriptions start with the journal, followed by the year.
+                names.append(re.split(r"\b(?:19|20)\d{2}\b", text, maxsplit=1)[0].strip(" .,:;"))
+    return names
+
+
 def prepare_reference(source: Path) -> dict:
     publications = []
     by_author = defaultdict(list)

app/services/publication_quartiles.py

@@ -55,20 +55,21 @@ def catalog_entries(doi_metadata, publication_year, allow_latest=False):
     if issues:
         return [], "journal_identity_conflict"
     metrics = [(record, metric) for record in records for metric in record["metrics"]]
-    selected_year = publication_year
-    if allow_latest and metrics and not any(metric["year"] == publication_year for _, metric in metrics):
-        selected_year = max(metric["year"] for _, metric in metrics)
+    selected_years = {}
+    for system in SYSTEM_FIELDS:
+        years = {metric["year"] for _, metric in metrics if metric["system"] == system}
+        selected_years[system] = publication_year if publication_year in years else max(years, default=None) if allow_latest else None
     output = []
     for record in records:
         for metric in record["metrics"]:
-            if metric["year"] == selected_year and metric["system"] in SYSTEM_FIELDS:
+            if metric["year"] == selected_years.get(metric["system"]) and metric["system"] in SYSTEM_FIELDS:
                 output.append({"system": metric["system"], "quartile": metric["quartile"],
                                "metric": metric["metric"], "metric_year": metric["year"],
                                "publication_year": publication_year, "category": metric["category"],
                                "journal": record["name"], "issns": doi_metadata["issns"],
                                "source": metric["source"], "source_url": safe_url(metric["source_url"]),
                                "checked_on": date.today().isoformat()})
-    return output, ("found" if selected_year == publication_year else "found_latest") if output else "historical_quartile_unavailable"
+    return output, ("found_latest" if any(item["metric_year"] != publication_year for item in output) else "found") if output else "historical_quartile_unavailable"
 
 
 def merge(existing, additions, lookup=None):

app/services/publication_vak.py

@@ -0,0 +1,71 @@
+"""Match a publication's journal to the supplied K1/K2 category snapshots."""
+
+from collections import defaultdict
+from datetime import date
+from functools import lru_cache
+import hashlib
+import json
+from pathlib import Path
+import re
+
+from app.services.council_catalog import issn
+from app.services.council_source import _name
+
+DATA = Path(__file__).resolve().parents[1] / "data"
+
+
+def journal_key(value):
+    return " ".join(re.findall(r"\w+", _name(value)))
+
+
+@lru_cache(maxsize=1)
+def category_index():
+    files = {category: DATA / filename for category, filename in
+             (("К1", "k1_journals.json"), ("К2", "k2_journals.json"))}
+    documents = {category: json.loads(path.read_text(encoding="utf8")) for category, path in files.items()}
+    if documents["К1"]["source_sha256"] != documents["К2"]["source_sha256"]:
+        raise ValueError("VAK category source mismatch")
+    identities = json.loads((DATA / "k1_specialties.json").read_text(encoding="utf8"))
+    if identities["k1_source_sha256"] != hashlib.sha256(files["К1"].read_bytes().replace(b"\r\n", b"\n")).hexdigest():
+        raise ValueError("VAK journal identity source mismatch")
+    known = {item["name"]: sorted({issn(entry["issn"]) for entry in item["vak_entries"]})
+             for item in identities["journals"] if item.get("status") == "matched"}
+    by_name, by_issn = defaultdict(list), defaultdict(list)
+    for category, document in documents.items():
+        for journal in document["journals"]:
+            record = {"category": category, "journal": journal["name"], "source": document["source"],
+                      "source_sha256": document["source_sha256"], "source_row": journal["source_row"],
+                      "categorized_on": journal["categorized_on"], "issns": known.get(journal["name"], []) if category == "К1" else []}
+            if record["issns"]:
+                record.update(identity_source=identities["source"], identity_source_sha256=identities["source_sha256"],
+                              identity_as_of=identities["as_of"])
+            for name in [journal["name"], *journal.get("aliases", [])]:
+                by_name[journal_key(name)].append(record)
+            for identifier in record["issns"]:
+                by_issn[identifier].append(record)
+    return dict(by_name), dict(by_issn)
+
+
+def assess_vak(names, identifiers):
+    by_name, by_issn = category_index()
+    matches = {}
+    normalized = {issn(value) for value in identifiers} - {""}
+    for identifier in normalized:
+        for record in by_issn.get(identifier, []):
+            matches[(record["category"], record["source_row"])] = {**record, "match_method": "issn", "identity_confirmed": True}
+    for name in names:
+        for record in by_name.get(journal_key(name), []):
+            if record["issns"] and normalized and not normalized.intersection(record["issns"]):
+                return {"schema_version": 1, "entries": [], "status": "journal_identity_conflict",
+                        "scope": "supplied_category_snapshot", "historical_category_confirmed": False, "checked_on": date.today().isoformat()}
+            key = record["category"], record["source_row"]
+            matches.setdefault(key, {**record, "match_method": "normalized_name", "identity_confirmed": False})
+    if len(matches) > 1:
+        status, output = "ambiguous_journal", []
+    elif matches:
+        status, output = "found", list(matches.values())
+    else:
+        status, output = ("not_found_in_k1_k2" if names or identifiers else "unknown_journal"), []
+    return {"schema_version": 1, "entries": output, "status": status,
+            "scope": "supplied_category_snapshot", "historical_category_confirmed": False,
+            "checked_on": date.today().isoformat()}

app/version.py

@@ -1,3 +1,3 @@
-APP_VERSION = "0.17.0"
-FRONTEND_VERSION = "0.17.0"
-BACKEND_VERSION = "0.17.0"
+APP_VERSION = "0.18.0"
+FRONTEND_VERSION = "0.18.0"
+BACKEND_VERSION = "0.18.0"

docs/publication-database.md

@@ -18,6 +18,7 @@
 | `document_url` | text | Нет | Ссылка на документ |
 | `citation_text` | text | Нет | Библиографическое описание |
 | `quartiles` | jsonb | Нет | Квартили Q1–Q4 с системой, годом метрики, источником и результатом поиска |
+| `vak_categories` | jsonb | Нет | Результат проверки журнала по предоставленным спискам К1/К2, способ идентификации и источник |
 | `annotation` | jsonb | Нет | Аннотация |
 | `description` | jsonb | Нет | Структурированное описание |
 | `authors` | jsonb | Нет | Список авторов |
@@ -65,7 +66,7 @@
 - Значения `Квартиль scopus` и `Квартиль wos` из `raw_data.csv_enrichment` переносятся в отдельные записи Q1–Q4. Их точная метрика и год метрики неизвестны: `metric` и `metric_year` остаются `null`. Год публикации не объявляется годом метрики. Исходное обогащение не удаляется.
 - При поиске новых значений журнал определяется по DOI через [Crossref](https://www.crossref.org/documentation/retrieve-metadata/rest-api/), а при отсутствии записи — через публичный [DataCite API](https://support.datacite.org/docs/rest-api). ISSN связывает журнал с рейтингом. Проверяются идентичность DOI, тип материала и соответствие года публикации данным DOI. Книге или главе не присваивается журнальный квартиль по одному похожему названию. Препринт DataCite не получает квартиль журнала связанной статьи автоматически.
 - Для старого ISSN допускается подтверждённая преемственность названия по официальному [списку Scopus за август 2026](https://downloads.ctfassets.net/o78em1y1w4i4/7xtaTxNiNcWRTeZkV86eNy/69cf2d506c905dc299531fdc93049dbb/ext_list_Aug_2026.xlsx). Нужны явная связь Continued as/Continues и единственное связанное название, совпадающее с журналом в ответе DOI. Похожесть названий не используется. Исходные ISSN Crossref сохраняются; дополнительные ISSN для рейтинга и ссылка на историю записываются отдельно в `lookup`, ссылка `journal_identity_source_url` также сохраняется в найденных квартилях.
-- Сначала используется метрика за год публикации. При явно разрешённом `--allow-latest` и отсутствии исторической метрики допускается последний доступный год: он остаётся в `metric_year`, а результат имеет статус `found_latest`. Это последний год в доступных проверенных источниках, а не обещание глобально самого нового рейтинга.
+- Для каждой системы сначала используется метрика за год публикации. При явно разрешённом `--allow-latest` и отсутствии исторической метрики допускается последний доступный год: он остаётся в `metric_year`, а результат имеет статус `found_latest`. Все категории CiteScore/JCR и SJR сохраняются вместе; разные системы могут иметь разные годы метрики. Это последний год в доступных проверенных источниках, а не обещание глобально самого нового рейтинга.
 - Источники: проверяемые каталоги проекта (CiteScore/JCR) и исторические SJR-данные [Universidad de Extremadura](https://opendata.unex.es/dataset/revistas-cientificas-2025) за 2009–2025. Портал указывает SCImago как автора и поддерживающую организацию и публикует данные по CC BY 3.0; атрибуция и ссылка сохраняются в каждой записи. [SJR](https://www.scimagolab.com/products/sjr-scimago-journal-country-rank/) хранится с системой SCImago, даже несмотря на происхождение исходных данных из Scopus.
 
 `lookup` также может содержать полученные по DOI тип материала, названия журнала, ISSN, годы и ссылку на регистратора DOI. Состояния поиска:
@@ -86,7 +87,7 @@
 
 ## Миграция и обогащение
 
-Миграция `migrations/010_publication_quartiles.sql` добавляет nullable JSONB-поле без удаления или перезаписи данных. Инициализация приложения также добавляет поле на существующей схеме PostgreSQL/SQLite. Обогащение не запускается при старте API или обработке HTTP-запроса.
+Миграции `migrations/010_publication_quartiles.sql` и `migrations/011_publication_vak_categories.sql` добавляют nullable JSONB-поля без удаления или перезаписи данных. Инициализация приложения также добавляет поля на существующей схеме PostgreSQL/SQLite. Обогащение не запускается при старте API или обработке HTTP-запроса.
 
 Перед изменением заполненной БД создать и проверить резервную копию. Подключение инструмента — только через `DATABASE_URL` из окружения; `.env` и файлы с credentials инструмент не читает. Не передавать секреты в аргументах и не коммитить отчёты, кэши и резервные копии.
 
@@ -100,11 +101,28 @@ python scripts/enrich_publication_quartiles.py --apply --cache-dir .impeccable/q
 Затем проверить DOI у записей без квартиля. Для предварительного сравнения убрать `--apply`:
 
 ```sh
-python scripts/enrich_publication_quartiles.py --apply --lookup-doi --sjr-history --allow-latest --workers 4 --cache-dir .impeccable/quartiles/cache --report .impeccable/quartiles/doi-applied.json
+python scripts/enrich_publication_quartiles.py --apply --lookup-doi --all-systems --check-vak --sjr-history --allow-latest --workers 4 --cache-dir .impeccable/quartiles/cache --report .impeccable/quartiles/doi-applied.json
 ```
 
 Для проверки смены ISSN добавить `--scopus-title-history /путь/ext_list_Aug_2026.xlsx`. Используется конкретный официальный снимок: SHA-256 `11e81f686401c89fbef28de31d1880388bb7122f35c89e09db6e1848237e9afb`; несоответствующий файл отклоняется. Это список названий и их истории, не источник квартилей.
 
+`--all-systems` проверяет также публикации с уже известным квартилем: добавляются остальные доступные системы и предметные категории. Без этого флага проверяются записи без квартиля и ранее найденные последние метрики, для которых может появиться исторический год. Новые доказательства дополняют список, не удаляя прежние. Несколько категорий с одинаковым Q1 сохраняются отдельно; один квартиль не заменяет остальные.
+
+## Поле категорий ВАК
+
+`vak_categories` имеет `schema_version: 1`, массив `entries`, `status`, `scope: "supplied_category_snapshot"`, `historical_category_confirmed: false` и дату `checked_on`. `NULL` — проверка ещё не проводилась. В каждом совпадении сохраняются `category` (К1/К2), каноническое название журнала, название и SHA-256 источника, строка источника, дата категоризации, ISSN при наличии, `match_method` и `identity_confirmed`.
+
+Проверяются `k1_journals.json` и `k2_journals.json`, извлечённые из предоставленного документа «ВАК категории.pdf». Для К1 доступна дополнительная идентификация по ISSN из `k1_specialties.json`; её источник, SHA-256 и дата списка сохраняются отдельно. Поиск К2 в текущих данных выполняется по нормализованному названию и явным административным переименованиям. Транслитерация, похожесть названий и предположение о соответствии русской и переводной версии не используются.
+
+- `match_method: "issn"`, `identity_confirmed: true` — журнал идентифицирован по ISSN.
+- `match_method: "normalized_name"`, `identity_confirmed: false` — название совпало со списком; идентичность по ISSN не подтверждена. Это предварительное совпадение.
+- `status: "found"` — одно совпадение; `ambiguous_journal` или `journal_identity_conflict` — неоднозначность/конфликт, категория автоматически не выбирается.
+- `not_found_in_k1_k2` — совпадений в этих двух снимках нет; это не доказывает отсутствие журнала в ВАК или наличие К3. `unknown_journal` — данных для идентификации журнала нет.
+
+Дата категоризации в документе не объявляется датой всего снимка, а категория по имеющемуся списку не подтверждает её наличие в год публикации. Книги и главы не становятся журнальными статьями из-за совпадения названия. Проверка К1/К2 не является подтверждением зачёта конкретной работы по критериям совета.
+
+`--check-vak` сохраняет этот результат в отдельном поле. Уже заполненные отличающиеся значения не перезаписываются; число таких случаев отражается в `vak_existing_value_kept`. Их нужно отдельно сравнить с актуальными источниками. Проверить результаты и причины неоднозначности до использования данных в заключении.
+
 Запросы DOI кэшируются по нормализованному DOI и переиспользуются для соавторов. Ошибки сети допускают повторную попытку; для полного обновления ранее найденных ответов использовать новый каталог кэша. Исторические рейтинги запрашиваются только для ISSN найденных журналов; ошибки источника отражаются в `ranking_failures` отчёта. Существующие квартильные записи не перезаписываются; повторный перенос не дублирует доказательства только из-за новой даты проверки. Перед записью инструмент проверяет, что публикация не изменилась параллельно; при изменении пропускает её и фиксирует `concurrent_change_skipped`. Коммит выполняется по публикациям: прерванную обработку можно возобновить. Проверить итоговый отчёт и полноту переноса после завершения.
 
 Парсер сохраняет `quartiles`, обновляя остальные исходные поля публикации. Страница совета читает отдельное поле с обратной совместимостью для исходных данных ВШЭ. В предварительные Q1/Q2 совета попадают Scopus/Web of Science с применимым годом; последние метрики другого года и SJR автоматически не превращаются в исторический зачёт или документальное подтверждение критериев.

migrations/011_publication_vak_categories.sql

@@ -0,0 +1,2 @@
+-- Matches to the supplied category snapshot, with provenance and identity state.
+ALTER TABLE employee_publications ADD COLUMN IF NOT EXISTS vak_categories JSONB;

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "miem-workers"
-version = "0.17.0"
+version = "0.18.0"
 description = "MIEM employees parser, admin API, and web admin"
 requires-python = ">=3.11"
 dependencies = [

scripts/enrich_publication_quartiles.py

@@ -18,9 +18,10 @@ from sqlalchemy import MetaData, Table, create_engine, select, update
 from openpyxl import load_workbook
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
-from app.services.council_source import _doi, _name
+from app.services.council_source import _doi, _name, _journal_names, _description_journals
 from app.services.council_catalog import issn
 from app.services.publication_quartiles import catalog_entries, entries, hse_entries, merge, sjr_entries, index_sjr, with_title_history
+from app.services.publication_vak import assess_vak
 
 SOURCE_TITLES_URL = "https://downloads.ctfassets.net/o78em1y1w4i4/7xtaTxNiNcWRTeZkV86eNy/69cf2d506c905dc299531fdc93049dbb/ext_list_Aug_2026.xlsx"
 SOURCE_TITLES_SHA256 = "11e81f686401c89fbef28de31d1880388bb7122f35c89e09db6e1848237e9afb"
@@ -119,7 +120,7 @@ def resolve_datacite(doi, timeout):
     for attempt in range(3):
         try:
             response = requests.get(url, timeout=timeout, allow_redirects=False,
-                                    headers={"User-Agent": "MIEM-workers/0.17 publication-quartiles", "Accept": "application/json"})
+                                    headers={"User-Agent": "MIEM-workers/0.18 publication-quartiles", "Accept": "application/json"})
             if response.status_code == 404:
                 return {**result, "status": "doi_not_found"}
             if response.status_code == 429 or response.status_code >= 500:
@@ -163,7 +164,7 @@ def resolve_doi(doi, cache, timeout=20):
     for attempt in range(3):
         try:
             response = requests.get(url, timeout=timeout, allow_redirects=False,
-                                    headers={"User-Agent": "MIEM-workers/0.17 publication-quartiles",
+                                    headers={"User-Agent": "MIEM-workers/0.18 publication-quartiles",
                                              "Accept": "application/json"})
             if response.status_code == 404:
                 result = resolve_datacite(doi, timeout)
@@ -200,6 +201,8 @@ def main():
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--apply", action="store_true", help="Commit changes; default is read-only")
     parser.add_argument("--lookup-doi", action="store_true")
+    parser.add_argument("--all-systems", action="store_true", help="Collect additional systems/categories even for publications with known quartiles")
+    parser.add_argument("--check-vak", action="store_true", help="Check the supplied K1/K2 snapshots")
     parser.add_argument("--allow-latest", action="store_true", help="Use latest available metric if the publication year is unavailable")
     parser.add_argument("--sjr-history", action="store_true", help="Check UNEx/SCImago historical quartiles by DOI ISSNs")
     parser.add_argument("--scopus-title-history", type=Path, help="Official Scopus August 2026 source-title XLSX for explicit ISSN succession")
@@ -218,13 +221,18 @@ def main():
     table = Table("employee_publications", MetaData(), autoload_with=engine)
     if "quartiles" not in table.c:
         raise ValueError("Apply migration 010 before enrichment")
-    columns = [table.c[field] for field in ("id", "year", "doi_url", "raw_data", "quartiles")]
+    fields = ["id", "year", "doi_url", "raw_data", "quartiles"]
+    if args.check_vak:
+        if "vak_categories" not in table.c:
+            raise ValueError("Apply migration 011 before VAK enrichment")
+        fields.extend(["description", "publication_type", "vak_categories"])
+    columns = [table.c[field] for field in fields]
     with engine.connect() as connection:
         rows = [dict(row) for row in connection.execute(select(*columns)).mappings()]
     def needs_lookup(row):
-        return not entries(row["quartiles"]) or row["quartiles"].get("lookup", {}).get("status") == "found_latest"
+        return args.all_systems or not entries(row["quartiles"]) or row["quartiles"].get("lookup", {}).get("status") == "found_latest"
     identifiers = sorted({_doi(row["doi_url"] or "") for row in rows
-                          if needs_lookup(row) and not hse_entries(row["raw_data"], row["year"])} - {""})
+                          if needs_lookup(row) and (args.all_systems or not hse_entries(row["raw_data"], row["year"]))} - {""})
     resolved = {}
     if args.lookup_doi:
         with ThreadPoolExecutor(max_workers=args.workers) as pool:
@@ -248,41 +256,61 @@ def main():
     counts = Counter()
     report = []
     for row in rows:
+        doi = _doi(row["doi_url"] or "")
+        metadata = resolved.get(doi, row["quartiles"].get("lookup", {}) if row["quartiles"] else {})
         additions = hse_entries(row["raw_data"], row["year"])
         payload = merge(row["quartiles"], additions)
         if additions:
             counts["hse_quartiles"] += 1
-        if args.lookup_doi and (not entries(payload) or payload.get("lookup", {}).get("status") == "found_latest"):
+        if args.lookup_doi and (args.all_systems or not entries(payload) or payload.get("lookup", {}).get("status") == "found_latest"):
             doi = _doi(row["doi_url"] or "")
             metadata = resolved.get(doi, {"status": "invalid_doi" if row["doi_url"] else "missing_doi"})
             status = metadata["status"]
             if status == "resolved":
-                found, status = catalog_entries(metadata, row["year"])
-                if not found and status == "historical_quartile_unavailable" and rankings:
-                    found, status = sjr_entries(metadata, row["year"], rankings)
-                if not found and status == "historical_quartile_unavailable" and args.allow_latest:
-                    found, status = catalog_entries(metadata, row["year"], True)
-                    if not found and rankings:
-                        found, status = sjr_entries(metadata, row["year"], rankings, True)
+                found, status = catalog_entries(metadata, row["year"], args.allow_latest)
+                if rankings:
+                    sjr, sjr_status = sjr_entries(metadata, row["year"], rankings, args.allow_latest)
+                    found.extend(sjr)
+                    if not found and status == "historical_quartile_unavailable":
+                        status = sjr_status
+                if found:
+                    status = "found_latest" if any(item["metric_year"] != row["year"] for item in found) else "found"
                 payload = merge(payload, found)
             payload["lookup"] = {**metadata, "status": status, "requested_year": row["year"],
                                  "checked_on": date.today().isoformat()}
             counts[status] += 1
         counts["with_quartile" if entries(payload) else "without_quartile"] += 1
-        if payload == row["quartiles"]:
+        vak_payload = row.get("vak_categories")
+        if args.check_vak:
+            raw = row["raw_data"] if isinstance(row["raw_data"], dict) else {}
+            imported = raw.get("csv_enrichment", {})
+            names = [*_journal_names(raw), *(_journal_names(imported) if isinstance(imported, dict) else []),
+                     *(metadata.get("journals", []) if metadata.get("type") in {"journal-article", "journal"} else [])]
+            if row["publication_type"] == "ARTICLE" and metadata.get("type") in {None, "journal-article", "journal"}:
+                description = row["description"] if isinstance(row["description"], dict) else raw.get("description", {})
+                if isinstance(description, dict):
+                    names.extend(_description_journals(description))
+            vak_payload = assess_vak(list(dict.fromkeys(names)), metadata.get("issns", []))
+            if row.get("vak_categories") is not None and row["vak_categories"] != vak_payload:
+                counts["vak_existing_value_kept"] += 1
+                vak_payload = row["vak_categories"]
+            counts["vak_" + vak_payload["status"]] += 1
+        if payload == row["quartiles"] and vak_payload == row.get("vak_categories"):
             continue
         if args.apply:
             # Only this field is changed; parser-owned data and existing evidence survive.
             with engine.begin() as connection:
                 current = dict(connection.execute(select(*columns).where(table.c.id == row["id"])
                                .with_for_update()).mappings().one())
-                if any(current[field] != row[field] for field in ("year", "doi_url", "raw_data", "quartiles")):
+                if any(current[field] != row[field] for field in fields if field != "id"):
                     counts["concurrent_change_skipped"] += 1
                     continue
-                connection.execute(update(table).where(table.c.id == row["id"]).values(
-                    quartiles=payload, updated_at=datetime.now(timezone.utc)))
+                values = {"quartiles": payload, "updated_at": datetime.now(timezone.utc)}
+                if args.check_vak:
+                    values["vak_categories"] = vak_payload
+                connection.execute(update(table).where(table.c.id == row["id"]).values(**values))
         counts["changed"] += 1
-        report.append({"id": row["id"], "quartiles": payload})
+        report.append({"id": row["id"], "quartiles": payload, **({"vak_categories": vak_payload} if args.check_vak else {})})
     summary = {"mode": "apply" if args.apply else "dry-run", "publications": len(rows),
                "unique_doi_checked": len(resolved), "counts": dict(counts), "ranking_failures": ranking_failures}
     pending.write_text(json.dumps({"summary": summary, "records": report}, ensure_ascii=False, indent=2), encoding="utf8")

tests/test_api.py

@@ -18,7 +18,7 @@ def test_health_returns_versions():
     response = TestClient(app).get("/api/health")
 
     assert response.status_code == 200
-    assert response.json()["backend_version"] == "0.17.0"
+    assert response.json()["backend_version"] == "0.18.0"
 
 
 def test_api_employees_and_stats_require_admin_session():

tests/test_council.py

@@ -218,7 +218,7 @@ def test_public_page_data_minimization_and_safe_links(db_session, export):
         assert "&lt;script&gt;" in response.text and "<script>" not in response.text
         assert "javascript:" not in response.text and "private@example.org" not in response.text
         assert "/admin/employees/" not in response.text and "/admin/logout" not in response.text
-        assert "Backend 0.17.0 · Frontend 0.17.0" in response.text
+        assert "Backend 0.18.0 · Frontend 0.18.0" in response.text
         assert "Вход в админку" in response.text
         assert client.get("/admin/directory", follow_redirects=False).status_code in (302, 303, 401, 403)
         assert "Сотрудники не найдены" in client.get("/admin/council?q=missing").text

tests/test_publication_quartiles.py

@@ -90,8 +90,10 @@ def test_schema_addition_is_repeatable_and_preserves_existing_data(monkeypatch):
 def test_crawler_preserves_quartiles_while_replacing_source_owned_raw_data():
     payload = quartiles.merge(None, quartiles.hse_entries({"csv_enrichment": {"Квартиль scopus": "Q4"}}, 2024))
     publication = EmployeePublication(title="Old", source_hash="old", quartiles=deepcopy(payload))
+    publication.vak_categories = {"schema_version": 1, "entries": [{"category": "К1"}], "status": "found"}
     _apply_publication(publication, {"id": "42", "title": "Updated", "year": 2024, "raw_data": {"title": "Updated"}}, "new")
     assert publication.quartiles == payload
+    assert publication.vak_categories["entries"] == [{"category": "К1"}]
     assert publication.title == "Updated" and publication.source_hash == "new"
 
 

tests/test_publication_vak.py

@@ -0,0 +1,80 @@
+import pytest
+
+from app.services import publication_vak as vak
+from app.services.publication_quartiles import catalog_entries, sjr_entries, index_sjr
+
+
+@pytest.fixture()
+def index(monkeypatch):
+    k1 = {"category": "К1", "journal": "Первый журнал", "source_row": 1, "issns": ["00063835"]}
+    k2 = {"category": "К2", "journal": "Второй журнал", "source_row": 2, "issns": []}
+    monkeypatch.setattr(vak, "category_index", lambda: ({"первый журнал": [k1], "второй журнал": [k2]}, {"00063835": [k1]}))
+    return k1, k2
+
+
+def test_vak_identifies_k1_by_issn_without_guessing_translated_names(index):
+    result = vak.assess_vak(["English title"], ["0006-3835"])
+    assert result["status"] == "found" and result["entries"][0]["category"] == "К1"
+    assert result["entries"][0]["identity_confirmed"] and not result["historical_category_confirmed"]
+
+
+def test_k2_name_match_is_explicitly_preliminary(index):
+    result = vak.assess_vak([" Второй журнал. "], [])
+    assert result["entries"][0]["category"] == "К2"
+    assert not result["entries"][0]["identity_confirmed"]
+    assert result["entries"][0]["match_method"] == "normalized_name"
+
+
+def test_vak_conflicting_identifiers_and_multiple_journals_are_not_unioned(index):
+    assert vak.assess_vak(["Первый журнал"], ["1111-1111"])["status"] == "journal_identity_conflict"
+    assert vak.assess_vak(["Первый журнал", "Второй журнал"], [])["status"] == "ambiguous_journal"
+    assert vak.assess_vak([], [])["status"] == "unknown_journal"
+    assert vak.assess_vak(["Другой журнал"], [])["status"] == "not_found_in_k1_k2"
+
+
+def test_committed_category_snapshots_match_known_journal_and_record_sources():
+    vak.category_index.cache_clear()
+    result = vak.assess_vak(["Бизнес-информатика"], [])
+    assert result["status"] == "found" and result["entries"][0]["category"] == "К1"
+    assert result["entries"][0]["source_sha256"] and result["entries"][0]["categorized_on"]
+
+
+def test_all_catalog_systems_keep_their_own_metric_year_and_multiple_categories(monkeypatch):
+    metrics = [{"year": year, "system": system, "quartile": quarter, "metric": metric,
+                "category": category, "source": "Snapshot", "source_url": "https://example.org/source"}
+               for year, system, quarter, metric, category in [(2024, "Scopus", "Q1", "CiteScore", "Analysis"),
+                 (2024, "Scopus", "Q2", "CiteScore", "Applied Mathematics"), (2025, "Web of Science", "Q3", "JCR", "Mathematics")]]
+    monkeypatch.setattr("app.services.publication_quartiles.publication_catalog", lambda ids, names: ([{"name": "Journal", "metrics": metrics}], []))
+    metadata = {"type": "journal-article", "years": [2024], "issns": ["0006-3835"]}
+    result, status = catalog_entries(metadata, 2024, True)
+    assert status == "found_latest" and len(result) == 3
+    assert {(r["system"], r["quartile"], r["metric_year"]) for r in result} == {
+        ("Scopus", "Q1", 2024), ("Scopus", "Q2", 2024), ("Web of Science", "Q3", 2025)}
+
+
+def test_book_container_title_is_not_treated_as_a_vak_journal(monkeypatch, tmp_path):
+    from sqlalchemy import create_engine, select
+    from sqlalchemy.orm import Session
+    from app.db import Base
+    from app.models import Employee, EmployeePublication
+    from scripts.enrich_publication_quartiles import main
+    url = "sqlite:///" + str(tmp_path / "books.db")
+    engine = create_engine(url)
+    Base.metadata.create_all(engine)
+    with Session(engine) as session:
+        employee = Employee(profile_key="test", canonical_url="https://example.org/person")
+        session.add(employee)
+        session.flush()
+        session.add(EmployeePublication(employee_id=employee.id, title="Chapter", year=2024, publication_type="CHAPTER",
+            source_hash="hash", doi_url="10.1234/chapter", raw_data={}))
+        session.commit()
+    monkeypatch.setenv("DATABASE_URL", url)
+    monkeypatch.setattr("scripts.enrich_publication_quartiles.resolve_doi", lambda *args: {
+        "status": "resolved", "type": "book-chapter", "journals": ["Бизнес-информатика"], "issns": [], "years": [2024]})
+    monkeypatch.setattr("sys.argv", ["enrich", "--apply", "--lookup-doi", "--check-vak", "--all-systems",
+        "--cache-dir", str(tmp_path / "cache"), "--report", str(tmp_path / "report.json")])
+    main()
+    with Session(engine) as session:
+        publication = session.scalar(select(EmployeePublication))
+        assert publication.vak_categories["status"] == "unknown_journal"
+        assert publication.vak_categories["entries"] == [] and publication.quartiles["entries"] == []