fix: prepare indexed council data and skip ineligible publication parsing

README.md

@@ -92,7 +92,7 @@ Compose запускает `api`, `worker` и `postgres`. API привязан
 
 Счётчики **Q1**, **Q2**, **К1** и **ВСЕГО** учитывают уникальные публикации за последние пять лет по году. Q1/Q2 берутся из отметок Scopus или Web of Science в CSV и сохранённом CSV-обогащении БД. К1 определяется по совпадению нормализованного названия журнала с итоговой категорией в предоставленном справочнике ВАК. ВСЕГО — объединение Q1 ИЛИ Q2 ИЛИ К1: работа в нескольких категориях учитывается один раз. При расхождении годов источников каждый критерий проверяется по году своей записи; неизвестный год наследуется только при единственном согласованном годе. Счётчики «В БД» и «Все публикации» включают все годы независимо от категорий.
 
-CSV `app/data/council_articles.csv` сопоставляется по полному ФИО только при единственном совпадении. Публикации объединяются по DOI, затем по нормализованным названию и году, если нет конфликтующих DOI. Импорт CSV в БД не выдаётся за данные сайта. Для К1 поддерживаются необязательные колонки CSV «Название журнала» или «Журнал»; в исходном CSV их нет. В БД название берётся из поля `journal` в `raw_data` или короткого описания ВШЭ (`description.short`/`shortLeft`, название перед годом). Название самой публикации для определения журнала не используется. Неизвестный или несовпавший журнал не считается К1; перевод и нечёткое сопоставление не применяются. Явно указанное в PDF прежнее название журнала сохраняется как проверенный алиас; служебное примечание о переименовании хранится отдельно от названия.
+Подготовленный JSON `app/data/council_articles.json` сопоставляется по полному ФИО только при единственном совпадении. Публикации объединяются по DOI, затем по нормализованным названию и году, если нет конфликтующих DOI. Импорт CSV в БД не выдаётся за данные сайта. Для К1 поддерживаются необязательные колонки CSV «Название журнала» или «Журнал»; в исходном CSV их нет. В БД название берётся из поля `journal` в `raw_data` или короткого описания ВШЭ (`description.short`/`shortLeft`, название перед годом). Название самой публикации для определения журнала не используется. Неизвестный или несовпавший журнал не считается К1; перевод и нечёткое сопоставление не применяются. Явно указанное в PDF прежнее название журнала сохраняется как проверенный алиас; служебное примечание о переименовании хранится отдельно от названия.
 
 Справочник `app/data/k1_journals.json` содержит 773 журнала К1, номера строк, даты категорирования и SHA256 исходного «ВАК категории.pdf». При извлечении проверены все 3134 строки; для записей с историей смены категории взята последняя категория. PDF при запросах не читается. Для обновления справочника разово установите `pdfplumber` в среду разработки и выполните:
 
@@ -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.15.1`. Админка всегда показывает версии backend и frontend в footer.
+Версия сервиса: `0.15.2`. Админка всегда показывает версии backend и frontend в footer.
 
 ## Разовый импорт публикаций
 
@@ -197,3 +197,28 @@ python scripts/extract_k1_specialties.py "app/data/ВАК Перечень.pdf"
 сохранены в `scripts/k1_specialty_matches.json`; каждое сопоставление проверяется по
 номеру строки и точному исходному названию. Нечёткий поиск не применяется.
 Сервис читает готовый JSON и не требует PDF или `pdfplumber` при запуске.
+
+### Подготовка публикаций для портала
+
+Портал читает `app/data/council_articles.json`: публикации представлены отдельными
+объектами с названием, годом, DOI, массивами авторов, квартилей и журналов.
+Дополнительные поля исходного экспорта сохранены в `metadata`; `source_row` связывает
+запись со строкой исходного CSV. `by_author` — готовый индекс нормализованных ФИО
+со ссылками на публикации, без копирования одной работы для каждого соавтора.
+
+Исходный `app/data/council_articles.csv` сохранён для проверки и отдельного импорта
+в БД. Данные файла и индекс кэшируются на процесс, CSV не преобразуется при
+HTTP-запросах. Название источника «Файл CSV» и номера строк в интерфейсе и XLSX
+сохранены как происхождение данных. Перед разбором публикаций сотрудников из БД
+проверяется докторская степень; публикации остальных сотрудников не разбираются.
+
+После обновления CSV пересоздайте JSON до сборки/развёртывания сервиса:
+
+```sh
+python scripts/prepare_council_articles.py app/data/council_articles.csv
+```
+
+Нужна только стандартная библиотека Python. Скрипт проверяет структуру CSV и индекс,
+сохраняет число записей и SHA-256 источника и заменяет JSON атомарно после проверки.
+После замены файла перезапустите процессы API, чтобы сбросить их кэш. Обновления
+публикаций из БД продолжают отображаться без перезапуска.

app/services/council.py

@@ -1,10 +1,8 @@
 """Public publication evidence from the council export and employee database."""
 
 import calendar
-import csv
 import json
 import re
-import unicodedata
 from collections import Counter, defaultdict
 from datetime import date, datetime
 from functools import lru_cache
@@ -18,27 +16,18 @@ 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
 
 
-EXPORT = Path(__file__).resolve().parents[1] / "data" / "council_articles.csv"
+EXPORT = Path(__file__).resolve().parents[1] / "data" / "council_articles.json"
 K1_JOURNALS = EXPORT.with_name("k1_journals.json")
 K1_SPECIALTIES = EXPORT.with_name("k1_specialties.json")
 DOCTOR = re.compile(r"^доктор\b.+\bнаук$", re.IGNORECASE)
-DOI = re.compile(r"10\.\d{4,9}/\S+", re.IGNORECASE)
 TYPES = {"ARTICLE": "Статья", "BOOK": "Книга", "CHAPTER": "Глава книги",
          "BOOK_CHAPTER": "Глава книги", "REPORT": "Доклад", "PATENT": "Патент",
          "PREPRINT": "Препринт", "CONFERENCE_PAPER": "Материал конференции"}
 
 
-def _name(value: str) -> str:
-    return " ".join(unicodedata.normalize("NFKC", value).casefold().replace("ё", "е").split())
-
-
-def _doi(value: str) -> str:
-    value = re.sub(r"^(?:https?://(?:dx\.)?doi\.org/|doi:\s*)", "", value.strip(), flags=re.I).lower()
-    return value if DOI.fullmatch(value) else ""
-
-
 def _safe_url(value: str | None) -> str | None:
     try:
         url = urlsplit(value or "")
@@ -54,12 +43,6 @@ def period(today: date) -> tuple[date, date]:
     return date(year, today.month, min(today.day, calendar.monthrange(year, today.month)[1])), today
 
 
-def _quartiles(row: dict) -> list[str]:
-    return [f"{label} {str(row.get(field) or '').strip().upper()}"
-            for label, field in (("Scopus", "Квартиль scopus"), ("Web of Science", "Квартиль wos"))
-            if str(row.get(field) or '').strip().upper() in {"Q1", "Q2"}]
-
-
 def _journal_key(value: str) -> str:
     return " ".join(re.findall(r"\w+", _name(value)))
 
@@ -83,17 +66,6 @@ def _k1_specialties() -> dict:
     return {**data, "by_journal": {j["name"]: j["specialty_codes"] for j in data["journals"]}}
 
 
-def _journal_names(data: dict) -> list[str]:
-    names = []
-    for field in ("Название журнала", "Журнал", "journal"):
-        value = data.get(field)
-        if isinstance(value, dict):
-            value = value.get("name")
-        if isinstance(value, str) and value.strip():
-            names.append(value.strip())
-    return names
-
-
 def _description_journals(description: dict) -> list[str]:
     names = []
     for field in ("short", "shortLeft"):
@@ -109,26 +81,7 @@ def _description_journals(description: dict) -> list[str]:
 
 @lru_cache(maxsize=1)
 def _articles_by_author() -> dict[str, list[dict]]:
-    articles = defaultdict(list)
-    with EXPORT.open(encoding="utf-8-sig", newline="") as source:
-        reader = csv.DictReader(source, delimiter=";")
-        required = {"Название публикации", "Год", "DOI", "Авторы НИУ ВШЭ", "Квартиль scopus", "Квартиль wos"}
-        if not required <= set(reader.fieldnames or []):
-            raise ValueError("Missing council CSV headers")
-        for row_number, row in enumerate(reader, 2):
-            if None in row or any(value is None for value in row.values()):
-                raise ValueError("Malformed council CSV record")
-            title = row["Название публикации"].strip()
-            if not title:
-                continue
-            year = int(row["Год"]) if row["Год"].strip().isdigit() else None
-            article = {"title": title, "year": year, "doi": _doi(row["DOI"]),
-                       "types": [], "quartiles": _quartiles(row), "journals": _journal_names(row), "origins": ["Файл CSV"],
-                       "rows": [row_number], "in_csv": True, "in_db": False, "url": None}
-            for author in set(_name(value) for value in row["Авторы НИУ ВШЭ"].split(";")) - {""}:
-                articles[author].append(article)
-    return articles
-
+    return load_reference(EXPORT)
 
 def _database_article(publication) -> dict:
     raw = publication.raw_data if isinstance(publication.raw_data, dict) else {}
@@ -245,9 +198,6 @@ def council_candidates(db: Session, today: date | None = None, q: str = "", offs
         author = _name(employee.full_name or "")
         ambiguous = bool(author and name_counts[author] > 1 and by_author.get(author))
         csv_articles = by_author.get(author, []) if author and name_counts[author] == 1 else []
-        articles = _unique_articles([*csv_articles, *(_database_article(p) for p in employee.publications)], start, end)
-        if not articles:
-            continue
         data = employee.current_data if isinstance(employee.current_data, dict) else {}
         try:
             degrees = academic_degrees(data)
@@ -255,6 +205,9 @@ def council_candidates(db: Session, today: date | None = None, q: str = "", offs
             degrees = []
         if not any(DOCTOR.fullmatch(degree) for degree in degrees):
             continue
+        articles = _unique_articles([*csv_articles, *(_database_article(p) for p in employee.publications)], start, end)
+        if not articles:
+            continue
         specialty_count = sum(specialty in a["specialty_codes"] for a in articles) if specialty else 0
         for article in articles:
             article["matches_specialty"] = bool(specialty and specialty in article["specialty_codes"])

app/services/council_source.py

@@ -0,0 +1,103 @@
+"""Prepared council publication data and shared source normalization."""
+
+import csv
+import hashlib
+import json
+from io import StringIO
+from collections import defaultdict
+from pathlib import Path
+import re
+import unicodedata
+
+DOI = re.compile(r"10\.\d{4,9}/\S+", re.IGNORECASE)
+
+
+def _name(value: str) -> str:
+    return " ".join(unicodedata.normalize("NFKC", value).casefold().replace("ё", "е").split())
+
+
+def _doi(value: str) -> str:
+    value = re.sub(r"^(?:https?://(?:dx\.)?doi\.org/|doi:\s*)", "", value.strip(), flags=re.I).lower()
+    return value if DOI.fullmatch(value) else ""
+
+
+def _quartiles(row: dict) -> list[str]:
+    return [f"{label} {str(row.get(field) or '').strip().upper()}"
+            for label, field in (("Scopus", "Квартиль scopus"), ("Web of Science", "Квартиль wos"))
+            if str(row.get(field) or '').strip().upper() in {"Q1", "Q2"}]
+
+
+def _journal_names(data: dict) -> list[str]:
+    names = []
+    for field in ("Название журнала", "Журнал", "journal"):
+        value = data.get(field)
+        if isinstance(value, dict):
+            value = value.get("name")
+        if isinstance(value, str) and value.strip():
+            names.append(value.strip())
+    return names
+
+
+def prepare_reference(source: Path) -> dict:
+    publications = []
+    by_author = defaultdict(list)
+    source_bytes = source.read_bytes()
+    with StringIO(source_bytes.decode("utf-8-sig"), newline="") as stream:
+        reader = csv.DictReader(stream, delimiter=";")
+        required = {"Название публикации", "Год", "DOI", "Авторы НИУ ВШЭ", "Квартиль scopus", "Квартиль wos"}
+        if not required <= set(reader.fieldnames or []):
+            raise ValueError("Missing council CSV headers")
+        if len(reader.fieldnames) != len(set(reader.fieldnames)):
+            raise ValueError("Duplicate council CSV headers")
+        for row_number, row in enumerate(reader, 2):
+            if None in row or any(value is None for value in row.values()):
+                raise ValueError("Malformed council CSV record")
+            authors = list(dict.fromkeys(value.strip() for value in row["Авторы НИУ ВШЭ"].split(";") if value.strip()))
+            record = {"source_row": row_number, "title": row["Название публикации"].strip(),
+                      "year": int(row["Год"]) if row["Год"].strip().isdigit() else None,
+                      "doi": _doi(row["DOI"]), "authors": authors, "quartiles": _quartiles(row),
+                      "journals": _journal_names(row),
+                      "metadata": {key: value for key, value in row.items()
+                                   if key not in {"Название публикации", "Год", "DOI", "Авторы НИУ ВШЭ"}}}
+            index = len(publications)
+            publications.append(record)
+            if record["title"]:
+                for author in sorted({_name(value) for value in authors} - {""}):
+                    by_author[author].append(index)
+    return {"schema_version": 1, "source_file": source.name,
+            "source_sha256": hashlib.sha256(source_bytes).hexdigest(),
+            "source_rows": len(publications), "publications": publications, "by_author": dict(by_author)}
+
+
+def load_reference(path: Path) -> dict[str, list[dict]]:
+    data = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(data, dict) or type(data.get("schema_version")) is not int or data.get("schema_version") != 1:
+        raise ValueError("Unsupported council reference schema")
+    publications = data.get("publications")
+    by_author = data.get("by_author")
+    if not isinstance(publications, list) or not isinstance(by_author, dict) or data.get("source_rows") != len(publications):
+        raise ValueError("Invalid council reference records")
+    articles = []
+    expected = defaultdict(list)
+    for index, record in enumerate(publications):
+        if not isinstance(record, dict) or not isinstance(record.get("title"), str) or not isinstance(record.get("doi"), str):
+            raise ValueError("Invalid council publication text")
+        if record["doi"] != _doi(record["doi"]) or (record.get("year") is not None and type(record["year"]) is not int):
+            raise ValueError("Invalid council publication DOI/year")
+        if record.get("source_row") != index + 2:
+            raise ValueError("Invalid council source row")
+        for field in ("authors", "quartiles", "journals"):
+            if not isinstance(record.get(field), list) or not all(isinstance(value, str) for value in record[field]):
+                raise ValueError("Invalid council publication list")
+        if not set(record["quartiles"]) <= {"Scopus Q1", "Scopus Q2", "Web of Science Q1", "Web of Science Q2"}:
+            raise ValueError("Invalid council quartiles")
+        if record["title"]:
+            for author in sorted({_name(value) for value in record["authors"]} - {""}):
+                expected[author].append(index)
+        articles.append({"title": record["title"], "year": record["year"], "doi": record["doi"],
+                         "types": [], "quartiles": record["quartiles"], "journals": record["journals"],
+                         "origins": ["Файл CSV"], "rows": [record["source_row"]],
+                         "in_csv": True, "in_db": False, "url": None})
+    if by_author != dict(expected) or any(type(i) is not int for indexes in by_author.values() for i in indexes):
+        raise ValueError("Council author index disagrees with publication authors")
+    return {author: [articles[index] for index in indexes] for author, indexes in by_author.items()}

app/version.py

@@ -1,3 +1,3 @@
-APP_VERSION = "0.15.1"
-FRONTEND_VERSION = "0.15.1"
-BACKEND_VERSION = "0.15.1"
+APP_VERSION = "0.15.2"
+FRONTEND_VERSION = "0.15.2"
+BACKEND_VERSION = "0.15.2"

pyproject.toml

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

scripts/prepare_council_articles.py

@@ -0,0 +1,33 @@
+"""Prepare a readable, indexed council JSON from the owner-supplied CSV.
+
+python scripts/prepare_council_articles.py app/data/council_articles.csv
+No database access or app startup; only Python's standard library is required.
+"""
+import argparse
+import json
+from pathlib import Path
+import sys
+from tempfile import NamedTemporaryFile
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from app.services.council_source import load_reference, prepare_reference
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("source", type=Path)
+    parser.add_argument("--output", type=Path, default=Path("app/data/council_articles.json"))
+    args = parser.parse_args()
+    if args.source.resolve() == args.output.resolve():
+        parser.error("Output must not overwrite the source CSV")
+    result = prepare_reference(args.source)
+    text = json.dumps(result, ensure_ascii=False, indent=2) + "\n"
+    with NamedTemporaryFile(mode="w", encoding="utf-8", dir=args.output.parent,
+                            prefix=args.output.name + ".", suffix=".tmp", delete=False) as stream:
+        stream.write(text)
+        temporary = Path(stream.name)
+    try:
+        load_reference(temporary)
+        temporary.replace(args.output)
+    finally:
+        temporary.unlink(missing_ok=True)
+    print(f"Prepared {result['source_rows']} publications and {len(result['by_author'])} author names")

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

tests/test_council.py

@@ -19,6 +19,7 @@ from app.services.council_export import _append
 @pytest.fixture()
 def export(monkeypatch, tmp_path):
     path = tmp_path / "articles.csv"
+    prepared = tmp_path / "articles.json"
     def write(rows):
         fields = ["Название публикации", "Год", "DOI", "EID (Scopus)", "UT (WoS)", "Название журнала",
                   "Авторы НИУ ВШЭ", "Квартиль scopus", "Квартиль wos"]
@@ -26,8 +27,11 @@ def export(monkeypatch, tmp_path):
             writer = csv.DictWriter(output, fieldnames=fields, delimiter=";")
             writer.writeheader()
             writer.writerows(rows)
+        from app.services.council_source import prepare_reference
+        import json
+        prepared.write_text(json.dumps(prepare_reference(path), ensure_ascii=False), encoding="utf-8")
         council._articles_by_author.cache_clear()
-    monkeypatch.setattr(council, "EXPORT", path)
+    monkeypatch.setattr(council, "EXPORT", prepared)
     write([])
     yield write
     council._articles_by_author.cache_clear()
@@ -211,7 +215,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.15.1 · Frontend 0.15.1" in response.text
+        assert "Backend 0.15.2 · Frontend 0.15.2" 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
@@ -565,3 +569,13 @@ def test_xlsx_specialties_are_explicit_with_and_without_filter(db_session, expor
             assert unmatched['Специальности ВАК'] == 'Не подтверждены'
         assert book['Сотрудники'].auto_filter.ref.endswith(f'M{len(staff)}')
         assert book['Публикации'].auto_filter.ref.endswith(f'Q{len(works)}')
+
+
+def test_non_doctor_publications_are_not_parsed(db_session, export, monkeypatch):
+    person = employee(db_session, degree='Кандидат технических наук')
+    publication(db_session, person, kind='ARTICLE', description={'short': {'ru': '<b>Journal</b>. 2024.'}})
+    db_session.commit()
+    def unexpected(publication):
+        pytest.fail('Ineligible employee publications must not be parsed')
+    monkeypatch.setattr(council, '_database_article', unexpected)
+    assert council.council_candidates(db_session)['total'] == 0

tests/test_council_source.py

@@ -0,0 +1,102 @@
+import csv
+import hashlib
+import json
+from pathlib import Path
+
+import pytest
+
+from app.services import council
+from app.services.council_source import load_reference, prepare_reference
+
+
+def write_csv(path, rows):
+    fields = ['Название публикации', 'Год', 'DOI', 'Авторы НИУ ВШЭ', 'Квартиль scopus', 'Квартиль wos', 'Название журнала', 'EID (Scopus)']
+    with path.open('w', encoding='utf-8-sig', newline='') as stream:
+        writer = csv.DictWriter(stream, fieldnames=fields, delimiter=';')
+        writer.writeheader()
+        writer.writerows(rows)
+
+
+def test_prepared_reference_preserves_quoted_fields_metadata_and_matching(tmp_path):
+    source = tmp_path / 'source.csv'
+    write_csv(source, [
+        {'Название публикации': 'Quoted; title\ncontinuation', 'Год': '2024', 'DOI': 'https://doi.org/10.1234/TEST',
+         'Авторы НИУ ВШЭ': ' Ёлкин  Пётр; ЕЛКИН ПЕТР; Другой автор ', 'Квартиль scopus': ' q1 ',
+         'Квартиль wos': 'Q2', 'Название журнала': 'Бизнес-информатика', 'EID (Scopus)': 'source-id'},
+        {'Название публикации': '', 'Авторы НИУ ВШЭ': 'Другой автор'},
+        {'Название публикации': 'Unknown', 'Год': '', 'DOI': 'invalid', 'Авторы НИУ ВШЭ': ''},
+    ])
+    before = source.read_bytes()
+    data = prepare_reference(source)
+    assert source.read_bytes() == before
+    assert data['source_rows'] == 3
+    assert data['source_sha256'] == hashlib.sha256(before).hexdigest()
+    assert data['by_author'] == {'другой автор': [0], 'елкин петр': [0]}
+    record = data['publications'][0]
+    assert record['title'] == 'Quoted; title\ncontinuation'
+    assert record['doi'] == '10.1234/test' and record['year'] == 2024
+    assert record['quartiles'] == ['Scopus Q1', 'Web of Science Q2']
+    assert record['journals'] == ['Бизнес-информатика']
+    assert record['metadata']['EID (Scopus)'] == 'source-id'
+    output = tmp_path / 'data.json'
+    output.write_text(json.dumps(data, ensure_ascii=False), encoding='utf-8')
+    articles = load_reference(output)
+    assert articles['елкин петр'][0]['rows'] == [2]
+    assert articles['другой автор'][0]['title'] == record['title']
+    assert len(articles['елкин петр']) == 1
+    assert articles['елкин петр'][0]['origins'] == ['Файл CSV']
+
+
+@pytest.mark.parametrize('change', ['author', 'index', 'bool_index', 'year', 'schema', 'source_row', 'list'])
+def test_corrupt_prepared_reference_is_rejected(tmp_path, change):
+    source = tmp_path / 'source.csv'
+    write_csv(source, [{'Название публикации': 'First', 'Авторы НИУ ВШЭ': 'Author'},
+                       {'Название публикации': 'Second', 'Авторы НИУ ВШЭ': 'Other'}])
+    data = prepare_reference(source)
+    if change == 'author': data['by_author'] = {'stranger': [0], 'other': [1]}
+    if change == 'index': data['by_author']['author'] = [3]
+    if change == 'bool_index': data['by_author']['other'] = [True]
+    if change == 'year': data['publications'][0]['year'] = True
+    if change == 'schema': data['schema_version'] = True
+    if change == 'source_row': data['publications'][0]['source_row'] = 77
+    if change == 'list': data['publications'][0]['journals'] = 'Journal'
+    path = tmp_path / 'data.json'
+    path.write_text(json.dumps(data), encoding='utf8')
+    with pytest.raises(ValueError): load_reference(path)
+
+
+def test_committed_reference_is_reproducible_and_loaded_once(monkeypatch):
+    source = council.EXPORT.with_suffix('.csv')
+    data = json.loads(council.EXPORT.read_text(encoding='utf-8'))
+    assert data == prepare_reference(source)
+    assert data['source_rows'] == 1797 and len(data['by_author']) == 825
+    calls = []
+    original = council.load_reference
+    def counted(path):
+        calls.append(path)
+        return original(path)
+    monkeypatch.setattr(council, 'load_reference', counted)
+    council._articles_by_author.cache_clear()
+    try:
+        first = council._articles_by_author()
+        assert council._articles_by_author() is first
+        assert calls == [council.EXPORT]
+    finally:
+        council._articles_by_author.cache_clear()
+
+
+def test_converter_does_not_replace_output_when_input_is_invalid(tmp_path):
+    import subprocess
+    import sys
+    source = tmp_path / 'invalid.csv'
+    source.write_text('Wrong header\nvalue\n', encoding='utf8')
+    output = tmp_path / 'prepared.json'
+    output.write_text('previous reference', encoding='utf8')
+    result = subprocess.run([sys.executable, 'scripts/prepare_council_articles.py', str(source), '--output', str(output)],
+                            capture_output=True, text=True)
+    assert result.returncode != 0
+    assert output.read_text(encoding='utf8') == 'previous reference'
+    assert not list(tmp_path.glob('*.tmp'))
+    same = subprocess.run([sys.executable, 'scripts/prepare_council_articles.py', str(source), '--output', str(source)],
+                          capture_output=True, text=True)
+    assert same.returncode != 0 and source.read_text(encoding='utf8') == 'Wrong header\nvalue\n'