feat: filter employees by academic degree

This commit is contained in:
2026-08-26 13:39:30 +03:00
parent 3cfd1d5532
commit f3403b371e
9 changed files with 114 additions and 12 deletions

View File

@@ -1,5 +1,6 @@
from __future__ import annotations
import re
from datetime import date, datetime, time
from math import ceil
from typing import Any
@@ -19,6 +20,8 @@ EMPLOYEE_SORTS = {
"hse_start_year": Employee.current_data["hse_start_year"].as_integer(),
}
_ACADEMIC_DEGREE_PATTERN = re.compile(r"\b(?:кандидат|доктор)\s+[\w\s-]{0,80}?\s+наук\b|\bph\.?\s*d\.?\b", re.IGNORECASE)
def employee_display_payload(employee: Employee) -> dict[str, Any]:
data = _as_dict(employee.current_data)
@@ -28,6 +31,7 @@ def employee_display_payload(employee: Employee) -> dict[str, Any]:
positions = _clean_list(data.get("positions"))
emails = _clean_list(contacts.get("emails"))
phones = _clean_list(contacts.get("phones"))
academic_degrees = _academic_degrees(sections)
return {
"id": employee.id,
"full_name": employee.full_name,
@@ -42,6 +46,7 @@ def employee_display_payload(employee: Employee) -> dict[str, Any]:
"phones": phones,
"phone_text": ", ".join(phones),
"address": contacts.get("address"),
"academic_degree_text": "; ".join(academic_degrees),
"publications_count": _count_section_items(sections, "publications"),
"courses_count": _count_section_items(sections, "courses_by_year"),
"news_count": len(stored_news_links) or _count_section_items(sections, "news"),
@@ -81,6 +86,7 @@ def build_employee_query(
started_from: date | None = None,
started_to: date | None = None,
has_email: bool | None = None,
has_academic_degree: bool | None = None,
) -> Select[tuple[Employee]]:
stmt = select(Employee)
filters = []
@@ -97,6 +103,15 @@ def build_employee_query(
filters.append(Employee.current_data.cast(Text).ilike("%@%"))
elif has_email is False:
filters.append(or_(Employee.current_data.is_(None), ~Employee.current_data.cast(Text).ilike("%@%")))
if has_academic_degree is not None:
data_text = Employee.current_data.cast(Text)
degree_condition = or_(
and_(_json_text_contains(data_text, "кандидат"), _json_text_contains(data_text, "наук")),
and_(_json_text_contains(data_text, "доктор"), _json_text_contains(data_text, "наук")),
_json_text_contains(data_text, "phd"),
_json_text_contains(data_text, "ph.d."),
)
filters.append(degree_condition if has_academic_degree else or_(Employee.current_data.is_(None), ~degree_condition))
if filters:
stmt = stmt.where(and_(*filters))
return stmt
@@ -110,6 +125,7 @@ def list_employees_page(
started_from: date | None = None,
started_to: date | None = None,
has_email: bool | None = None,
has_academic_degree: bool | None = None,
sort: str = "full_name",
direction: str = "asc",
limit: int = 50,
@@ -123,6 +139,7 @@ def list_employees_page(
started_from=started_from,
started_to=started_to,
has_email=has_email,
has_academic_degree=has_academic_degree,
)
total = db.scalar(select(func.count()).select_from(base_stmt.subquery())) or 0
sort_column = EMPLOYEE_SORTS.get(sort, Employee.full_name)
@@ -214,6 +231,36 @@ def format_admin_datetime(value: Any) -> str:
return value.strftime("%d.%m.%Y %H:%M")
def _academic_degrees(sections: list[Any]) -> list[str]:
degrees = []
for section in sections:
section_data = _as_dict(section)
title = str(section_data.get("title") or "")
if not re.search(r"уч[её]н.*степен|academic degree", title, re.IGNORECASE):
continue
values = [
*(_as_dict(entry).get("text") for entry in _as_list(section_data.get("year_entries"))),
*_clean_list(section_data.get("paragraphs")),
*_clean_list(section_data.get("items")),
section_data.get("raw_text"),
]
for value in values:
text = str(value or "").strip()
if text and _ACADEMIC_DEGREE_PATTERN.search(text) and text not in degrees:
degrees.append(text)
return degrees
def _json_text_contains(data_text: Any, value: str) -> Any:
escaped = value.encode("unicode_escape").decode("ascii")
escaped_capitalized = value.capitalize().encode("unicode_escape").decode("ascii")
return or_(
data_text.ilike(f"%{value}%"),
data_text.ilike(f"%{escaped}%"),
data_text.ilike(f"%{escaped_capitalized}%"),
)
def _employee_status_display(status: str | None) -> str:
labels = {"active": "Работает", "dismissed": "Уволен"}
return labels.get(status or "", status or "Не указано")