feat: add council employee sorting by name and publication counts

Reviewed-on: #57
Reviewed-by: admin <av_progs@mail.ru>

.gitea/workflows/ci.yml

@@ -6,6 +6,11 @@ on:
   pull_request:
     branches: [main]
   workflow_dispatch:
+    inputs:
+      diagnostic_only:
+        description: 'Print key fingerprint; production deployment is temporarily disabled'
+        type: boolean
+        default: true
 
 jobs:
   test:
@@ -27,7 +32,9 @@ jobs:
       - run: sh -n infra/install-postgres-backup-cron.sh
 
   deploy:
-    if: (github.event_name == 'push' && github.ref == 'refs/heads/main') || github.event_name == 'workflow_dispatch'
+    # Production deployment is disabled until separate Workers deployment access is installed.
+    name: Fingerprint diagnostics (production deployment temporarily disabled)
+    if: github.event_name == 'workflow_dispatch' && github.event.inputs.diagnostic_only == 'true'
     needs: test
     runs-on: ubuntu-latest
     concurrency:
@@ -42,6 +49,7 @@ jobs:
         id: prepare
         shell: bash
         run: |
+          set +x
           set -euo pipefail
           backend_version="$(sed -n 's/^BACKEND_VERSION = "\(.*\)"$/\1/p' app/version.py)"
           case "$backend_version" in
@@ -55,7 +63,15 @@ jobs:
         env:
           PROD_SSH_KEY: ${{ secrets.PROD_SSH_KEY }}
           PROD_KNOWN_HOSTS: ${{ secrets.PROD_KNOWN_HOSTS }}
+      - name: Deployment key fingerprint
+        shell: bash
+        run: |
+          set +x
+          set -euo pipefail
+          trap 'rm -f "$RUNNER_TEMP/deploy_key" "$RUNNER_TEMP/known_hosts"' EXIT
+          ssh-keygen -y -P '' -f "$RUNNER_TEMP/deploy_key" | ssh-keygen -lf - -E sha256 | awk '{print $2}'
       - name: Upload release
+        if: ${{ false }}
         shell: bash
         run: |
           set -euo pipefail
@@ -68,6 +84,7 @@ jobs:
           PROD_HOST: ${{ secrets.PROD_HOST }}
           PROD_USER: ${{ secrets.PROD_USER }}
       - name: Deploy and verify
+        if: ${{ false }}
         shell: bash
         run: |
           set -euo pipefail

.gitignore

@@ -9,3 +9,5 @@ pytest-cache-files-*/
 htmlcov/
 postgres_data/
 MCP_DESCRIPTION.md
+
+.impeccable/review/

README.md

@@ -88,7 +88,9 @@ Compose запускает `api`, `worker` и `postgres`. API привязан
 
 ## Кандидаты в диссертационный совет
 
-Ссылка в админке открывает `/admin/council`. Страница использует действующих сотрудников и докторские степени из их профилей, а публикации — из предоставленной выгрузки `app/data/council_articles.csv`. Публикации сопоставляются только по полному ФИО при единственном совпадении; повторный DOI (или одинаковые название и год при отсутствии DOI) учитывается один раз. Порог — 10 предварительно подходящих публикаций за скользящие пять лет от текущей даты по Москве.
+Страница `/admin/council` доступна без авторизации. Она показывает сотрудников с докторской степенью в профиле и хотя бы одной публикацией в CSV или таблице `employee_publications`. Порог количества работ не применяется; сотрудники без степени, с кандидатской степенью или только PhD в список не включаются; уволенные сотрудники отмечены отдельно. Есть поиск по ФИО, сортировка по ФИО и каждому из трёх счётчиков публикаций в обоих направлениях и страницы по 25 сотрудников. Сортировка применяется ко всему списку и сохраняется при поиске и переходе между страницами. У каждого сотрудника видны степень, число уникальных работ из CSV с Q1/Q2 за скользящие пять лет, число уникальных работ в БД и общий перечень без дублей с источниками, типами и подтверждёнными критериями.
+
+CSV `app/data/council_articles.csv` сопоставляется по полному ФИО только при единственном совпадении. Публикации объединяются по DOI, затем по нормализованным названию и году, если нет конфликтующих DOI. Совпадение в двух источниках сохраняет оба источника, но учитывается один раз в общем количестве. Импорт CSV в БД не выдаётся за данные сайта. CSV не содержит типа материала, точной даты и подтверждения К1/RSCI: неизвестные сведения обозначены явно; счётчик CSV предварительный. Публичная страница не открывает контакты, служебные данные или операции админки.
 
 Список **предварительный**: в CSV указаны год и квартиль Scopus/WoS, но нет точной даты, типа материала, журнала, источника квартиля, подтверждения К1 и RSCI. На границах периода год не позволяет проверить точную дату. Статьи с Q1/Q2 учитываются с пометкой источника; отсутствие кандидата не доказывает недостаточную публикационную активность. Для подтверждённого отбора нужны проверка журналов и дат и дополнительные источники публикаций. Выгрузка поставляется вместе с релизом и обновляется отдельным изменением репозитория.
 
@@ -128,7 +130,9 @@ docker compose exec postgres pg_dump -U miem miem_workers > backup.sql
 docker compose down
 ```
 
-Production deploy выполняется job `deploy` workflow `CI` после успешного `test` на push в `main` или вручную. Для него нужны Actions secrets `PROD_HOST`, `PROD_USER`, `PROD_SSH_KEY` и `PROD_KNOWN_HOSTS`; `.env` и ключи в репозиторий не добавляются. PostgreSQL backup выполняется на VPS раз в неделю user cron deploy-пользователя через `infra/postgres-backup.sh`: `pg_dumpall` сохраняет весь кластер в `~/backups/postgres/postgres-latest.sql`. Новый dump сначала создаётся во временный файл и заменяет старый только после успешного завершения; предыдущие dump-файлы `miem_workers` удаляются после успешного создания нового. Одновременно выполняется только один production deploy.
+Production deployment временно отключён до установки отдельного deployment-доступа Workers. Workflow `CI` запускает тесты на push в `main` и pull_request в `main`; шаги Upload release и Deploy and verify безусловно отключены для всех событий и значений input. Стандартные имена deployment secrets: `PROD_HOST`, `PROD_USER`, `PROD_SSH_KEY`, `PROD_KNOWN_HOSTS`; `.env` и ключи в репозиторий не добавляются. PostgreSQL backup выполняется на VPS раз в неделю user cron deploy-пользователя через `infra/postgres-backup.sh`: `pg_dumpall` сохраняет весь кластер в `~/backups/postgres/postgres-latest.sql`. Новый dump сначала создаётся во временный файл и заменяет старый только после успешного завершения; предыдущие dump-файлы `miem_workers` удаляются после успешного создания нового. Эта правка не меняет существующий backup cron.
+
+Ручной запуск с `diagnostic_only=true` (по умолчанию) выводит только SHA256 fingerprint ключа `PROD_SSH_KEY`, без подключения к VPS. Публичный ключ передаётся через pipe, не сохраняется в файл или artifacts; shell tracing отключён. При `diagnostic_only=false` диагностический job пропускается: это значение также не разрешает deployment. Возврат production deployment требует отдельной reviewed правки после установки нового доступа.
 
 ## Проверки
 
@@ -141,8 +145,12 @@ node --check app/static/admin.js
 
 Тесты с данными используют временную SQLite; отдельные API smoke-тесты запускают приложение с его текущей конфигурацией. Браузерная проверка: `pip install playwright`, `python -m playwright install chromium`, затем `python tests/browser_admin.py`. Она проверяет реальный рендеринг страниц, клавиатуру, диалог, фильтры и восстановление прогресса после ошибки. Скриншоты сохраняются во временную папку; путь выводится в конце.
 
-Версия сервиса: `0.10.2`. Админка всегда показывает версии backend и frontend в footer.
+Версия сервиса: `0.10.6`. Админка всегда показывает версии backend и frontend в footer.
 
 ## Разовый импорт публикаций
 
 Инструкция, минимальные права временной роли и ограниченный SSH-запуск: [docs/publication-import.md](docs/publication-import.md). По умолчанию импортер выполняет dry-run; production-применение требует отдельного решения владельца.
+
+Интеграционная проверка импорта на отдельном PostgreSQL: [docs/postgresql-import-verification.md](docs/postgresql-import-verification.md).
+
+Отдельный тестовый стенд на VPS и read-only проверка ресурсов: [docs/postgresql-import-vps.md](docs/postgresql-import-vps.md). Стенд запускает только владелец после оценки ресурсов.

app/admin.py

@@ -134,14 +134,14 @@ def directory(
 
 
 @router.get("/council", response_class=HTMLResponse)
-def council(request: Request, db: Session = Depends(get_db), settings: Settings = Depends(get_settings)):
-    require_admin(request, settings)
+def council(request: Request, q: str = "", offset: int = 0, sort: str = "csv_count", direction: str = "desc",
+            db: Session = Depends(get_db)):
     try:
-        result = council_candidates(db)
+        result = council_candidates(db, q=q, offset=offset, sort=sort, direction=direction)
     except (SQLAlchemyError, OSError, ValueError, KeyError):
         db.rollback()
-        return _render(request, "council.html", {"error": True}, status_code=503)
-    return _render(request, "council.html", {"error": False, **result})
+        return _render(request, "council.html", {"error": True, "public_page": True}, status_code=503)
+    return _render(request, "council.html", {"error": False, "public_page": True, **result})
 
 
 @router.get("/employees/{employee_id}", response_class=HTMLResponse)

app/services/council.py

@@ -1,17 +1,18 @@
-"""Preliminary council candidates from the supplied publication export."""
+"""Public publication evidence from the council export and employee database."""
 
 import calendar
 import csv
 import re
+import unicodedata
 from collections import Counter, defaultdict
 from datetime import date, datetime
 from functools import lru_cache
 from pathlib import Path
-from urllib.parse import quote
+from urllib.parse import quote, urlsplit
 from zoneinfo import ZoneInfo
 
 from sqlalchemy import select
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import Session, selectinload
 
 from app.models import Employee
 from app.services.academic_degrees import academic_degrees
@@ -20,18 +21,28 @@ from app.services.academic_degrees import academic_degrees
 EXPORT = Path(__file__).resolve().parents[1] / "data" / "council_articles.csv"
 DOCTOR = re.compile(r"^доктор\b.+\bнаук$", re.IGNORECASE)
 DOI = re.compile(r"10\.\d{4,9}/\S+", re.IGNORECASE)
+TYPES = {"ARTICLE": "Статья", "BOOK": "Книга", "CHAPTER": "Глава книги",
+         "BOOK_CHAPTER": "Глава книги", "REPORT": "Доклад", "PATENT": "Патент",
+         "PREPRINT": "Препринт", "CONFERENCE_PAPER": "Материал конференции"}
 
 
 def _name(value: str) -> str:
-    return " ".join(value.casefold().replace("ё", "е").split())
+    return " ".join(unicodedata.normalize("NFKC", value).casefold().replace("ё", "е").split())
 
 
 def _doi(value: str) -> str:
-    value = value.strip().lower()
-    for prefix in ("https://doi.org/", "http://doi.org/", "doi:"):
-        if value.startswith(prefix):
-            value = value[len(prefix):]
-    return value
+    value = re.sub(r"^(?:https?://(?:dx\.)?doi\.org/|doi:\s*)", "", value.strip(), flags=re.I).lower()
+    return value if DOI.fullmatch(value) else ""
+
+
+def _safe_url(value: str | None) -> str | None:
+    try:
+        url = urlsplit(value or "")
+        if url.scheme in {"http", "https"} and url.hostname and not url.username and not url.password:
+            return value
+    except ValueError:
+        pass
+    return None
 
 
 def period(today: date) -> tuple[date, date]:
@@ -39,79 +50,142 @@ def period(today: date) -> tuple[date, date]:
     return date(year, today.month, min(today.day, calendar.monthrange(year, today.month)[1])), today
 
 
+def _quartiles(row: dict) -> list[str]:
+    return [f"{label} {str(row.get(field) or '').strip().upper()}"
+            for label, field in (("Scopus", "Квартиль scopus"), ("Web of Science", "Квартиль wos"))
+            if str(row.get(field) or '').strip().upper() in {"Q1", "Q2"}]
+
+
 @lru_cache(maxsize=1)
 def _articles_by_author() -> dict[str, list[dict]]:
     articles = defaultdict(list)
     with EXPORT.open(encoding="utf-8-sig", newline="") as source:
-        for row_number, row in enumerate(csv.DictReader(source, delimiter=";"), 2):
-            sources = []
-            for label, quartile, identifier in (
-                ("Scopus", "Квартиль scopus", "EID (Scopus)"),
-                ("Web of Science", "Квартиль wos", "UT (WoS)"),
-            ):
-                value = (row[quartile] or "").strip().upper()
-                if value in {"Q1", "Q2"}:
-                    sources.append({"name": label, "quartile": value, "identifier": (row[identifier] or "").strip()})
-            if not sources:
-                continue
-            try:
-                year = int(row["Год"])
-            except (TypeError, ValueError):
-                continue
-            title = (row["Название публикации"] or "").strip()
+        reader = csv.DictReader(source, delimiter=";")
+        required = {"Название публикации", "Год", "DOI", "Авторы НИУ ВШЭ", "Квартиль scopus", "Квартиль wos"}
+        if not required <= set(reader.fieldnames or []):
+            raise ValueError("Missing council CSV headers")
+        for row_number, row in enumerate(reader, 2):
+            if None in row or any(value is None for value in row.values()):
+                raise ValueError("Malformed council CSV record")
+            title = row["Название публикации"].strip()
             if not title:
                 continue
-            doi = _doi(row["DOI"] or "")
-            article = {
-                "title": title,
-                "year": year,
-                "doi": doi,
-                "doi_url": f"https://doi.org/{quote(doi, safe='/')}" if DOI.fullmatch(doi) else None,
-                "sources": sources,
-                "row_number": row_number,
-            }
-            for author in set(_name(value) for value in (row["Авторы НИУ ВШЭ"] or "").split(";")) - {""}:
+            year = int(row["Год"]) if row["Год"].strip().isdigit() else None
+            article = {"title": title, "year": year, "doi": _doi(row["DOI"]),
+                       "types": [], "quartiles": _quartiles(row), "origins": ["Файл CSV"],
+                       "rows": [row_number], "in_csv": True, "in_db": False, "url": None}
+            for author in set(_name(value) for value in row["Авторы НИУ ВШЭ"].split(";")) - {""}:
                 articles[author].append(article)
     return articles
 
 
+def _database_article(publication) -> dict:
+    raw = publication.raw_data if isinstance(publication.raw_data, dict) else {}
+    imported = raw.get("csv_enrichment")
+    from_site = bool(publication.publication_id or publication.url or any(key != "csv_enrichment" for key in raw))
+    origins = ["БД miem-workers · сайт ВШЭ"] if from_site else []
+    if imported:
+        origins.append("БД miem-workers · импорт CSV")
+    kind = publication.publication_type
+    return {"title": publication.title, "year": publication.year, "doi": _doi(publication.doi_url or ""),
+            "types": [TYPES.get(kind.upper(), f"Другой тип: {kind}")] if kind else [],
+            "quartiles": _quartiles(imported) if isinstance(imported, dict) else [],
+            "origins": origins or ["БД miem-workers"], "rows": [], "in_csv": False, "in_db": True,
+            "url": _safe_url(publication.url)}
+
+
 def _unique_articles(articles: list[dict], start: date, end: date) -> list[dict]:
+    title_years = defaultdict(set)
+    for article in articles:
+        if article["year"] is not None:
+            title_years[_name(article["title"])].add(article["year"])
+    def title_key(article):
+        title = _name(article["title"])
+        years = title_years[title]
+        year = article["year"]
+        return title, next(iter(years)) if year is None and len(years) == 1 else year
+
+    # A title/year fallback is safe only when it identifies at most one DOI.
+    title_dois = defaultdict(set)
+    for article in articles:
+        if article["doi"]:
+            title_dois[title_key(article)].add(article["doi"])
     unique = {}
     for article in articles:
-        if not start.year <= article["year"] <= end.year:
-            continue
-        key = article["doi"] or (_name(article["title"]), article["year"])
+        fallback = title_key(article)
+        identifiers = title_dois[fallback]
+        identifier = article["doi"] or (next(iter(identifiers)) if len(identifiers) == 1 else "")
+        key = identifier or fallback
         if key not in unique:
-            unique[key] = {**article, "sources": list(article["sources"])}
+            unique[key] = {**article, "doi": identifier,
+                           **{field: list(article[field]) for field in ("types", "quartiles", "origins", "rows")},
+                           "csv_qualifies": False}
+        merged = unique[key]
+        for field in ("types", "quartiles", "origins", "rows"):
+            merged[field] = list(dict.fromkeys([*merged[field], *article[field]]))
+        merged["in_csv"] |= article["in_csv"]
+        merged["in_db"] |= article["in_db"]
+        merged["url"] = merged["url"] or article["url"]
+        merged["year"] = merged["year"] or article["year"]
+        in_period = article["year"] is not None and start.year <= article["year"] <= end.year
+        merged["csv_qualifies"] |= bool(article["in_csv"] and article["quartiles"] and in_period)
+    for article in unique.values():
+        year = article["year"]
+        article["criteria"] = list(article["quartiles"])
+        if year is None:
+            article["criteria"].append("Год не указан — период не подтверждён")
+        elif start.year <= year <= end.year:
+            article["criteria"].append("В пределах пяти лет по году; точную дату нужно проверить" if year in {start.year, end.year}
+                                       else "В пределах пяти лет")
         else:
-            known = {item["name"] for item in unique[key]["sources"]}
-            unique[key]["sources"].extend(
-                item for item in article["sources"] if item["name"] not in known
-            )
-    return sorted(unique.values(), key=lambda item: (-len(item["sources"]), -item["year"], item["title"].casefold()))
+            article["criteria"].append("Вне пятилетнего периода")
+        if not article["quartiles"]:
+            article["criteria"].append("Q1/Q2, К1 и RSCI не подтверждены доступными данными")
+        article["doi_url"] = f"https://doi.org/{quote(article['doi'], safe='/')}" if article["doi"] else None
+        article["type_label"] = " / ".join(article["types"]) or "Тип не указан — статья не подтверждена"
+        if article["types"] and article["types"] != ["Статья"]:
+            article["csv_qualifies"] = False
+            article["criteria"].append("Тип материала не подтверждён как статья — не включён в счётчик CSV")
+    return sorted(unique.values(), key=lambda item: (not item["csv_qualifies"], -(item["year"] or 0), item["title"].casefold()))
 
 
-def council_candidates(db: Session, today: date | None = None) -> dict:
+def council_candidates(db: Session, today: date | None = None, q: str = "", offset: int = 0,
+                       sort: str = "csv_count", direction: str = "desc") -> dict:
     today = today or datetime.now(ZoneInfo("Europe/Moscow")).date()
     start, end = period(today)
-    employees = db.scalars(select(Employee).where(Employee.status == "active")).all()
+    employees = db.scalars(select(Employee).options(selectinload(Employee.publications))).all()
     name_counts = Counter(_name(employee.full_name or "") for employee in employees)
     by_author = _articles_by_author()
     candidates = []
     for employee in employees:
         author = _name(employee.full_name or "")
-        if not author or name_counts[author] != 1:
+        ambiguous = bool(author and name_counts[author] > 1 and by_author.get(author))
+        csv_articles = by_author.get(author, []) if author and name_counts[author] == 1 else []
+        articles = _unique_articles([*csv_articles, *(_database_article(p) for p in employee.publications)], start, end)
+        if not articles:
             continue
+        data = employee.current_data if isinstance(employee.current_data, dict) else {}
         try:
-            degrees = academic_degrees(employee.current_data if isinstance(employee.current_data, dict) else {})
+            degrees = academic_degrees(data)
         except (AttributeError, TypeError):
+            degrees = []
+        if not any(DOCTOR.fullmatch(degree) for degree in degrees):
             continue
-        degree = next((value for value in degrees if DOCTOR.fullmatch(value)), None)
-        if not degree:
-            continue
-        articles = _unique_articles(by_author.get(author, []), start, end)
-        if len(articles) >= 10:
-            candidates.append({"id": employee.id, "name": employee.full_name, "degree": degree,
-                               "count": len(articles), "articles": articles})
-    candidates.sort(key=lambda item: (-item["count"], item["name"].casefold()))
-    return {"candidates": candidates, "start": start, "end": end, "total": len(candidates)}
+        candidates.append({"name": employee.full_name or "ФИО не указано", "degree": "; ".join(degrees) or "Учёная степень не указана",
+                           "count": len(articles), "csv_count": sum(a["csv_qualifies"] for a in articles),
+                           "db_count": sum(a["in_db"] for a in articles), "articles": articles,
+                           "profile_url": _safe_url(employee.canonical_url), "ambiguous": ambiguous,
+                           "dismissed": employee.status == "dismissed"})
+    sort = sort if sort in {"full_name", "csv_count", "db_count", "count"} else "csv_count"
+    direction = direction if direction in {"asc", "desc"} else "desc"
+    candidates.sort(key=lambda item: _name(item["name"]), reverse=sort == "full_name" and direction == "desc")
+    if sort != "full_name":
+        candidates.sort(key=lambda item: (item[sort], item["count"]), reverse=direction == "desc")
+    total = len(candidates)
+    query = _name(q)
+    candidates = [item for item in candidates if query in _name(item["name"])]
+    matched = len(candidates)
+    limit = 25
+    offset = min(max(0, offset), ((matched - 1) // limit) * limit) if matched else 0
+    return {"candidates": candidates[offset:offset + limit], "start": start, "end": end, "total": total,
+            "matched": matched, "q": q, "offset": offset, "limit": limit, "sort": sort, "direction": direction}

app/static/admin.css

@@ -819,6 +819,7 @@
 
 .council {
   display: grid;
+  grid-template-columns: minmax(0, 1fr);
   gap: 24px;
   color: #25313a;
 }
@@ -888,6 +889,29 @@
   .council__total strong { display: inline; margin-inline-end: 8px; font-size: 28px; }
   .council__criteria { grid-template-columns: 1fr; gap: 14px; }
   .council__candidate-summary { grid-template-columns: minmax(0, 1fr) auto 14px; gap: 10px; padding: 14px; }
-  .council__candidate-count small { max-width: 72px; }
+  .council__candidate-summary .council__candidate-identity { grid-column: 1; }
+  .council__candidate-cue { grid-column: 2; grid-row: 1; }
+  .council__candidate-summary { grid-template-columns: minmax(0, 1fr) 14px; }
+  .council__counts { grid-column: 1 / -1; justify-content: space-between; }
+  .council__candidate-count { text-align: start; }
   .council__candidate-body { padding: 14px; }
 }
+
+.council__search { display: flex; align-items: end; flex-wrap: wrap; gap: 12px; }
+.council__search .directory__field { flex: 1 1 240px; max-width: 520px; }
+.council__notice summary { cursor: pointer; padding-block: 6px; }
+.council__notice p { max-width: 75ch; margin-bottom: 0; }
+.council__counts { display: grid; grid-template-columns: repeat(3, minmax(0, 1fr)); gap: clamp(12px, 3vw, 36px); }
+.council__candidate-count small { max-width: 105px; line-height: 1.4; }
+.council__hint { margin: 0; color: #4c5b64; font-size: 14px; line-height: 1.5; }
+.council__candidate-body .council__hint { margin-top: 12px; }
+.council__evidence { display: flex; flex-wrap: wrap; gap: 4px 20px; padding-inline-start: 18px; margin: 10px 0 0; color: #4c5b64; font-size: 13px; line-height: 1.5; }
+.council__article-heading span { flex-shrink: 0; }
+.council ::selection { background: #ccfbf1; color: #25313a; }
+.council input { caret-color: #0f766e; }
+.council a { text-underline-offset: 3px; }
+.council__article { overflow-wrap: anywhere; }
+@media (max-width: 400px) {
+  .council__title { font-size: clamp(23px, 7vw, 29px); overflow-wrap: anywhere; }
+  .council__article-heading { flex-direction: column; gap: 4px; }
+}

app/templates/base.html

@@ -9,8 +9,12 @@
   <body class="admin">
     <a class="admin__skip-link" href="#main-content">Перейти к содержимому</a>
     <header class="admin__header">
-      <h1 class="admin__brand"><a class="admin__brand-link" href="/admin">MIEM Employees</a></h1>
+      <h1 class="admin__brand"><a class="admin__brand-link" href="{{ '/admin/council' if public_page else '/admin' }}">MIEM Employees</a></h1>
       <nav class="admin__nav">
+        {% if public_page %}
+        <a class="admin__link" href="/admin/council" aria-current="page">Диссертационный совет</a>
+        <a class="admin__link" href="/admin/login">Вход в админку</a>
+        {% else %}
         <a class="admin__link" href="/admin"{% if request.url.path == "/admin" %} aria-current="page"{% endif %}>Обзор</a>
         <a class="admin__link" href="/admin/directory"{% if request.url.path == "/admin/directory" %} aria-current="page"{% endif %}>Сотрудники</a>
         <a class="admin__link" href="/admin/council"{% if request.url.path == "/admin/council" %} aria-current="page"{% endif %}>Диссертационный совет</a>
@@ -18,6 +22,7 @@
         <form method="post" action="/admin/logout">
           <button class="button button--ghost" type="submit">Выйти</button>
         </form>
+        {% endif %}
       </nav>
     </header>
     <main class="admin__main" id="main-content">

app/templates/council.html

@@ -1,59 +1,85 @@
 {% extends "base.html" %}
-{% block title %}Кандидаты в диссертационный совет · MIEM Employees{% endblock %}
+{% block title %}Диссертационный совет · MIEM Employees{% endblock %}
 {% block content %}
-<section class="council">
+<section class="council" aria-labelledby="council-title">
   <header class="council__hero">
     <div class="council__hero-copy">
-      <h2 class="council__title">Кандидаты в диссертационный совет</h2>
-      <p class="council__intro">Сотрудники с докторской степенью в профиле и публикациями с отметками Q1 или Q2 в имеющейся выгрузке. Список помогает начать проверку состава комиссии.</p>
+      <h2 class="council__title" id="council-title">Диссертационный совет</h2>
+      <p class="council__intro">Публикации сотрудников для проверки состава совета. В списке сотрудники с докторской степенью в профиле и хотя бы одной публикацией в файле CSV или базе miem-workers. Минимальный порог количества работ не применяется.</p>
     </div>
-    {% if not error %}<p class="council__total"><strong>{{ total }}</strong> предварительно подходят</p>{% endif %}
   </header>
 
   {% if error %}
   <div class="council__message" role="alert">
-    <h3>Не удалось загрузить список</h3>
-    <p>Проверьте доступность базы сотрудников и файла публикаций, затем обновите страницу.</p>
+    <h3>Не удалось загрузить публикации</h3>
+    <p>Данные временно недоступны. Попробуйте обновить страницу позже.</p>
+    <a class="admin__link" href="/admin/council">Повторить загрузку</a>
   </div>
   {% else %}
-  <div class="council__criteria" aria-label="Правила предварительного отбора">
-    <p><strong>10 и более</strong><span>различных публикаций</span></p>
-    <p><strong>{{ start.strftime('%d.%m.%Y') }} — {{ end.strftime('%d.%m.%Y') }}</strong><span>скользящие пять лет</span></p>
-    <p><strong>К1 · RSCI · Q1/Q2</strong><span>итоговые основания отбора</span></p>
-  </div>
-  <div class="council__notice" role="note">
-    <strong>Предварительный список.</strong> Сейчас учитываются только отметки Q1/Q2 в CSV для Scopus и Web of Science. В нём нет названия журнала, точной даты статьи, типа материала и подтверждения К1 или RSCI. Год на границе периода и принадлежность к журналу нужно проверить. Сотрудники с публикациями только в К1 или RSCI могут отсутствовать.
-  </div>
+  <details class="council__notice">
+    <summary><strong>Как читать счётчики и критерии</strong></summary>
+    <p>«CSV по критериям» — уникальные публикации с отметкой Scopus или Web of Science Q1/Q2 за {{ start.strftime('%d.%m.%Y') }} — {{ end.strftime('%d.%m.%Y') }}. «В БД» — все уникальные публикации сотрудника из miem-workers, включая импортированные из CSV. Общий перечень объединяет оба источника; совпадения не увеличивают итоговое количество.</p>
+    <p>В CSV нет типа материала, точной даты и подтверждения К1 или RSCI. Счётчик CSV предварительный: для пограничных лет нужно проверить дату, для материалов без типа — подтвердить, что это статьи. Материалы, для которых в БД указан другой тип, в этот счётчик не входят. Отсутствие отметки означает недостаток данных. Наличие публикаций не подтверждает соответствие всем требованиям совета.</p>
+  </details>
 
+  <form class="council__search" method="get" action="/admin/council" role="search">
+    <label class="directory__field" for="council-search">Поиск по ФИО
+      <input class="directory__input" id="council-search" name="q" type="search" value="{{ q }}" placeholder="Фамилия или полное имя">
+    </label>
+    <label class="directory__field">Сортировать по
+      <select class="directory__input" name="sort">
+        {% for value, label in [('full_name', 'ФИО'), ('csv_count', 'Публикациям CSV по критериям'), ('db_count', 'Публикациям в БД'), ('count', 'Всего публикаций без дублей')] %}
+        <option value="{{ value }}"{% if sort == value %} selected{% endif %}>{{ label }}</option>
+        {% endfor %}
+      </select>
+    </label>
+    <label class="directory__field">Направление
+      <select class="directory__input" name="direction">
+        <option value="asc"{% if direction == 'asc' %} selected{% endif %}>По возрастанию</option>
+        <option value="desc"{% if direction == 'desc' %} selected{% endif %}>По убыванию</option>
+      </select>
+    </label>
+    <button class="button" type="submit">Применить</button>
+    {% if q %}<a class="admin__link" href="?sort={{ sort }}&amp;direction={{ direction }}">Сбросить поиск</a>{% endif %}
+  </form>
   <div class="council__list-heading">
     <h3>Сотрудники</h3>
-    <p>Откройте строку, чтобы увидеть публикации и основания.</p>
+    <p role="status">{% if matched %}Показаны {{ offset + 1 }}–{{ [offset + limit, matched]|min }} из {{ matched }}{% else %}Найдено: 0{% endif %}{% if q %} · Всего в списке: {{ total }}{% endif %}</p>
   </div>
   {% if candidates %}
+  <p class="council__hint">Откройте сотрудника, чтобы увидеть публикации, источники и основания.</p>
   <div class="council__list">
     {% for candidate in candidates %}
     <details class="council__candidate">
       <summary class="council__candidate-summary">
-        <span class="council__candidate-identity"><strong>{{ candidate.name }}</strong><small>{{ candidate.degree }}</small></span>
-        <span class="council__candidate-count"><strong>{{ candidate.count }}</strong><small>публикаций по выгрузке</small></span>
+        <span class="council__candidate-identity"><strong>{{ candidate.name }}</strong><small>{{ candidate.degree }}{% if candidate.dismissed %} · Уволен{% endif %}</small></span>
+        <span class="council__counts">
+          <span class="council__candidate-count"><strong>{{ candidate.csv_count }}</strong><small>CSV по критериям</small></span>
+          <span class="council__candidate-count"><strong>{{ candidate.db_count }}</strong><small>В БД</small></span>
+          <span class="council__candidate-count"><strong>{{ candidate.count }}</strong><small>Всего без дублей</small></span>
+        </span>
         <span class="council__candidate-cue" aria-hidden="true"></span>
       </summary>
       <div class="council__candidate-body">
-        <a class="admin__link" href="/admin/employees/{{ candidate.id }}">Открыть профиль сотрудника</a>
+        {% if candidate.profile_url %}<a class="admin__link" href="{{ candidate.profile_url }}" target="_blank" rel="noopener noreferrer">Профиль на сайте ВШЭ</a>{% endif %}
+        {% if candidate.ambiguous %}<p class="council__hint">В базе есть сотрудники с одинаковым ФИО. Публикации из файла не присвоены автоматически; показаны записи, связанные с этим сотрудником в БД.</p>{% endif %}
         <ol class="council__articles">
           {% for article in candidate.articles %}
           <li class="council__article">
             <div class="council__article-heading">
               <strong>{{ article.title }}</strong>
-              <span>{{ article.year }}</span>
+              <span>{{ article.year or 'Год не указан' }}</span>
             </div>
             <div class="council__article-meta">
-              {% for source in article.sources %}
-              <span class="council__source">{{ source.name }} {{ source.quartile }}{% if source.identifier %} · {{ source.identifier }}{% else %} · ID не указан{% endif %}</span>
-              {% endfor %}
+              {% for origin in article.origins %}<span class="council__source">{{ origin }}</span>{% endfor %}
+              <span>{{ article.type_label }}</span>
               {% if article.doi_url %}<a class="admin__link" href="{{ article.doi_url }}" target="_blank" rel="noopener noreferrer">DOI</a>{% endif %}
-              <span>Строка {{ article.row_number }} в CSV</span>
+              {% if article.url %}<a class="admin__link" href="{{ article.url }}" target="_blank" rel="noopener noreferrer">Публикация на сайте</a>{% endif %}
+              {% if article.rows %}<span>Строки CSV: {{ article.rows|join(', ') }}</span>{% endif %}
             </div>
+            <ul class="council__evidence" aria-label="Критерии и подтверждения">
+              {% for criterion in article.criteria %}<li>{{ criterion }}</li>{% endfor %}
+            </ul>
           </li>
           {% endfor %}
         </ol>
@@ -61,10 +87,18 @@
     </details>
     {% endfor %}
   </div>
+  {% if matched > limit %}
+  <nav class="directory__pagination" aria-label="Страницы сотрудников">
+    {% if offset > 0 %}<a class="admin__link" href="?q={{ q|urlencode }}&amp;sort={{ sort }}&amp;direction={{ direction }}&amp;offset={{ offset - limit }}">Назад</a>{% endif %}
+    <span class="directory__page">{{ offset // limit + 1 }} / {{ (matched + limit - 1) // limit }}</span>
+    {% if offset + limit < matched %}<a class="admin__link" href="?q={{ q|urlencode }}&amp;sort={{ sort }}&amp;direction={{ direction }}&amp;offset={{ offset + limit }}">Далее</a>{% endif %}
+  </nav>
+  {% endif %}
   {% else %}
   <div class="council__message">
-    <h4>Кандидаты пока не найдены</h4>
-    <p>По доступной выгрузке нет действующих сотрудников с докторской степенью, указанной в профиле, и десятью предварительно подходящими публикациями. Это не означает, что таких сотрудников нет.</p>
+    <h4>{% if q %}Сотрудники не найдены{% else %}Сотрудники с докторской степенью и публикациями пока не найдены{% endif %}</h4>
+    <p>{% if q %}Попробуйте другую фамилию или сбросьте поиск.{% else %}Для включения в список нужны докторская степень в профиле и хотя бы одна публикация, однозначно связанная с сотрудником.{% endif %}</p>
+    {% if q %}<a class="admin__link" href="?sort={{ sort }}&amp;direction={{ direction }}">Сбросить поиск</a>{% endif %}
   </div>
   {% endif %}
   {% endif %}

app/version.py

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

docs/postgresql-import-verification.md

@@ -0,0 +1,99 @@
+# Проверка импорта на изолированном PostgreSQL
+
+Запускается средствами проекта: pytest, SQLAlchemy, psycopg и отдельный Docker
+Compose. Не использует production Compose, `.env`, SSH или данные сотрудников.
+PostgreSQL 16 и runner работают в отдельной `internal`-сети без опубликованных
+портов. Данные PostgreSQL находятся в tmpfs, без постоянных volumes. Базы:
+`miem_import_test` и `miem_crawl_test`. Все сотрудники, статьи и credentials
+синтетические; тестовые пароли непригодны для реальных систем.
+
+Dockerfile копирует только нужный код и requirements. Его отдельный dockerignore
+исключает `.git`, `.env`, файлы ключей/credentials, CSV владельца, данные
+`app/data` и остальные файлы workspace. Runtime не получает переменные хоста.
+CLI subprocess дополнительно получает только явно заданные тестовые libpq
+параметры; `.pgpass`/service-файлы отключены. Не изменяйте адрес `postgres-test`
+или имена тестовых БД для подключения к существующему серверу.
+
+Синтетическая роль `import_test` создаётся только init-скриптом одноразового
+тестового кластера. Fixture выдаёт ей документированные column grants и USAGE
+sequence. Проверяется отсутствие CREATE схемы, INSERT сотрудников, DELETE
+публикаций и sequence UPDATE. Это не создание временного доступа к рабочей БД.
+Таблицы создаются моделью приложения и очищаются только в тестовых БД.
+
+HTTP заблокирован test fixture; входные профили полного обхода подставляются
+локально. Это проверяет настоящий SQL, транзакции, savepoint и сериализацию JSON,
+но не парсинг сайта ВШЭ. Fixtures SQLite и подмена psycopg для интеграционных
+тестов не используются.
+
+## Что проверяется
+
+1. Dry-run: все строки всех таблиц и состояния sequences совпадают до/после.
+2. Apply: одна вставка, одно обновление, пропуск отсутствующего сотрудника.
+3. Повторный импорт: ноль INSERT/UPDATE, неизменные таблицы/sequences.
+4. CHECK constraint тестовой БД вызывает ошибку третьего действия после UPDATE
+   и успешного INSERT; все строки возвращаются в исходное состояние. Sequence
+   может продвинуться при rollback — это ожидаемое поведение PostgreSQL.
+5. Обновление только года без raw_data сохраняет существующий JSONB.
+6. Поздние строки CSV дополняют метаданные; первое непустое значение сохраняется
+   при конфликте, отчёт содержит названия конфликтующих полей. Повтор идемпотентен.
+7. Два последовательных crawl сохраняют CSV-статью, отсутствующую у ВШЭ;
+   появившаяся статья получает HSE ID без второй записи и потери метаданных.
+8. Полный `run_crawl` обрабатывает двух сотрудников. Два HSE ID на одну CSV-запись
+   откатывают savepoint первого, сохраняя JSONB, авторов, ID, хеш и updated_at;
+   публикация второго сохраняется. Проверяются обе очередности ID и уже
+   установленная привязка; неоднозначность отражается в CrawlError.
+
+Всего 10 PostgreSQL-тестов (восьмой сценарий параметризован). В обычном
+`python -m pytest -q` они пропускаются. Opt-in `MIEM_POSTGRES_INTEGRATION=1`
+разрешён только с точным тестовым DATABASE_URL Compose-проекта.
+
+## Команды владельцу
+
+Выполняйте на локальной машине с Docker Engine/Compose v2 из корня репозитория
+после независимого ревью. Команды явно выбирают локальный Unix socket и очищают
+наследуемое окружение. Для Docker Desktop/Windows используйте Linux/WSL с
+доступным локальным `/var/run/docker.sock`. Для отдельного стенда на VPS используйте
+[отдельную инструкцию](postgresql-import-vps.md) и `compose.vps.yml`: этот локальный
+Compose не предназначен для запуска на сервере с работающими сервисами.
+Первый шаг:
+
+```sh
+env -i PATH="$PATH" docker --host unix:///var/run/docker.sock version
+```
+
+Ожидается exit code 0 и секции `Client` и `Server`. `command not found`,
+`permission denied` или только Client с ошибкой подключения означает, что
+до запуска этого варианта нужен локальный Docker. При его отсутствии сначала
+выполните только read-only диагностику из инструкции VPS.
+
+Дальнейшие точные команды сохранены здесь для воспроизводимого запуска:
+
+```sh
+env -i PATH="$PATH" docker --host unix:///var/run/docker.sock compose --env-file /dev/null --project-name miem-publication-integration -f tests/postgres/compose.yml config --quiet
+env -i PATH="$PATH" docker --host unix:///var/run/docker.sock compose --env-file /dev/null --project-name miem-publication-integration -f tests/postgres/compose.yml build tests
+env -i PATH="$PATH" docker --host unix:///var/run/docker.sock compose --env-file /dev/null --project-name miem-publication-integration -f tests/postgres/compose.yml up --abort-on-container-exit --exit-code-from tests
+```
+
+Первые две команды проверяют Compose и собирают runner с уже используемыми
+зависимостями. Сборка/скачивание образов требуют registry/PyPI, но тестовая
+runtime-сеть не имеет внешнего доступа и запросов к ВШЭ нет. При успехе третьей
+команды ожидаются `10 passed` и exit code 0. Фактическую совместимость образа,
+column grants и PostgreSQL можно считать подтверждённой **только после такого
+выполнения**, не по unit-тестам или статическому чтению файлов.
+
+После сохранения вывода тестов удалите именно этот тестовый проект:
+
+```sh
+env -i PATH="$PATH" docker --host unix:///var/run/docker.sock compose --env-file /dev/null --project-name miem-publication-integration -f tests/postgres/compose.yml down
+```
+
+Это уничтожит только синтетические данные в tmpfs тестового PostgreSQL. Не
+используйте `docker system prune`, production Compose или удаление рабочих
+volumes. Перед повторным прогоном выполните этот же `down`, чтобы следующий
+кластер начинался с чистого init.sql; не запускайте два прогона одновременно.
+
+## Статус подготовки
+
+В среде подготовки Docker отсутствует; настоящие PostgreSQL-тесты не запускались.
+Локально доступны только unit-тесты, проверка Python и статические проверки.
+Production/VPS и доступ к рабочей БД не изменялись.

docs/postgresql-import-vps.md

@@ -0,0 +1,153 @@
+# PostgreSQL-проверка на отдельном стенде VPS
+
+Команды выполняет владелец под root. Подготовка не подключается к VPS и не
+запускает стенд. Не используйте production Compose/`.env`, Docker context,
+volumes, сети, mounts или production credentials. Никаких production-ролей БД,
+SSH-ключей и правил firewall эти команды не создают и не изменяют.
+
+## Сначала только read-only проверка
+
+Следующий блок не создаёт контейнеров, не скачивает образов и не меняет данные.
+Docker CLI использует локальный Unix socket и очищенное окружение; конфигурация
+CLI указывает на отдельный каталог, который здесь не создаётся.
+
+```sh
+free -h
+nproc
+uptime
+df -h / /var/lib/docker
+env -i PATH="$PATH" docker --config /root/miem-publication-integration/docker-config --host unix:///var/run/docker.sock info --format 'CPUs={{.NCPU}} RAM_bytes={{.MemTotal}} cgroup={{.CgroupVersion}} DockerRootDir={{.DockerRootDir}} warnings={{json .Warnings}}'
+env -i PATH="$PATH" docker --config /root/miem-publication-integration/docker-config --host unix:///var/run/docker.sock compose version
+env -i PATH="$PATH" docker --config /root/miem-publication-integration/docker-config --host unix:///var/run/docker.sock stats --no-stream --format 'table {{.Name}}\t{{.CPUPerc}}\t{{.MemUsage}}\t{{.PIDs}}'
+env -i PATH="$PATH" docker --config /root/miem-publication-integration/docker-config --host unix:///var/run/docker.sock image inspect postgres:16-alpine python:3.12-slim miem-publication-integration-postgres:pr55 miem-publication-integration-runner:pr55 --format '{{.RepoTags}} id={{.Id}} revision={{index .Config.Labels "org.opencontainers.image.revision"}}'
+```
+
+Как читать результат:
+
+- `free`: нужна колонка **available**, а не только free. Предварительный запас —
+  не менее 2 GiB: стенд ограничен суммарно 1 GiB, ещё минимум 1 GiB остаётся
+  системе/сервисам. Это ориентир, решение зависит от их нагрузки и пиков.
+- `nproc` и `uptime`: сравните число CPU и load average. Постоянный load на уровне
+  числа CPU или выше — основание отложить тест, а не останавливать сервисы.
+- `df`: желательно не менее 4 GiB свободно для загрузки образов и запаса работы
+  Docker. Если DockerRootDir отличается от `/var/lib/docker`, отдельно выполните
+  `df -h` для показанного каталога. Не используйте prune для освобождения места.
+- `stats`: показывает потребление уже работающих контейнеров, но не их пиковые
+  требования. Docker warnings о недоступных memory/swap/cgroup limits нужно
+  устранить до запуска; без действующих лимитов этот вариант не запускать.
+- `image inspect`: существующие образы показывают tags, ID и revision.
+  `No such image` означает отсутствие образа, **не** разрешение на pull/build.
+  Для запуска нужны два образа `miem-publication-integration-*:pr55` с revision
+  проверенного commit PR #55. Base-образы нужны только для сборки, не для самого
+  запуска уже подготовленного стенда.
+
+Пришлите только этот диагностический вывод. Не присылайте `docker inspect`
+production-контейнеров целиком, env, конфигурации подключения или секреты.
+До оценки ресурсов и завершения независимого review остальные действия ниже
+не выполнять.
+
+## Изоляция и лимиты подготовленного стенда
+
+Используется только `tests/postgres/compose.vps.yml`, не override production или
+локального Compose. Project name фиксирован:
+`miem-publication-integration-vps`.
+
+| Ресурс | Лимит / область |
+|---|---|
+| PostgreSQL | Отдельный контейнер `postgres-test`, 0.5 CPU, 512 MiB RAM, без дополнительного swap, 128 процессов |
+| PostgreSQL настройки | shared_buffers 64 MiB, max_connections 12 |
+| Test runner | 0.5 CPU, 512 MiB RAM, без дополнительного swap, 128 процессов; UID 65534, read-only rootfs, no-new-privileges, без capabilities |
+| Данные PostgreSQL | Только named volume `miem-publication-integration-vps_pgdata`; local driver tmpfs до 256 MiB |
+| Временные файлы runner | tmpfs `/tmp` до 128 MiB; входят в memory limit контейнера |
+| Сеть | Только `miem-publication-integration-vps_isolated`, internal; опубликованных портов нет |
+| Логи | json-file, до 5 MiB на контейнер, один файл |
+
+Volume с tmpfs выбран намеренно: отдельное имя для адресной очистки и предел
+размера синтетических данных. Данные не сохраняются после остановки/размонтирования.
+Образы содержат только код и синтетический init.sql. Host bind mounts,
+Docker socket, production networks и env_file отсутствуют. Docker socket
+доступен только CLI владельца на хосте, ни одному контейнеру он не передаётся.
+Нет restart policy. Pull запрещён; ресурсы не создаются автоматически при
+отсутствии заранее подготовленного образа.
+
+Лимиты процесса сборки/загрузки образов отличаются от runtime: Compose CPU/RAM
+limits **не ограничивают Docker build**. Поэтому на VPS с рабочими сервисами
+не выполняйте незапланированную сборку. Если готовых образов нет, отдельно
+согласуйте место сборки и окно загрузки после read-only диагностики.
+
+## Подготовка образов после review (не запуск стенда)
+
+На отдельной согласованной машине сборки с Docker, из проверенной копии PR #55,
+укажите полный SHA проверенного commit в переменной `TEST_COMMIT` (не секрет).
+Сначала `git status --short` должен быть пустым, `git rev-parse HEAD` — совпадать
+с TEST_COMMIT. Это не команды для немедленного выполнения на VPS.
+
+```sh
+git status --short
+git rev-parse HEAD
+env -i PATH="$PATH" docker --host unix:///var/run/docker.sock build --label org.opencontainers.image.revision="$TEST_COMMIT" -t miem-publication-integration-runner:pr55 -f tests/postgres/Dockerfile .
+env -i PATH="$PATH" docker --host unix:///var/run/docker.sock build --label org.opencontainers.image.revision="$TEST_COMMIT" -t miem-publication-integration-postgres:pr55 -f tests/postgres/Dockerfile.postgres tests/postgres
+env -i PATH="$PATH" docker --host unix:///var/run/docker.sock image save --output miem-publication-integration-images.tar miem-publication-integration-postgres:pr55 miem-publication-integration-runner:pr55
+sha256sum miem-publication-integration-images.tar tests/postgres/compose.vps.yml
+```
+
+Dockerfile-specific ignore-файлы исключают секреты, реальные CSV и остальные
+файлы workspace из обоих контекстов сборки. Передайте владельцу только архив
+этих двух образов и `compose.vps.yml` с проверенными SHA256, без полного workspace
+и `.env`. Транспорт выбирает владелец; инструкция не получает SSH/production
+credentials. Base-образы могут скачиваться только на согласованной машине сборки.
+
+На VPS после оценки ресурсов/отдельного разрешения владельца: файлы размещаются
+в `/root/miem-publication-integration`, Compose переименован в `compose.vps.yml`,
+архив — в `images.tar`. Проверка checksum должна совпасть с источником.
+
+```sh
+install -d -m 700 /root/miem-publication-integration /root/miem-publication-integration/docker-config
+sha256sum /root/miem-publication-integration/images.tar /root/miem-publication-integration/compose.vps.yml
+env -i PATH="$PATH" docker --config /root/miem-publication-integration/docker-config --host unix:///var/run/docker.sock image load --input /root/miem-publication-integration/images.tar
+env -i PATH="$PATH" docker --config /root/miem-publication-integration/docker-config --host unix:///var/run/docker.sock image inspect miem-publication-integration-postgres:pr55 miem-publication-integration-runner:pr55 --format '{{.RepoTags}} id={{.Id}} revision={{index .Config.Labels "org.opencontainers.image.revision"}}'
+env -i PATH="$PATH" docker --config /root/miem-publication-integration/docker-config --host unix:///var/run/docker.sock compose --env-file /dev/null --project-name miem-publication-integration-vps -f /root/miem-publication-integration/compose.vps.yml config --quiet
+```
+
+Проверьте отсутствие уже существующих ресурсов с этим project name. Если они
+есть, выясните их назначение; не запускайте второй стенд и не удаляйте чужие
+ресурсы. Следующие команды только читают labels Docker Compose:
+
+```sh
+env -i PATH="$PATH" docker --config /root/miem-publication-integration/docker-config --host unix:///var/run/docker.sock ps -a --filter label=com.docker.compose.project=miem-publication-integration-vps
+env -i PATH="$PATH" docker --config /root/miem-publication-integration/docker-config --host unix:///var/run/docker.sock network ls --filter label=com.docker.compose.project=miem-publication-integration-vps
+env -i PATH="$PATH" docker --config /root/miem-publication-integration/docker-config --host unix:///var/run/docker.sock volume ls --filter label=com.docker.compose.project=miem-publication-integration-vps
+```
+
+## Запуск и адресная очистка (пока не выполнять)
+
+Только после согласования владельцем ресурсов и проверенного состава стенда:
+
+```sh
+env -i PATH="$PATH" docker --config /root/miem-publication-integration/docker-config --host unix:///var/run/docker.sock compose --env-file /dev/null --project-name miem-publication-integration-vps -f /root/miem-publication-integration/compose.vps.yml up --no-build --pull never --abort-on-container-exit --exit-code-from tests
+```
+
+Ожидаемый успех — `10 passed`, exit code 0. Сохраните вывод/код выхода и логи:
+
+```sh
+env -i PATH="$PATH" docker --config /root/miem-publication-integration/docker-config --host unix:///var/run/docker.sock compose --env-file /dev/null --project-name miem-publication-integration-vps -f /root/miem-publication-integration/compose.vps.yml logs --no-color > /root/miem-publication-integration/test.log
+```
+
+**Следующая команда удаляет контейнеры, внутреннюю сеть и named volume именно
+этого теста; все его синтетические данные будут потеряны.** Перед ней сохраните
+отчёт. Она не затрагивает ресурсы других Compose-проектов:
+
+```sh
+env -i PATH="$PATH" docker --config /root/miem-publication-integration/docker-config --host unix:///var/run/docker.sock compose --env-file /dev/null --project-name miem-publication-integration-vps -f /root/miem-publication-integration/compose.vps.yml down --volumes
+```
+
+Не используйте `--remove-orphans`, `docker system prune`, общие rm/rmi/volume prune
+или down рабочего проекта. Base-образы и образы теста оставьте: они могут быть
+общими для других проверок. Архив и журнал также не удаляются этой командой.
+Для подтверждения очистки повторите три read-only команды с project label:
+в выводе должны остаться только заголовки.
+
+## Статус
+
+VPS-стенд не запускался, Docker/Compose-проверка не выполнялась в среде
+подготовки из-за отсутствия Docker. PostgreSQL-интеграция ещё не подтверждена.

pyproject.toml

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

tests/browser_council.py

@@ -0,0 +1,139 @@
+"""Offline public council UX check: python tests/browser_council.py.
+
+Requires Playwright/Chromium; BROWSER_EXECUTABLE may select an installed browser.
+Synthetic employees and intercepted requests; no production access.
+"""
+import os
+from pathlib import Path
+import sys
+from urllib.parse import urlsplit
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from fastapi.testclient import TestClient
+from playwright.sync_api import expect, sync_playwright
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import StaticPool
+
+from app.db import Base, get_db
+from app.main import app
+from app.models import Employee, EmployeePublication
+from app.services import council
+
+
+def main():
+    engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+    Base.metadata.create_all(engine)
+    factory = sessionmaker(bind=engine)
+    with factory() as db:
+        for number in range(28):
+            person = Employee(profile_key=f"test:{number}", full_name=f"Сотрудник {number:02d}",
+                              canonical_url="https://www.hse.ru/staff/test",
+                              current_data={"sections": [{"title": "Учёные степени",
+                                            "items": ["Доктор технических наук"]}]})
+            db.add(person)
+            db.flush()
+            db.add(EmployeePublication(employee_id=person.id, title="Исследование методов передачи данных в фотонных интегральных схемах",
+                                       year=2024, publication_type="ARTICLE", publication_id=f"{number}",
+                                       url=f"https://publications.hse.ru/view/{number}", source_hash=f"test:{number}"))
+            if number == 27:
+                db.add(EmployeePublication(employee_id=person.id, title="Дополнительная статья", year=2024,
+                                           publication_type="ARTICLE", source_hash="test:extra"))
+        for name, degree in [("Без степени", None), ("Кандидат наук", "Кандидат технических наук")]:
+            person = Employee(profile_key=name, full_name=name, canonical_url="https://www.hse.ru/staff/test",
+                              current_data={"sections": [{"title": "Учёные степени", "items": [degree] if degree else []}]})
+            db.add(person)
+            db.flush()
+            db.add(EmployeePublication(employee_id=person.id, title="Не включать в совет", year=2024, source_hash=name))
+        db.commit()
+
+    def database():
+        with factory() as db:
+            yield db
+
+    original = council._articles_by_author
+    council._articles_by_author = lambda: {"сотрудник 26": [
+        {"title": f"Публикация из CSV {number}", "year": 2024, "doi": f"10.1234/test-{number}",
+         "types": [], "quartiles": ["Scopus Q1"], "origins": ["Файл CSV"], "rows": [number + 2],
+         "in_csv": True, "in_db": False, "url": None} for number in range(2)]}
+    app.dependency_overrides[get_db] = database
+    client = TestClient(app)
+    output = Path(".impeccable/review")
+    output.mkdir(parents=True, exist_ok=True)
+
+    def serve(route):
+        url = urlsplit(route.request.url)
+        response = client.get(url.path + ("?" + url.query if url.query else ""))
+        route.fulfill(status=response.status_code, body=response.content,
+                      content_type=response.headers.get("content-type", "text/plain"))
+
+    try:
+        with sync_playwright() as playwright:
+            browser = playwright.chromium.launch(executable_path=os.getenv("BROWSER_EXECUTABLE"))
+            page = browser.new_page(viewport={"width": 1440, "height": 1000})
+            page.route("**/*", serve)
+            errors = []
+            page.on("pageerror", lambda error: errors.append(str(error)))
+            page.goto("http://miem.test/admin/council")
+            expect(page.get_by_role("status")).to_have_text("Показаны 1–25 из 28")
+            assert "Без степени" not in page.content() and "Кандидат наук" not in page.content()
+            expect(page.get_by_role("link", name="Вход в админку")).to_be_visible()
+            page.get_by_role("link", name="Далее").click()
+            expect(page.get_by_role("status")).to_have_text("Показаны 26–28 из 28")
+            page.get_by_role("link", name="Назад").click()
+            summary = page.locator(".council__candidate-summary").first
+            summary.focus()
+            page.keyboard.press("Enter")
+            expect(page.get_by_text("БД miem-workers · сайт ВШЭ").first).to_be_visible()
+            expect(page.get_by_text("Q1/Q2, К1 и RSCI не подтверждены доступными данными").first).to_be_visible()
+            assert page.locator(".council__candidate").first.evaluate("e => e.open")
+            page.screenshot(path=str(output / "desktop.png"), full_page=True)
+            page.screenshot(path=str(output / "sorting-desktop.png"), full_page=False)
+            for width in (1024, 768, 700, 390, 320):
+                page.set_viewport_size({"width": width, "height": 844})
+                assert page.evaluate("document.documentElement.scrollWidth <= innerWidth"), width
+                assert summary.evaluate("e => e.scrollWidth <= e.clientWidth"), width
+            page.set_viewport_size({"width": 390, "height": 844})
+            page.screenshot(path=str(output / "mobile.png"), full_page=True)
+            page.screenshot(path=str(output / "sorting-mobile.png"), full_page=False)
+            for sort, direction, first in [("full_name", "asc", "Сотрудник 00"),
+                                            ("full_name", "desc", "Сотрудник 27"),
+                                            ("db_count", "desc", "Сотрудник 27"),
+                                            ("count", "desc", "Сотрудник 26"),
+                                            ("csv_count", "desc", "Сотрудник 26")]:
+                page.get_by_label("Сортировать по").select_option(sort)
+                page.get_by_label("Направление").select_option(direction)
+                page.get_by_role("button", name="Применить", exact=True).click()
+                expect(page.locator(".council__candidate-identity strong").first).to_have_text(first)
+            page.get_by_label("Сортировать по").select_option("full_name")
+            page.get_by_label("Направление").select_option("desc")
+            page.get_by_role("button", name="Применить", exact=True).click()
+            page.get_by_role("link", name="Далее").click()
+            expect(page.get_by_label("Сортировать по")).to_have_value("full_name")
+            expect(page.get_by_label("Направление")).to_have_value("desc")
+            expect(page.locator(".council__candidate-identity strong").first).to_have_text("Сотрудник 02")
+            page.get_by_label("Поиск по ФИО").fill("Сотрудник 27")
+            page.get_by_role("button", name="Применить", exact=True).click()
+            expect(page.get_by_role("status")).to_have_text("Показаны 1–1 из 1 · Всего в списке: 28")
+            expect(page.get_by_label("Сортировать по")).to_have_value("full_name")
+            expect(page.get_by_label("Направление")).to_have_value("desc")
+            page.get_by_label("Поиск по ФИО").fill("Нет совпадений")
+            page.get_by_role("button", name="Применить", exact=True).click()
+            expect(page.get_by_text("Сотрудники не найдены", exact=True)).to_be_visible()
+            page.get_by_role("link", name="Сбросить поиск").first.click()
+            expect(page.get_by_role("status")).to_have_text("Показаны 1–25 из 28")
+            expect(page.get_by_label("Сортировать по")).to_have_value("full_name")
+            expect(page.get_by_label("Направление")).to_have_value("desc")
+            assert not errors, errors
+            browser.close()
+    finally:
+        council._articles_by_author = original
+        app.dependency_overrides.clear()
+        client.close()
+        engine.dispose()
+    print(f"Council browser checks passed; screenshots: {output}")
+
+
+if __name__ == "__main__":
+    main()

tests/postgres/Dockerfile

@@ -0,0 +1,7 @@
+FROM python:3.12-slim
+WORKDIR /app
+COPY requirements.txt .
+RUN python -m pip install --no-cache-dir -r requirements.txt
+COPY app ./app
+COPY scripts/import_publications.py ./scripts/import_publications.py
+COPY tests/conftest.py tests/test_publication_import_postgres.py ./tests/

tests/postgres/Dockerfile.dockerignore

@@ -0,0 +1,18 @@
+**
+!requirements.txt
+!app/
+!app/**
+!scripts/
+!scripts/import_publications.py
+!tests/
+!tests/conftest.py
+!tests/test_publication_import_postgres.py
+!tests/postgres/
+!tests/postgres/Dockerfile
+!tests/postgres/Dockerfile.dockerignore
+app/data/
+**/__pycache__/
+**/.env*
+**/*credentials*
+**/*.key
+**/*.pem

tests/postgres/Dockerfile.postgres

@@ -0,0 +1,2 @@
+FROM postgres:16-alpine
+COPY init.sql /docker-entrypoint-initdb.d/10-test.sql

tests/postgres/Dockerfile.postgres.dockerignore

@@ -0,0 +1,4 @@
+**
+!init.sql
+!Dockerfile.postgres
+!Dockerfile.postgres.dockerignore

tests/postgres/compose.vps.yml

@@ -0,0 +1,69 @@
+name: miem-publication-integration-vps
+
+services:
+  postgres-test:
+    image: miem-publication-integration-postgres:pr55
+    pull_policy: never
+    environment:
+      POSTGRES_USER: test_admin
+      POSTGRES_PASSWORD: isolated_test_admin
+      POSTGRES_DB: miem_import_test
+    command: ["postgres", "-c", "shared_buffers=64MB", "-c", "max_connections=12"]
+    cpus: 0.5
+    mem_limit: 512m
+    memswap_limit: 512m
+    pids_limit: 128
+    volumes:
+      - pgdata:/var/lib/postgresql/data
+    healthcheck:
+      test: ["CMD", "pg_isready", "-h", "127.0.0.1", "-U", "test_admin", "-d", "miem_import_test"]
+      interval: 2s
+      timeout: 5s
+      retries: 30
+    logging:
+      driver: json-file
+      options:
+        max-size: "5m"
+        max-file: "1"
+    networks: [isolated]
+
+  tests:
+    image: miem-publication-integration-runner:pr55
+    pull_policy: never
+    user: "65534:65534"
+    read_only: true
+    cap_drop: [ALL]
+    security_opt: ["no-new-privileges:true"]
+    cpus: 0.5
+    mem_limit: 512m
+    memswap_limit: 512m
+    pids_limit: 128
+    tmpfs:
+      - /tmp:size=128m,mode=1777
+    environment:
+      DATABASE_URL: postgresql+psycopg://test_admin:isolated_test_admin@postgres-test:5432/miem_import_test
+      MIEM_POSTGRES_INTEGRATION: "1"
+      PYTHONDONTWRITEBYTECODE: "1"
+      PYTEST_ADDOPTS: "-p no:cacheprovider"
+    command: ["python", "-m", "pytest", "-q", "tests/test_publication_import_postgres.py"]
+    depends_on:
+      postgres-test:
+        condition: service_healthy
+    logging:
+      driver: json-file
+      options:
+        max-size: "5m"
+        max-file: "1"
+    networks: [isolated]
+
+networks:
+  isolated:
+    internal: true
+
+volumes:
+  pgdata:
+    driver: local
+    driver_opts:
+      type: tmpfs
+      device: tmpfs
+      o: size=256m

tests/postgres/compose.yml

@@ -0,0 +1,36 @@
+name: miem-publication-integration
+
+services:
+  postgres-test:
+    image: postgres:16-alpine
+    environment:
+      POSTGRES_USER: test_admin
+      POSTGRES_PASSWORD: isolated_test_admin
+      POSTGRES_DB: miem_import_test
+    tmpfs:
+      - /var/lib/postgresql/data
+    volumes:
+      - ./init.sql:/docker-entrypoint-initdb.d/10-test.sql:ro
+    healthcheck:
+      test: ["CMD", "pg_isready", "-h", "127.0.0.1", "-U", "test_admin", "-d", "miem_import_test"]
+      interval: 2s
+      timeout: 5s
+      retries: 30
+    networks: [isolated]
+
+  tests:
+    build:
+      context: ../..
+      dockerfile: tests/postgres/Dockerfile
+    environment:
+      DATABASE_URL: postgresql+psycopg://test_admin:isolated_test_admin@postgres-test:5432/miem_import_test
+      MIEM_POSTGRES_INTEGRATION: "1"
+    command: ["python", "-m", "pytest", "-q", "tests/test_publication_import_postgres.py"]
+    depends_on:
+      postgres-test:
+        condition: service_healthy
+    networks: [isolated]
+
+networks:
+  isolated:
+    internal: true

tests/postgres/init.sql

@@ -0,0 +1,6 @@
+-- Synthetic credentials for this ephemeral, private test cluster only.
+CREATE DATABASE miem_crawl_test OWNER test_admin;
+CREATE ROLE import_test LOGIN PASSWORD 'isolated_test_import'
+    NOSUPERUSER NOCREATEDB NOCREATEROLE NOINHERIT NOREPLICATION NOBYPASSRLS
+    CONNECTION LIMIT 1;
+GRANT CONNECT ON DATABASE miem_import_test, miem_crawl_test TO import_test;

tests/test_admin_templates.py

@@ -1,8 +1,13 @@
-from pathlib import Path
+from pathlib import Path
+from types import SimpleNamespace
+
+from jinja2 import Environment, FileSystemLoader
 
 
 def test_base_navigation_is_russian_and_has_no_legacy_employees_link():
-    template = Path("app/templates/base.html").read_text(encoding="utf-8")
+    template = Environment(loader=FileSystemLoader("app/templates")).get_template("base.html").render(
+        request=SimpleNamespace(url=SimpleNamespace(path="/admin"))
+    )
 
     assert "Обзор" in template
     assert "Сотрудники" in template

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

tests/test_council.py

@@ -1,76 +1,229 @@
 import csv
 from datetime import date
 
+import pytest
 from fastapi.testclient import TestClient
+from sqlalchemy.exc import OperationalError
 
-from app.config import Settings, get_settings
 from app.db import get_db
 from app.main import app
-from app.models import Employee
-from app.security import SESSION_COOKIE, sign_session
+from app.models import Employee, EmployeePublication
 from app.services import council
 
 
-def test_council_counts_unique_articles_and_only_active_doctors(db_session, monkeypatch, tmp_path):
-    fields = ["Название публикации", "Год", "DOI", "EID (Scopus)", "UT (WoS)",
-              "Авторы НИУ ВШЭ", "Квартиль scopus", "Квартиль wos"]
-    export = tmp_path / "articles.csv"
-    with export.open("w", encoding="utf-8-sig", newline="") as output:
-        writer = csv.DictWriter(output, fieldnames=fields, delimiter=";")
-        writer.writeheader()
-        for number in range(10):
-            writer.writerow({"Название публикации": f"Article {number}", "Год": 2024,
-                             "DOI": f"10.1234/{number}", "EID (Scopus)": f"eid-{number}",
-                             "Авторы НИУ ВШЭ": "Иванов Иван Иванович; Петров Пётр Петрович",
-                             "Квартиль scopus": "Q1"})
-        writer.writerow({"Название публикации": "Article 0", "Год": 2024, "DOI": "10.1234/0",
-                         "UT (WoS)": "ut-0", "Авторы НИУ ВШЭ": "Иванов Иван Иванович",
-                         "Квартиль wos": "Q2"})
-    monkeypatch.setattr(council, "EXPORT", export)
+@pytest.fixture()
+def export(monkeypatch, tmp_path):
+    path = tmp_path / "articles.csv"
+    def write(rows):
+        fields = ["Название публикации", "Год", "DOI", "EID (Scopus)", "UT (WoS)",
+                  "Авторы НИУ ВШЭ", "Квартиль scopus", "Квартиль wos"]
+        with path.open("w", encoding="utf-8-sig", newline="") as output:
+            writer = csv.DictWriter(output, fieldnames=fields, delimiter=";")
+            writer.writeheader()
+            writer.writerows(rows)
+        council._articles_by_author.cache_clear()
+    monkeypatch.setattr(council, "EXPORT", path)
+    write([])
+    yield write
     council._articles_by_author.cache_clear()
-    db_session.add_all([
-        Employee(profile_key="doctor", canonical_url="https://example.org/1", full_name="Иванов Иван Иванович",
-                 status="active", current_data={"sections": [{"title": "Учёные степени",
-                 "items": ["Доктор технических наук"]}]}),
-        Employee(profile_key="candidate", canonical_url="https://example.org/2", full_name="Петров Пётр Петрович",
-                 status="active", current_data={"sections": [{"title": "Учёные степени",
-                 "items": ["Кандидат технических наук"]}]}),
-    ])
+
+
+def employee(db, name="Иванов Иван Иванович", degree="Доктор технических наук", status="active", key="doctor"):
+    person = Employee(profile_key=key, canonical_url="https://www.hse.ru/staff/test", full_name=name,
+                      status=status, current_data={"contacts": {"emails": ["private@example.org"]},
+                      "sections": [{"title": "Учёные степени", "items": [degree] if degree else []}]})
+    db.add(person)
+    db.flush()
+    return person
+
+
+def publication(db, person, title="Shared work", doi=None, year=2024, kind=None, **extra):
+    item = EmployeePublication(employee_id=person.id, title=title, year=year, doi_url=doi,
+                               publication_type=kind, source_hash=f"hash-{len(db.new)}-{title}-{doi}", **extra)
+    db.add(item)
+    return item
+
+
+def row(title="Shared work", doi="10.1234/shared", year="2024", author="Иванов Иван Иванович", **extra):
+    return {"Название публикации": title, "Год": year, "DOI": doi, "Авторы НИУ ВШЭ": author,
+            "Квартиль scopus": "Q1", **extra}
+
+
+def test_council_union_counts_provenance_and_types(db_session, export):
+    export([row(), row(**{"Квартиль scopus": "", "Квартиль wos": "Q2"}),
+            row(title="CSV only", doi="", year="2022"),
+            row(title="Not eligible", doi="10.1234/old", year="2019")])
+    person = employee(db_session, degree="Доктор технических наук")
+    publication(db_session, person, doi="https://dx.doi.org/10.1234/SHARED", kind="ARTICLE",
+                publication_id="100", url="https://publications.hse.ru/view/100")
+    publication(db_session, person, title=" shared   WORK ", kind="ARTICLE")
+    publication(db_session, person, title="Book only", kind="BOOK", year=2020)
+    publication(db_session, person, title="Imported", raw_data={"csv_enrichment": {"Квартиль wos": "Q2"}})
     db_session.commit()
     result = council.council_candidates(db_session, date(2026, 9, 30))
     assert result["total"] == 1
-    assert result["candidates"][0]["count"] == 10
-    assert len(result["candidates"][0]["articles"][0]["sources"]) == 2
+    candidate = result["candidates"][0]
+    assert (candidate["csv_count"], candidate["db_count"], candidate["count"]) == (2, 3, 5)
+    shared = next(a for a in candidate["articles"] if a["doi"] == "10.1234/shared")
+    assert shared["type_label"] == "Статья"
+    assert shared["quartiles"] == ["Scopus Q1", "Web of Science Q2"]
+    assert shared["rows"] == [2, 3]
+    assert "Файл CSV" in shared["origins"] and "БД miem-workers · сайт ВШЭ" in shared["origins"]
+    imported = next(a for a in candidate["articles"] if a["title"] == "Imported")
+    assert imported["origins"] == ["БД miem-workers · импорт CSV"]
+    assert not imported["csv_qualifies"]
+    assert "Тип не указан" in imported["type_label"]
     assert council.period(date(2024, 2, 29))[0] == date(2019, 2, 28)
-    settings = Settings(_env_file=None, session_secret="test-session-secret")
-    monkeypatch.setattr("app.admin.council_candidates", lambda db: council.council_candidates(db, date(2026, 9, 30)))
-    app.dependency_overrides[get_db] = lambda: db_session
-    app.dependency_overrides[get_settings] = lambda: settings
-    try:
-        client = TestClient(app)
-        client.cookies.set(SESSION_COOKIE, sign_session(settings.admin_username, settings))
-        response = client.get("/admin/council")
-        assert response.status_code == 200
-        assert "Иванов Иван Иванович" in response.text
-        assert "Article 0" in response.text
-        assert "публикаций по выгрузке" in response.text
-    finally:
-        app.dependency_overrides.clear()
-    council._articles_by_author.cache_clear()
 
 
-def test_council_page_requires_admin_and_explains_preliminary_result(db_session, monkeypatch):
-    monkeypatch.setattr(council, "_articles_by_author", lambda: {})
-    settings = Settings(_env_file=None, session_secret="test-session-secret")
+def test_only_doctors_with_a_publication_no_count_threshold(db_session, export):
+    export([row(author="Доктор", title="Single qualifying"),
+            row(author="Без степени", title="No degree"),
+            row(author="Кандидат", title="Candidate work"),
+            row(author="PhD", title="PhD work"),
+            row(author="Две степени", title="Mixed degrees", **{"Квартиль scopus": "Q3"}),
+            row(author="Уволен", title="Old", year="2010")])
+    employee(db_session, "Доктор", key="csv-only")
+    for name, degree in [("Без степени", None), ("Кандидат", "Кандидат технических наук"), ("PhD", "Ph.D.")]:
+        person = employee(db_session, name, degree, key=name)
+        publication(db_session, person)
+    employee(db_session, "Две степени", "Кандидат технических наук; Доктор физико-математических наук", key="mixed")
+    employee(db_session, "Уволен", status="dismissed", key="dismissed")
+    only_db = employee(db_session, "Только БД", key="database")
+    publication(db_session, only_db, kind="PATENT")
+    employee(db_session, "Без публикаций", key="empty")
+    db_session.commit()
+    items = council.council_candidates(db_session, date(2026, 9, 30))["candidates"]
+    assert {c["name"] for c in items} == {"Доктор", "Две степени", "Уволен", "Только БД"}
+    assert next(c for c in items if c["name"] == "Доктор")["csv_count"] == 1
+    assert next(c for c in items if c["name"] == "Только БД")["db_count"] == 1
+    assert next(c for c in items if c["name"] == "Две степени")["csv_count"] == 0
+    assert next(c for c in items if c["name"] == "Уволен")["dismissed"]
+
+
+def test_no_false_merge_when_title_has_conflicting_dois(db_session, export):
+    export([row(doi="10.1234/one"), row(doi="10.1234/two"), row(doi="")])
+    person = employee(db_session)
+    publication(db_session, person, title="Renamed work", doi="https://doi.org/10.1234/one")
+    db_session.commit()
+    item = council.council_candidates(db_session, date(2026, 9, 30))["candidates"][0]
+    assert item["count"] == 3
+    assert item["csv_count"] == 3 and item["db_count"] == 1
+
+
+def test_csv_same_name_does_not_attribute_to_wrong_employee(db_session, export):
+    export([row()])
+    for key in ["one", "two"]:
+        person = employee(db_session, key=key)
+        publication(db_session, person, title=key)
+    db_session.commit()
+    items = council.council_candidates(db_session, date(2026, 9, 30))["candidates"]
+    assert len(items) == 2
+    assert all(c["ambiguous"] and c["csv_count"] == 0 and c["count"] == 1 for c in items)
+
+
+def test_missing_database_year_merges_only_unambiguous_title_and_non_article_excluded(db_session, export):
+    export([row(title="Known book"), row(title="Article", doi="10.1234/article")])
+    person = employee(db_session)
+    publication(db_session, person, title="Known book", year=None, kind="BOOK")
+    publication(db_session, person, title="Article", year=None, kind="ARTICLE")
+    db_session.commit()
+    candidate = council.council_candidates(db_session, date(2026, 9, 30))["candidates"][0]
+    assert (candidate["count"], candidate["csv_count"], candidate["db_count"]) == (2, 1, 2)
+    book = next(a for a in candidate["articles"] if a["type_label"] == "Книга")
+    assert not book["csv_qualifies"]
+    assert "не включён в счётчик CSV" in book["criteria"][-1]
+
+
+def test_search_and_pagination_include_all_workers(db_session, export):
+    for number in range(28):
+        person = employee(db_session, f"Петров {number:02d}", key=f"person-{number}")
+        publication(db_session, person, title=f"Work {number}")
+    db_session.commit()
+    first = council.council_candidates(db_session)
+    last = council.council_candidates(db_session, offset=25)
+    assert first["total"] == 28 and len(first["candidates"]) == 25
+    assert len(last["candidates"]) == 3
+    assert {c["name"] for c in first["candidates"]}.isdisjoint(c["name"] for c in last["candidates"])
+    assert council.council_candidates(db_session, q="петров 27")["matched"] == 1
+    assert council.council_candidates(db_session, offset=-1)["offset"] == 0
+    assert council.council_candidates(db_session, offset=999)["offset"] == 25
+    descending = council.council_candidates(db_session, sort="full_name", direction="desc")
+    last_descending = council.council_candidates(db_session, sort="full_name", direction="desc", offset=25)
+    assert descending["candidates"][0]["name"] == "Петров 27"
+    assert [c["name"] for c in last_descending["candidates"]] == ["Петров 02", "Петров 01", "Петров 00"]
+
+
+def test_sort_by_name_and_each_unique_publication_count(db_session, export):
+    export([row(author="Анна", title=f"CSV {number}", doi=f"10.1234/csv-{number}") for number in range(3)]
+           + [row(author="Борис", title="CSV Boris", doi="10.1234/boris")])
+    anna = employee(db_session, "Анна", key="anna")
+    publication(db_session, anna, title="CSV 0", doi="https://doi.org/10.1234/csv-0")
+    for name, count in [("Борис", 2), ("Вера", 4)]:
+        person = employee(db_session, name, key=name)
+        for number in range(count):
+            publication(db_session, person, title=f"DB {name} {number}")
+    db_session.commit()
+    expected = {
+        "full_name": {"asc": ["Анна", "Борис", "Вера"], "desc": ["Вера", "Борис", "Анна"]},
+        "csv_count": {"asc": ["Вера", "Борис", "Анна"], "desc": ["Анна", "Борис", "Вера"]},
+        "db_count": {"asc": ["Анна", "Борис", "Вера"], "desc": ["Вера", "Борис", "Анна"]},
+        "count": {"asc": ["Анна", "Борис", "Вера"], "desc": ["Вера", "Анна", "Борис"]},
+    }
+    for sort, directions in expected.items():
+        for direction, names in directions.items():
+            result = council.council_candidates(db_session, sort=sort, direction=direction)
+            assert [c["name"] for c in result["candidates"]] == names
+    invalid = council.council_candidates(db_session, sort="unknown", direction="unknown")
+    assert (invalid["sort"], invalid["direction"]) == ("csv_count", "desc")
+    assert [c["name"] for c in invalid["candidates"]] == expected["csv_count"]["desc"]
     app.dependency_overrides[get_db] = lambda: db_session
-    app.dependency_overrides[get_settings] = lambda: settings
     try:
-        client = TestClient(app)
-        assert client.get("/admin/council", follow_redirects=False).status_code in (302, 303, 401, 403)
-        client.cookies.set(SESSION_COOKIE, sign_session(settings.admin_username, settings))
-        response = client.get("/admin/council")
+        response = TestClient(app).get("/admin/council?sort=full_name&direction=desc")
         assert response.status_code == 200
-        assert "Предварительный список" in response.text
-        assert "К1 или RSCI могут отсутствовать" in response.text
+        assert response.text.index("Вера") < response.text.index("Борис") < response.text.index("Анна")
+        assert '<option value="full_name" selected>' in response.text
+        assert '<option value="desc" selected>' in response.text
+    finally:
+        app.dependency_overrides.clear()
+
+
+def test_public_page_data_minimization_and_safe_links(db_session, export):
+    export([row(title='<script>alert("csv")</script>', doi="javascript:bad")])
+    person = employee(db_session)
+    person.canonical_url = "javascript:bad"
+    publication(db_session, person, url="javascript:alert(1)", doi="javascript:bad", kind="BOOK")
+    db_session.commit()
+    app.dependency_overrides[get_db] = lambda: db_session
+    try:
+        client = TestClient(app)
+        response = client.get("/admin/council", follow_redirects=False)
+        assert response.status_code == 200
+        assert "Иванов Иван Иванович" in response.text and "Доктор технических наук" in response.text
+        assert "Минимальный порог количества работ не применяется" in response.text
+        assert "CSV по критериям" in response.text and "В БД" in response.text and "Всего без дублей" in response.text
+        assert "Книга" in response.text and "Тип не указан" in response.text
+        assert "&lt;script&gt;" in response.text and "<script>" not in response.text
+        assert "javascript:" not in response.text and "private@example.org" not in response.text
+        assert "/admin/employees/" not in response.text and "/admin/logout" not in response.text
+        assert "Backend 0.12.0 · Frontend 0.12.0" in response.text
+        assert "Вход в админку" in response.text
+        assert client.get("/admin/directory", follow_redirects=False).status_code in (302, 303, 401, 403)
+        assert "Сотрудники не найдены" in client.get("/admin/council?q=missing").text
+    finally:
+        app.dependency_overrides.clear()
+
+
+def test_public_error_state_missing_export_and_database_failure(db_session, export, monkeypatch):
+    app.dependency_overrides[get_db] = lambda: db_session
+    try:
+        council.EXPORT.unlink()
+        response = TestClient(app).get("/admin/council")
+        assert response.status_code == 503 and "Повторить загрузку" in response.text
+        def fail(*args, **kwargs):
+            raise OperationalError("query", {}, Exception("private error"))
+        monkeypatch.setattr("app.admin.council_candidates", fail)
+        response = TestClient(app).get("/admin/council")
+        assert response.status_code == 503 and "private error" not in response.text
     finally:
         app.dependency_overrides.clear()

tests/test_deploy_diagnostic.py

@@ -0,0 +1,70 @@
+import os
+from pathlib import Path
+import re
+import subprocess
+import textwrap
+
+import pytest
+
+
+WORKFLOW = (Path(__file__).parents[1] / ".gitea/workflows/ci.yml").read_text()
+
+
+@pytest.mark.parametrize("keygen_exit", [0, 1])
+def test_fingerprint_output_and_cleanup(tmp_path, keygen_exit):
+    step = WORKFLOW.split("      - name: Deployment key fingerprint\n", 1)[1]
+    step = step.split("      - name:", 1)[0]
+    script = textwrap.dedent(step.split("        run: |\n", 1)[1])
+    keygen = tmp_path / "ssh-keygen"
+    keygen.write_text(
+        "#!/bin/sh\n"
+        'case "$1" in\n'
+        '-y) printf "PUBLIC_MATERIAL\\n"; exit "$KEYGEN_EXIT" ;;\n'
+        '-lf) cat >/dev/null; printf "256 SHA256:test comment (ED25519)\\n" ;;\n'
+        '*) exit 99 ;;\n'
+        "esac\n"
+    )
+    keygen.chmod(0o700)
+    (tmp_path / "deploy_key").write_text("PRIVATE_MATERIAL")
+    (tmp_path / "known_hosts").write_text("HOST_MATERIAL")
+    result = subprocess.run(
+        ["bash", "-x", "-c", script], capture_output=True, text=True,
+        env={**os.environ, "RUNNER_TEMP": str(tmp_path),
+             "PATH": f"{tmp_path}:{os.environ['PATH']}",
+             "KEYGEN_EXIT": str(keygen_exit)},
+    )
+    assert (result.returncode == 0) == (keygen_exit == 0)
+    if keygen_exit == 0:
+        assert result.stdout == "SHA256:test\n"
+    assert "MATERIAL" not in result.stdout + result.stderr
+    assert "comment" not in result.stdout + result.stderr
+    assert not (tmp_path / "deploy_key").exists()
+    assert not (tmp_path / "known_hosts").exists()
+
+
+@pytest.mark.parametrize("event", ["push", "pull_request", "workflow_dispatch"])
+@pytest.mark.parametrize("diagnostic_only", ["true", "false", ""])
+def test_launch_conditions(event, diagnostic_only):
+    job = WORKFLOW.split("\n  deploy:\n", 1)[1]
+    condition = re.search(r"^    if: (.+)$", job, re.MULTILINE).group(1)
+    terms = re.findall(r"(github\.[a-z_.]+) == '([^']*)'", condition)
+    assert condition == " && ".join(f"{name} == '{value}'" for name, value in terms)
+    context = {"github.event_name": event,
+               "github.event.inputs.diagnostic_only": diagnostic_only}
+    runs = all(context[name] == value for name, value in terms)
+    assert runs == (event == "workflow_dispatch" and diagnostic_only == "true")
+
+
+def test_both_vps_steps_are_unconditionally_disabled():
+    for name in ("Upload release", "Deploy and verify"):
+        step = WORKFLOW.split(f"      - name: {name}\n", 1)[1]
+        assert step.startswith("        if: ${{ false }}\n")
+
+
+def test_push_and_pull_request_keep_tests():
+    triggers = WORKFLOW.split("\njobs:\n", 1)[0]
+    for event in ("push", "pull_request"):
+        assert f"  {event}:\n    branches: [main]\n" in triggers
+    test_job = WORKFLOW.split("\n  test:\n", 1)[1].split("\n  deploy:\n", 1)[0]
+    assert "    if:" not in test_job
+    assert "python -m pytest -q" in test_job

tests/test_publication_import_postgres.py

@@ -0,0 +1,280 @@
+"""Real PostgreSQL checks; opt in only inside tests/postgres/compose.yml."""
+
+import copy
+import csv
+import json
+import os
+import subprocess
+import sys
+from contextlib import contextmanager
+from pathlib import Path
+
+import pytest
+import requests
+from sqlalchemy import create_engine, select, text
+from sqlalchemy.orm import Session
+
+from app.config import Settings
+from app.db import Base
+from app.models import CrawlError, CrawlRun, Employee, EmployeePublication
+from app.services.crawler import _try_sync_employee_publications, run_crawl
+
+
+ADMIN_URL = "postgresql+psycopg://test_admin:isolated_test_admin@postgres-test:5432/miem_import_test"
+pytestmark = pytest.mark.skipif(
+    os.environ.get("MIEM_POSTGRES_INTEGRATION") != "1", reason="Requires isolated PostgreSQL Compose project"
+)
+
+
+@pytest.fixture(autouse=True)
+def isolated_environment(monkeypatch):
+    assert os.environ.get("DATABASE_URL") == ADMIN_URL, "Use the dedicated test Compose project"
+
+    def forbidden_request(*args, **kwargs):
+        raise AssertionError("External HTTP requests are forbidden in PostgreSQL integration tests")
+
+    monkeypatch.setattr(requests.sessions.Session, "request", forbidden_request)
+
+
+@contextmanager
+def database(name):
+    assert name in {"miem_import_test", "miem_crawl_test"}
+    engine = create_engine(ADMIN_URL.replace("/miem_import_test", "/" + name))
+    try:
+        with engine.connect() as connection:
+            assert connection.scalar(text("SELECT current_database()")) == name
+        Base.metadata.create_all(engine)
+        with engine.begin() as connection:
+            connection.execute(text("GRANT USAGE ON SCHEMA public TO import_test"))
+            connection.execute(text("GRANT SELECT (id, full_name) ON public.employees TO import_test"))
+            connection.execute(text("GRANT SELECT (id, employee_id, title, year, doi_url, raw_data) ON public.employee_publications TO import_test"))
+            connection.execute(text("GRANT INSERT (employee_id, title, year, doi_url, raw_data, source_hash, created_at, updated_at) ON public.employee_publications TO import_test"))
+            connection.execute(text("GRANT UPDATE (title, year, doi_url, raw_data, updated_at) ON public.employee_publications TO import_test"))
+            # Created from the actual model: not an assumed sequence name.
+            sequence = connection.scalar(text("SELECT pg_get_serial_sequence('public.employee_publications', 'id')"))
+            assert sequence == "public.employee_publications_id_seq"
+            connection.execute(text("GRANT USAGE ON SEQUENCE public.employee_publications_id_seq TO import_test"))
+            permissions = connection.execute(text("""SELECT
+                has_schema_privilege('import_test', 'public', 'CREATE'),
+                has_table_privilege('import_test', 'public.employees', 'INSERT'),
+                has_table_privilege('import_test', 'public.employee_publications', 'DELETE'),
+                has_sequence_privilege('import_test', 'public.employee_publications_id_seq', 'UPDATE')
+            """)).one()
+            assert not any(permissions)
+        with Session(engine) as session:
+            yield session, name
+    finally:
+        # Only the dedicated ephemeral test DB; never DROP DATABASE or roles.
+        Base.metadata.drop_all(engine)
+        engine.dispose()
+
+
+@pytest.fixture
+def pg_import_db():
+    with database("miem_import_test") as db:
+        yield db
+
+
+@pytest.fixture
+def pg_crawl_db():
+    with database("miem_crawl_test") as db:
+        yield db
+
+
+def employee(session, name="Synthetic One", key="one"):
+    record = Employee(full_name=name, profile_key="staff:" + key,
+                      canonical_url="https://synthetic.invalid/staff/" + key)
+    session.add(record)
+    session.commit()
+    return record
+
+
+def article(title="Synthetic article", identifier="10.1234/test", year="2024", author="Synthetic One", **metadata):
+    return {"Название публикации": title, "Год": year, "DOI": identifier,
+            "Авторы НИУ ВШЭ": author, **metadata}
+
+
+def write_csv(path, rows):
+    columns = list(dict.fromkeys(key for row in rows for key in row))
+    with path.open("w", encoding="utf-8-sig", newline="") as output:
+        writer = csv.DictWriter(output, fieldnames=columns, delimiter=";")
+        writer.writeheader()
+        writer.writerows(rows)
+
+
+def cli(tmp_path, db, rows, *, apply=False, expected_code=0):
+    _, name = db
+    source = tmp_path / "articles.csv"
+    report = tmp_path / "report.json"
+    write_csv(source, rows)
+    command = [sys.executable, str(Path(__file__).resolve().parents[1] / "scripts/import_publications.py"),
+               str(source), "--report", str(report)]
+    if apply:
+        command.append("--apply")
+    # Explicit whitelist: no production env, .env, pg_service or ~/.pgpass.
+    environment = {"PATH": os.environ["PATH"], "LANG": "C.UTF-8", "PGHOST": "postgres-test",
+                   "PGPORT": "5432", "PGDATABASE": name, "PGUSER": "import_test",
+                   "PGPASSWORD": "isolated_test_import", "PGPASSFILE": "/dev/null",
+                   "PGSERVICEFILE": "/dev/null", "PGCONNECT_TIMEOUT": "5"}
+    result = subprocess.run(command, env=environment, cwd=tmp_path, capture_output=True, text=True, timeout=60)
+    assert result.returncode == expected_code, result.stdout + result.stderr
+    assert "Synthetic One" not in result.stdout and "isolated_test_import" not in result.stdout
+    return json.loads(report.read_text()) if report.exists() else None
+
+
+def snapshot(session):
+    session.expire_all()
+    tables = {table.name: session.execute(select(table).order_by(*table.primary_key.columns)).all()
+              for table in Base.metadata.sorted_tables}
+    sequences = session.execute(text("SELECT sequencename, last_value FROM pg_sequences WHERE schemaname = 'public' ORDER BY sequencename")).all()
+    session.commit()
+    return tables, sequences
+
+
+def test_postgres_dry_run_changes_neither_rows_nor_sequences(pg_import_db, tmp_path):
+    session, _ = pg_import_db
+    employee(session)
+    before = snapshot(session)
+    report = cli(tmp_path, pg_import_db, [article()])
+    assert report["summary"]["counts"] == {"insert": 1, "update": 0, "skip": 0}
+    assert snapshot(session) == before
+
+
+def test_postgres_apply_and_reimport(pg_import_db, tmp_path):
+    session, _ = pg_import_db
+    person = employee(session)
+    existing = EmployeePublication(employee_id=person.id, title="Existing", year=2024,
+                                   source_hash="existing", raw_data={"hse": "keep"})
+    session.add(existing)
+    session.commit()
+    rows = [article(title="Existing"), article(title="New", identifier="10.1234/new"),
+            article(title="Unknown author", identifier="10.1234/unknown", author="Synthetic Missing")]
+    report = cli(tmp_path, pg_import_db, rows, apply=True)
+    assert report["summary"]["counts"] == {"insert": 1, "update": 1, "skip": 1}
+    session.expire_all()
+    assert session.query(Employee).count() == 1
+    assert session.query(EmployeePublication).count() == 2
+    assert existing.doi_url == "https://doi.org/10.1234/test" and existing.raw_data["hse"] == "keep"
+    before = snapshot(session)
+    repeated = cli(tmp_path, pg_import_db, rows, apply=True)
+    assert repeated["summary"]["counts"] == {"insert": 0, "update": 0, "skip": 3}
+    assert snapshot(session) == before
+
+
+def test_postgres_apply_failure_rolls_back_update_and_insert(pg_import_db, tmp_path):
+    session, _ = pg_import_db
+    person = employee(session)
+    existing = EmployeePublication(employee_id=person.id, title="Existing", year=2024,
+                                   source_hash="existing", raw_data={"hse": "keep"})
+    session.add(existing)
+    session.commit()
+    # Test-only constraint forces a real server error after update + successful INSERT.
+    session.execute(text("ALTER TABLE public.employee_publications ADD CONSTRAINT test_failure CHECK (title <> 'Reject')"))
+    session.commit()
+    before, _ = snapshot(session)
+    rows = [article(title="Existing"), article(title="Accepted", identifier="10.1234/accepted"),
+            article(title="Reject", identifier="10.1234/reject")]
+    assert cli(tmp_path, pg_import_db, rows, apply=True, expected_code=1) is None
+    after, _ = snapshot(session)
+    assert after == before  # nextval may advance on rollback; rows must not change.
+    assert (tmp_path / "report.json.pending").exists()
+
+
+def test_postgres_year_only_update_preserves_json(pg_import_db, tmp_path):
+    session, _ = pg_import_db
+    person = employee(session)
+    row = article()
+    raw = {"hse": {"keep": True}, "csv_enrichment": row.copy()}
+    record = EmployeePublication(employee_id=person.id, title=row["Название публикации"],
+                                 year=None, doi_url="https://doi.org/10.1234/test",
+                                 source_hash="existing", raw_data=raw)
+    session.add(record)
+    session.commit()
+    report = cli(tmp_path, pg_import_db, [row], apply=True)
+    session.expire_all()
+    assert report["summary"]["counts"]["update"] == 1
+    assert record.year == 2024 and record.raw_data == raw
+
+
+def test_postgres_late_csv_metadata_and_conflicts(pg_import_db, tmp_path):
+    session, _ = pg_import_db
+    employee(session)
+    rows = [article(), article(**{"EID (Scopus)": "late-id", "Квартиль scopus": "Q1"}),
+            article(**{"EID (Scopus)": "conflict-id", "UT (WoS)": "late-wos"})]
+    report = cli(tmp_path, pg_import_db, rows, apply=True)
+    session.expire_all()
+    assert session.query(EmployeePublication).count() == 1
+    metadata = session.query(EmployeePublication).one().raw_data["csv_enrichment"]
+    assert metadata["EID (Scopus)"] == "late-id" and metadata["Квартиль scopus"] == "Q1"
+    assert metadata["UT (WoS)"] == "late-wos"
+    assert report["summary"]["metadata_conflicts"] == 1
+    assert report["records"][2]["conflict_fields"] == ["EID (Scopus)"]
+    before = snapshot(session)
+    repeated = cli(tmp_path, pg_import_db, rows, apply=True)
+    assert repeated["summary"]["counts"]["insert"] == repeated["summary"]["counts"]["update"] == 0
+    assert snapshot(session) == before
+
+
+def test_postgres_crawler_preserves_import_and_adopts_hse_id(pg_crawl_db, tmp_path):
+    session, _ = pg_crawl_db
+    person = employee(session)
+    cli(tmp_path, pg_crawl_db, [article(), article(title="CSV only", identifier="10.1234/only")], apply=True)
+    session.expire_all()
+    ids = {p.title: p.id for p in session.query(EmployeePublication).all()}
+    run = CrawlRun(source_url="https://synthetic.invalid/persons", status="running")
+    session.add(run)
+    session.commit()
+    parsed = {"sections": [{"type": "publications", "publications": [
+        {"publication_id": "hse-one", "title": "Synthetic article", "year": 2024,
+         "doi_url": "https://doi.org/10.1234/test", "raw_data": {"id": "hse-one"}}]}]}
+    for _ in range(2):
+        _try_sync_employee_publications(session, run, person, copy.deepcopy(parsed))
+        session.commit()
+        session.expire_all()
+    assert session.query(EmployeePublication).count() == 2
+    kept = session.get(EmployeePublication, ids["Synthetic article"])
+    assert kept.publication_id == "hse-one" and "csv_enrichment" in kept.raw_data
+    assert session.get(EmployeePublication, ids["CSV only"]).raw_data["csv_enrichment"]
+    assert session.query(CrawlError).count() == 0
+
+
+@pytest.mark.parametrize("bound_id", [None, "hse-A"])
+@pytest.mark.parametrize("incoming_ids", [("hse-A", "hse-B"), ("hse-B", "hse-A")])
+def test_postgres_hse_id_conflict_does_not_block_other_employee(pg_crawl_db, tmp_path, monkeypatch, bound_id, incoming_ids):
+    session, _ = pg_crawl_db
+    first = employee(session)
+    second = employee(session, "Synthetic Two", "two")
+    cli(tmp_path, pg_crawl_db, [article(identifier="")], apply=True)
+    session.expire_all()
+    original = session.query(EmployeePublication).filter_by(employee_id=first.id).one()
+    original.publication_id = bound_id
+    original.authors = [{"id": "original-author"}]
+    session.commit()
+    fields = ("publication_id", "source_hash", "raw_data", "authors", "updated_at")
+    before = {field: copy.deepcopy(getattr(original, field)) for field in fields}
+    session.commit()
+
+    def profile(person, publications):
+        return {"source_url": person.canonical_url, "profile_type": "staff",
+                "profile_id": person.profile_key.split(":")[1], "full_name": person.full_name,
+                "parser_version": "integration-test", "tabs": [],
+                "sections": [{"type": "publications", "publications": publications}]}
+
+    profiles = {
+        first.canonical_url: profile(first, [{"publication_id": key, "title": "Synthetic article", "year": 2024,
+                                             "authors": [{"id": key}], "raw_data": {"id": key}} for key in incoming_ids]),
+        second.canonical_url: profile(second, [{"publication_id": "second-article", "title": "Other employee article",
+                                               "year": 2024, "raw_data": {"id": "second-article"}}]),
+    }
+    monkeypatch.setattr("app.services.crawler.collect_profile_links", lambda *a, **k: list(profiles))
+    monkeypatch.setattr("app.services.crawler.parse_person_profile", lambda _, url, *a, **k: copy.deepcopy(profiles[url]))
+    settings = Settings(_env_file=None, database_url=ADMIN_URL.replace("miem_import_test", "miem_crawl_test"),
+                        source_url="https://synthetic.invalid/persons", request_delay_seconds=0)
+    run = run_crawl(session, settings)
+    session.expire_all()
+    assert run.status == "completed" and run.parsed_count == 2
+    kept = session.query(EmployeePublication).filter_by(employee_id=first.id).one()
+    assert {field: getattr(kept, field) for field in fields} == before
+    assert session.query(EmployeePublication).filter_by(employee_id=second.id).one().publication_id == "second-article"
+    error = session.query(CrawlError).filter_by(crawl_run_id=run.id).one()
+    assert "Ambiguous HSE IDs" in error.message