From 05a62d18e651bd37a0b82256b1ea1d079ab95f34 Mon Sep 17 00:00:00 2001 From: codex Date: Mon, 5 Oct 2026 11:43:14 +0300 Subject: [PATCH 1/3] feat: prepare safe publication CSV enrichment import --- README.md | 6 +- app/version.py | 6 +- docs/publication-import.md | 283 +++++++++++++++++++++++++++++++ pyproject.toml | 2 +- scripts/import-publications-run | 25 +++ scripts/import-publications-ssh | 8 + scripts/import_publications.py | 189 +++++++++++++++++++++ tests/test_api.py | 2 +- tests/test_publication_import.py | 154 +++++++++++++++++ 9 files changed, 669 insertions(+), 6 deletions(-) create mode 100644 docs/publication-import.md create mode 100644 scripts/import-publications-run create mode 100644 scripts/import-publications-ssh create mode 100644 scripts/import_publications.py create mode 100644 tests/test_publication_import.py diff --git a/README.md b/README.md index 8c13fdc..2671c9e 100644 --- a/README.md +++ b/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.0`. Админка всегда показывает версии backend и frontend в footer. + +## Разовый импорт публикаций + +Инструкция, минимальные права временной роли и ограниченный SSH-запуск: [docs/publication-import.md](docs/publication-import.md). По умолчанию импортер выполняет dry-run; production-применение требует отдельного решения владельца. diff --git a/app/version.py b/app/version.py index 3a1f80e..d6851c2 100644 --- a/app/version.py +++ b/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.0" +FRONTEND_VERSION = "0.10.0" +BACKEND_VERSION = "0.10.0" diff --git a/docs/publication-import.md b/docs/publication-import.md new file mode 100644 index 0000000..85c8a01 --- /dev/null +++ b/docs/publication-import.md @@ -0,0 +1,283 @@ +# Разовый импорт публикаций + +Инструмент: `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`); первая +непустая метаинформация имеет приоритет. Если строка без DOI была запланирована +первой, следующая строка с DOI может дополнить ту же вставку +(`merged_csv_duplicate`). При повторном запуске уже заполненные записи дают +`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 БД.** Во время импорта владелец должен +исключить crawl и другие записи публикаций. Текущий crawler удаляет отсутствующие +в ответе ВШЭ записи и заменяет `raw_data` существующих публикаций. Поэтому +обогащение может исчезнуть после следующего crawl: сохранение сторонних данных +нужно решить отдельным изменением до возобновления обходов. Импортер не меняет +эту логику. Не выполняйте production apply до независимого ревью и решения +вопроса сохранности. Здесь не приведены команды рестарта сервисов. + +## Команды владельцу (не выполнялись) + +Ниже — подготовка разового доступа после ревью. Выполняет владелец на сервере. +Рабочая директория — проверенная копия этого 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` добавьте **только публичный** ключ +владельца с ограничениями (замените `` самим публичным ключом): + +```text +restrict,command="/usr/local/sbin/miem-publication-import-ssh" +``` + +```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/удаление данных. diff --git a/pyproject.toml b/pyproject.toml index e95cfbf..ce3adff 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "miem-workers" -version = "0.9.0" +version = "0.10.0" description = "MIEM employees parser, admin API, and web admin" requires-python = ">=3.11" dependencies = [ diff --git a/scripts/import-publications-run b/scripts/import-publications-run new file mode 100644 index 0000000..25ee9a1 --- /dev/null +++ b/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 diff --git a/scripts/import-publications-ssh b/scripts/import-publications-ssh new file mode 100644 index 0000000..26c67ca --- /dev/null +++ b/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 diff --git a/scripts/import_publications.py b/scripts/import_publications.py new file mode 100644 index 0000000..f5b77df --- /dev/null +++ b/scripts/import_publications.py @@ -0,0 +1,189 @@ +"""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)) + if key in seen: + skip("csv_duplicate") + continue + 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 + 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("already_present") + continue + if isinstance(existing["id"], str): + actions[int(existing["id"].split("-")[1])]["values"].update(values) + existing.update(values) + skip("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": ""}) + 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"]) + 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")}, + "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) diff --git a/tests/test_api.py b/tests/test_api.py index 0ce79e8..9cae74e 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.9.0" + assert response.json()["backend_version"] == "0.10.0" def test_api_employees_and_stats_require_admin_session(): diff --git a/tests/test_publication_import.py b/tests/test_publication_import.py new file mode 100644 index 0000000..07780be --- /dev/null +++ b/tests/test_publication_import.py @@ -0,0 +1,154 @@ +"""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" -- 2.54.0 From 72afb3104935635edca423db4fe7c4e1c8e6c406 Mon Sep 17 00:00:00 2001 From: codex Date: Mon, 5 Oct 2026 11:57:44 +0300 Subject: [PATCH 2/3] fix: preserve imported publication enrichment during crawls --- README.md | 2 +- app/services/crawler.py | 62 ++++++++++++++++++++++++----- app/version.py | 6 +-- docs/publication-import.md | 27 ++++++++++--- pyproject.toml | 2 +- tests/test_api.py | 2 +- tests/test_crawler.py | 81 +++++++++++++++++++++++++++++++++++++- 7 files changed, 160 insertions(+), 22 deletions(-) diff --git a/README.md b/README.md index 2671c9e..bd16256 100644 --- a/README.md +++ b/README.md @@ -141,7 +141,7 @@ node --check app/static/admin.js Тесты с данными используют временную SQLite; отдельные API smoke-тесты запускают приложение с его текущей конфигурацией. Браузерная проверка: `pip install playwright`, `python -m playwright install chromium`, затем `python tests/browser_admin.py`. Она проверяет реальный рендеринг страниц, клавиатуру, диалог, фильтры и восстановление прогресса после ошибки. Скриншоты сохраняются во временную папку; путь выводится в конце. -Версия сервиса: `0.10.0`. Админка всегда показывает версии backend и frontend в footer. +Версия сервиса: `0.10.1`. Админка всегда показывает версии backend и frontend в footer. ## Разовый импорт публикаций diff --git a/app/services/crawler.py b/app/services/crawler.py index 40074c1..6a742e8 100644 --- a/app/services/crawler.py +++ b/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,7 +484,9 @@ def _publication_payloads(parsed: dict) -> list[dict]: return publications -def _apply_publication(target: EmployeePublication, publication: dict, source_hash: str) -> None: +def _apply_publication(target: EmployeePublication, publication: dict, source_hash: str) -> None: + enrichment = _csv_enrichment(target) + preserved = {field: getattr(target, field) for field in ("title", "year", "doi_url")} target.publication_id = _clean_optional(publication.get("publication_id") or publication.get("id")) target.title = _publication_title(publication) target.year = _int_or_none(publication.get("year")) @@ -462,7 +501,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 diff --git a/app/version.py b/app/version.py index d6851c2..8b1b005 100644 --- a/app/version.py +++ b/app/version.py @@ -1,3 +1,3 @@ -APP_VERSION = "0.10.0" -FRONTEND_VERSION = "0.10.0" -BACKEND_VERSION = "0.10.0" +APP_VERSION = "0.10.1" +FRONTEND_VERSION = "0.10.1" +BACKEND_VERSION = "0.10.1" diff --git a/docs/publication-import.md b/docs/publication-import.md index 85c8a01..3f859bd 100644 --- a/docs/publication-import.md +++ b/docs/publication-import.md @@ -91,12 +91,27 @@ commit также возможна. Безопасный повторный dry- При необходимости владелец перемещает `.pending` в архив перед повтором. **До применения нужен свежий backup БД.** Во время импорта владелец должен -исключить crawl и другие записи публикаций. Текущий crawler удаляет отсутствующие -в ответе ВШЭ записи и заменяет `raw_data` существующих публикаций. Поэтому -обогащение может исчезнуть после следующего crawl: сохранение сторонних данных -нужно решить отдельным изменением до возобновления обходов. Импортер не меняет -эту логику. Не выполняйте production apply до независимого ревью и решения -вопроса сохранности. Здесь не приведены команды рестарта сервисов. +исключить конкурентные записи публикаций на время транзакции, согласовав окно +работы; эта подготовка не отключает 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-записей остаётся прежним. Здесь +не приведены команды рестарта или отключения сервисов. ## Команды владельцу (не выполнялись) diff --git a/pyproject.toml b/pyproject.toml index ce3adff..1978fae 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "miem-workers" -version = "0.10.0" +version = "0.10.1" description = "MIEM employees parser, admin API, and web admin" requires-python = ">=3.11" dependencies = [ diff --git a/tests/test_api.py b/tests/test_api.py index 9cae74e..34dff5e 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.10.0" + assert response.json()["backend_version"] == "0.10.1" def test_api_employees_and_stats_require_admin_session(): diff --git a/tests/test_crawler.py b/tests/test_crawler.py index c1c2256..7adbb6e 100644 --- a/tests/test_crawler.py +++ b/tests/test_crawler.py @@ -434,7 +434,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 +573,82 @@ 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 -- 2.54.0 From 25eb6481f79c9cfdb2779a8af1b845e955ce984a Mon Sep 17 00:00:00 2001 From: codex Date: Mon, 5 Oct 2026 12:12:35 +0300 Subject: [PATCH 3/3] fix: address publication import review findings --- README.md | 2 +- app/services/crawler.py | 8 +++- app/version.py | 6 +-- docs/publication-import.md | 22 +++++++-- pyproject.toml | 2 +- scripts/import_publications.py | 17 ++++--- tests/test_api.py | 2 +- tests/test_crawler.py | 38 ++++++++++++++- tests/test_publication_import.py | 79 ++++++++++++++++++++++++++++++++ 9 files changed, 157 insertions(+), 19 deletions(-) diff --git a/README.md b/README.md index bd16256..552b27c 100644 --- a/README.md +++ b/README.md @@ -141,7 +141,7 @@ node --check app/static/admin.js Тесты с данными используют временную SQLite; отдельные API smoke-тесты запускают приложение с его текущей конфигурацией. Браузерная проверка: `pip install playwright`, `python -m playwright install chromium`, затем `python tests/browser_admin.py`. Она проверяет реальный рендеринг страниц, клавиатуру, диалог, фильтры и восстановление прогресса после ошибки. Скриншоты сохраняются во временную папку; путь выводится в конце. -Версия сервиса: `0.10.1`. Админка всегда показывает версии backend и frontend в footer. +Версия сервиса: `0.10.2`. Админка всегда показывает версии backend и frontend в footer. ## Разовый импорт публикаций diff --git a/app/services/crawler.py b/app/services/crawler.py index 6a742e8..8c2570d 100644 --- a/app/services/crawler.py +++ b/app/services/crawler.py @@ -486,8 +486,14 @@ def _publication_payloads(parsed: dict) -> list[dict]: 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 = _clean_optional(publication.get("publication_id") or publication.get("id")) + 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")) diff --git a/app/version.py b/app/version.py index 8b1b005..5f2e035 100644 --- a/app/version.py +++ b/app/version.py @@ -1,3 +1,3 @@ -APP_VERSION = "0.10.1" -FRONTEND_VERSION = "0.10.1" -BACKEND_VERSION = "0.10.1" +APP_VERSION = "0.10.2" +FRONTEND_VERSION = "0.10.2" +BACKEND_VERSION = "0.10.2" diff --git a/docs/publication-import.md b/docs/publication-import.md index 3f859bd..0224232 100644 --- a/docs/publication-import.md +++ b/docs/publication-import.md @@ -31,10 +31,16 @@ SHA256 `1f236be56d99677b1403df29a4c1d3a7e6d10cb90bd355d678201a027921b0ce`. удаляются. Одна статья у разных сотрудников — предусмотренные моделью авторские связи, а не дубли внутри карточки. -Повтор CSV внутри сотрудника не создаёт записи (`csv_duplicate`); первая -непустая метаинформация имеет приоритет. Если строка без DOI была запланирована -первой, следующая строка с DOI может дополнить ту же вставку -(`merged_csv_duplicate`). При повторном запуске уже заполненные записи дают +Повтор CSV внутри сотрудника не создаёт записи. Каждая поздняя строка дополняет +ещё пустые метаданные; первый непустой источник имеет приоритет, в том числе +ранее сохранённое значение БД. Повтор без изменений — `csv_duplicate`, дополнение +запланированной вставки — `merged_csv_duplicate`. При конфликте непустых +метаданных первое значение сохраняется; остальные пустые поля всё равно +дополняются. Отчёт содержит `metadata_conflict_first_value_kept`, список +`conflict_fields` (только названия полей, без значений) и сводный счётчик +`metadata_conflicts`. Конфликт DOI/года идентификации по-прежнему пропускает +всю связь. Если строка без DOI была запланирована первой, следующая строка с +DOI может дополнить ту же вставку. При повторном запуске уже заполненные записи дают `already_present`. Хеш новой записи детерминирован по сотруднику и DOI либо названию/году. `source_hash` существующих записей не меняется. @@ -113,6 +119,14 @@ title/year/doi_url, если ВШЭ вернул пустые значения. обрабатываться. Удаление обычных устаревших HSE-записей остаётся прежним. Здесь не приведены команды рестарта или отключения сервисов. +Уже установленный HSE publication_id CSV-записи нельзя заменить другим HSE ID, +даже когда DOI/название/год совпадают. Это проверяется перед изменением полей +при любом пути сопоставления (ID, source_hash, CSV-поля). Поэтому две HSE-статьи +не могут по очереди перезаписать одну CSV-запись: весь savepoint откатывается +независимо от порядка статей. Отсутствующий ID во входе не очищает существующую +CSV-привязку. Отчёт обхода (`errors` в деталях запуска) содержит CrawlError с +идентификатором записи БД и конфликтующими HSE ID, без ФИО и названий статей. + ## Команды владельцу (не выполнялись) Ниже — подготовка разового доступа после ревью. Выполняет владелец на сервере. diff --git a/pyproject.toml b/pyproject.toml index 1978fae..a7b6182 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "miem-workers" -version = "0.10.1" +version = "0.10.2" description = "MIEM employees parser, admin API, and web admin" requires-python = ">=3.11" dependencies = [ diff --git a/scripts/import_publications.py b/scripts/import_publications.py index f5b77df..b89b3bd 100644 --- a/scripts/import_publications.py +++ b/scripts/import_publications.py @@ -72,9 +72,7 @@ def plan(rows, employees, publications): skip("year_conflict") continue key = (employee_id, identifier or (normalized(title), year)) - if key in seen: - skip("csv_duplicate") - continue + 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 @@ -86,6 +84,9 @@ def plan(rows, employees, publications): 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): @@ -101,12 +102,12 @@ def plan(rows, employees, publications): if existing: values = {k: v for k, v in values.items() if existing.get(k) != v} if not values: - skip("already_present") + 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("merged_csv_duplicate") + skip("metadata_conflict_first_value_kept" if conflicts else "merged_csv_duplicate") continue action = {"result": "update", "id": existing["id"], "values": values} existing.update(values) @@ -116,7 +117,7 @@ def plan(rows, employees, publications): 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": ""}) + audit.append({**entry, "result": action["result"], "reason": "metadata_conflict_first_value_kept" if conflicts else ""}) return actions, audit @@ -155,7 +156,8 @@ def main(): if args.apply: for action in actions: values = dict(action["values"]) - values["raw_data"] = Jsonb(values["raw_data"]) + 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( @@ -172,6 +174,7 @@ def main(): 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: diff --git a/tests/test_api.py b/tests/test_api.py index 34dff5e..6ddf9f5 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.10.1" + assert response.json()["backend_version"] == "0.10.2" def test_api_employees_and_stats_require_admin_session(): diff --git a/tests/test_crawler.py b/tests/test_crawler.py index 7adbb6e..d5b1476 100644 --- a/tests/test_crawler.py +++ b/tests/test_crawler.py @@ -1,4 +1,5 @@ -import gzip +import gzip +import pytest from datetime import datetime, timezone from app.models import ( @@ -652,3 +653,38 @@ def test_ambiguous_csv_match_rolls_back_publication_sync(db_session): 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"] diff --git a/tests/test_publication_import.py b/tests/test_publication_import.py index 07780be..b1f41d6 100644 --- a/tests/test_publication_import.py +++ b/tests/test_publication_import.py @@ -152,3 +152,82 @@ def test_report_cannot_overwrite_source(tmp_path, monkeypatch): 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} -- 2.54.0