fix: validate publication evidence identity and original journal ISSNs

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## 0.18.1
+
+- Проверка К1/К2 учитывает ISSN/eISSN исходной записи вместе с идентификаторами DOI, в том числе без сетевого поиска. Противоречие названия и ISSN не превращается в предварительное совпадение.
+- Квартильные основания привязаны к собственному DOI и журналу. После изменения публикации парсером старые основания сохраняются, но исключаются из счётчиков при конфликте; обновление общего результата поиска не перепривязывает старые записи.
+
 ## 0.18.0
 
 - Добавлено поле `vak_categories` с результатом проверки журнала по предоставленным спискам К1/К2, источником и способом сопоставления. Совпадение по названию отмечается отдельно от подтверждённого 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.18.0`. Админка всегда показывает версии backend и frontend в footer.
+Версия сервиса: `0.18.1`. Админка всегда показывает версии backend и frontend в footer.
 
 ## Разовый импорт публикаций
 

app/services/council.py

@@ -16,9 +16,9 @@ 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, _description_journals
+from app.services.council_source import _doi, _journal_names, _name, 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.publication_quartiles import labels as quartile_labels, identity_issues, hse_entries, merge as merge_quartiles
 from app.services.council_assessment import (
     BRANCHES, assess_candidate, degree_level, evidence, publication_proofs,
     publication_type_allowed, sourced,
@@ -110,15 +110,24 @@ def _database_article(publication) -> dict:
     kind = publication.publication_type
     description = publication.description if isinstance(publication.description, dict) else raw.get("description")
     journals = _journal_names(raw)
-    if isinstance(imported, dict):
-        journals.extend(_journal_names(imported))
     if isinstance(description, dict):
         journals.extend(_description_journals(description))
+    doi = _doi(publication.doi_url or "")
+    identifiers = [raw[key] for key in ("issn", "eissn", "ISSN", "EISSN") if isinstance(raw.get(key), str)]
+    current_journals = list(journals)
+    imported_quartiles = merge_quartiles(None, hse_entries(raw, publication.year))
+    issues = identity_issues(publication.quartiles, doi, current_journals, identifiers)
+    imported_issues = identity_issues(imported_quartiles, doi, current_journals, identifiers)
+    imported_doi = _doi(str(imported.get("DOI") or "")) if isinstance(imported, dict) else ""
+    if isinstance(imported, dict) and not imported_issues and (not imported_doi or imported_doi == doi):
+        journals.extend(_journal_names(imported))
+    issues = list(dict.fromkeys([*issues, *imported_issues]))
     return {"title": publication.title, "year": publication.year, "doi": _doi(publication.doi_url or ""),
             "types": [TYPES.get(kind.upper(), f"Другой тип: {kind}")] if kind 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)],
+            "quartiles": list(dict.fromkeys([*quartile_labels(publication.quartiles, publication.year, doi, current_journals, identifiers),
+                          *quartile_labels(imported_quartiles, publication.year, doi, current_journals, identifiers)])),
+            "quartile_issues": issues,
+            "journals": journals, "issns": [raw[key] for key in ("issn", "eissn", "ISSN", "EISSN") if isinstance(raw.get(key), str)],
             "source_hashes": [publication.source_hash],
             "origins": origins, "rows": [], "in_csv": bool(imported), "in_db": from_site,
             "url": _safe_url(publication.url)}
@@ -151,9 +160,10 @@ def _unique_articles(articles: list[dict], start: date, end: date) -> list[dict]
         if key not in unique:
             unique[key] = {**article, "doi": identifier,
                            **{field: list(article[field]) for field in ("types", "quartiles", "journals", "origins", "rows")},
-                           "evidence": [], "issns": [], "source_hashes": []}
+                           "evidence": [], "issns": [], "source_hashes": [], "quartile_issues": []}
         merged = unique[key]
         merged["evidence"].append(article)
+        merged["quartile_issues"] = list(dict.fromkeys([*merged["quartile_issues"], *article.get("quartile_issues", [])]))
         for field in ("issns", "source_hashes"):
             merged[field] = list(dict.fromkeys([*merged[field], *article.get(field, [])]))
         for field in ("types", "quartiles", "journals", "origins", "rows"):
@@ -167,7 +177,7 @@ def _unique_articles(articles: list[dict], start: date, end: date) -> list[dict]
         # Never pick a conflicting source year merely because that row came first.
         year = article["year"] = years[0] if len(years) == 1 else None
         article["source_years"] = years
-        article["criteria"] = list(article["quartiles"])
+        article["criteria"] = [*article["quartiles"], *article["quartile_issues"]]
         matches = list(dict.fromkeys(k1_journals[key] for name in article["journals"]
                                     if (key := _journal_key(name)) in k1_journals))
         article["criteria"].extend(f"К1 · {name} (справочник ВАК)" for name in matches)

app/services/council_assessment.py

@@ -126,7 +126,7 @@ def publication_proofs(employee_evidence, article):
 
 
 def assess_article(article, proofs, start, end, level, branch, specialty, policy):
-    reasons = list(article.get("catalog_issues", []))
+    reasons = [*article.get("catalog_issues", []), *article.get("quartile_issues", [])]
     years = article["source_years"]
     dates = set()
     date_conflict = False

app/services/publication_quartiles.py

@@ -4,8 +4,8 @@ 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
+from app.services.council_catalog import publication_catalog, safe_url, issn
+from app.services.council_source import _name, _doi
 
 
 SYSTEM_FIELDS = {"Scopus": "Квартиль scopus", "Web of Science": "Квартиль wos"}
@@ -17,8 +17,16 @@ def entries(payload):
         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 (payload.get("lookup") is not None and not isinstance(payload["lookup"], dict))
+            or (isinstance(payload.get("lookup"), dict) and payload["lookup"].get("doi") is not None
+                and not isinstance(payload["lookup"]["doi"], str))
             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"])):
+                       and item.get("system") in {*SYSTEM_FIELDS, "SCImago"}
+                       and (item.get("doi") is None or isinstance(item["doi"], str))
+                       and (item.get("journal") is None or isinstance(item["journal"], str))
+                       and isinstance(item.get("issns", []), list)
+                       and all(isinstance(value, str) for value in item.get("issns", []))
+                       for item in payload["entries"])):
         raise ValueError("Invalid publication quartiles")
     return payload["entries"]
 
@@ -34,6 +42,7 @@ def hse_entries(raw, publication_year):
         quartile = str(metadata.get(field) or "").strip().upper()
         if quartile in QUARTILES:
             output.append({"system": system, "quartile": quartile, "metric": None,
+                           "doi": _doi(str(metadata.get("DOI") or "")) or None,
                            "metric_year": None, "publication_year": applicable_year,
                            "category": None, "journal": metadata.get("Название журнала") or None,
                            "issns": [], "source": "Данные ВШЭ", "source_url": None,
@@ -64,6 +73,7 @@ def catalog_entries(doi_metadata, publication_year, allow_latest=False):
         for metric in record["metrics"]:
             if metric["year"] == selected_years.get(metric["system"]) and metric["system"] in SYSTEM_FIELDS:
                 output.append({"system": metric["system"], "quartile": metric["quartile"],
+                               "doi": doi_metadata.get("doi"),
                                "metric": metric["metric"], "metric_year": metric["year"],
                                "publication_year": publication_year, "category": metric["category"],
                                "journal": record["name"], "issns": doi_metadata["issns"],
@@ -74,11 +84,13 @@ def catalog_entries(doi_metadata, publication_year, allow_latest=False):
 
 def merge(existing, additions, lookup=None):
     current = entries(existing)
-    output = list(current)
+    previous_doi = _doi(((existing or {}).get("lookup") or {}).get("doi") or "") or None
+    # Freeze the previous lookup's identity on legacy evidence before replacing it.
+    output = [{**item, "doi": previous_doi} if "doi" not in item else dict(item) for item in 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"))
+                     "publication_year", "category", "journal", "source", "source_url", "doi"))
     seen = {key(item) for item in output}
     for item in additions:
         if key(item) not in seen:
@@ -90,12 +102,40 @@ def merge(existing, additions, lookup=None):
     return result
 
 
-def labels(payload, publication_year):
+def binding_issue(item, payload, doi="", journals=(), identifiers=()):
+    value = item.get("doi") if "doi" in item else ((payload or {}).get("lookup") or {}).get("doi")
+    source_doi = _doi(value or "")
+    if value and not source_doi:
+        return "DOI сохранённого квартиля некорректен — требуется перепроверка"
+    current_doi = _doi(doi or "")
+    if source_doi and source_doi != current_doi:
+        return "Сохранённый квартиль относится к другому DOI — требуется перепроверка"
+    source_ids = {issn(value) for value in item.get("issns", [])} - {""}
+    current_ids = {issn(value) for value in identifiers} - {""}
+    ids_match = bool(source_ids & current_ids)
+    if source_ids and current_ids and not ids_match:
+        return "ISSN сохранённого квартиля и публикации расходятся — требуется перепроверка"
+    source_name = _name(item.get("journal") or "")
+    names_match = bool(source_name and source_name in {_name(name) for name in journals if name})
+    if source_name and journals and not names_match and not ids_match:
+        return "Журнал сохранённого квартиля и публикации расходятся — требуется перепроверка"
+    if not (source_doi and source_doi == current_doi) and not ids_match and not names_match:
+        return "Привязка сохранённого квартиля к публикации не подтверждена — требуется перепроверка"
+    return None
+
+
+def identity_issues(payload, doi="", journals=(), identifiers=()):
+    return list(dict.fromkeys(issue for item in entries(payload)
+                             if (issue := binding_issue(item, payload, doi, journals, identifiers))))
+
+
+def labels(payload, publication_year, doi="", journals=(), identifiers=()):
     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}))
+                             and item.get("metric_year") in {None, publication_year}
+                             and not binding_issue(item, payload, doi, journals, identifiers)))
 
 
 def index_sjr(rankings):
@@ -165,6 +205,7 @@ def sjr_entries(metadata, publication_year, rankings, allow_latest=False):
     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",
+               "doi": metadata.get("doi"),
                "metric_year": year, "publication_year": publication_year, "category": category,
                "journal": record["journal"], "issns": record["issns"],
                "source": "SCImago Journal & Country Rank · Universidad de Extremadura",

app/version.py

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

docs/publication-database.md

@@ -42,6 +42,7 @@
     {
       "system": "Scopus",
       "quartile": "Q2",
+      "doi": "10.1234/example",
       "metric": "CiteScore",
       "metric_year": 2024,
       "publication_year": 2024,
@@ -63,6 +64,8 @@
 ```
 
 - `entries` сохраняет все найденные категории и системы отдельно, без выбора одного «лучшего» квартиля вместо остальных. Возможные системы: Scopus, Web of Science, SCImago. Метрики CiteScore, JCR и SJR не подменяют друг друга. При наличии только лучшего квартиля SJR используется метрика `SJR best quartile`, категория остаётся `null`.
+- Каждое основание хранит собственный `doi`. При дополнении старого payload основания без этого ключа получают DOI из прежнего `lookup` до его обновления; при отсутствии прежнего DOI сохраняется `null`. Новый общий `lookup` не перепривязывает прежние записи. DOI данных ВШЭ берётся из их собственных исходных метаданных.
+- Перед использованием в счётчиках основание сверяется с текущими DOI, журналом и ISSN публикации. Конфликт исключает это основание и выводится как причина перепроверки; доказательства сохраняются. Совпадающий ISSN позволяет подтвердить переименование журнала, но не снимает конфликт DOI. Основание без проверяемой привязки не засчитывается по одному совпадению года. Это также относится к обратному чтению `raw_data.csv_enrichment`.
 - Значения `Квартиль 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` также сохраняется в найденных квартилях.
@@ -106,13 +109,13 @@ python scripts/enrich_publication_quartiles.py --apply --lookup-doi --all-system
 
 Для проверки смены ISSN добавить `--scopus-title-history /путь/ext_list_Aug_2026.xlsx`. Используется конкретный официальный снимок: SHA-256 `11e81f686401c89fbef28de31d1880388bb7122f35c89e09db6e1848237e9afb`; несоответствующий файл отклоняется. Это список названий и их истории, не источник квартилей.
 
-`--all-systems` проверяет также публикации с уже известным квартилем: добавляются остальные доступные системы и предметные категории. Без этого флага проверяются записи без квартиля и ранее найденные последние метрики, для которых может появиться исторический год. Новые доказательства дополняют список, не удаляя прежние. Несколько категорий с одинаковым Q1 сохраняются отдельно; один квартиль не заменяет остальные.
+`--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 в текущих данных выполняется по нормализованному названию и явным административным переименованиям. Транслитерация, похожесть названий и предположение о соответствии русской и переводной версии не используются.
+Проверяются `k1_journals.json` и `k2_journals.json`, извлечённые из предоставленного документа «ВАК категории.pdf». Для К1 доступна дополнительная идентификация по ISSN из `k1_specialties.json`; её источник, SHA-256 и дата списка сохраняются отдельно. В проверку передаются ISSN/eISSN исходного `raw_data`, данных ВШЭ и актуального ответа DOI; исходные идентификаторы используются и без сетевого поиска или DOI. Ответ, сохранённый для другого DOI, не считается актуальным источником идентификаторов. Поиск К2 в текущих данных выполняется по нормализованному названию и явным административным переименованиям. Транслитерация, похожесть названий и предположение о соответствии русской и переводной версии не используются.
 
 - `match_method: "issn"`, `identity_confirmed: true` — журнал идентифицирован по ISSN.
 - `match_method: "normalized_name"`, `identity_confirmed: false` — название совпало со списком; идентичность по ISSN не подтверждена. Это предварительное совпадение.

pyproject.toml

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

scripts/enrich_publication_quartiles.py

@@ -20,7 +20,7 @@ from openpyxl import load_workbook
 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 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_quartiles import catalog_entries, entries, hse_entries, merge, sjr_entries, index_sjr, with_title_history, identity_issues
 from app.services.publication_vak import assess_vak
 
 SOURCE_TITLES_URL = "https://downloads.ctfassets.net/o78em1y1w4i4/7xtaTxNiNcWRTeZkV86eNy/69cf2d506c905dc299531fdc93049dbb/ext_list_Aug_2026.xlsx"
@@ -221,18 +221,26 @@ def main():
     table = Table("employee_publications", MetaData(), autoload_with=engine)
     if "quartiles" not in table.c:
         raise ValueError("Apply migration 010 before enrichment")
-    fields = ["id", "year", "doi_url", "raw_data", "quartiles"]
+    fields = ["id", "year", "doi_url", "raw_data", "quartiles", "description"]
     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"])
+        fields.extend(["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 args.all_systems or not entries(row["quartiles"]) or row["quartiles"].get("lookup", {}).get("status") == "found_latest"
+        raw = row["raw_data"] if isinstance(row["raw_data"], dict) else {}
+        names = _journal_names(raw)
+        description = row["description"] if isinstance(row["description"], dict) else raw.get("description", {})
+        if isinstance(description, dict):
+            names.extend(_description_journals(description))
+        ids = [raw[key] for key in ("issn", "eissn", "ISSN", "EISSN") if isinstance(raw.get(key), str)]
+        payload = merge(row["quartiles"], hse_entries(row["raw_data"], row["year"]))
+        return (args.all_systems or not entries(payload) or (payload.get("lookup") or {}).get("status") == "found_latest"
+                or bool(identity_issues(payload, row["doi_url"], names, ids)))
     identifiers = sorted({_doi(row["doi_url"] or "") for row in rows
-                          if needs_lookup(row) and (args.all_systems or not hse_entries(row["raw_data"], row["year"]))} - {""})
+                          if needs_lookup(row)} - {""})
     resolved = {}
     if args.lookup_doi:
         with ThreadPoolExecutor(max_workers=args.workers) as pool:
@@ -257,12 +265,13 @@ def main():
     report = []
     for row in rows:
         doi = _doi(row["doi_url"] or "")
-        metadata = resolved.get(doi, row["quartiles"].get("lookup", {}) if row["quartiles"] else {})
+        previous_lookup = (row["quartiles"].get("lookup") or {}) if row["quartiles"] else {}
+        metadata = resolved.get(doi, previous_lookup if doi and _doi(previous_lookup.get("doi") or "") == doi 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 (args.all_systems or not entries(payload) or payload.get("lookup", {}).get("status") == "found_latest"):
+        if args.lookup_doi and needs_lookup(row):
             doi = _doi(row["doi_url"] or "")
             metadata = resolved.get(doi, {"status": "invalid_doi" if row["doi_url"] else "missing_doi"})
             status = metadata["status"]
@@ -290,7 +299,11 @@ def main():
                 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", []))
+            identifiers = list(metadata.get("issns", []))
+            for source in (raw, imported):
+                if isinstance(source, dict):
+                    identifiers.extend(source[key] for key in ("issn", "eissn", "ISSN", "EISSN") if isinstance(source.get(key), str))
+            vak_payload = assess_vak(list(dict.fromkeys(names)), list(dict.fromkeys(identifiers)))
             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"]

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.18.0"
+    assert response.json()["backend_version"] == "0.18.1"
 
 
 def test_api_employees_and_stats_require_admin_session():

tests/test_council.py

@@ -67,7 +67,8 @@ def test_council_union_counts_provenance_and_types(db_session, export):
                 publication_id="100", url="https://publications.hse.ru/view/100")
     publication(db_session, person, title=" shared   WORK ", kind="ARTICLE")
     publication(db_session, person, title="Book only", kind="BOOK", year=2020)
-    publication(db_session, person, title="Imported", raw_data={"csv_enrichment": {"Квартиль wos": "Q2"}})
+    publication(db_session, person, title="Imported", doi="https://doi.org/10.1234/imported",
+                raw_data={"csv_enrichment": {"Квартиль wos": "Q2", "DOI": "10.1234/imported"}})
     db_session.commit()
     result = council.council_candidates(db_session, date(2026, 9, 30))
     assert result["total"] == 1
@@ -218,7 +219,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.18.0 · Frontend 0.18.0" in response.text
+        assert "Backend 0.18.1 · Frontend 0.18.1" 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

@@ -29,11 +29,11 @@ def test_invalid_hse_value_does_not_invent_a_quartile(value):
 
 
 def test_merge_is_idempotent_and_preserves_conflicting_evidence():
-    first = quartiles.hse_entries({"csv_enrichment": {"Квартиль scopus": "Q1"}}, 2024)[0]
+    first = quartiles.hse_entries({"csv_enrichment": {"Квартиль scopus": "Q1", "DOI": "10.1234/work"}}, 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, 2024, doi="10.1234/work") == ["Scopus Q1"]
     assert quartiles.labels(payload, 2023) == []
 
 
@@ -99,9 +99,9 @@ def test_crawler_preserves_quartiles_while_replacing_source_owned_raw_data():
 
 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)
+    values = quartiles.hse_entries({"csv_enrichment": {"Квартиль scopus": "Q2", "DOI": "10.1234/work"}}, 2024)
     publication = EmployeePublication(title="Work", year=2024, publication_type="ARTICLE", source_hash="hash",
-        quartiles=quartiles.merge(None, values), raw_data={})
+        quartiles=quartiles.merge(None, values), raw_data={}, doi_url="https://doi.org/10.1234/work")
     assert _database_article(publication)["quartiles"] == ["Scopus Q2"]
     publication.quartiles = quartiles.merge(None, [{**values[0], "metric_year": 2025}])
     assert _database_article(publication)["quartiles"] == []
@@ -217,3 +217,83 @@ def test_crossref_not_found_falls_back_to_datacite_journal_metadata(monkeypatch,
     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
+
+
+@pytest.mark.parametrize("new_doi,new_journal", [("10.1234/new", "Different journal"), ("10.1234/old", "Different journal")])
+def test_crawler_doi_or_journal_change_excludes_preserved_quartile(new_doi, new_journal):
+    from app.services.council import _database_article
+    evidence = {"system": "Scopus", "quartile": "Q1", "publication_year": 2024, "metric_year": 2024,
+                "journal": "Old journal", "issns": [], "source": "Checked"}
+    payload = {"schema_version": 1, "entries": [evidence], "lookup": {"doi": "10.1234/old"}}
+    publication = EmployeePublication(title="Work", year=2024, source_hash="old", publication_type="ARTICLE",
+        doi_url="https://doi.org/10.1234/old", quartiles=deepcopy(payload), raw_data={"journal": "Old journal"})
+    assert _database_article(publication)["quartiles"] == ["Scopus Q1"]
+    _apply_publication(publication, {"id": "42", "title": "Work", "year": 2024, "type": "ARTICLE",
+        "doi_url": "https://doi.org/" + new_doi, "raw_data": {"journal": new_journal}}, "new")
+    article = _database_article(publication)
+    assert publication.quartiles == payload
+    assert article["quartiles"] == [] and article["quartile_issues"]
+    assert all("перепроверка" in issue for issue in article["quartile_issues"])
+
+
+def test_updating_lookup_freezes_legacy_binding_instead_of_rebinding_old_entries():
+    old = {"system": "Scopus", "quartile": "Q1", "publication_year": 2024, "metric_year": 2024,
+           "journal": "Old journal", "issns": [], "source": "Checked"}
+    payload = {"schema_version": 1, "entries": [old], "lookup": {"doi": "10.1234/old"}}
+    updated = quartiles.merge(payload, [], lookup={"doi": "10.1234/new"})
+    assert updated["entries"][0]["doi"] == "10.1234/old"
+    assert quartiles.labels(updated, 2024, "10.1234/new", ["Old journal"]) == []
+    fresh = {**old, "doi": "10.1234/new", "quartile": "Q2"}
+    refreshed = quartiles.merge(updated, [fresh], lookup={"doi": "10.1234/new"})
+    assert quartiles.labels(refreshed, 2024, "10.1234/new", ["Old journal"]) == ["Scopus Q2"]
+    assert len(refreshed["entries"]) == 2
+
+
+def test_unbound_hse_fallback_does_not_bypass_quartile_identity_checks():
+    from app.services.council import _database_article
+    publication = EmployeePublication(title="Work", year=2024, source_hash="hash", publication_type="ARTICLE",
+        doi_url="10.1234/new", raw_data={"journal": "New journal", "csv_enrichment": {
+            "DOI": "10.1234/old", "Название журнала": "Old journal", "Квартиль scopus": "Q1"}})
+    article = _database_article(publication)
+    assert article["quartiles"] == [] and article["quartile_issues"]
+    assert "Old journal" not in article["journals"]
+
+
+def test_journal_issn_allows_explicit_rename_but_conflicting_issn_is_rejected():
+    value = {"system": "Scopus", "quartile": "Q1", "publication_year": 2024, "metric_year": 2024,
+             "doi": "10.1234/work", "journal": "Old title", "issns": ["0006-3835"]}
+    payload = quartiles.merge(None, [value])
+    assert quartiles.labels(payload, 2024, "10.1234/work", ["Renamed title"], ["00063835"]) == ["Scopus Q1"]
+    assert quartiles.labels(payload, 2024, "10.1234/work", ["Old title"], ["11111111"]) == []
+
+
+def test_default_enrichment_rechecks_filled_evidence_after_doi_change(monkeypatch, tmp_path):
+    url = "sqlite:///" + str(tmp_path / "stale.db")
+    engine = create_engine(url)
+    Base.metadata.create_all(engine)
+    old = {"system": "Scopus", "quartile": "Q1", "publication_year": 2024, "metric_year": 2024,
+           "journal": "Old journal", "issns": [], "source": "Checked"}
+    payload = {"schema_version": 1, "entries": [old], "lookup": {"doi": "10.1234/old", "status": "found"}}
+    with Session(engine) as session:
+        person = Employee(profile_key="test", canonical_url="https://example.org/person")
+        session.add(person)
+        session.flush()
+        session.add(EmployeePublication(employee_id=person.id, title="Work", year=2024, source_hash="hash",
+            doi_url="10.1234/new", raw_data={"journal": "New journal"}, quartiles=payload))
+        session.commit()
+    calls = []
+    def resolve(doi, cache):
+        calls.append(doi)
+        return {"status": "resolved", "doi": doi, "type": "journal-article", "journals": ["New journal"]}
+    monkeypatch.setattr("scripts.enrich_publication_quartiles.resolve_doi", resolve)
+    monkeypatch.setattr("scripts.enrich_publication_quartiles.catalog_entries", lambda metadata, year, latest: ([
+        {**old, "doi": metadata["doi"], "journal": "New journal", "quartile": "Q2"}], "found"))
+    monkeypatch.setenv("DATABASE_URL", url)
+    monkeypatch.setattr("sys.argv", ["enrich", "--apply", "--lookup-doi", "--cache-dir", str(tmp_path / "cache"),
+                                    "--report", str(tmp_path / "report.json")])
+    main()
+    with Session(engine) as session:
+        stored = session.scalar(select(EmployeePublication)).quartiles
+        assert calls == ["10.1234/new"] and len(stored["entries"]) == 2
+        assert stored["entries"][0]["doi"] == "10.1234/old"
+        assert quartiles.labels(stored, 2024, "10.1234/new", ["New journal"]) == ["Scopus Q2"]

tests/test_publication_vak.py

@@ -78,3 +78,35 @@ def test_book_container_title_is_not_treated_as_a_vak_journal(monkeypatch, tmp_p
         publication = session.scalar(select(EmployeePublication))
         assert publication.vak_categories["status"] == "unknown_journal"
         assert publication.vak_categories["entries"] == [] and publication.quartiles["entries"] == []
+
+
+@pytest.mark.parametrize("raw,expected", [
+    ({"journal": "Первый журнал", "issn": "1111-1111"}, "journal_identity_conflict"),
+    ({"issn": "0006-3835"}, "found"),
+    ({"eissn": "0006-3835"}, "found"),
+    ({"csv_enrichment": {"ISSN": "0006-3835"}}, "found"),
+])
+def test_cli_checks_original_issn_even_without_doi(index, monkeypatch, tmp_path, raw, expected):
+    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 / "vak.db")
+    engine = create_engine(url)
+    Base.metadata.create_all(engine)
+    with Session(engine) as session:
+        person = Employee(profile_key="test", canonical_url="https://example.org/person")
+        session.add(person)
+        session.flush()
+        session.add(EmployeePublication(employee_id=person.id, title="Work", year=2024, source_hash="hash", raw_data=raw))
+        session.commit()
+    monkeypatch.setenv("DATABASE_URL", url)
+    monkeypatch.setattr("sys.argv", ["enrich", "--apply", "--check-vak", "--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"] == expected
+        if expected == "found":
+            assert publication.vak_categories["entries"][0]["identity_confirmed"]