fix: speed up academic degree filter

app/services/admin_data.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import re
 from datetime import date, datetime, time
 from math import ceil
 from typing import Any
@@ -10,6 +9,7 @@ from sqlalchemy import Select, Text, and_, desc, func, or_, select
 from sqlalchemy.orm import Session
 
 from app.models import CrawlError, CrawlRun, CrawlRunEmployeeChange, Employee, EmployeeNewsLink
+from app.services.academic_degrees import academic_degrees
 
 EMPLOYEE_SORTS = {
     "full_name": Employee.full_name,
@@ -20,9 +20,6 @@ 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)
     contacts = _as_dict(data.get("contacts"))
@@ -31,7 +28,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)
+    degree_values = academic_degrees(data)
     return {
         "id": employee.id,
         "full_name": employee.full_name,
@@ -46,7 +43,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),
+        "academic_degree_text": "; ".join(degree_values),
         "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"),
@@ -104,14 +101,7 @@ def build_employee_query(
     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))
+        filters.append(Employee.has_academic_degree.is_(has_academic_degree))
     if filters:
         stmt = stmt.where(and_(*filters))
     return stmt
@@ -235,36 +225,6 @@ 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": "Работает", "verification_required": "Требует проверки", "dismissed": "Уволен"}
     return labels.get(status or "", status or "Не указано")