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

This commit is contained in:
agent committed 2026-10-07 11:35:28 +03:00
1 parent 66c8e9e0c3
commit c6c6866235
13 files changed
+487 -153

No files matched your search

+2
View File
@@ -9,3 +9,5 @@ pytest-cache-files-*/
htmlcov/
postgres_data/
MCP_DESCRIPTION.md
.impeccable/review/
+3 -1
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`, независимо от степени и количества работ; уволенные сотрудники отмечены отдельно. Есть поиск по ФИО и страницы по 25 сотрудников. У каждого сотрудника видны степень, число уникальных работ из CSV с Q1/Q2 за скользящие пять лет, число уникальных работ в БД и общий перечень без дублей с источниками, типами и подтверждёнными критериями.
CSV `app/data/council_articles.csv` сопоставляется по полному ФИО только при единственном совпадении. Публикации объединяются по DOI, затем по нормализованным названию и году, если нет конфликтующих DOI. Совпадение в двух источниках сохраняет оба источника, но учитывается один раз в общем количестве. Импорт CSV в БД не выдаётся за данные сайта. CSV не содержит типа материала, точной даты и подтверждения К1/RSCI: неизвестные сведения обозначены явно; счётчик CSV предварительный. Публичная страница не открывает контакты, служебные данные или операции админки.
Список **предварительный**: в CSV указаны год и квартиль Scopus/WoS, но нет точной даты, типа материала, журнала, источника квартиля, подтверждения К1 и RSCI. На границах периода год не позволяет проверить точную дату. Статьи с Q1/Q2 учитываются с пометкой источника; отсутствие кандидата не доказывает недостаточную публикационную активность. Для подтверждённого отбора нужны проверка журналов и дат и дополнительные источники публикаций. Выгрузка поставляется вместе с релизом и обновляется отдельным изменением репозитория.
+4 -5
View File
@@ -134,14 +134,13 @@ 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, db: Session = Depends(get_db)):
try:
result = council_candidates(db)
result = council_candidates(db, q=q, offset=offset)
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)
+125 -59
View File
@@ -1,37 +1,47 @@
"""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
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 +49,135 @@ 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) -> 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):
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)}
degrees = []
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"})
candidates.sort(key=lambda item: (-item["csv_count"], -item["count"], item["name"].casefold()))
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}
+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">
+47 -26
View File
@@ -1,59 +1,72 @@
{% 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>
<button class="button" type="submit">Найти</button>
{% if q %}<a class="admin__link" href="/admin/council">Сбросить поиск</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 +74,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;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;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="/admin/council">Сбросить поиск</a>{% endif %}
</div>
{% endif %}
{% endif %}
+3 -3
View File
@@ -1,3 +1,3 @@
APP_VERSION = "0.10.6"
FRONTEND_VERSION = "0.10.6"
BACKEND_VERSION = "0.10.6"
APP_VERSION = "0.11.0"
FRONTEND_VERSION = "0.11.0"
BACKEND_VERSION = "0.11.0"
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "miem-workers"
version = "0.10.6"
version = "0.11.0"
description = "MIEM employees parser, admin API, and web admin"
requires-python = ">=3.11"
dependencies = [
+104
View File
@@ -0,0 +1,104 @@
"""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}"))
db.commit()
def database():
with factory() as db:
yield db
original = council._articles_by_author
council._articles_by_author = lambda: {}
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")
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)
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.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")
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")
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.6"
assert response.json()["backend_version"] == "0.11.0"
def test_api_employees_and_stats_require_admin_session():
+159 -53
View File
@@ -1,76 +1,182 @@
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=None, 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_every_employee_with_a_publication_no_degree_or_count_threshold(db_session, export):
export([row(author="Кандидат", title="Single qualifying"),
row(author="Без степени", title="Not Q1/Q2", **{"Квартиль scopus": "Q3"}),
row(author="Уволен", title="Old", year="2010")])
employee(db_session, "Кандидат", "Кандидат технических наук", key="candidate")
employee(db_session, "Без степени", key="none")
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"] == 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
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, degree="Кандидат технических наук")
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
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 = client.get("/admin/council", follow_redirects=False)
assert response.status_code == 200
assert "Предварительный список" in response.text
assert "К1 или RSCI могут отсутствовать" in response.text
assert "Иванов Иван Иванович" in response.text and "Кандидат технических наук" 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.11.0 · Frontend 0.11.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()