Compare commits

..
Author SHA1 Message Date
codex 32e26ccdd2 feat: add council employee sorting by name and publication counts
CI / test (pull_request) Successful in 9m15s
CI / Fingerprint diagnostics (production deployment temporarily disabled) (pull_request) Has been skipped
2026-10-07 16:50:29 +03:00
admin fac0431a03 Merge pull request 'Диссертационный совет: публичные публикации из CSV и БД' (#57) from feature/public-council-publications into main
CI / test (push) Successful in 9m10s
CI / Fingerprint diagnostics (production deployment temporarily disabled) (push) Has been skipped
Reviewed-on: #57
Reviewed-by: admin <av_progs@mail.ru>
2026-10-07 09:56:33 +00:00
agent ed9d0b48c9 fix: restore doctoral degree requirement for council candidates
CI / test (pull_request) Successful in 9m13s
CI / Fingerprint diagnostics (production deployment temporarily disabled) (pull_request) Has been skipped
2026-10-07 11:58:18 +03:00
agent c6c6866235 feat: publish council publication evidence from CSV and database
CI / test (pull_request) Successful in 9m13s
CI / Fingerprint diagnostics (production deployment temporarily disabled) (pull_request) Has been skipped
2026-10-07 11:35:28 +03:00
admin 66c8e9e0c3 Merge pull request 'Временно отключить production deploy и оставить диагностику fingerprint' (#56) from fix/deploy-key-fingerprint-diagnostic into main
CI / test (push) Successful in 9m12s
CI / Fingerprint diagnostics (production deployment temporarily disabled) (push) Has been skipped
Reviewed-on: #56
Reviewed-by: agent <agent@vakanaut.ru>
2026-10-06 07:08:04 +00:00
codex f39b5dd512 fix: temporarily disable production deployment for all events
CI / test (pull_request) Successful in 9m11s
CI / Fingerprint diagnostics (production deployment temporarily disabled) (pull_request) Has been skipped
2026-10-05 18:35:19 +03:00
codex b8a27758b6 fix: add deployment key fingerprint diagnostic mode
CI / test (pull_request) Successful in 9m13s
CI / deploy (pull_request) Has been skipped
2026-10-05 16:28:06 +03:00
admin c975e7418b Merge pull request 'Изолированная PostgreSQL-проверка импорта и сохранности публикаций' (#55) from feature/postgresql-import-verification into main
CI / test (push) Successful in 9m12s
CI / deploy (push) Failing after 11s
Reviewed-on: #55
Reviewed-by: agent <agent@vakanaut.ru>
2026-10-05 12:12:32 +00:00
15 changed files with 682 additions and 155 deletions

No files matched your search

+18 -1
View File
@@ -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
+2
View File
@@ -9,3 +9,5 @@ pytest-cache-files-*/
htmlcov/
postgres_data/
MCP_DESCRIPTION.md
.impeccable/review/
+7 -3
View File
@@ -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,7 +145,7 @@ node --check app/static/admin.js
Тесты с данными используют временную SQLite; отдельные API smoke-тесты запускают приложение с его текущей конфигурацией. Браузерная проверка: `pip install playwright`, `python -m playwright install chromium`, затем `python tests/browser_admin.py`. Она проверяет реальный рендеринг страниц, клавиатуру, диалог, фильтры и восстановление прогресса после ошибки. Скриншоты сохраняются во временную папку; путь выводится в конце.
Версия сервиса: `0.10.4`. Админка всегда показывает версии backend и frontend в footer.
Версия сервиса: `0.10.6`. Админка всегда показывает версии backend и frontend в footer.
## Разовый импорт публикаций
+5 -5
View File
@@ -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)
+131 -57
View File
@@ -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}
+25 -1
View File
@@ -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; }
}
+6 -1
View File
@@ -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">
+60 -26
View File
@@ -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 %}
+3 -3
View File
@@ -1,3 +1,3 @@
APP_VERSION = "0.10.4"
FRONTEND_VERSION = "0.10.4"
BACKEND_VERSION = "0.10.4"
APP_VERSION = "0.12.0"
FRONTEND_VERSION = "0.12.0"
BACKEND_VERSION = "0.12.0"
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "miem-workers"
version = "0.10.4"
version = "0.12.0"
description = "MIEM employees parser, admin API, and web admin"
requires-python = ">=3.11"
dependencies = [
+139
View File
@@ -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()
+7 -2
View File
@@ -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
+1 -1
View File
@@ -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.4"
assert response.json()["backend_version"] == "0.12.0"
def test_api_employees_and_stats_require_admin_session():
+207 -54
View File
@@ -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()
+70
View File
@@ -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