README.md

@@ -141,4 +141,8 @@ node --check app/static/admin.js
 
 Тесты с данными используют временную SQLite; отдельные API smoke-тесты запускают приложение с его текущей конфигурацией. Браузерная проверка: `pip install playwright`, `python -m playwright install chromium`, затем `python tests/browser_admin.py`. Она проверяет реальный рендеринг страниц, клавиатуру, диалог, фильтры и восстановление прогресса после ошибки. Скриншоты сохраняются во временную папку; путь выводится в конце.
 
-Версия сервиса: `0.9.0`. Админка всегда показывает версии backend и frontend в footer.
+Версия сервиса: `0.10.2`. Админка всегда показывает версии backend и frontend в footer.
+
+## Разовый импорт публикаций
+
+Инструкция, минимальные права временной роли и ограниченный SSH-запуск: [docs/publication-import.md](docs/publication-import.md). По умолчанию импортер выполняет dry-run; production-применение требует отдельного решения владельца.

app/services/crawler.py

@@ -2,7 +2,8 @@ import gzip
 import hashlib
 import json
 import re
-import time
+import time
+import unicodedata
 from datetime import datetime, timezone
 
 import requests
@@ -398,8 +399,11 @@ def _employee_publications_table_exists(db: Session) -> bool:
     return inspect(db.connection()).has_table(EmployeePublication.__tablename__)
 
 
-def _sync_employee_publications(db: Session, employee: Employee, parsed: dict) -> None:
-    publications = _publication_payloads(parsed)
+def _sync_employee_publications(db: Session, employee: Employee, parsed: dict) -> None:
+    publications = _publication_payloads(parsed)
+    imported = [item for item in db.scalars(select(EmployeePublication).where(
+        EmployeePublication.employee_id == employee.id
+    )).all() if _csv_enrichment(item) is not None]
     seen_hashes = set()
     for publication in publications:
         source_hash = _publication_hash(publication)
@@ -420,8 +424,10 @@ def _sync_employee_publications(db: Session, employee: Employee, parsed: dict) -
                     EmployeePublication.source_hash == source_hash,
                 )
             )
-        if not existing:
-            existing = EmployeePublication(employee_id=employee.id, source_hash=source_hash, title=_publication_title(publication))
+        if not existing:
+            existing = _match_imported_publication(imported, publication)
+        if not existing:
+            existing = EmployeePublication(employee_id=employee.id, source_hash=source_hash, title=_publication_title(publication))
             db.add(existing)
         _apply_publication(existing, publication, source_hash)
 
@@ -432,8 +438,39 @@ def _sync_employee_publications(db: Session, employee: Employee, parsed: dict) -
                 EmployeePublication.source_hash.not_in(seen_hashes),
             )
         ).all()
-        for item in stale:
-            db.delete(item)
+        for item in stale:
+            if _csv_enrichment(item) is None:
+                db.delete(item)
+
+
+def _csv_enrichment(publication: EmployeePublication) -> dict | None:
+    raw = publication.raw_data
+    metadata = raw.get("csv_enrichment") if isinstance(raw, dict) else None
+    return metadata if isinstance(metadata, dict) else None
+
+
+def _match_imported_publication(imported: list[EmployeePublication], publication: dict) -> EmployeePublication | None:
+    def doi(value):
+        return re.sub(r"^(?:https?://(?:dx\.)?doi\.org/|doi:\s*)", "", (value or "").strip(), flags=re.I).lower()
+
+    def title(value):
+        return " ".join(unicodedata.normalize("NFKC", value or "").casefold().replace("ё", "е").split())
+
+    identifier = doi(publication.get("doi_url"))
+    year = _int_or_none(publication.get("year"))
+    name = title(publication.get("title") or publication.get("text"))
+    matches = [item for item in imported if
+               (identifier and doi(item.doi_url) == identifier) or
+               (name and title(item.title) == name and item.year in (None, year))]
+    if len(matches) > 1:
+        raise ValueError("Ambiguous imported publication; synchronization rolled back")
+    if not matches:
+        return None
+    match = matches[0]
+    if ((identifier and doi(match.doi_url) and identifier != doi(match.doi_url)) or
+            (year and match.year and year != match.year)):
+        raise ValueError("Conflicting imported publication; synchronization rolled back")
+    return match
 
 
 def _publication_payloads(parsed: dict) -> list[dict]:
@@ -447,8 +484,16 @@ def _publication_payloads(parsed: dict) -> list[dict]:
     return publications
 
 
-def _apply_publication(target: EmployeePublication, publication: dict, source_hash: str) -> None:
-    target.publication_id = _clean_optional(publication.get("publication_id") or publication.get("id"))
+def _apply_publication(target: EmployeePublication, publication: dict, source_hash: str) -> None:
+    enrichment = _csv_enrichment(target)
+    publication_id = _clean_optional(publication.get("publication_id") or publication.get("id"))
+    if enrichment is not None and target.publication_id and publication_id and target.publication_id != publication_id:
+        raise ValueError(
+            f"Ambiguous HSE IDs for CSV publication {target.id}: "
+            f"{target.publication_id} != {publication_id}; publication synchronization rolled back"
+        )
+    preserved = {field: getattr(target, field) for field in ("title", "year", "doi_url")}
+    target.publication_id = (publication_id or target.publication_id) if enrichment is not None else publication_id
     target.title = _publication_title(publication)
     target.year = _int_or_none(publication.get("year"))
     target.publication_type = _clean_optional(publication.get("publication_type") or publication.get("type"))
@@ -462,7 +507,12 @@ def _apply_publication(target: EmployeePublication, publication: dict, source_ha
     target.annotation = publication.get("annotation") if isinstance(publication.get("annotation"), dict) else None
     target.description = publication.get("description") if isinstance(publication.get("description"), dict) else None
     target.authors = publication.get("authors") if isinstance(publication.get("authors"), list) else None
-    target.raw_data = publication.get("raw_data") if isinstance(publication.get("raw_data"), dict) else publication
+    target.raw_data = publication.get("raw_data") if isinstance(publication.get("raw_data"), dict) else publication
+    if enrichment is not None:
+        target.raw_data = {**target.raw_data, "csv_enrichment": enrichment}
+        for field in ("title", "year", "doi_url"):
+            if not getattr(target, field) or (field == "title" and not (publication.get("title") or publication.get("text"))):
+                setattr(target, field, preserved[field])
     target.source_hash = source_hash
 
 

app/version.py

@@ -1,3 +1,3 @@
-APP_VERSION = "0.9.0"
-FRONTEND_VERSION = "0.9.0"
-BACKEND_VERSION = "0.9.0"
+APP_VERSION = "0.10.2"
+FRONTEND_VERSION = "0.10.2"
+BACKEND_VERSION = "0.10.2"

docs/publication-import.md

@@ -0,0 +1,312 @@
+# Разовый импорт публикаций
+
+Инструмент: `scripts/import_publications.py`. Он не импортирует приложение, не читает
+`.env`, не вызывает `init_db`, не создаёт таблиц и не удаляет записи. Подключение
+берётся только из libpq environment/service, настроенных владельцем. Production в
+рамках подготовки не проверялся и не изменялся.
+
+Локальные проверки: синтетические сотрудники, повторный запуск, дубликаты,
+неоднозначности, конфликты DOI/года, сохранение заполненных значений и
+транзакционный путь на тестовом соединении (read-only / rollback при второй
+ошибочной вставке / успешный commit). Прогон на настоящем PostgreSQL и проверка
+grants не выполнялись; первый owner-run dry-run с read-only ролью обязателен.
+
+## Вход и сопоставление
+
+Файл владельца `Статьи.csv`: UTF-8 с возможным BOM, разделитель `;`, 1797 записей,
+SHA256 `1f236be56d99677b1403df29a4c1d3a7e6d10cb90bd355d678201a027921b0ce`.
+Обязательные колонки: `Название публикации`, `Год`, `DOI`, `Авторы НИУ ВШЭ`.
+Номер строки в отчёте — номер CSV-записи с заголовком, а не физической строки.
+
+Сотрудник сопоставляется по полному ФИО: Unicode NFKC, регистр, пробелы и ё/е.
+Учитываются все существующие статусы; новые сотрудники не создаются. Ноль
+совпадений → `employee_not_found`, несколько → `ambiguous_employee`. Инициалы,
+переводы, перестановки слов и приблизительные совпадения не используются.
+
+Публикация сопоставляется **внутри конкретного сотрудника** по нормализованному DOI
+или полному названию и году. Отсутствующий год БД допускает заполнение при
+совпадении названия. Если эти способы указывают на разные записи, импорт
+пропускается (`ambiguous_publication`). Разные заполненные DOI или годы —
+`doi_conflict` / `year_conflict`. Существующие дубликаты не объединяются и не
+удаляются. Одна статья у разных сотрудников — предусмотренные моделью авторские
+связи, а не дубли внутри карточки.
+
+Повтор CSV внутри сотрудника не создаёт записи. Каждая поздняя строка дополняет
+ещё пустые метаданные; первый непустой источник имеет приоритет, в том числе
+ранее сохранённое значение БД. Повтор без изменений — `csv_duplicate`, дополнение
+запланированной вставки — `merged_csv_duplicate`. При конфликте непустых
+метаданных первое значение сохраняется; остальные пустые поля всё равно
+дополняются. Отчёт содержит `metadata_conflict_first_value_kept`, список
+`conflict_fields` (только названия полей, без значений) и сводный счётчик
+`metadata_conflicts`. Конфликт DOI/года идентификации по-прежнему пропускает
+всю связь. Если строка без DOI была запланирована первой, следующая строка с
+DOI может дополнить ту же вставку. При повторном запуске уже заполненные записи дают
+`already_present`. Хеш новой записи детерминирован по сотруднику и DOI либо
+названию/году. `source_hash` существующих записей не меняется.
+
+Пустые ячейки ничего не стирают. Пустые название/авторы, некорректный год
+(допустимо 1000–9999) или DOI пропускаются как `invalid_fields`. Повреждённая
+структура CSV прерывает весь запуск до подключения. Несовместимый JSON БД
+пропускается (`unsupported_raw_data`, `unsupported_csv_metadata`).
+
+## Таблицы, столбцы и права
+
+| Объект | Операция / столбцы |
+|---|---|
+| `public.employees` | Только SELECT: `id`, `full_name` |
+| `public.employee_publications` | SELECT: `id`, `employee_id`, `title`, `year`, `doi_url`, `raw_data` |
+| Новая публикация | INSERT: `employee_id`, `title`, `year`, `doi_url` при наличии, `raw_data`, `source_hash`, `created_at`, `updated_at` |
+| Существующая публикация | UPDATE: только пустые `title`, `year`, `doi_url`; `raw_data`; служебный `updated_at` |
+| Sequence идентификатора публикации | USAGE для `nextval`; имя определяется `pg_get_serial_sequence`, обычно `public.employee_publications_id_seq` |
+
+Остальные таблицы и sequences не затрагиваются. В `raw_data` сохраняются прежние
+ключи; добавляется/дополняется `csv_enrichment` только отсутствующими или пустыми
+значениями. Уже заполненные CSV-метаданные не заменяются. Поля `authors`,
+`publication_id`, тип материала, язык, аннотация, ссылки HSE и статус не меняются.
+CSV содержит **только авторов НИУ ВШЭ**, поэтому он не считается полным составом
+авторов и не заменяет `authors`. Идентификаторы Scopus/WoS, квартиль и остальные
+непустые исходные ячейки сохраняются в `raw_data.csv_enrichment` без интерпретации.
+Карточки `employees.current_data` и snapshots не изменяются; отчёт и данные
+таблицы не означают обновления всех представлений интерфейса.
+
+Минимальные собственные grants роли: CONNECT к БД, USAGE схемы public,
+перечисленные SELECT/INSERT/UPDATE по столбцам, USAGE найденной sequence.
+Не нужны DELETE, TRUNCATE, CREATE, ALTER, ownership, superuser, доступ к прочим
+таблицам и sequence UPDATE/setval. Роль новая, без членства в других ролях.
+Права PUBLIC всё равно наследуются: владелец должен отдельно проверить их;
+этот набор команд не меняет общие права PUBLIC и не гарантирует изоляцию при
+уже существующих широких общих grants. В частности TEMP может быть выдан PUBLIC.
+
+## Применение и ограничения
+
+Без `--apply` используется PostgreSQL `SERIALIZABLE READ ONLY`, без INSERT,
+UPDATE и nextval, с rollback. С `--apply` весь план применяется одной транзакцией
+SERIALIZABLE. Advisory lock предотвращает одновременное применение этим
+импортером. Таймаут запроса — 60 секунд, ожидания блокировок — 5 секунд.
+Ошибка записи/конфликт сериализации откатывает все изменения таблиц; автоматических
+повторов нет. Sequence может получить пропуски при откате — это нормальное
+поведение PostgreSQL. Изменение входного файла во время запуска прерывает импорт.
+
+JSON-отчёт содержит количества insert/update/skip, причины и пары
+`номер CSV-записи — employee_id`, без ФИО, названий статей и секретов. Счётчики
+относятся к авторским связям, а не уникальным статьям. Консоль выводит только
+сводку. Отчёт сначала пишется в `.pending` до commit, после успеха переименовывается.
+Если остался `.pending`, следующий запуск блокируется: сначала проверьте исход
+транзакции. Отчёт `.pending` не доказывает commit; ошибка переименования после
+commit также возможна. Безопасный повторный dry-run позволяет проверить состояние.
+При необходимости владелец перемещает `.pending` в архив перед повтором.
+
+**До применения нужен свежий backup БД.** Во время импорта владелец должен
+исключить конкурентные записи публикаций на время транзакции, согласовав окно
+работы; эта подготовка не отключает crawler. Не выполняйте production apply до
+независимого ревью и развёртывания исправления crawler из этого PR.
+
+До исправления `_upsert_employee` запускал синхронизацию публикаций даже для
+неизменённой карточки. При пустом списке публикаций или отсутствии таблицы
+синхронизация пропускалась. При непустом списке `_sync_employee_publications`
+удалял все записи сотрудника, чей `source_hash` отсутствовал среди хешей ответа
+ВШЭ, включая CSV-вставки. Найденные по `publication_id`/хешу записи полностью
+получали новый `raw_data`, поэтому теряли CSV-метаданные даже без удаления.
+
+Исправление сохраняет записи с dict `raw_data.csv_enrichment` при очистке
+устаревших публикаций. При обновлении сохраняется этот ключ и заполненные
+title/year/doi_url, если ВШЭ вернул пустые значения. Непустые значения ВШЭ
+продолжают обновлять канонические поля, а исходные CSV-значения остаются в
+метаданных. Если статья впервые появилась у ВШЭ, crawler сопоставляет её с
+импортированной записью по DOI или названию/году до новой вставки, сохраняет id
+и добавляет HSE publication_id. Конфликт/неоднозначность откатывает весь savepoint
+синхронизации публикаций и регистрирует CrawlError; карточка сотрудника продолжает
+обрабатываться. Удаление обычных устаревших HSE-записей остаётся прежним. Здесь
+не приведены команды рестарта или отключения сервисов.
+
+Уже установленный HSE publication_id CSV-записи нельзя заменить другим HSE ID,
+даже когда DOI/название/год совпадают. Это проверяется перед изменением полей
+при любом пути сопоставления (ID, source_hash, CSV-поля). Поэтому две HSE-статьи
+не могут по очереди перезаписать одну CSV-запись: весь savepoint откатывается
+независимо от порядка статей. Отсутствующий ID во входе не очищает существующую
+CSV-привязку. Отчёт обхода (`errors` в деталях запуска) содержит CrawlError с
+идентификатором записи БД и конфликтующими HSE ID, без ФИО и названий статей.
+
+## Команды владельцу (не выполнялись)
+
+Ниже — подготовка разового доступа после ревью. Выполняет владелец на сервере.
+Рабочая директория — проверенная копия этого PR; файлы программы и CSV должны
+быть защищены от записи временным SSH-пользователем. Пароли/ключи не включать
+в репозиторий или чат. Имена контейнеров соответствуют текущему Compose-проекту;
+владелец проверяет их локально перед выполнением.
+
+Сначала резервная копия в приватный каталог владельца:
+
+```sh
+install -d -m 700 /var/backups/miem-publication-import
+docker exec miem_workers-postgres-1 pg_dump -U miem -d miem_workers -Fc -f /tmp/publications-before.dump
+docker cp miem_workers-postgres-1:/tmp/publications-before.dump /var/backups/miem-publication-import/before.dump
+chmod 600 /var/backups/miem-publication-import/before.dump
+```
+
+При повторной операции используйте новый путь backup; не перезаписывайте нужную
+предыдущую копию. Проверьте exit codes и архив `pg_restore --list` до выдачи write.
+
+Откройте psql владельцем БД:
+
+```sh
+docker exec -it miem_workers-postgres-1 psql -U miem -d miem_workers
+```
+
+Выполните SQL (без пароля в SQL/истории):
+
+```sql
+\set ON_ERROR_STOP on
+SELECT format('CREATE ROLE miem_publication_import LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOINHERIT NOREPLICATION NOBYPASSRLS CONNECTION LIMIT 1 VALID UNTIL %L', CURRENT_TIMESTAMP + interval '24 hours') \gexec
+\password miem_publication_import
+GRANT CONNECT ON DATABASE miem_workers TO miem_publication_import;
+GRANT USAGE ON SCHEMA public TO miem_publication_import;
+GRANT SELECT (id, full_name) ON public.employees TO miem_publication_import;
+GRANT SELECT (id, employee_id, title, year, doi_url, raw_data) ON public.employee_publications TO miem_publication_import;
+-- Сначала выдаются только read-права для dry-run.
+SELECT pg_get_serial_sequence('public.employee_publications', 'id');
+SELECT has_schema_privilege('miem_publication_import', 'public', 'CREATE') AS unexpected_create,
+       has_table_privilege('miem_publication_import', 'public.employees', 'INSERT') AS unexpected_employee_insert,
+       has_table_privilege('miem_publication_import', 'public.employee_publications', 'DELETE') AS unexpected_delete;
+```
+
+Три проверки должны вернуть false. Не выдавайте write-права при неожиданном
+доступе. Если sequence не найдена, выясните фактический механизм id перед импортом.
+Срок роли ограничивает парольную авторизацию, но не заменяет отзыв доступа и
+завершение открытых сессий.
+
+После принятого dry-run, непосредственно перед разрешённым apply:
+
+```sql
+GRANT INSERT (employee_id, title, year, doi_url, raw_data, source_hash, created_at, updated_at) ON public.employee_publications TO miem_publication_import;
+GRANT UPDATE (title, year, doi_url, raw_data, updated_at) ON public.employee_publications TO miem_publication_import;
+SELECT format('GRANT USAGE ON SEQUENCE %s TO miem_publication_import', pg_get_serial_sequence('public.employee_publications', 'id')) \gexec
+```
+
+Подготовьте внутри API-контейнера защищённые файлы libpq `pg_service.conf` и
+`pgpass` из окружения/секрет-хранилища владельца: service
+`miem_publication_import`, БД `miem_workers`, пользователь
+`miem_publication_import`. Используйте внутренний PostgreSQL endpoint Compose,
+порт 5432; не публикуйте порт БД. Поле password хранится только в `pgpass`,
+доступном UID 65534 с mode 600. Не копируйте `.env` приложения. Этот шаг
+зависит от секрет-хранилища владельца; универсальной команды получения секрета нет.
+
+Установите файлы (команды root; каталог контейнера не должен быть volume с
+непроверенными файлами). `Статьи.csv` расположен рядом с каталогом репозитория:
+
+```sh
+docker exec miem_workers-api-1 mkdir -p /opt/miem-import /tmp/miem-import
+docker cp scripts/import_publications.py miem_workers-api-1:/opt/miem-import/import_publications.py
+docker cp ../Статьи.csv miem_workers-api-1:/opt/miem-import/articles.csv
+docker exec miem_workers-api-1 chown -R root:root /opt/miem-import
+docker exec miem_workers-api-1 chmod 755 /opt/miem-import
+docker exec miem_workers-api-1 chmod 644 /opt/miem-import/import_publications.py /opt/miem-import/articles.csv
+docker exec miem_workers-api-1 chown 65534:65534 /opt/miem-import/pg_service.conf /opt/miem-import/pgpass /tmp/miem-import
+docker exec miem_workers-api-1 chmod 600 /opt/miem-import/pg_service.conf /opt/miem-import/pgpass
+docker exec miem_workers-api-1 chmod 700 /tmp/miem-import
+docker exec miem_workers-api-1 sha256sum /opt/miem-import/articles.csv
+install -m 755 -o root -g root scripts/import-publications-run /usr/local/sbin/miem-publication-import
+install -m 755 -o root -g root scripts/import-publications-ssh /usr/local/sbin/miem-publication-import-ssh
+useradd --create-home --shell /bin/sh miem-pub-import
+install -d -m 700 -o miem-pub-import -g miem-pub-import /home/miem-pub-import/.ssh
+```
+
+Проверьте SHA256 CSV по указанному выше значению и script по проверенной копии PR.
+Секретные файлы должны существовать перед командами chown/chmod. Root-владелец
+каталога не позволяет UID 65534 заменить программу/CSV; этот UID может писать
+только отчёт и свои файлы конфигурации. Не предоставляйте SSH-пользователю группу
+docker или общий sudo. Каталоги и executables выше должны быть root-owned и
+недоступны для записи этому пользователю. Убедитесь, что абсолютный путь Docker
+на сервере — `/usr/bin/docker`.
+
+Создайте `/etc/sudoers.d/miem-publication-import` с mode 440:
+
+```sudoers
+miem-pub-import ALL=(root) NOPASSWD: /usr/local/sbin/miem-publication-import dry-run, /usr/local/sbin/miem-publication-import apply
+```
+
+```sh
+visudo -cf /etc/sudoers.d/miem-publication-import
+```
+
+В `/home/miem-pub-import/.ssh/authorized_keys` добавьте **только публичный** ключ
+владельца с ограничениями (замените `<PUBLIC_KEY>` самим публичным ключом):
+
+```text
+restrict,command="/usr/local/sbin/miem-publication-import-ssh" <PUBLIC_KEY>
+```
+
+```sh
+chown root:root /home/miem-pub-import/.ssh/authorized_keys
+chmod 644 /home/miem-pub-import/.ssh/authorized_keys
+chown root:root /home/miem-pub-import /home/miem-pub-import/.ssh
+chmod 755 /home/miem-pub-import
+chmod 755 /home/miem-pub-import/.ssh
+```
+
+Root-владение home/.ssh предотвращает замену ограниченного authorized_keys.
+Убедитесь, что SSH не разрешает этому пользователю другие ключи, пароль,
+keyboard-interactive, пользовательские environment/rc или общий shell.
+При необходимости задайте в sshd `Match User miem-pub-import` с
+`AuthenticationMethods publickey`, `PasswordAuthentication no`,
+`KbdInteractiveAuthentication no`, `DisableForwarding yes`, `PermitTTY no`,
+`PermitUserRC no` и указанным ForceCommand. Проверка/reload SSH зависит от
+конфигурации сервера; перед reload сохраните действующую сессию владельца.
+
+С клиентской машины параметры хоста/ключа берутся из окружения владельца:
+
+```sh
+ssh -i "$IMPORT_SSH_KEY" -o IdentitiesOnly=yes -o BatchMode=yes -o StrictHostKeyChecking=yes "miem-pub-import@$PROD_HOST" 'import-publications dry-run'
+```
+
+После проверки сводки, выдачи write-grants и отдельного решения владельца:
+
+```sh
+ssh -i "$IMPORT_SSH_KEY" -o IdentitiesOnly=yes -o BatchMode=yes -o StrictHostKeyChecking=yes "miem-pub-import@$PROD_HOST" 'import-publications apply'
+ssh -i "$IMPORT_SSH_KEY" -o IdentitiesOnly=yes -o BatchMode=yes -o StrictHostKeyChecking=yes "miem-pub-import@$PROD_HOST" 'import-publications dry-run'
+```
+
+Повторный dry-run должен показать ноль insert/update. При concurrent changes
+сводки могут различаться; без production-проверки фактические количества неизвестны.
+Владелец забирает отчёты без предоставления SCP/general shell временному ключу:
+
+```sh
+docker cp miem_workers-api-1:/tmp/miem-import/apply.json ./publication-import-apply.json
+docker cp miem_workers-api-1:/tmp/miem-import/dry-run.json ./publication-import-dry-run.json
+```
+
+## Отзыв доступа
+
+Владелец прекращает новые SSH-входы: перемещает временный authorized_keys в свой
+закрытый архив и удаляет только этот sudoers-файл. Не удаляйте данные БД или volumes.
+
+```sh
+mv /home/miem-pub-import/.ssh/authorized_keys /var/backups/miem-publication-import/authorized_keys.disabled
+rm /etc/sudoers.d/miem-publication-import
+```
+
+В psql владельцем:
+
+```sql
+ALTER ROLE miem_publication_import NOLOGIN;
+SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE usename = 'miem_publication_import' AND pid <> pg_backend_pid();
+REVOKE ALL PRIVILEGES ON DATABASE miem_workers FROM miem_publication_import;
+REVOKE ALL PRIVILEGES ON SCHEMA public FROM miem_publication_import;
+REVOKE SELECT (id, full_name) ON public.employees FROM miem_publication_import;
+REVOKE SELECT (id, employee_id, title, year, doi_url, raw_data),
+       INSERT (employee_id, title, year, doi_url, raw_data, source_hash, created_at, updated_at),
+       UPDATE (title, year, doi_url, raw_data, updated_at)
+       ON public.employee_publications FROM miem_publication_import;
+SELECT format('REVOKE ALL PRIVILEGES ON SEQUENCE %s FROM miem_publication_import', pg_get_serial_sequence('public.employee_publications', 'id')) \gexec
+DROP ROLE miem_publication_import;
+```
+
+После сохранения отчётов и backup удалите только временные секреты подключения:
+
+```sh
+docker exec miem_workers-api-1 rm /opt/miem-import/pgpass /opt/miem-import/pg_service.conf
+```
+
+Эти действия не используют `DROP OWNED` и не удаляют публикации. SQL роли не
+является миграцией схемы приложения. Если DROP ROLE не проходит, проверьте
+оставшиеся grants; не применяйте CASCADE/удаление данных.

pyproject.toml

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

scripts/import-publications-run

@@ -0,0 +1,25 @@
+#!/bin/sh
+# Install root-owned; caller cannot choose container, executable, input or env.
+set -eu
+case "${1:-}" in
+    dry-run) mode=dry-run ;;
+    apply) mode=apply ;;
+    *) exit 1 ;;
+esac
+[ "$#" -eq 1 ] || exit 1
+if [ "$mode" = apply ]; then
+    exec /usr/bin/docker exec --user 65534:65534 \
+        -e PGSERVICEFILE=/opt/miem-import/pg_service.conf \
+        -e PGSERVICE=miem_publication_import \
+        -e PGPASSFILE=/opt/miem-import/pgpass \
+        -e PYTHONNOUSERSITE=1 \
+        miem_workers-api-1 python -I /opt/miem-import/import_publications.py \
+        /opt/miem-import/articles.csv --apply --report /tmp/miem-import/apply.json
+fi
+exec /usr/bin/docker exec --user 65534:65534 \
+    -e PGSERVICEFILE=/opt/miem-import/pg_service.conf \
+    -e PGSERVICE=miem_publication_import \
+    -e PGPASSFILE=/opt/miem-import/pgpass \
+    -e PYTHONNOUSERSITE=1 \
+    miem_workers-api-1 python -I /opt/miem-import/import_publications.py \
+    /opt/miem-import/articles.csv --report /tmp/miem-import/dry-run.json

scripts/import-publications-ssh

@@ -0,0 +1,8 @@
+#!/bin/sh
+# Forced SSH command: no arguments, paths, forwarding or shell from the client.
+set -eu
+case "${SSH_ORIGINAL_COMMAND:-}" in
+    'import-publications dry-run') exec sudo -n /usr/local/sbin/miem-publication-import dry-run ;;
+    'import-publications apply') exec sudo -n /usr/local/sbin/miem-publication-import apply ;;
+    *) printf '%s\n' 'Allowed: import-publications dry-run | import-publications apply' >&2; exit 1 ;;
+esac

scripts/import_publications.py

@@ -0,0 +1,192 @@
+"""Owner-run CSV enrichment; no app startup, .env loading, DDL or deletes."""
+
+import argparse
+import csv
+import hashlib
+import json
+import re
+import unicodedata
+from collections import Counter, defaultdict
+from pathlib import Path
+
+
+def normalized(value):
+    return " ".join(unicodedata.normalize("NFKC", value or "").casefold().replace("ё", "е").split())
+
+
+def doi(value):
+    return re.sub(r"^(?:https?://(?:dx\.)?doi\.org/|doi:\s*)", "", (value or "").strip(), flags=re.I).lower()
+
+
+def read_rows(path):
+    with path.open(encoding="utf-8-sig", newline="") as source:
+        reader = csv.DictReader(source, delimiter=";")
+        required = {"Название публикации", "Год", "DOI", "Авторы НИУ ВШЭ"}
+        if not required <= set(reader.fieldnames or []):
+            raise ValueError("Missing required CSV headers")
+        rows = []
+        for number, row in enumerate(reader, 2):
+            if None in row or any(v is None for v in row.values()):
+                raise ValueError(f"Malformed CSV record {number}")
+            rows.append((number, {k.strip(): v.strip() for k, v in row.items()}))
+        return rows
+
+
+def plan(rows, employees, publications):
+    names = defaultdict(list)
+    by_employee = defaultdict(list)
+    for employee in employees:
+        if normalized(employee["full_name"]):
+            names[normalized(employee["full_name"])].append(employee["id"])
+    for publication in publications:
+        by_employee[publication["employee_id"]].append(dict(publication))
+    actions, audit, seen = [], [], set()
+    for number, row in rows:
+        title, identifier = row["Название публикации"], doi(row["DOI"])
+        author_names = sorted(set(normalized(a) for a in row["Авторы НИУ ВШЭ"].split(";")) - {""})
+        year = int(row["Год"]) if row["Год"].isdigit() else None
+        if not title or not year or not 1000 <= year <= 9999 or not author_names or (identifier and not re.fullmatch(r"10\.\d{4,9}/\S+", identifier)):
+            audit.append({"row": number, "employee_id": None, "result": "skip", "reason": "invalid_fields"})
+            continue
+        for author in author_names:
+            ids = names.get(author, [])
+            entry = {"row": number, "employee_id": ids[0] if len(ids) == 1 else None}
+            def skip(reason):
+                audit.append({**entry, "result": "skip", "reason": reason})
+            if len(ids) != 1:
+                skip("employee_not_found" if not ids else "ambiguous_employee")
+                continue
+            employee_id = ids[0]
+            candidates = by_employee[employee_id]
+            exact = [p for p in candidates if identifier and doi(p.get("doi_url")) == identifier]
+            title_matches = [p for p in candidates if normalized(p["title"]) == normalized(title) and p.get("year") in (None, year)]
+            matches = {p["id"]: p for p in exact + title_matches}
+            if len(matches) > 1:
+                skip("ambiguous_publication")
+                continue
+            existing = next(iter(matches.values()), None)
+            if existing and identifier and doi(existing.get("doi_url")) and doi(existing["doi_url"]) != identifier:
+                skip("doi_conflict")
+                continue
+            if existing and existing.get("year") and existing["year"] != year:
+                skip("year_conflict")
+                continue
+            key = (employee_id, identifier or (normalized(title), year))
+            duplicate = key in seen
+            seen.add(key)
+            metadata = {k: v for k, v in row.items() if v}
+            raw = existing.get("raw_data") if existing else None
+            if raw is not None and not isinstance(raw, dict):
+                skip("unsupported_raw_data")
+                continue
+            raw = dict(raw or {})
+            old_metadata = raw.get("csv_enrichment", {})
+            if not isinstance(old_metadata, dict):
+                skip("unsupported_csv_metadata")
+                continue
+            conflicts = sorted(k for k, v in metadata.items() if old_metadata.get(k) and old_metadata[k] != v)
+            if conflicts:
+                entry["conflict_fields"] = conflicts
+            merged = dict(old_metadata)
+            for k, v in metadata.items():
+                if not merged.get(k):
+                    merged[k] = v
+            raw["csv_enrichment"] = merged
+            values = {"raw_data": raw}
+            if identifier and (not existing or not existing.get("doi_url")):
+                values["doi_url"] = "https://doi.org/" + identifier
+            if not existing or not existing.get("year"):
+                values["year"] = year
+            if not existing or not existing.get("title"):
+                values["title"] = title
+            if existing:
+                values = {k: v for k, v in values.items() if existing.get(k) != v}
+                if not values:
+                    skip("metadata_conflict_first_value_kept" if conflicts else "csv_duplicate" if duplicate else "already_present")
+                    continue
+                if isinstance(existing["id"], str):
+                    actions[int(existing["id"].split("-")[1])]["values"].update(values)
+                    existing.update(values)
+                    skip("metadata_conflict_first_value_kept" if conflicts else "merged_csv_duplicate")
+                    continue
+                action = {"result": "update", "id": existing["id"], "values": values}
+                existing.update(values)
+            else:
+                source_hash = hashlib.sha256(json.dumps(key, ensure_ascii=False, sort_keys=True).encode()).hexdigest()
+                values.update(employee_id=employee_id, source_hash=source_hash)
+                action = {"result": "insert", "values": values}
+                by_employee[employee_id].append({**values, "id": f"planned-{len(actions)}"})
+            actions.append(action)
+            audit.append({**entry, "result": action["result"], "reason": "metadata_conflict_first_value_kept" if conflicts else ""})
+    return actions, audit
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("file", type=Path)
+    parser.add_argument("--apply", action="store_true", help="Commit; default is read-only dry-run")
+    parser.add_argument("--report", type=Path, required=True)
+    args = parser.parse_args()
+    pending_report = args.report.with_suffix(args.report.suffix + ".pending")
+    if args.file.resolve() in (args.report.resolve(), pending_report.resolve()):
+        raise ValueError("Report must not overwrite input")
+    source_digest = hashlib.sha256(args.file.read_bytes()).hexdigest()
+    rows = read_rows(args.file)
+    # libpq environment/service only; never load application settings or .env.
+    import psycopg
+    from psycopg.rows import dict_row
+    from psycopg.types.json import Jsonb
+    from psycopg import sql
+
+    # Check report access before any possible commit; pending file is not a success report.
+    with pending_report.open("x", encoding="utf-8"):
+        pass
+    with psycopg.connect("", row_factory=dict_row) as connection:
+        connection.execute("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE" + ("" if args.apply else " READ ONLY"))
+        connection.execute("SET LOCAL search_path = pg_catalog, public")
+        connection.execute("SET LOCAL statement_timeout = '60s'")
+        connection.execute("SET LOCAL lock_timeout = '5s'")
+        if args.apply:
+            locked = connection.execute("SELECT pg_try_advisory_xact_lock(716203841)").fetchone()
+            if not next(iter(locked.values())):
+                raise RuntimeError("Another import is running")
+        employees = connection.execute("SELECT id, full_name FROM public.employees").fetchall()
+        publications = connection.execute("SELECT id, employee_id, title, year, doi_url, raw_data FROM public.employee_publications").fetchall()
+        actions, audit = plan(rows, employees, publications)
+        if args.apply:
+            for action in actions:
+                values = dict(action["values"])
+                if "raw_data" in values:
+                    values["raw_data"] = Jsonb(values["raw_data"])
+                columns = list(values)
+                if action["result"] == "insert":
+                    query = sql.SQL("INSERT INTO public.employee_publications ({}, created_at, updated_at) VALUES ({}, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)").format(
+                        sql.SQL(", ").join(map(sql.Identifier, columns)), sql.SQL(", ").join(sql.Placeholder() for _ in columns))
+                    connection.execute(query, list(values.values()))
+                else:
+                    query = sql.SQL("UPDATE public.employee_publications SET {}, updated_at = CURRENT_TIMESTAMP WHERE id = %s").format(
+                        sql.SQL(", ").join(sql.SQL("{} = %s").format(sql.Identifier(k)) for k in columns))
+                    result = connection.execute(query, [*values.values(), action["id"]])
+                    if result.rowcount != 1:
+                        raise RuntimeError("Publication changed concurrently")
+        if hashlib.sha256(args.file.read_bytes()).hexdigest() != source_digest:
+            raise RuntimeError("Input changed during import")
+        summary = {"mode": "apply" if args.apply else "dry-run", "csv_rows": len(rows),
+                   "sha256": source_digest,
+                   "counts": {key: sum(item["result"] == key for item in audit) for key in ("insert", "update", "skip")},
+                   "metadata_conflicts": sum(bool(item.get("conflict_fields")) for item in audit),
+                   "skip_reasons": dict(Counter(item["reason"] for item in audit if item["result"] == "skip"))}
+        pending_report.write_text(json.dumps({"summary": summary, "records": audit}, ensure_ascii=False, indent=2), encoding="utf-8")
+        if not args.apply:
+            connection.rollback()
+    pending_report.replace(args.report)
+    print(json.dumps(summary, ensure_ascii=False))
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception as error:
+        # Driver errors may contain connection credentials or personal data.
+        print(f"Import failed ({type(error).__name__}); transaction not partially applied. Check configuration and permissions.")
+        raise SystemExit(1)

tests/test_api.py

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

tests/test_crawler.py

@@ -1,4 +1,5 @@
-import gzip
+import gzip
+import pytest
 from datetime import datetime, timezone
 
 from app.models import (
@@ -434,7 +435,7 @@ def test_upsert_employee_saves_publications_and_reuses_existing_rows(db_session)
     assert publications[0].authors == [{"id": "1", "title_ru": "Автор"}]
 
 
-def test_upsert_employee_records_publication_errors_without_failing_employee(monkeypatch, db_session):
+def test_upsert_employee_records_publication_errors_without_failing_employee(monkeypatch, db_session):
     run = CrawlRun(source_url="https://miem.hse.ru/persons", status="running")
     db_session.add(run)
     db_session.commit()
@@ -573,3 +574,117 @@ def _parsed_employee_with_news(profile_id: str) -> dict:
         }
     ]
     return parsed
+
+
+def test_csv_enrichment_survives_repeated_crawl_without_duplicate(db_session):
+    from scripts.import_publications import plan
+    from app.services.crawler import _sync_employee_publications
+
+    run = CrawlRun(source_url="https://miem.hse.ru/persons", status="running")
+    db_session.add(run)
+    db_session.commit()
+    employee, _ = _upsert_employee(db_session, run, _parsed_employee_with_publication("csv-preserved"))
+    db_session.commit()
+    existing = db_session.query(EmployeePublication).filter_by(employee_id=employee.id).one()
+    rows = [(number, {"Название публикации": title, "Год": str(year), "DOI": identifier,
+                      "Авторы НИУ ВШЭ": employee.full_name, "Квартиль scopus": "Q1"})
+            for number, title, year, identifier in (
+                (2, existing.title, existing.year, ""),
+                (3, "CSV only", 2024, "10.1234/csv-only"),
+                (4, "Later HSE", 2024, "10.1234/later"))]
+    actions, _ = plan(rows, [{"id": employee.id, "full_name": employee.full_name}],
+                      [{field: getattr(existing, field) for field in
+                        ("id", "employee_id", "title", "year", "doi_url", "raw_data")}])
+    for action in actions:
+        if action["result"] == "insert":
+            db_session.add(EmployeePublication(**action["values"]))
+        else:
+            for field, value in action["values"].items():
+                setattr(existing, field, value)
+    db_session.add(EmployeePublication(employee_id=employee.id, title="Obsolete HSE",
+                                       source_hash="obsolete", raw_data={"id": "obsolete"}))
+    db_session.commit()
+    original_id = existing.id
+    imported_ids = {p.title: p.id for p in db_session.query(EmployeePublication).all()}
+
+    parsed = _parsed_employee_with_publication("csv-preserved")
+    publication = parsed["sections"][0]["publications"][0]
+    publication.pop("year")
+    publication.pop("doi_url")
+    publication["raw_data"]["fresh"] = True
+    for _ in range(2):
+        _sync_employee_publications(db_session, employee, parsed)
+        db_session.commit()
+    remaining = db_session.query(EmployeePublication).filter_by(employee_id=employee.id).all()
+    assert len(remaining) == 3
+    assert all(p.raw_data["csv_enrichment"]["Квартиль scopus"] == "Q1" for p in remaining)
+    refreshed = db_session.get(EmployeePublication, original_id)
+    assert refreshed.raw_data["fresh"] is True
+    assert refreshed.year == 2023 and refreshed.doi_url == "https://doi.org/10.1/test"
+    assert db_session.get(EmployeePublication, imported_ids["CSV only"]) is not None
+
+    parsed["sections"][0]["publications"].append({
+        "publication_id": "later-id", "title": "Later HSE", "year": 2024,
+        "doi_url": "http://dx.doi.org/10.1234/LATER", "raw_data": {"id": "later-id"}})
+    for _ in range(2):
+        _sync_employee_publications(db_session, employee, parsed)
+        db_session.commit()
+    assert db_session.query(EmployeePublication).filter_by(employee_id=employee.id).count() == 3
+    adopted = db_session.get(EmployeePublication, imported_ids["Later HSE"])
+    assert adopted.publication_id == "later-id"
+    assert adopted.raw_data["csv_enrichment"]["Квартиль scopus"] == "Q1"
+
+
+def test_ambiguous_csv_match_rolls_back_publication_sync(db_session):
+    from app.services.crawler import _try_sync_employee_publications
+    run = CrawlRun(source_url="https://miem.hse.ru/persons", status="running")
+    db_session.add(run)
+    db_session.commit()
+    employee, _ = _upsert_employee(db_session, run, _parsed_employee("csv-ambiguous"))
+    for index in range(2):
+        db_session.add(EmployeePublication(employee_id=employee.id, title="Same", year=2024,
+            doi_url="https://doi.org/10.1234/same", source_hash=f"csv-{index}",
+            raw_data={"csv_enrichment": {"Год": "2024"}}))
+    db_session.commit()
+    parsed = {"sections": [{"type": "publications", "publications": [
+        {"publication_id": "new", "title": "New"},
+        {"publication_id": "same", "title": "Same", "doi_url": "https://doi.org/10.1234/same", "year": 2024}]}]}
+    _try_sync_employee_publications(db_session, run, employee, parsed)
+    db_session.commit()
+    assert db_session.query(EmployeePublication).filter_by(employee_id=employee.id).count() == 2
+    assert db_session.query(CrawlError).filter_by(crawl_run_id=run.id, error_type="ValueError").count() == 1
+
+
+@pytest.mark.parametrize("bound_id", [None, "hse-A"])
+@pytest.mark.parametrize("incoming_ids", [("hse-A", "hse-B"), ("hse-B", "hse-A")])
+def test_two_hse_ids_cannot_overwrite_one_csv_record(db_session, bound_id, incoming_ids):
+    from app.services.crawler import _try_sync_employee_publications
+    from app.services.admin_data import run_detail_payload
+    run = CrawlRun(source_url="https://miem.hse.ru/persons", status="running")
+    db_session.add(run)
+    db_session.commit()
+    employee, _ = _upsert_employee(db_session, run, _parsed_employee("csv-hse-id-conflict"))
+    original_raw = {"csv_enrichment": {"Год": "2024"}, "original": "keep"}
+    original_authors = [{"title_ru": "Synthetic original author"}]
+    record = EmployeePublication(employee_id=employee.id, title="Same", year=2024,
+        publication_id=bound_id, source_hash="csv-original", raw_data=original_raw,
+        authors=original_authors)
+    db_session.add(record)
+    db_session.commit()
+    original_id = record.id
+    parsed = {"sections": [{"type": "publications", "publications": [
+        {"publication_id": identifier, "title": "Same", "year": 2024,
+         "authors": [{"id": identifier}], "raw_data": {"id": identifier}}
+        for identifier in incoming_ids]}]}
+    _try_sync_employee_publications(db_session, run, employee, parsed)
+    db_session.commit()
+    db_session.expire_all()
+    kept = db_session.get(EmployeePublication, original_id)
+    assert db_session.query(EmployeePublication).filter_by(employee_id=employee.id).count() == 1
+    assert kept.publication_id == bound_id and kept.source_hash == "csv-original"
+    assert kept.raw_data == original_raw and kept.authors == original_authors
+    errors = run_detail_payload(db_session, run)["errors"]
+    assert len(errors) == 1
+    assert errors[0]["error_type"] == "ValueError"
+    assert "Ambiguous HSE IDs" in errors[0]["message"]
+    assert "hse-A" in errors[0]["message"] and "hse-B" in errors[0]["message"]

tests/test_publication_import.py

@@ -0,0 +1,233 @@
+"""Offline checks: synthetic identities only, no settings or database access."""
+from scripts.import_publications import plan, read_rows
+from pathlib import Path
+
+
+def row(title="Test", doi="10.1234/test", year="2024", authors="Test Person"):
+    return {"Название публикации": title, "DOI": doi, "Год": year,
+            "Авторы НИУ ВШЭ": authors}
+
+
+EMPLOYEES = [{"id": 1, "full_name": "Test Person"}]
+
+
+def test_repeat_and_no_new_employees():
+    rows = [(2, row()), (3, row()), (4, row(authors="Missing Person"))]
+    actions, audit = plan(rows, EMPLOYEES, [])
+    assert len(actions) == 1 and actions[0]["result"] == "insert"
+    assert [a["reason"] for a in audit] == ["", "csv_duplicate", "employee_not_found"]
+    stored = [{"id": 11, **actions[0]["values"]}]
+    repeat, _ = plan(rows, EMPLOYEES, stored)
+    assert repeat == []
+
+
+def test_ambiguity_and_conflicts():
+    assert not plan([(2, row())], EMPLOYEES * 2, [])[0]
+    stored = [{"id": 1, "employee_id": 1, "title": "Test", "year": 2024,
+               "doi_url": "https://doi.org/10.1234/other", "raw_data": {}}]
+    assert plan([(2, row())], EMPLOYEES, stored)[1][0]["reason"] == "doi_conflict"
+    stored[0]["doi_url"] = "https://doi.org/10.1234/test"
+    assert plan([(2, row(year="2025"))], EMPLOYEES, stored)[1][0]["reason"] == "year_conflict"
+    assert plan([(2, row())], EMPLOYEES, stored * 2)[0]  # same id is one identity
+    stored.append({**stored[0], "id": 2})
+    assert plan([(2, row())], EMPLOYEES, stored)[1][0]["reason"] == "ambiguous_publication"
+
+
+def test_fill_only_preserves_metadata_and_empty_values():
+    stored = [{"id": 1, "employee_id": 1, "title": "Test", "year": 2024,
+               "doi_url": None, "raw_data": {"hse": "keep", "csv_enrichment": {"extra": "keep"}}}]
+    actions, _ = plan([(2, {**row(), "extra": "", "new": "added"})], EMPLOYEES, stored)
+    values = actions[0]["values"]
+    assert set(values) == {"doi_url", "raw_data"}
+    assert values["raw_data"]["hse"] == "keep"
+    assert values["raw_data"]["csv_enrichment"]["extra"] == "keep"
+    assert values["raw_data"]["csv_enrichment"]["new"] == "added"
+    assert "new" not in stored[0]["raw_data"]["csv_enrichment"]
+
+
+def test_missing_doi_then_doi_does_not_update_synthetic_id():
+    actions, _ = plan([(2, row(doi="")), (3, row())], EMPLOYEES, [])
+    assert len(actions) == 1 and actions[0]["result"] == "insert"
+    assert actions[0]["values"]["doi_url"] == "https://doi.org/10.1234/test"
+
+
+def test_invalid_rows_and_owner_file():
+    assert not plan([(2, row(year="oops")), (3, row(doi="bad"))], EMPLOYEES, [])[0]
+    path = Path(__file__).resolve().parents[2] / "Статьи.csv"
+    if path.exists():
+        rows = read_rows(path)
+        assert len(rows) == 1797
+        actions, audit = plan(rows, [], [])
+        assert not actions and audit and all(a["result"] == "skip" for a in audit)
+
+
+def test_cli_read_only_and_atomic_failure(tmp_path, monkeypatch):
+    import sys
+    import psycopg
+    from scripts import import_publications
+
+    source = tmp_path / "input.csv"
+    source.write_text("Название публикации;Год;DOI;Авторы НИУ ВШЭ\nFirst;2024;10.1234/first;Test Person\nSecond;2024;10.1234/second;Test Person\n", encoding="utf-8")
+
+    class Result:
+        rowcount = 1
+
+        def __init__(self, rows):
+            self.rows = rows
+
+        def fetchall(self):
+            return self.rows
+
+        def fetchone(self):
+            return self.rows[0]
+
+    class Connection:
+        def __init__(self, fail=False):
+            self.queries, self.writes = [], []
+            self.fail, self.rolled_back, self.committed = fail, False, False
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, kind, *_):
+            if kind:
+                self.rollback()
+            elif not self.rolled_back:
+                self.committed = True
+
+        def rollback(self):
+            self.rolled_back = True
+            self.writes.clear()
+
+        def execute(self, query, params=None):
+            query = query if isinstance(query, str) else query.as_string()
+            self.queries.append(query)
+            if query.startswith("SELECT id, full_name"):
+                return Result(EMPLOYEES)
+            if query.startswith("SELECT id, employee_id"):
+                return Result([])
+            if "pg_try_advisory" in query:
+                return Result([{"locked": True}])
+            if query.startswith("INSERT"):
+                self.writes.append(params)
+                if self.fail and len(self.writes) == 2:
+                    raise RuntimeError("Synthetic second-write failure")
+            return Result([])
+
+    report = tmp_path / "dry.json"
+    dry = Connection()
+    monkeypatch.setattr(psycopg, "connect", lambda *a, **k: dry)
+    monkeypatch.setattr(sys, "argv", ["import_publications", str(source), "--report", str(report)])
+    import_publications.main()
+    assert dry.rolled_back and not dry.committed and not dry.writes
+    assert not any(q.startswith(("INSERT", "UPDATE", "DELETE")) for q in dry.queries)
+    assert "READ ONLY" in dry.queries[0]
+    assert report.exists()
+
+    failed = Connection(fail=True)
+    monkeypatch.setattr(psycopg, "connect", lambda *a, **k: failed)
+    report = tmp_path / "apply.json"
+    monkeypatch.setattr(sys, "argv", ["import_publications", str(source), "--report", str(report), "--apply"])
+    import pytest
+    with pytest.raises(RuntimeError):
+        import_publications.main()
+    assert failed.rolled_back and not failed.committed and not failed.writes
+    assert not report.exists()
+
+    success = Connection()
+    monkeypatch.setattr(psycopg, "connect", lambda *a, **k: success)
+    monkeypatch.setattr(sys, "argv", ["import_publications", str(source), "--report", str(tmp_path / "success.json"), "--apply"])
+    import_publications.main()
+    assert success.committed and len(success.writes) == 2
+    assert not any(q.startswith(("DELETE", "ALTER", "CREATE")) for q in success.queries)
+
+
+def test_report_cannot_overwrite_source(tmp_path, monkeypatch):
+    import sys
+    import pytest
+    from scripts import import_publications
+    source = tmp_path / "input.csv"
+    source.write_text("unchanged", encoding="utf-8")
+    monkeypatch.setattr(sys, "argv", ["import_publications", str(source), "--report", str(source)])
+    with pytest.raises(ValueError):
+        import_publications.main()
+    assert source.read_text() == "unchanged"
+
+
+def test_duplicate_rows_fill_late_metadata_and_report_conflicts():
+    rows = [(2, {**row(), "EID (Scopus)": "", "Квартиль scopus": ""}),
+            (3, {**row(), "EID (Scopus)": "new-id", "Квартиль scopus": "Q1"}),
+            (4, {**row(), "EID (Scopus)": "different-id", "Квартиль scopus": "Q2", "UT (WoS)": "late-wos"})]
+    actions, audit = plan(rows, EMPLOYEES, [])
+    assert len(actions) == 1 and actions[0]["result"] == "insert"
+    metadata = actions[0]["values"]["raw_data"]["csv_enrichment"]
+    assert metadata["EID (Scopus)"] == "new-id" and metadata["Квартиль scopus"] == "Q1"
+    assert metadata["UT (WoS)"] == "late-wos"
+    assert audit[1]["reason"] == "merged_csv_duplicate"
+    assert audit[2]["conflict_fields"] == ["EID (Scopus)", "Квартиль scopus"]
+    assert audit[2]["result"] == "skip"
+    repeated, repeated_audit = plan(rows, EMPLOYEES, [{"id": 11, **actions[0]["values"]}])
+    assert repeated == []
+    assert repeated_audit[2]["reason"] == "metadata_conflict_first_value_kept"
+    assert "different-id" not in str(audit)
+
+
+def test_cli_apply_year_only_preserves_existing_json(tmp_path, monkeypatch):
+    import json
+    import sys
+    import psycopg
+    from scripts import import_publications
+    source = tmp_path / "input.csv"
+    fields = row()
+    source.write_text(";".join(fields) + "\n" + ";".join(fields.values()) + "\n", encoding="utf-8")
+    original_raw = {"hse": "keep", "csv_enrichment": fields.copy()}
+    stored = {"id": 11, "employee_id": 1, "title": "Test", "year": None,
+              "doi_url": "https://doi.org/10.1234/test", "raw_data": original_raw}
+    actions, _ = plan([(2, fields)], EMPLOYEES, [stored])
+    assert actions[0]["values"] == {"year": 2024}
+
+    class Result:
+        rowcount = 1
+
+        def __init__(self, rows):
+            self.rows = rows
+
+        def fetchall(self):
+            return self.rows
+
+        def fetchone(self):
+            return self.rows[0]
+
+    class Connection:
+        committed = False
+        updates = []
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, error, *_):
+            self.committed = error is None
+
+        def execute(self, query, params=None):
+            query = query if isinstance(query, str) else query.as_string()
+            if query.startswith("SELECT id, full_name"):
+                return Result(EMPLOYEES)
+            if query.startswith("SELECT id, employee_id"):
+                return Result([stored])
+            if "pg_try_advisory" in query:
+                return Result([{"locked": True}])
+            if query.startswith("UPDATE"):
+                self.updates.append((query, params))
+            return Result([])
+
+    connection = Connection()
+    report = tmp_path / "report.json"
+    monkeypatch.setattr(psycopg, "connect", lambda *a, **k: connection)
+    monkeypatch.setattr(sys, "argv", ["import_publications", str(source), "--report", str(report), "--apply"])
+    import_publications.main()
+    assert connection.committed and len(connection.updates) == 1
+    query, params = connection.updates[0]
+    assert '"year" = %s' in query and "raw_data" not in query
+    assert params == [2024, 11]
+    assert stored["raw_data"] == original_raw and stored["year"] is None
+    assert json.loads(report.read_text())["summary"]["counts"] == {"insert": 0, "update": 1, "skip": 0}