diff --git a/CHANGELOG.md b/CHANGELOG.md
index 9f4d20a..26eab12 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,21 @@
# Changelog
+## 0.18.1
+
+- Проверка К1/К2 учитывает ISSN/eISSN исходной записи вместе с идентификаторами DOI, в том числе без сетевого поиска. Противоречие названия и ISSN не превращается в предварительное совпадение.
+- Квартильные основания привязаны к собственному DOI и журналу. После изменения публикации парсером старые основания сохраняются, но исключаются из счётчиков при конфликте; обновление общего результата поиска не перепривязывает старые записи.
+
+## 0.18.0
+
+- Добавлено поле `vak_categories` с результатом проверки журнала по предоставленным спискам К1/К2, источником и способом сопоставления. Совпадение по названию отмечается отдельно от подтверждённого ISSN; текущий снимок не объявляется исторической категорией.
+- Инструмент обогащения сохраняет все доступные квартильные категории всех систем, включая дополнительные сведения для публикаций с уже известным квартилем.
+
+## 0.17.0
+
+- Добавлено отдельное поле публикации `quartiles` с квартилями Q1–Q4, системой, годом метрики и источником. Данные ВШЭ переносятся без удаления исходных значений; новые сведения сопоставляются по DOI и ISSN.
+- Добавлена возобновляемая проверка DOI и обогащение по квартильным каталогам: при отсутствии исторической метрики можно явно разрешить последний доступный год. Недоступные сведения и конфликты фиксируются отдельно; текущий квартиль не подменяет исторический зачёт в совете.
+- Состав полей публикации, миграция и порядок обогащения описаны в документации.
+
## 0.16.5
- Закреплён обязательный перенос недостающих данных перед будущим production-выпуском: резервная копия, предварительная сверка, сохранение заполненных значений, отдельная обработка конфликтов и проверка полноты до обновления приложения.
diff --git a/README.md b/README.md
index a9624cc..c40a8a7 100644
--- a/README.md
+++ b/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.18.1`. Админка всегда показывает версии 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` и `011_publication_vak_categories.sql`; данные переносятся отдельным инструментом, а не при старте приложения.
+
+Квартили Q1–Q4 из данных ВШЭ сохраняются с исходной системой; неизвестный год метрики остаётся неизвестным. Для остальных публикаций журнал устанавливается по DOI, рейтинг — по ISSN. Исторические SJR доступны через открытые данные университета Эстремадуры за 2009–2025; CiteScore/JCR читаются из проверенных каталогов. При отсутствии метрики за год публикации допускается явно разрешённый последний доступный год, с сохранением его года и источника. Отсутствие DOI, неоднозначность и недоступная метрика фиксируются без присвоения вымышленного квартиля. Парсер сохраняет новое поле при обновлении записи.
+
### Подготовка публикаций для портала
Портал читает `app/data/council_articles.json`: публикации представлены отдельными
diff --git a/app/db.py b/app/db.py
index 103e709..4b99494 100644
--- a/app/db.py
+++ b/app/db.py
@@ -37,6 +37,14 @@ 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")}
+ for column in ("quartiles", "vak_categories"):
+ if column not in publication_columns:
+ with engine.begin() as connection:
+ connection.execute(text("ALTER TABLE employee_publications ADD COLUMN " +
+ ("IF NOT EXISTS " if engine.dialect.name == "postgresql" else "") +
+ column + " " + ("JSONB" if engine.dialect.name == "postgresql" else "JSON")))
if "employees" in table_names and "employee_news_links" not in table_names:
models.EmployeeNewsLink.__table__.create(bind=engine, checkfirst=True)
inspector = inspect(engine)
diff --git a/app/models.py b/app/models.py
index df4a67a..0ac7b1c 100644
--- a/app/models.py
+++ b/app/models.py
@@ -108,6 +108,8 @@ 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)
+ vak_categories: Mapped[dict | None] = mapped_column(json_type)
annotation: Mapped[dict | None] = mapped_column(json_type)
description: Mapped[dict | None] = mapped_column(json_type)
authors: Mapped[list | None] = mapped_column(json_type)
diff --git a/app/services/council.py b/app/services/council.py
index 19445a4..ad1f3cf 100644
--- a/app/services/council.py
+++ b/app/services/council.py
@@ -11,14 +11,14 @@ from pathlib import Path
from urllib.parse import quote, urlsplit
from zoneinfo import ZoneInfo
-from bs4 import BeautifulSoup
from sqlalchemy import select
from sqlalchemy.orm import Session, selectinload
from app.models import Employee
from app.services.academic_degrees import academic_degrees
-from app.services.council_source import _doi, _journal_names, _name, _quartiles, load_reference
+from app.services.council_source import _doi, _journal_names, _name, load_reference, _description_journals
from app.services.council_catalog import catalog, publication_catalog
+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,
@@ -96,19 +96,6 @@ def _k2_journals() -> dict:
return result
-def _description_journals(description: dict) -> list[str]:
- names = []
- for field in ("short", "shortLeft"):
- value = description.get(field)
- values = value.values() if isinstance(value, dict) else [value]
- for text in values:
- if isinstance(text, str):
- text = BeautifulSoup(text, "html.parser").get_text(" ", strip=True)
- # HSE short descriptions start with the journal, followed by the year.
- names.append(re.split(r"\b(?:19|20)\d{2}\b", text, maxsplit=1)[0].strip(" .,:;"))
- return names
-
-
@lru_cache(maxsize=1)
def _articles_by_author() -> dict[str, list[dict]]:
return load_reference(EXPORT)
@@ -123,14 +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": _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)}
@@ -163,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"):
@@ -179,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)
diff --git a/app/services/council_assessment.py b/app/services/council_assessment.py
index 0460fb9..b6acd05 100644
--- a/app/services/council_assessment.py
+++ b/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
diff --git a/app/services/council_source.py b/app/services/council_source.py
index 8b09497..a2b8aef 100644
--- a/app/services/council_source.py
+++ b/app/services/council_source.py
@@ -38,6 +38,21 @@ def _journal_names(data: dict) -> list[str]:
return names
+def _description_journals(description: dict) -> list[str]:
+ from bs4 import BeautifulSoup
+
+ names = []
+ for field in ("short", "shortLeft"):
+ value = description.get(field)
+ values = value.values() if isinstance(value, dict) else [value]
+ for text in values:
+ if isinstance(text, str):
+ text = BeautifulSoup(text, "html.parser").get_text(" ", strip=True)
+ # HSE short descriptions start with the journal, followed by the year.
+ names.append(re.split(r"\b(?:19|20)\d{2}\b", text, maxsplit=1)[0].strip(" .,:;"))
+ return names
+
+
def prepare_reference(source: Path) -> dict:
publications = []
by_author = defaultdict(list)
diff --git a/app/services/publication_quartiles.py b/app/services/publication_quartiles.py
new file mode 100644
index 0000000..c547d0e
--- /dev/null
+++ b/app/services/publication_quartiles.py
@@ -0,0 +1,217 @@
+"""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, issn
+from app.services.council_source import _name, _doi
+
+
+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 (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"}
+ 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"]
+
+
+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,
+ "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,
+ "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_years = {}
+ for system in SYSTEM_FIELDS:
+ years = {metric["year"] for _, metric in metrics if metric["system"] == system}
+ selected_years[system] = publication_year if publication_year in years else max(years, default=None) if allow_latest else None
+ output = []
+ for record in records:
+ for metric in record["metrics"]:
+ if metric["year"] == selected_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"],
+ "source": metric["source"], "source_url": safe_url(metric["source_url"]),
+ "checked_on": date.today().isoformat()})
+ return output, ("found_latest" if any(item["metric_year"] != publication_year for item in output) else "found") if output else "historical_quartile_unavailable"
+
+
+def merge(existing, additions, lookup=None):
+ current = entries(existing)
+ 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", "doi"))
+ 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 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 not binding_issue(item, payload, doi, journals, identifiers)))
+
+
+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",
+ "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",
+ "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"
diff --git a/app/services/publication_vak.py b/app/services/publication_vak.py
new file mode 100644
index 0000000..9a7328f
--- /dev/null
+++ b/app/services/publication_vak.py
@@ -0,0 +1,71 @@
+"""Match a publication's journal to the supplied K1/K2 category snapshots."""
+
+from collections import defaultdict
+from datetime import date
+from functools import lru_cache
+import hashlib
+import json
+from pathlib import Path
+import re
+
+from app.services.council_catalog import issn
+from app.services.council_source import _name
+
+DATA = Path(__file__).resolve().parents[1] / "data"
+
+
+def journal_key(value):
+ return " ".join(re.findall(r"\w+", _name(value)))
+
+
+@lru_cache(maxsize=1)
+def category_index():
+ files = {category: DATA / filename for category, filename in
+ (("К1", "k1_journals.json"), ("К2", "k2_journals.json"))}
+ documents = {category: json.loads(path.read_text(encoding="utf8")) for category, path in files.items()}
+ if documents["К1"]["source_sha256"] != documents["К2"]["source_sha256"]:
+ raise ValueError("VAK category source mismatch")
+ identities = json.loads((DATA / "k1_specialties.json").read_text(encoding="utf8"))
+ if identities["k1_source_sha256"] != hashlib.sha256(files["К1"].read_bytes().replace(b"\r\n", b"\n")).hexdigest():
+ raise ValueError("VAK journal identity source mismatch")
+ known = {item["name"]: sorted({issn(entry["issn"]) for entry in item["vak_entries"]})
+ for item in identities["journals"] if item.get("status") == "matched"}
+ by_name, by_issn = defaultdict(list), defaultdict(list)
+ for category, document in documents.items():
+ for journal in document["journals"]:
+ record = {"category": category, "journal": journal["name"], "source": document["source"],
+ "source_sha256": document["source_sha256"], "source_row": journal["source_row"],
+ "categorized_on": journal["categorized_on"], "issns": known.get(journal["name"], []) if category == "К1" else []}
+ if record["issns"]:
+ record.update(identity_source=identities["source"], identity_source_sha256=identities["source_sha256"],
+ identity_as_of=identities["as_of"])
+ for name in [journal["name"], *journal.get("aliases", [])]:
+ by_name[journal_key(name)].append(record)
+ for identifier in record["issns"]:
+ by_issn[identifier].append(record)
+ return dict(by_name), dict(by_issn)
+
+
+def assess_vak(names, identifiers):
+ by_name, by_issn = category_index()
+ matches = {}
+ normalized = {issn(value) for value in identifiers} - {""}
+ for identifier in normalized:
+ for record in by_issn.get(identifier, []):
+ matches[(record["category"], record["source_row"])] = {**record, "match_method": "issn", "identity_confirmed": True}
+ for name in names:
+ for record in by_name.get(journal_key(name), []):
+ if record["issns"] and normalized and not normalized.intersection(record["issns"]):
+ return {"schema_version": 1, "entries": [], "status": "journal_identity_conflict",
+ "scope": "supplied_category_snapshot", "historical_category_confirmed": False, "checked_on": date.today().isoformat()}
+ key = record["category"], record["source_row"]
+ matches.setdefault(key, {**record, "match_method": "normalized_name", "identity_confirmed": False})
+ if len(matches) > 1:
+ status, output = "ambiguous_journal", []
+ elif matches:
+ status, output = "found", list(matches.values())
+ else:
+ status, output = ("not_found_in_k1_k2" if names or identifiers else "unknown_journal"), []
+ return {"schema_version": 1, "entries": output, "status": status,
+ "scope": "supplied_category_snapshot", "historical_category_confirmed": False,
+ "checked_on": date.today().isoformat()}
diff --git a/app/version.py b/app/version.py
index 9265bb2..3ccdace 100644
--- a/app/version.py
+++ b/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.18.1"
+FRONTEND_VERSION = "0.18.1"
+BACKEND_VERSION = "0.18.1"
diff --git a/docs/publication-database.md b/docs/publication-database.md
new file mode 100644
index 0000000..3950c01
--- /dev/null
+++ b/docs/publication-database.md
@@ -0,0 +1,133 @@
+# Запись публикации в БД
+
+Объект `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 с системой, годом метрики, источником и результатом поиска |
+| `vak_categories` | jsonb | Нет | Результат проверки журнала по предоставленным спискам К1/К2, способ идентификации и источник |
+| `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",
+ "doi": "10.1234/example",
+ "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`.
+- Каждое основание хранит собственный `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` также сохраняется в найденных квартилях.
+- Для каждой системы сначала используется метрика за год публикации. При явно разрешённом `--allow-latest` и отсутствии исторической метрики допускается последний доступный год: он остаётся в `metric_year`, а результат имеет статус `found_latest`. Все категории CiteScore/JCR и SJR сохраняются вместе; разные системы могут иметь разные годы метрики. Это последний год в доступных проверенных источниках, а не обещание глобально самого нового рейтинга.
+- Источники: проверяемые каталоги проекта (CiteScore/JCR) и исторические SJR-данные [Universidad de Extremadura](https://opendata.unex.es/dataset/revistas-cientificas-2025) за 2009–2025. Портал указывает SCImago как автора и поддерживающую организацию и публикует данные по CC BY 3.0; атрибуция и ссылка сохраняются в каждой записи. [SJR](https://www.scimagolab.com/products/sjr-scimago-journal-country-rank/) хранится с системой SCImago, даже несмотря на происхождение исходных данных из Scopus.
+
+`lookup` также может содержать полученные по DOI тип материала, названия журнала, ISSN, годы и ссылку на регистратора DOI. Состояния поиска:
+
+| Статус | Значение |
+|---|---|
+| `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` и `migrations/011_publication_vak_categories.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 --all-systems --check-vak --sjr-history --allow-latest --workers 4 --cache-dir .impeccable/quartiles/cache --report .impeccable/quartiles/doi-applied.json
+```
+
+Для проверки смены ISSN добавить `--scopus-title-history /путь/ext_list_Aug_2026.xlsx`. Используется конкретный официальный снимок: SHA-256 `11e81f686401c89fbef28de31d1880388bb7122f35c89e09db6e1848237e9afb`; несоответствующий файл отклоняется. Это список названий и их истории, не источник квартилей.
+
+`--all-systems` проверяет также публикации с уже известным квартилем: добавляются остальные доступные системы и предметные категории. Без этого флага проверяются записи без квартиля, ранее найденные последние метрики и заполненные основания с изменившейся или неподтверждённой привязкой. Новые доказательства дополняют список, не удаляя прежние. Несколько категорий с одинаковым Q1 сохраняются отдельно; один квартиль не заменяет остальные.
+
+## Поле категорий ВАК
+
+`vak_categories` имеет `schema_version: 1`, массив `entries`, `status`, `scope: "supplied_category_snapshot"`, `historical_category_confirmed: false` и дату `checked_on`. `NULL` — проверка ещё не проводилась. В каждом совпадении сохраняются `category` (К1/К2), каноническое название журнала, название и SHA-256 источника, строка источника, дата категоризации, ISSN при наличии, `match_method` и `identity_confirmed`.
+
+Проверяются `k1_journals.json` и `k2_journals.json`, извлечённые из предоставленного документа «ВАК категории.pdf». Для К1 доступна дополнительная идентификация по ISSN из `k1_specialties.json`; её источник, SHA-256 и дата списка сохраняются отдельно. В проверку передаются ISSN/eISSN исходного `raw_data`, данных ВШЭ и актуального ответа DOI; исходные идентификаторы используются и без сетевого поиска или DOI. Ответ, сохранённый для другого DOI, не считается актуальным источником идентификаторов. Поиск К2 в текущих данных выполняется по нормализованному названию и явным административным переименованиям. Транслитерация, похожесть названий и предположение о соответствии русской и переводной версии не используются.
+
+- `match_method: "issn"`, `identity_confirmed: true` — журнал идентифицирован по ISSN.
+- `match_method: "normalized_name"`, `identity_confirmed: false` — название совпало со списком; идентичность по ISSN не подтверждена. Это предварительное совпадение.
+- `status: "found"` — одно совпадение; `ambiguous_journal` или `journal_identity_conflict` — неоднозначность/конфликт, категория автоматически не выбирается.
+- `not_found_in_k1_k2` — совпадений в этих двух снимках нет; это не доказывает отсутствие журнала в ВАК или наличие К3. `unknown_journal` — данных для идентификации журнала нет.
+
+Дата категоризации в документе не объявляется датой всего снимка, а категория по имеющемуся списку не подтверждает её наличие в год публикации. Книги и главы не становятся журнальными статьями из-за совпадения названия. Проверка К1/К2 не является подтверждением зачёта конкретной работы по критериям совета.
+
+`--check-vak` сохраняет этот результат в отдельном поле. Уже заполненные отличающиеся значения не перезаписываются; число таких случаев отражается в `vak_existing_value_kept`. Их нужно отдельно сравнить с актуальными источниками. Проверить результаты и причины неоднозначности до использования данных в заключении.
+
+Запросы DOI кэшируются по нормализованному DOI и переиспользуются для соавторов. Ошибки сети допускают повторную попытку; для полного обновления ранее найденных ответов использовать новый каталог кэша. Исторические рейтинги запрашиваются только для ISSN найденных журналов; ошибки источника отражаются в `ranking_failures` отчёта. Существующие квартильные записи не перезаписываются; повторный перенос не дублирует доказательства только из-за новой даты проверки. Перед записью инструмент проверяет, что публикация не изменилась параллельно; при изменении пропускает её и фиксирует `concurrent_change_skipped`. Коммит выполняется по публикациям: прерванную обработку можно возобновить. Проверить итоговый отчёт и полноту переноса после завершения.
+
+Парсер сохраняет `quartiles`, обновляя остальные исходные поля публикации. Страница совета читает отдельное поле с обратной совместимостью для исходных данных ВШЭ. В предварительные Q1/Q2 совета попадают Scopus/Web of Science с применимым годом; последние метрики другого года и SJR автоматически не превращаются в исторический зачёт или документальное подтверждение критериев.
+
+Миграция и обогащение локальной БД не обновляют production. Порядок переноса только недостающих данных в production и сохранения его заполненных значений описан в README.
diff --git a/migrations/010_publication_quartiles.sql b/migrations/010_publication_quartiles.sql
new file mode 100644
index 0000000..34281c6
--- /dev/null
+++ b/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;
diff --git a/migrations/011_publication_vak_categories.sql b/migrations/011_publication_vak_categories.sql
new file mode 100644
index 0000000..0c370c8
--- /dev/null
+++ b/migrations/011_publication_vak_categories.sql
@@ -0,0 +1,2 @@
+-- Matches to the supplied category snapshot, with provenance and identity state.
+ALTER TABLE employee_publications ADD COLUMN IF NOT EXISTS vak_categories JSONB;
diff --git a/pyproject.toml b/pyproject.toml
index 67586a7..658634c 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
[project]
name = "miem-workers"
-version = "0.16.5"
+version = "0.18.1"
description = "MIEM employees parser, admin API, and web admin"
requires-python = ">=3.11"
dependencies = [
diff --git a/scripts/enrich_publication_quartiles.py b/scripts/enrich_publication_quartiles.py
new file mode 100644
index 0000000..8765c98
--- /dev/null
+++ b/scripts/enrich_publication_quartiles.py
@@ -0,0 +1,340 @@
+"""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, _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, 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"
+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:
+PREFIX dct:
+PREFIX ou:
+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.18 publication-quartiles", "Accept": "application/json"})
+ if response.status_code == 404:
+ return {**result, "status": "doi_not_found"}
+ if response.status_code == 429 or response.status_code >= 500:
+ 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.18 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("--all-systems", action="store_true", help="Collect additional systems/categories even for publications with known quartiles")
+ parser.add_argument("--check-vak", action="store_true", help="Check the supplied K1/K2 snapshots")
+ parser.add_argument("--allow-latest", action="store_true", help="Use latest available metric if the publication year is unavailable")
+ parser.add_argument("--sjr-history", action="store_true", help="Check UNEx/SCImago historical quartiles by DOI ISSNs")
+ parser.add_argument("--scopus-title-history", type=Path, help="Official Scopus August 2026 source-title XLSX for explicit ISSN succession")
+ 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")
+ 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(["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):
+ 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)} - {""})
+ 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:
+ doi = _doi(row["doi_url"] or "")
+ 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 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"]
+ if status == "resolved":
+ found, status = catalog_entries(metadata, row["year"], args.allow_latest)
+ if rankings:
+ sjr, sjr_status = sjr_entries(metadata, row["year"], rankings, args.allow_latest)
+ found.extend(sjr)
+ if not found and status == "historical_quartile_unavailable":
+ status = sjr_status
+ if found:
+ status = "found_latest" if any(item["metric_year"] != row["year"] for item in found) else "found"
+ payload = merge(payload, found)
+ payload["lookup"] = {**metadata, "status": status, "requested_year": row["year"],
+ "checked_on": date.today().isoformat()}
+ counts[status] += 1
+ counts["with_quartile" if entries(payload) else "without_quartile"] += 1
+ vak_payload = row.get("vak_categories")
+ if args.check_vak:
+ raw = row["raw_data"] if isinstance(row["raw_data"], dict) else {}
+ imported = raw.get("csv_enrichment", {})
+ names = [*_journal_names(raw), *(_journal_names(imported) if isinstance(imported, dict) else []),
+ *(metadata.get("journals", []) if metadata.get("type") in {"journal-article", "journal"} else [])]
+ if row["publication_type"] == "ARTICLE" and metadata.get("type") in {None, "journal-article", "journal"}:
+ description = row["description"] if isinstance(row["description"], dict) else raw.get("description", {})
+ if isinstance(description, dict):
+ names.extend(_description_journals(description))
+ 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"]
+ counts["vak_" + vak_payload["status"]] += 1
+ if payload == row["quartiles"] and vak_payload == row.get("vak_categories"):
+ continue
+ if args.apply:
+ # Only this field is changed; parser-owned data and existing evidence survive.
+ with engine.begin() as connection:
+ current = dict(connection.execute(select(*columns).where(table.c.id == row["id"])
+ .with_for_update()).mappings().one())
+ if any(current[field] != row[field] for field in fields if field != "id"):
+ counts["concurrent_change_skipped"] += 1
+ continue
+ values = {"quartiles": payload, "updated_at": datetime.now(timezone.utc)}
+ if args.check_vak:
+ values["vak_categories"] = vak_payload
+ connection.execute(update(table).where(table.c.id == row["id"]).values(**values))
+ counts["changed"] += 1
+ report.append({"id": row["id"], "quartiles": payload, **({"vak_categories": vak_payload} if args.check_vak else {})})
+ summary = {"mode": "apply" if args.apply else "dry-run", "publications": len(rows),
+ "unique_doi_checked": len(resolved), "counts": dict(counts), "ranking_failures": ranking_failures}
+ pending.write_text(json.dumps({"summary": summary, "records": report}, ensure_ascii=False, indent=2), encoding="utf8")
+ 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)
diff --git a/tests/test_api.py b/tests/test_api.py
index c0a2e8e..538eb7a 100644
--- a/tests/test_api.py
+++ b/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.18.1"
def test_api_employees_and_stats_require_admin_session():
diff --git a/tests/test_council.py b/tests/test_council.py
index f34c664..7fd9575 100644
--- a/tests/test_council.py
+++ b/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 "<script>" in response.text and "