feat: store and enrich publication quartiles with provenance

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## 0.17.0
+
+- Добавлено отдельное поле публикации `quartiles` с квартилями Q1–Q4, системой, годом метрики и источником. Данные ВШЭ переносятся без удаления исходных значений; новые сведения сопоставляются по DOI и ISSN.
+- Добавлена возобновляемая проверка DOI и обогащение по квартильным каталогам: при отсутствии исторической метрики можно явно разрешить последний доступный год. Недоступные сведения и конфликты фиксируются отдельно; текущий квартиль не подменяет исторический зачёт в совете.
+- Состав полей публикации, миграция и порядок обогащения описаны в документации.
+
 ## 0.16.5
 
 - Закреплён обязательный перенос недостающих данных перед будущим production-выпуском: резервная копия, предварительная сверка, сохранение заполненных значений, отдельная обработка конфликтов и проверка полноты до обновления приложения.

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.16.5`. Админка всегда показывает версии backend и frontend в footer.
+Версия сервиса: `0.17.0`. Админка всегда показывает версии backend и frontend в footer.
 
 ## Разовый импорт публикаций
 
@@ -200,6 +200,12 @@ python scripts/extract_k1_specialties.py "app/data/ВАК Перечень.pdf"
 `k1_source_sha256` рассчитывается для исходного JSON с окончаниями строк LF; оба
 JSON-справочника записываются с LF для воспроизводимости на Windows и Linux.
 
+### Квартили публикаций и схема БД
+
+Полный состав полей `employee_publications`, структура отдельного поля `quartiles`, ограничения сопоставления и команды обогащения описаны в [docs/publication-database.md](docs/publication-database.md). Для существующей БД добавлено nullable JSONB-поле миграцией `010_publication_quartiles.sql`; данные переносятся отдельным инструментом, а не при старте приложения.
+
+Квартили Q1–Q4 из данных ВШЭ сохраняются с исходной системой; неизвестный год метрики остаётся неизвестным. Для остальных публикаций журнал устанавливается по DOI, рейтинг — по ISSN. Исторические SJR доступны через открытые данные университета Эстремадуры за 2009–2025; CiteScore/JCR читаются из проверенных каталогов. При отсутствии метрики за год публикации допускается явно разрешённый последний доступный год, с сохранением его года и источника. Отсутствие DOI, неоднозначность и недоступная метрика фиксируются без присвоения вымышленного квартиля. Парсер сохраняет новое поле при обновлении записи.
+
 ### Подготовка публикаций для портала
 
 Портал читает `app/data/council_articles.json`: публикации представлены отдельными

app/db.py

@@ -37,6 +37,13 @@ def _ensure_runtime_schema() -> None:
         models.EmployeePublication.__table__.create(bind=engine, checkfirst=True)
         inspector = inspect(engine)
         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")))
     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

@@ -108,6 +108,7 @@ class EmployeePublication(Base):
     other_url: Mapped[str | None] = mapped_column(Text)
     document_url: Mapped[str | None] = mapped_column(Text)
     citation_text: Mapped[str | None] = mapped_column(Text)
+    quartiles: 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

@@ -19,6 +19,7 @@ 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_catalog import catalog, publication_catalog
+from app.services.publication_quartiles import labels as quartile_labels
 from app.services.council_assessment import (
     BRANCHES, assess_candidate, degree_level, evidence, publication_proofs,
     publication_type_allowed, sourced,
@@ -129,7 +130,8 @@ def _database_article(publication) -> dict:
         journals.extend(_description_journals(description))
     return {"title": publication.title, "year": publication.year, "doi": _doi(publication.doi_url or ""),
             "types": [TYPES.get(kind.upper(), f"Другой тип: {kind}")] if kind else [],
-            "quartiles": _quartiles(imported) if isinstance(imported, dict) else [],
+            "quartiles": list(dict.fromkeys([*quartile_labels(publication.quartiles, publication.year),
+                          *(_quartiles(imported) if isinstance(imported, dict) else [])])),
             "journals": journals, "issns": [raw[key] for key in ("issn", "eissn") if isinstance(raw.get(key), str)],
             "source_hashes": [publication.source_hash],
             "origins": origins, "rows": [], "in_csv": bool(imported), "in_db": from_site,

app/services/publication_quartiles.py

@@ -0,0 +1,175 @@
+"""Publication quartiles with source, system and year kept separately."""
+
+from datetime import date
+from collections import defaultdict
+import re
+
+from app.services.council_catalog import publication_catalog, safe_url
+from app.services.council_source import _name
+
+
+SYSTEM_FIELDS = {"Scopus": "Квартиль scopus", "Web of Science": "Квартиль wos"}
+QUARTILES = {"Q1", "Q2", "Q3", "Q4"}
+
+
+def entries(payload):
+    if payload is None:
+        return []
+    if (not isinstance(payload, dict) or type(payload.get("schema_version")) is not int or payload["schema_version"] != 1
+            or not isinstance(payload.get("entries"), list)
+            or not all(isinstance(item, dict) and item.get("quartile") in QUARTILES
+                       and item.get("system") in {*SYSTEM_FIELDS, "SCImago"} for item in payload["entries"])):
+        raise ValueError("Invalid publication quartiles")
+    return payload["entries"]
+
+
+def hse_entries(raw, publication_year):
+    metadata = raw.get("csv_enrichment", {}) if isinstance(raw, dict) else {}
+    if not isinstance(metadata, dict):
+        return []
+    source_year = str(metadata.get("Год") or "").strip()
+    applicable_year = int(source_year) if source_year.isdigit() else publication_year
+    output = []
+    for system, field in SYSTEM_FIELDS.items():
+        quartile = str(metadata.get(field) or "").strip().upper()
+        if quartile in QUARTILES:
+            output.append({"system": system, "quartile": quartile, "metric": None,
+                           "metric_year": None, "publication_year": applicable_year,
+                           "category": None, "journal": metadata.get("Название журнала") or None,
+                           "issns": [], "source": "Данные ВШЭ", "source_url": None,
+                           "checked_on": date.today().isoformat()})
+    return output
+
+
+def catalog_entries(doi_metadata, publication_year, allow_latest=False):
+    """Only a journal identified by DOI/ISSN and the exact requested metric year."""
+    if doi_metadata.get("type") != "journal-article":
+        return [], "not_journal_article"
+    if publication_year is None:
+        return [], "missing_publication_year"
+    if publication_year not in doi_metadata.get("years", []):
+        return [], "publication_year_conflict"
+    if not doi_metadata.get("issns"):
+        return [], "missing_journal_issn"
+    records, issues = publication_catalog(doi_metadata["issns"], [])
+    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)
+    output = []
+    for record in records:
+        for metric in record["metrics"]:
+            if metric["year"] == selected_year 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"
+
+
+def merge(existing, additions, lookup=None):
+    current = entries(existing)
+    output = list(current)
+    # Repeating a lookup must not duplicate evidence merely because its date changed.
+    def key(item):
+        return tuple(item.get(field) for field in ("system", "quartile", "metric", "metric_year",
+                     "publication_year", "category", "journal", "source", "source_url"))
+    seen = {key(item) for item in output}
+    for item in additions:
+        if key(item) not in seen:
+            output.append(item)
+            seen.add(key(item))
+    result = {**(existing or {}), "schema_version": 1, "entries": output}
+    if lookup is not None:
+        result["lookup"] = lookup
+    return result
+
+
+def labels(payload, publication_year):
+    return list(dict.fromkeys(f"{item['system']} {item['quartile']}" for item in entries(payload)
+                             if item["quartile"] in {"Q1", "Q2"}
+                             and item["system"] in SYSTEM_FIELDS
+                             and item.get("publication_year") == publication_year
+                             and item.get("metric_year") in {None, publication_year}))
+
+
+def index_sjr(rankings):
+    from app.services.council_catalog import issn
+    indexed = {}
+    for year, records in rankings.items():
+        by_issn = defaultdict(list)
+        for record in records:
+            for identifier in set(map(issn, record["issns"])) - {""}:
+                by_issn[identifier].append(record)
+        indexed[int(year)] = dict(by_issn)
+    return indexed
+
+
+def with_title_history(metadata, by_issn, by_title, source_url):
+    """Use explicit Scopus title succession, never a fuzzy journal-name match."""
+    from app.services.council_catalog import issn
+    if not safe_url(source_url):
+        return metadata
+    extra = set()
+    names = {_name(name) for name in metadata.get("journals", [])}
+    for identifier in metadata.get("issns", []):
+        for old in by_issn.get(issn(identifier), []):
+            if old["history"] not in {"Continued as", "Continues"}:
+                continue
+            for related in old["related"]:
+                candidates = by_title.get(_name(related), [])
+                if _name(related) in names and len(candidates) == 1:
+                    extra.update(candidates[0]["issns"])
+    if not extra:
+        return metadata
+    return {**metadata, "ranking_issns": sorted(set(metadata["issns"]) | extra),
+            "journal_identity_source_url": source_url}
+
+
+def sjr_entries(metadata, publication_year, rankings, allow_latest=False):
+    from app.services.council_catalog import issn
+    if metadata.get("type") != "journal-article":
+        return [], "not_journal_article"
+    if publication_year is None or publication_year not in metadata.get("years", []):
+        return [], "publication_year_conflict"
+    matches = {}
+    identifiers = set(map(issn, metadata.get("ranking_issns", metadata.get("issns", [])))) - {""}
+    for year, by_issn in rankings.items():
+        journals = {record["source_url"]: record for identifier in identifiers for record in by_issn.get(identifier, [])}
+        if journals:
+            matches[int(year)] = journals
+    year = publication_year if publication_year in matches else max(matches, default=None) if allow_latest else None
+    if year is None:
+        return [], "historical_quartile_unavailable"
+    if len(matches[year]) != 1:
+        return [], "journal_identity_conflict"
+    record = next(iter(matches[year].values()))
+    if record.get("conflict"):
+        return [], "ranking_category_conflict"
+    categories = []
+    for category in record.get("categories", "").split(";"):
+        if not category.strip():
+            continue
+        match = re.fullmatch(r"\s*(.+?)\s*\((Q[1-4])\)\s*", category)
+        if match:
+            categories.append((match[1].strip(), match[2]))
+        else:
+            return [], "ranking_category_conflict"
+    if categories and min(quartile for _, quartile in categories) != record["quartile"]:
+        return [], "ranking_category_conflict"
+    if not categories and record.get("quartile") in QUARTILES:
+        categories = [(None, record["quartile"])]
+    result = [{"system": "SCImago", "quartile": quartile, "metric": "SJR" if category else "SJR best quartile",
+               "metric_year": year, "publication_year": publication_year, "category": category,
+               "journal": record["journal"], "issns": record["issns"],
+               "source": "SCImago Journal & Country Rank · Universidad de Extremadura",
+               "source_url": record["source_url"], "checked_on": date.today().isoformat()}
+              for category, quartile in categories]
+    if metadata.get("journal_identity_source_url"):
+        for item in result:
+            item["journal_identity_source_url"] = metadata["journal_identity_source_url"]
+    return result, "found" if year == publication_year else "found_latest"

app/version.py

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

docs/publication-database.md

@@ -0,0 +1,112 @@
+# Запись публикации в БД
+
+Объект `EmployeePublication`, таблица `employee_publications`. Одна запись относится к одному сотруднику; одна работа у нескольких соавторов может иметь несколько записей с одним DOI. Типы ниже указаны для PostgreSQL; в SQLite JSONB представлен JSON.
+
+| Поле | Тип | Обязательное | Назначение |
+|---|---|:---:|---|
+| `id` | integer | Да | Первичный ключ |
+| `employee_id` | integer | Да | Внешний ключ `employees.id`; удаление сотрудника удаляет связанные записи |
+| `publication_id` | varchar(64) | Нет | Идентификатор публикации в источнике ВШЭ |
+| `title` | text | Да | Название |
+| `year` | integer | Нет | Год публикации |
+| `publication_type` | varchar(64) | Нет | Тип материала: ARTICLE, CHAPTER, BOOK, PREPRINT и другие |
+| `language` | varchar(16) | Нет | Язык |
+| `status` | integer | Нет | Код статуса из источника |
+| `url` | text | Нет | Основная ссылка |
+| `doi_url` | text | Нет | DOI или ссылка DOI |
+| `other_url` | text | Нет | Дополнительная ссылка |
+| `document_url` | text | Нет | Ссылка на документ |
+| `citation_text` | text | Нет | Библиографическое описание |
+| `quartiles` | jsonb | Нет | Квартили Q1–Q4 с системой, годом метрики, источником и результатом поиска |
+| `annotation` | jsonb | Нет | Аннотация |
+| `description` | jsonb | Нет | Структурированное описание |
+| `authors` | jsonb | Нет | Список авторов |
+| `raw_data` | jsonb | Нет | Исходные данные, включая обогащение из данных ВШЭ |
+| `source_hash` | varchar(64) | Да | Хеш исходной записи |
+| `created_at` | timestamptz | Да | Создание записи |
+| `updated_at` | timestamptz | Да | Обновление записи |
+
+Уникальны пары `(employee_id, publication_id)` и `(employee_id, source_hash)`. Индексы: `employee_id`, `publication_id`, `doi_url`, `year`, `publication_type`. Атрибут ORM `employee` ссылается на объект сотрудника и отдельным столбцом не является.
+
+## Поле квартилей
+
+`NULL` означает, что обогащение ещё не выполнялось. После проверки поле содержит объект со схемой версии 1. Пустой `entries` означает отсутствие найденного квартиля; причина сохраняется в `lookup`. Пустой результат не равнозначен Q4 и не доказывает отсутствие индексации.
+
+Пример структуры (иллюстративные данные):
+
+```json
+{
+  "schema_version": 1,
+  "entries": [
+    {
+      "system": "Scopus",
+      "quartile": "Q2",
+      "metric": "CiteScore",
+      "metric_year": 2024,
+      "publication_year": 2024,
+      "category": "Applied Mathematics",
+      "journal": "Название журнала",
+      "issns": ["0006-3835"],
+      "source": "Название проверенного источника",
+      "source_url": "https://example.org/metrics",
+      "checked_on": "2026-10-08"
+    }
+  ],
+  "lookup": {
+    "doi": "10.1234/example",
+    "status": "found",
+    "requested_year": 2024,
+    "checked_on": "2026-10-08"
+  }
+}
+```
+
+- `entries` сохраняет все найденные категории и системы отдельно, без выбора одного «лучшего» квартиля вместо остальных. Возможные системы: Scopus, Web of Science, SCImago. Метрики CiteScore, JCR и SJR не подменяют друг друга. При наличии только лучшего квартиля SJR используется метрика `SJR best quartile`, категория остаётся `null`.
+- Значения `Квартиль 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`. Это последний год в доступных проверенных источниках, а не обещание глобально самого нового рейтинга.
+- Источники: проверяемые каталоги проекта (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. Состояния поиска:
+
+| Статус | Значение |
+|---|---|
+| `found` | Найдена метрика за год публикации |
+| `found_latest` | Использован другой, последний доступный год |
+| `missing_doi` / `invalid_doi` | DOI отсутствует или некорректен |
+| `doi_not_found` / `doi_lookup_failed` | DOI не найден в Crossref и DataCite / источник не ответил успешно; DOI может относиться к другому регистратору |
+| `doi_identity_conflict` | Ответ источника относится к другому DOI |
+| `not_journal_article` | DOI относится к материалу, не определённому как журнальная статья |
+| `missing_publication_year` / `publication_year_conflict` | Год неизвестен / отличается от годов источника DOI |
+| `missing_journal_issn` | Журнал по ISSN не установлен |
+| `journal_identity_conflict` | Идентификаторы связываются с несколькими журналами |
+| `ranking_category_conflict` | Сведения о категориях SJR противоречат друг другу |
+| `historical_quartile_unavailable` | Проверенные источники не дали допустимой метрики |
+
+## Миграция и обогащение
+
+Миграция `migrations/010_publication_quartiles.sql` добавляет nullable JSONB-поле без удаления или перезаписи данных. Инициализация приложения также добавляет поле на существующей схеме PostgreSQL/SQLite. Обогащение не запускается при старте API или обработке HTTP-запроса.
+
+Перед изменением заполненной БД создать и проверить резервную копию. Подключение инструмента — только через `DATABASE_URL` из окружения; `.env` и файлы с credentials инструмент не читает. Не передавать секреты в аргументах и не коммитить отчёты, кэши и резервные копии.
+
+Сначала выполнить предварительное сравнение без записи, затем перенос известных значений:
+
+```sh
+python scripts/enrich_publication_quartiles.py --cache-dir .impeccable/quartiles/cache --report .impeccable/quartiles/hse-plan.json
+python scripts/enrich_publication_quartiles.py --apply --cache-dir .impeccable/quartiles/cache --report .impeccable/quartiles/hse-applied.json
+```
+
+Затем проверить 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
+```
+
+Для проверки смены ISSN добавить `--scopus-title-history /путь/ext_list_Aug_2026.xlsx`. Используется конкретный официальный снимок: SHA-256 `11e81f686401c89fbef28de31d1880388bb7122f35c89e09db6e1848237e9afb`; несоответствующий файл отклоняется. Это список названий и их истории, не источник квартилей.
+
+Запросы DOI кэшируются по нормализованному DOI и переиспользуются для соавторов. Ошибки сети допускают повторную попытку; для полного обновления ранее найденных ответов использовать новый каталог кэша. Исторические рейтинги запрашиваются только для ISSN найденных журналов; ошибки источника отражаются в `ranking_failures` отчёта. Существующие квартильные записи не перезаписываются; повторный перенос не дублирует доказательства только из-за новой даты проверки. Перед записью инструмент проверяет, что публикация не изменилась параллельно; при изменении пропускает её и фиксирует `concurrent_change_skipped`. Коммит выполняется по публикациям: прерванную обработку можно возобновить. Проверить итоговый отчёт и полноту переноса после завершения.
+
+Парсер сохраняет `quartiles`, обновляя остальные исходные поля публикации. Страница совета читает отдельное поле с обратной совместимостью для исходных данных ВШЭ. В предварительные Q1/Q2 совета попадают Scopus/Web of Science с применимым годом; последние метрики другого года и SJR автоматически не превращаются в исторический зачёт или документальное подтверждение критериев.
+
+Миграция и обогащение локальной БД не обновляют production. Порядок переноса только недостающих данных в production и сохранения его заполненных значений описан в README.

migrations/010_publication_quartiles.sql

@@ -0,0 +1,2 @@
+-- Source/year-aware quartiles; NULL means enrichment has not run yet.
+ALTER TABLE employee_publications ADD COLUMN IF NOT EXISTS quartiles JSONB;

pyproject.toml

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

scripts/enrich_publication_quartiles.py

@@ -0,0 +1,299 @@
+"""Backfill HSE quartiles and resolve remaining DOI journals. No .env or DDL."""
+
+import argparse
+from collections import Counter, defaultdict
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from datetime import date, datetime, timezone
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import sys
+import time
+from urllib.parse import quote
+
+import requests
+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_catalog import issn
+from app.services.publication_quartiles import catalog_entries, entries, hse_entries, merge, sjr_entries, index_sjr, with_title_history
+
+SOURCE_TITLES_URL = "https://downloads.ctfassets.net/o78em1y1w4i4/7xtaTxNiNcWRTeZkV86eNy/69cf2d506c905dc299531fdc93049dbb/ext_list_Aug_2026.xlsx"
+SOURCE_TITLES_SHA256 = "11e81f686401c89fbef28de31d1880388bb7122f35c89e09db6e1848237e9afb"
+
+
+def load_title_history(path):
+    if hashlib.sha256(path.read_bytes()).hexdigest() != SOURCE_TITLES_SHA256:
+        raise ValueError("Scopus title-list snapshot checksum mismatch")
+    by_issn, by_title = defaultdict(list), defaultdict(list)
+    workbook = load_workbook(path, read_only=True, data_only=True)
+    try:
+        rows = workbook.worksheets[0].iter_rows(values_only=True)
+        columns = next(rows)
+        required = {"Source Title", "ISSN", "EISSN", "Source Type", "Title History Indication", "Related Title 1"}
+        if not required <= set(columns):
+            raise ValueError("Invalid Scopus title list")
+        for row in rows:
+            data = dict(zip(columns, row))
+            if data["Source Type"] != "Journal":
+                continue
+            identifiers = {issn(data.get(key)) for key in ("ISSN", "EISSN")} - {""}
+            identifiers = {identifier[:4] + "-" + identifier[4:] for identifier in identifiers if len(identifier) == 8}
+            record = {"title": str(data["Source Title"] or ""), "issns": identifiers,
+                      "history": data["Title History Indication"],
+                      "related": [str(value) for key, value in data.items() if key and "Related Title" in key and value]}
+            by_title[_name(record["title"])].append(record)
+            for identifier in identifiers:
+                by_issn[issn(identifier)].append(record)
+    finally:
+        workbook.close()
+    return dict(by_issn), dict(by_title)
+
+
+def load_sjr(identifiers, years, cache):
+    """Read the university's licensed SCImago dataset for the requested ISSNs."""
+    rankings = {}
+    failures = []
+    for year in sorted(set(years) & set(range(2009, 2026))):
+        records = {}
+        for offset in range(0, len(identifiers), 200):
+            batch = identifiers[offset:offset + 200]
+            digest = hashlib.sha256(json.dumps([year, batch]).encode()).hexdigest()
+            path = cache / f"sjr-{digest}.json"
+            if path.exists():
+                bindings = json.loads(path.read_text(encoding="utf8"))
+            else:
+                values = " ".join(json.dumps(value) for value in batch)
+                query = f'''PREFIX bibo: <http://purl.org/ontology/bibo/>
+PREFIX dct: <http://purl.org/dc/terms/>
+PREFIX ou: <https://opendata.unex.es/def/ontouniversidad#>
+SELECT DISTINCT ?uri ?title ?issn ?eissn ?quartile ?categories WHERE {{
+ VALUES ?wanted {{ {values} }}
+ {{ ?uri bibo:issn ?wanted }} UNION {{ ?uri bibo:eissn ?wanted }}
+ ?uri a bibo:Journal; dct:title ?title; dct:identifier ?id; ou:sjrMejorCuartil ?quartile .
+ FILTER(STRSTARTS(STR(?id), "{year}-"))
+ OPTIONAL {{ ?uri bibo:issn ?issn }} OPTIONAL {{ ?uri bibo:eissn ?eissn }}
+ OPTIONAL {{ ?uri ou:categoriaSJR ?categories }}
+}} LIMIT 10001'''
+                try:
+                    response = requests.get("https://opendata.unex.es/sparql/", timeout=45,
+                        params={"query": query, "format": "application/sparql-results+json"},
+                        headers={"Accept": "application/sparql-results+json"})
+                    response.raise_for_status()
+                    bindings = response.json()["results"]["bindings"]
+                    if len(bindings) >= 10000:
+                        raise ValueError("Truncated ranking query")
+                    path.write_text(json.dumps(bindings, ensure_ascii=False), encoding="utf8")
+                except (requests.RequestException, ValueError, KeyError, TypeError):
+                    failures.append({"year": year, "batch": offset, "status": "ranking_source_unavailable"})
+                    continue
+            for item in bindings:
+                value = lambda field: item.get(field, {}).get("value", "")
+                url = value("uri")
+                if not url.startswith(f"https://opendata.unex.es/recurso/ciencia-tecnologia/investigacion/revistas/Revista/{year}-"):
+                    raise ValueError("Ranking year/identity mismatch")
+                record = {"journal": value("title"), "issns": sorted({value("issn"), value("eissn")} - {""}),
+                          "quartile": value("quartile"), "categories": value("categories"), "source_url": url}
+                if record["quartile"] not in {"Q1", "Q2", "Q3", "Q4"}:
+                    continue
+                previous = records.get(url)
+                if previous:
+                    record["journal"] = min(previous["journal"], record["journal"])
+                    record["issns"] = sorted(set(previous["issns"]) | set(record["issns"]))
+                    record["categories"] = "; ".join(sorted({category.strip() for value in
+                        (previous["categories"], record["categories"]) for category in value.split(";") if category.strip()}))
+                    record["conflict"] = previous.get("conflict", False) or previous["quartile"] != record["quartile"]
+                records[url] = record
+        rankings[year] = list(records.values())
+        print(f"SJR {year}: {len(records)} matching journals", flush=True)
+    return index_sjr(rankings), failures
+
+
+def resolve_datacite(doi, timeout):
+    url = "https://api.datacite.org/dois/" + quote(doi, safe="")
+    result = {"doi": doi, "status": "doi_lookup_failed", "source_url": url, "datacite_checked": True}
+    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"})
+            if response.status_code == 404:
+                return {**result, "status": "doi_not_found"}
+            if response.status_code == 429 or response.status_code >= 500:
+                time.sleep(min(10, 2 ** attempt))
+                continue
+            response.raise_for_status()
+            metadata = response.json()["data"]["attributes"]
+            if _doi(metadata.get("doi", "")) != doi:
+                return {**result, "status": "doi_identity_conflict"}
+            types = metadata.get("types", {})
+            general = types.get("resourceTypeGeneral")
+            journal_article = general == "JournalArticle" or (general in {None, "Text"} and types.get("citeproc") == "article-journal")
+            container = metadata.get("container") or {}
+            identifiers = []
+            if container.get("identifierType") == "ISSN":
+                identifiers.append(container.get("identifier", ""))
+            for related in metadata.get("relatedIdentifiers", []):
+                if related.get("relatedIdentifierType") == "ISSN" and related.get("relationType") in {"IsPartOf", "IsPublishedIn"}:
+                    identifiers.append(related.get("relatedIdentifier", ""))
+            identifiers = sorted({value[:4] + "-" + value[-4:] for value in identifiers if isinstance(value, str)
+                                  and re.fullmatch(r"\d{4}-?\d{3}[\dX]", value, re.I)})
+            year = metadata.get("publicationYear")
+            years = [int(year)] if str(year).isdigit() else []
+            return {**result, "status": "resolved", "type": "journal-article" if journal_article else "non-journal-material",
+                    "issns": identifiers, "journals": [container["title"]] if isinstance(container.get("title"), str) else [], "years": years}
+        except (requests.RequestException, ValueError, KeyError, TypeError, AttributeError):
+            if attempt < 2:
+                time.sleep(2 ** attempt)
+    return result
+
+
+def resolve_doi(doi, cache, timeout=20):
+    path = cache / (hashlib.sha256(doi.encode()).hexdigest() + ".json")
+    if path.exists():
+        result = json.loads(path.read_text(encoding="utf8"))
+        if result.get("doi") == doi and (result.get("status") == "resolved"
+                or (result.get("status") == "doi_not_found" and result.get("datacite_checked"))):
+            return result
+    url = "https://api.crossref.org/works/" + quote(doi, safe="")
+    result = {"doi": doi, "status": "doi_lookup_failed", "source_url": url}
+    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"})
+            if response.status_code == 404:
+                result = resolve_datacite(doi, timeout)
+                break
+            if response.status_code == 429 or response.status_code >= 500:
+                time.sleep(min(10, 2 ** attempt))
+                continue
+            response.raise_for_status()
+            message = response.json()["message"]
+            if _doi(message.get("DOI", "")) != doi:
+                result["status"] = "doi_identity_conflict"
+                break
+            years = set()
+            for field in ("published", "published-print", "published-online", "issued"):
+                for parts in message.get(field, {}).get("date-parts", []):
+                    if parts and type(parts[0]) is int:
+                        years.add(parts[0])
+            result.update(status="resolved", type=message.get("type"),
+                          journals=[v for v in message.get("container-title", []) if isinstance(v, str)],
+                          issns=[v for v in message.get("ISSN", []) if isinstance(v, str)
+                                 and re.fullmatch(r"\d{4}-\d{3}[\dX]", v, re.I)], years=sorted(years))
+            break
+        except (requests.RequestException, ValueError, KeyError, TypeError, AttributeError):
+            if attempt < 2:
+                time.sleep(2 ** attempt)
+    if result["status"] in {"resolved", "doi_not_found"}:
+        temporary = path.with_suffix(".tmp")
+        temporary.write_text(json.dumps(result, ensure_ascii=False), encoding="utf8")
+        temporary.replace(path)
+    return result
+
+
+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("--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")
+    parser.add_argument("--cache-dir", type=Path, required=True)
+    parser.add_argument("--report", type=Path, required=True)
+    parser.add_argument("--workers", type=int, choices=range(1, 9), default=4)
+    args = parser.parse_args()
+    pending = args.report.with_suffix(args.report.suffix + ".pending")
+    if args.scopus_title_history and args.scopus_title_history.resolve() in {args.report.resolve(), pending.resolve()}:
+        raise ValueError("Report must not overwrite the source title list")
+    args.cache_dir.mkdir(parents=True, exist_ok=True)
+    args.report.parent.mkdir(parents=True, exist_ok=True)
+    # Verify report destination before changing any data.
+    pending.write_text("", encoding="utf8")
+    engine = create_engine(os.environ["DATABASE_URL"])
+    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")]
+    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"
+    identifiers = sorted({_doi(row["doi_url"] or "") for row in rows
+                          if needs_lookup(row) and not hse_entries(row["raw_data"], row["year"])} - {""})
+    resolved = {}
+    if args.lookup_doi:
+        with ThreadPoolExecutor(max_workers=args.workers) as pool:
+            tasks = {pool.submit(resolve_doi, doi, args.cache_dir): doi for doi in identifiers}
+            for task in as_completed(tasks):
+                resolved[tasks[task]] = task.result()
+                if len(resolved) % 100 == 0 or len(resolved) == len(identifiers):
+                    print(f"DOI checked: {len(resolved)}/{len(identifiers)}", flush=True)
+    rankings, ranking_failures = {}, []
+    if args.scopus_title_history:
+        by_issn, by_title = load_title_history(args.scopus_title_history)
+        resolved = {doi: with_title_history(metadata, by_issn, by_title, SOURCE_TITLES_URL)
+                    for doi, metadata in resolved.items()}
+    if args.lookup_doi and args.sjr_history:
+        issns = sorted({value for metadata in resolved.values() if metadata.get("type") == "journal-article"
+                        for value in metadata.get("ranking_issns", metadata.get("issns", []))})
+        years = {row["year"] for row in rows if needs_lookup(row) and row["year"] is not None}
+        if args.allow_latest:
+            years.add(2025)
+        rankings, ranking_failures = load_sjr(issns, years, args.cache_dir)
+    counts = Counter()
+    report = []
+    for row in rows:
+        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"):
+            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)
+                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"]:
+            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")):
+                    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)))
+        counts["changed"] += 1
+        report.append({"id": row["id"], "quartiles": payload})
+    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")
+    pending.replace(args.report)
+    print(json.dumps(summary, ensure_ascii=False), flush=True)
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception as error:
+        # Connection errors can contain credentials; do not print their text.
+        print(f"Enrichment failed ({type(error).__name__}); completed records are retained; retry is safe.", flush=True)
+        raise SystemExit(1)

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.16.5"
+    assert response.json()["backend_version"] == "0.17.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.16.5 · Frontend 0.16.5" in response.text
+        assert "Backend 0.17.0 · Frontend 0.17.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

@@ -0,0 +1,217 @@
+from copy import deepcopy
+from datetime import date
+
+import pytest
+from sqlalchemy import create_engine, inspect, text, select
+from sqlalchemy.orm import Session
+
+from app.db import _ensure_runtime_schema, Base
+from app.models import EmployeePublication, Employee
+from app.services import publication_quartiles as quartiles
+from app.services.crawler import _apply_publication
+from scripts.enrich_publication_quartiles import resolve_doi, load_sjr, main
+
+
+@pytest.mark.parametrize("scopus,wos", [("Q1", "Q2"), ("q3", "Q4"), ("Q4", "")])
+def test_hse_backfill_keeps_systems_q1_to_q4_and_unknown_metric_year(scopus, wos):
+    raw = {"other": "preserve", "csv_enrichment": {"Квартиль scopus": scopus, "Квартиль wos": wos, "Год": "2024"}}
+    before = deepcopy(raw)
+    result = quartiles.hse_entries(raw, 2024)
+    assert raw == before
+    assert [e["quartile"] for e in result] == [v.upper() for v in (scopus, wos) if v]
+    assert all(e["metric_year"] is None and e["publication_year"] == 2024 for e in result)
+    assert [e["system"] for e in result] == (["Scopus", "Web of Science"] if wos else ["Scopus"])
+
+
+@pytest.mark.parametrize("value", ["Q5", "1", "", None])
+def test_invalid_hse_value_does_not_invent_a_quartile(value):
+    assert quartiles.hse_entries({"csv_enrichment": {"Квартиль scopus": value}}, 2024) == []
+
+
+def test_merge_is_idempotent_and_preserves_conflicting_evidence():
+    first = quartiles.hse_entries({"csv_enrichment": {"Квартиль scopus": "Q1"}}, 2024)[0]
+    payload = quartiles.merge(None, [first])
+    assert quartiles.merge(payload, [{**first, "checked_on": "2099-01-01"}]) == payload
+    assert len(quartiles.merge(payload, [{**first, "quartile": "Q2"}])["entries"]) == 2
+    assert quartiles.labels(payload, 2024) == ["Scopus Q1"]
+    assert quartiles.labels(payload, 2023) == []
+
+
+def test_catalog_metric_requires_exact_year_unless_latest_explicitly_allowed(monkeypatch):
+    monkeypatch.setattr(quartiles, "publication_catalog", lambda ids, names: ([{
+        "name": "Journal", "metrics": [{"year": 2025, "system": "Scopus", "quartile": "Q1", "metric": "CiteScore",
+        "category": "Mathematics", "source": "Snapshot", "source_url": "https://example.org/metrics"}]}], []))
+    metadata = {"type": "journal-article", "years": [2024], "issns": ["0006-3835"]}
+    assert quartiles.catalog_entries(metadata, 2024)[0] == []
+    found, status = quartiles.catalog_entries(metadata, 2024, True)
+    assert status == "found_latest" and found[0]["metric_year"] == 2025 and found[0]["publication_year"] == 2024
+    assert quartiles.labels(quartiles.merge(None, found), 2024) == []
+    assert quartiles.catalog_entries(metadata, 2023, True)[1] == "publication_year_conflict"
+    assert quartiles.catalog_entries({**metadata, "type": "book-chapter"}, 2024, True)[1] == "not_journal_article"
+
+
+def sjr_record(year, categories="Analysis (Q1); Applied Mathematics (Q2)"):
+    return {"journal": "Journal", "issns": ["0006-3835"], "quartile": "Q1", "categories": categories,
+            "source_url": f"https://opendata.unex.es/recurso/ciencia-tecnologia/investigacion/revistas/Revista/{year}-1"}
+
+
+def test_sjr_keeps_historical_year_categories_and_system_distinct():
+    metadata = {"type": "journal-article", "years": [2020], "issns": ["0006-3835"]}
+    found, status = quartiles.sjr_entries(metadata, 2020, quartiles.index_sjr({2020: [sjr_record(2020)], 2025: [sjr_record(2025)]}), True)
+    assert status == "found" and [e["quartile"] for e in found] == ["Q1", "Q2"]
+    assert all(e["system"] == "SCImago" and e["metric_year"] == 2020 for e in found)
+    assert quartiles.labels(quartiles.merge(None, found), 2020) == []
+    latest, status = quartiles.sjr_entries(metadata, 2020, quartiles.index_sjr({2025: [sjr_record(2025)]}), True)
+    assert status == "found_latest" and latest[0]["metric_year"] == 2025
+    assert quartiles.sjr_entries(metadata, 2020, quartiles.index_sjr({2025: [sjr_record(2025)]}))[0] == []
+
+
+def test_sjr_identity_and_category_conflicts_do_not_resolve_by_order():
+    metadata = {"type": "journal-article", "years": [2020], "issns": ["0006-3835"]}
+    first = sjr_record(2020)
+    second = {**first, "source_url": first["source_url"] + "2"}
+    assert quartiles.sjr_entries(metadata, 2020, quartiles.index_sjr({2020: [first, second]}))[1] == "journal_identity_conflict"
+    assert quartiles.sjr_entries(metadata, 2020, quartiles.index_sjr({2020: [sjr_record(2020, "Analysis (Q3)")]} ))[1] == "ranking_category_conflict"
+
+
+def test_schema_addition_is_repeatable_and_preserves_existing_data(monkeypatch):
+    engine = create_engine("sqlite://")
+    with engine.begin() as connection:
+        connection.execute(text("CREATE TABLE employee_publications (id INTEGER PRIMARY KEY, title TEXT NOT NULL)"))
+        connection.execute(text("INSERT INTO employee_publications (id,title) VALUES (1,'Keep')"))
+    monkeypatch.setattr("app.db.engine", engine)
+    _ensure_runtime_schema()
+    _ensure_runtime_schema()
+    assert "quartiles" in {c["name"] for c in inspect(engine).get_columns("employee_publications")}
+    with engine.connect() as connection:
+        assert connection.execute(text("SELECT title,quartiles FROM employee_publications")).one() == ("Keep", None)
+
+
+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))
+    _apply_publication(publication, {"id": "42", "title": "Updated", "year": 2024, "raw_data": {"title": "Updated"}}, "new")
+    assert publication.quartiles == payload
+    assert publication.title == "Updated" and publication.source_hash == "new"
+
+
+def test_council_reads_new_field_without_raw_enrichment_and_keeps_historical_filter():
+    from app.services.council import _database_article
+    values = quartiles.hse_entries({"csv_enrichment": {"Квартиль scopus": "Q2"}}, 2024)
+    publication = EmployeePublication(title="Work", year=2024, publication_type="ARTICLE", source_hash="hash",
+        quartiles=quartiles.merge(None, values), raw_data={})
+    assert _database_article(publication)["quartiles"] == ["Scopus Q2"]
+    publication.quartiles = quartiles.merge(None, [{**values[0], "metric_year": 2025}])
+    assert _database_article(publication)["quartiles"] == []
+
+
+def test_doi_cache_uses_verified_identity_and_does_not_need_network_on_repeat(monkeypatch, tmp_path):
+    calls = []
+    class Response:
+        status_code = 200
+        def raise_for_status(self):
+            pass
+        def json(self):
+            return {"message": {"DOI": "10.1234/work", "type": "journal-article", "container-title": ["Journal"],
+                                "ISSN": ["0006-3835"], "published": {"date-parts": [[2024, 2, 1]]}}}
+    def get(url, **kwargs):
+        calls.append(url)
+        assert kwargs["allow_redirects"] is False
+        return Response()
+    monkeypatch.setattr("scripts.enrich_publication_quartiles.requests.get", get)
+    result = resolve_doi("10.1234/work", tmp_path)
+    assert result["status"] == "resolved" and result["years"] == [2024]
+    assert resolve_doi("10.1234/work", tmp_path) == result and len(calls) == 1
+    assert calls[0] == "https://api.crossref.org/works/10.1234%2Fwork"
+
+
+def test_sjr_source_combines_multiple_categories_of_the_same_resource(monkeypatch, tmp_path):
+    url = "https://opendata.unex.es/recurso/ciencia-tecnologia/investigacion/revistas/Revista/2020-1"
+    def binding(category, quartile="Q1"):
+        return {key: {"value": value} for key, value in {
+            "uri": url, "title": "Journal", "issn": "0006-3835", "quartile": quartile, "categories": category}.items()}
+    class Response:
+        def raise_for_status(self):
+            pass
+        def json(self):
+            return {"results": {"bindings": [binding("Analysis (Q1)"), binding("Applied Mathematics (Q2)")]}}
+    monkeypatch.setattr("scripts.enrich_publication_quartiles.requests.get", lambda *args, **kwargs: Response())
+    rankings, errors = load_sjr(["0006-3835"], [2020], tmp_path)
+    assert not errors
+    metadata = {"type": "journal-article", "years": [2020], "issns": ["0006-3835"]}
+    found, status = quartiles.sjr_entries(metadata, 2020, rankings)
+    assert status == "found" and {e["quartile"] for e in found} == {"Q1", "Q2"}
+
+
+def test_doi_identity_conflict_does_not_supply_journal_metrics(monkeypatch, tmp_path):
+    class Response:
+        status_code = 200
+        def raise_for_status(self):
+            pass
+        def json(self):
+            return {"message": {"DOI": "10.1234/another", "type": "journal-article", "ISSN": ["0006-3835"]}}
+    monkeypatch.setattr("scripts.enrich_publication_quartiles.requests.get", lambda *args, **kwargs: Response())
+    result = resolve_doi("10.1234/work", tmp_path)
+    assert result["status"] == "doi_identity_conflict" and "issns" not in result
+    assert not list(tmp_path.glob("*.json"))
+
+
+def test_enrichment_cli_dry_run_apply_and_repeat_preserve_original_data(monkeypatch, tmp_path):
+    url = "sqlite:///" + str(tmp_path / "publications.db")
+    engine = create_engine(url)
+    Base.metadata.create_all(engine)
+    raw = {"original": "keep", "csv_enrichment": {"Квартиль scopus": "Q3", "Год": "2024"}}
+    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="Work", year=2024, source_hash="hash", raw_data=raw))
+        session.commit()
+    monkeypatch.setenv("DATABASE_URL", url)
+    arguments = ["enrich", "--cache-dir", str(tmp_path / "cache"), "--report", str(tmp_path / "report.json")]
+    monkeypatch.setattr("sys.argv", arguments)
+    main()
+    with Session(engine) as session:
+        assert session.scalar(select(EmployeePublication)).quartiles is None
+    monkeypatch.setattr("sys.argv", arguments + ["--apply"])
+    main()
+    with Session(engine) as session:
+        publication = session.scalar(select(EmployeePublication))
+        payload = deepcopy(publication.quartiles)
+        assert publication.raw_data == raw and payload["entries"][0]["quartile"] == "Q3"
+    main()
+    with Session(engine) as session:
+        publication = session.scalar(select(EmployeePublication))
+        assert publication.quartiles == payload and publication.raw_data == raw
+
+
+def test_issn_succession_requires_explicit_relation_and_unique_doi_journal_title():
+    old = {"history": "Continued as", "related": ["Physical Review B"]}
+    new = {"issns": {"2469-9950", "2469-9969"}}
+    metadata = {"issns": ["1098-0121", "1550-235X"], "journals": ["Physical Review B"]}
+    result = quartiles.with_title_history(metadata, {"10980121": [old]}, {"physical review b": [new]}, "https://example.org/source-titles")
+    assert result["issns"] == metadata["issns"] and "2469-9950" in result["ranking_issns"]
+    assert metadata == {"issns": ["1098-0121", "1550-235X"], "journals": ["Physical Review B"]}
+    assert quartiles.with_title_history(metadata, {"10980121": [old]}, {"physical review b": [new, new]}, "url") == metadata
+    assert quartiles.with_title_history({**metadata, "journals": ["Other journal"]}, {"10980121": [old]}, {"physical review b": [new]}, "url") == {**metadata, "journals": ["Other journal"]}
+
+
+def test_crossref_not_found_falls_back_to_datacite_journal_metadata(monkeypatch, tmp_path):
+    class Response:
+        def __init__(self, status):
+            self.status_code = status
+        def raise_for_status(self):
+            pass
+        def json(self):
+            return {"data": {"attributes": {"doi": "10.1234/work", "publicationYear": 2024,
+                "types": {"resourceTypeGeneral": "JournalArticle"},
+                "container": {"title": "Journal", "identifierType": "ISSN", "identifier": "0006-3835"}}}}
+    calls = []
+    def get(url, **kwargs):
+        calls.append(url)
+        return Response(404 if "crossref.org" in url else 200)
+    monkeypatch.setattr("scripts.enrich_publication_quartiles.requests.get", get)
+    result = resolve_doi("10.1234/work", tmp_path)
+    assert result["status"] == "resolved" and result["type"] == "journal-article"
+    assert result["issns"] == ["0006-3835"] and result["years"] == [2024]
+    assert result["datacite_checked"] and len(calls) == 2