# backend/employees/admin.py from io import StringIO from django import forms from django.contrib import admin, messages from django.contrib.auth.admin import GroupAdmin as DefaultGroupAdmin from django.contrib.auth.admin import UserAdmin as DjangoUserAdmin from django.contrib.auth.forms import ( AdminPasswordChangeForm, ReadOnlyPasswordHashField, ) from django.contrib.auth.models import Group, Permission from django.core.exceptions import PermissionDenied from django.core.management import call_command from django.core.management.base import CommandError from django.template.response import TemplateResponse from django.urls import path, reverse from django.utils import timezone from django.utils.html import format_html from simple_history.admin import SimpleHistoryAdmin from .models import ( Department, DepartmentRole, Employee, EmployeeAction, EmployeeDepartment, EmployeeSkillProgress, LdapSyncState, Position, SecretSkillSettings, Skill, SkillExperienceEvent, ) class SecretSkillSettingsAdminForm(forms.ModelForm): class Meta: model = SecretSkillSettings fields = ("calculation_start_date",) def clean_calculation_start_date(self): value = self.cleaned_data["calculation_start_date"] if value > timezone.localdate(): raise forms.ValidationError( "Дата начала расчёта не может быть из будущего." ) return value class SecretSkillRecalculateAdminForm(forms.Form): employee = forms.ModelChoiceField( label="Сотрудник", queryset=Employee.objects.none(), required=False, help_text="Не выбирайте сотрудника, чтобы проверить или пересчитать всех.", ) dry_run = forms.BooleanField( label="Сухой прогон без сохранения", required=False, initial=True, ) confirm = forms.BooleanField( label="Подтверждаю реальный пересчёт и сохранение изменений", required=False, ) def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.fields["employee"].queryset = Employee.objects.order_by( "last_name", "first_name", "id", ) def clean(self): cleaned_data = super().clean() if not cleaned_data.get("dry_run") and not cleaned_data.get("confirm"): raise forms.ValidationError( "Для реального пересчёта подтвердите сохранение изменений." ) return cleaned_data class SecretSkillReportAdminForm(forms.Form): top = forms.IntegerField( label="Количество сотрудников в рейтинге", min_value=1, max_value=100, initial=3, ) @admin.register(SecretSkillSettings) class SecretSkillSettingsAdmin(admin.ModelAdmin): """Единая дата начала расчёта всех секретных навыков.""" form = SecretSkillSettingsAdminForm list_display = ("calculation_start_date", "updated_at", "tools_link") fields = ("calculation_start_date", "updated_at", "tools_link") readonly_fields = ("updated_at", "tools_link") def get_urls(self): custom_urls = [ path( "tools/", self.admin_site.admin_view(self.tools_view), name="employees_secretskillsettings_tools", ) ] return custom_urls + super().get_urls() @admin.display(description="Операции") def tools_link(self, obj=None): url = reverse("admin:employees_secretskillsettings_tools") return format_html('Открыть инструменты', url) def tools_view(self, request): if not self.has_change_permission(request): raise PermissionDenied operation = request.POST.get("operation") recalculate_form = SecretSkillRecalculateAdminForm( request.POST if operation == "recalculate" else None, prefix="recalculate", ) report_form = SecretSkillReportAdminForm( request.POST if operation == "report" else None, prefix="report", ) output = "" try: if operation == "recalculate" and recalculate_form.is_valid(): stream = StringIO() employee = recalculate_form.cleaned_data["employee"] command_options = { "dry_run": recalculate_form.cleaned_data["dry_run"], "stdout": stream, } if employee is not None: command_options["employee_id"] = employee.pk call_command("recalculate_secret_skills", **command_options) output = stream.getvalue() messages.success(request, "Пересчёт завершён.") elif operation == "report" and report_form.is_valid(): stream = StringIO() call_command( "secret_skills_report", top=report_form.cleaned_data["top"], stdout=stream, ) output = stream.getvalue() messages.success(request, "Отчёт сформирован.") except CommandError as error: messages.error(request, str(error)) context = { **self.admin_site.each_context(request), "opts": self.model._meta, "title": "Инструменты секретных навыков", "settings_object": SecretSkillSettings.get_solo(), "recalculate_form": recalculate_form, "report_form": report_form, "output": output, } return TemplateResponse( request, "admin/employees/secretskillsettings/tools.html", context, ) def has_add_permission(self, request): if SecretSkillSettings.objects.exists(): return False return super().has_add_permission(request) def has_delete_permission(self, request, obj=None): return False def _resolve_ldap_user_for_employee(employee): """Возвращает наиболее вероятную LDAP-модель сотрудника. Идет через общий resolver: сначала sync-state DN, затем employee_number, затем email/UPN fallback среди кандидатов. """ from api.v1.directory.services import _find_ldap_user return _find_ldap_user(employee) # ========================= # Формы для кастомного User (Employee) # ========================= class LdapSyncedFilter(admin.SimpleListFilter): """Фильтр для отображения сотрудников по состоянию LDAP синхронизации.""" title = "Статус LDAP синхронизации" parameter_name = "ldap_synced" def lookups(self, request, model_admin): return ( ("yes", "Есть запись синхронизации"), ("no", "Нет записи синхронизации"), ) def queryset(self, request, queryset): from django.db.models import Exists, OuterRef if self.value() == "yes": # Показываем только тех, у кого есть запись синхронизации return queryset.filter( Exists( LdapSyncState.objects.filter( model="employee", object_pk=OuterRef("pk") ) ) ) elif self.value() == "no": # Показываем только тех, у кого нет записи синхронизации return queryset.exclude( Exists( LdapSyncState.objects.filter( model="employee", object_pk=OuterRef("pk") ) ) ) return queryset class EmployeeCreationForm(forms.ModelForm): """ Создание сотрудника в админке: логин — email, без username. """ password1 = forms.CharField(label="Пароль", widget=forms.PasswordInput) password2 = forms.CharField( label="Подтверждение пароля", widget=forms.PasswordInput ) class Meta: model = Employee fields = ("email", "first_name", "last_name", "phone_number") def clean_password2(self): p1 = self.cleaned_data.get("password1") or "" p2 = self.cleaned_data.get("password2") or "" if not p1: raise forms.ValidationError("Пароль обязателен") if p1 != p2: raise forms.ValidationError("Пароли не совпадают") return p2 def save(self, commit=True): user = super().save(commit=False) user.set_password(self.cleaned_data["password1"]) # При создании через админку обычно активируем и помечаем e-mail # подтверждённым if not user.is_active: user.is_active = True if not user.email_verified: user.email_verified = True if commit: user.save() return user class EmployeeChangeForm(forms.ModelForm): """ Редактирование сотрудника: пароль только для чтения. Смена пароля выполняется отдельной кнопкой. """ password = ReadOnlyPasswordHashField( label="Хэш пароля", help_text=( "Пароль хранится в виде хэша — посмотреть его невозможно. " "Используйте кнопку «Изменить пароль» выше." ), required=False, ) class Meta: model = Employee fields = ( "email", "password", "first_name", "last_name", "patronymic", "gender", "birth_date", "avatar", "phone_number", "telegram", "whatsapp", "wechat", "position", "additional_positions", "skills", "is_active", "is_hidden_in_ui", "email_verified", "is_staff", "is_superuser", "groups", "user_permissions", "last_login", ) def clean_password(self): # Возвращаем исходное значение (хэш), даже если пользователь ничего не # ввёл return self.initial.get("password") # ========================= # Инлайны # ========================= class EmployeeDepartmentInline(admin.TabularInline): """ Принадлежности сотрудника к отделам (видно из карточки сотрудника). """ model = EmployeeDepartment fk_name = "employee" extra = 0 autocomplete_fields = ("department", "role") fields = ("department", "role", "is_active", "date_from", "date_to") show_change_link = True class DepartmentMembershipInline(admin.TabularInline): """ Состав отдела — список участников прямо в карточке отдела. """ model = EmployeeDepartment fk_name = "department" extra = 0 autocomplete_fields = ("employee", "role") fields = ("employee", "role", "is_active", "date_from", "date_to") show_change_link = True # ========================= # Админки моделей # ========================= @admin.register(Employee) class EmployeeAdmin(DjangoUserAdmin): """ Кастомный UserAdmin для модели Employee. Включает стандартную кнопку «Изменить пароль». """ add_form = EmployeeCreationForm form = EmployeeChangeForm model = Employee # Включаем стандартную форму смены пароля и шаблон, # который показывает ссылку «Изменить пароль» на форме изменения # пользователя. change_password_form = AdminPasswordChangeForm change_form_template = "admin/employees/employee/change_form.html" list_display = ( "email", "last_name", "first_name", "phone_number", "position", "is_active", "is_hidden_in_ui", "email_verified", "is_ldap_managed", "ldap_sync_status", "is_staff", ) list_filter = ( "is_active", "is_hidden_in_ui", "email_verified", "is_staff", "is_superuser", "is_ldap_managed", LdapSyncedFilter, "position", "groups", ) search_fields = ("email", "first_name", "last_name", "phone_number") ordering = ("last_name", "first_name") list_select_related = ("position",) filter_horizontal = ( "groups", "user_permissions", "skills", "additional_positions", ) readonly_fields = ( "last_login", "created_at", "updated_at", "ldap_sync_info", "email_activation_code", "sms_activation_code", ) def ldap_sync_status(self, obj): """Краткий статус LDAP синхронизации для списка.""" if not obj.pk: return "-" import logging from django.utils.html import format_html logger = logging.getLogger(__name__) # Если не управляется LDAP if not obj.is_ldap_managed: return format_html('Локальный') try: # Логируем попытку поиска logger.debug( "LDAP sync status lookup started", extra={ "event": "ldap.admin.sync_status_lookup", "employee_id": obj.pk, }, ) sync_state = LdapSyncState.objects.filter( model="employee", object_pk=str(obj.pk) ).first() logger.debug( "LDAP sync status lookup completed", extra={ "event": "ldap.admin.sync_status_result", "employee_id": obj.pk, "sync_state_found": bool(sync_state), }, ) if not sync_state: # Попробуем найти любые записи для этого employee logger.warning( "LDAP sync state is missing", extra={ "event": "ldap.admin.sync_state_missing", "employee_id": obj.pk, }, ) # Создаем ссылку для быстрого создания записи from django.urls import reverse create_url = reverse("admin:employees_ldapsyncstate_add") return format_html( '' '❌ Нет записи (ID: {})' "", obj.pk, create_url, obj.pk, obj.pk, ) # Проверяем полноту данных has_guid = bool(sync_state.ldap_guid) has_dn = bool(sync_state.ldap_dn) has_updated = bool(sync_state.updated_at) # Полные данные if has_guid and has_dn and has_updated: return format_html( '✅ {}', obj.pk, sync_state.ldap_guid, sync_state.ldap_dn[:80] + "..." if len(sync_state.ldap_dn) > 80 else sync_state.ldap_dn, sync_state.updated_at.strftime("%d.%m.%Y %H:%M"), ) # Неполные данные elif has_updated: missing = [] if not has_guid: missing.append("GUID") if not has_dn: missing.append("DN") return format_html( '⚠️ {}', obj.pk, ", ".join(missing), sync_state.updated_at.strftime("%d.%m.%Y %H:%M"), ) else: return format_html( '' "⚠️ Неполные данные", obj.pk, ) except Exception as error: logger.exception( "LDAP sync status lookup failed", extra={ "event": "ldap.admin.sync_status_failed", "employee_id": obj.pk, "error_type": type(error).__name__, }, ) return format_html( '❌ Ошибка', obj.pk, str(error), ) ldap_sync_status.short_description = "LDAP статус" ldap_sync_status.admin_order_field = "id" # Сортировка по ID fieldsets = ( (None, {"fields": ("email", "password")}), ( "Личная информация", { "fields": ( "first_name", "last_name", "patronymic", "gender", "birth_date", "avatar", ) }, ), ( "Контакты", {"fields": ("phone_number", "telegram", "whatsapp", "wechat")}, ), ( "Работа", {"fields": ("position", "additional_positions", "skills")}, ), ( "Права и статусы", { "fields": ( "is_active", "is_hidden_in_ui", "email_verified", "email_activation_code", "sms_activation_code", "is_staff", "is_superuser", "groups", "user_permissions", ) }, ), ( "LDAP", { "fields": ("is_ldap_managed", "ldap_sync_info"), "classes": ("collapse",), }, ), ("Служебное", {"fields": ("last_login", "created_at", "updated_at")}), ) def ldap_sync_info(self, obj): """Отображает информацию о состоянии синхронизации LDAP.""" if not obj.pk: return "-" try: sync_state = LdapSyncState.objects.filter( model="employee", object_pk=str(obj.pk) ).first() if not sync_state: return "Нет записи синхронизации" from django.urls import reverse from django.utils.html import format_html # Ссылка на запись синхронизации в админке sync_url = reverse( "admin:employees_ldapsyncstate_change", args=[sync_state.pk] ) info = [ format_html( '📋 Открыть запись синхронизации', sync_url, ) ] if sync_state.ldap_dn: info.append(f"DN: {sync_state.ldap_dn}") if sync_state.ldap_guid: info.append(f"GUID: {sync_state.ldap_guid}") if sync_state.last_sync_dir: info.append(f"Направление: {sync_state.last_sync_dir}") if sync_state.last_ldap_modify_ts: info.append(f"LDAP изменен: {sync_state.last_ldap_modify_ts}") if sync_state.last_django_modify_ts: info.append( (f"Django изменен: {sync_state.last_django_modify_ts}") ) if sync_state.updated_at: info.append(f"Обновлено: {sync_state.updated_at}") return format_html("
".join(info)) if info else "-" except Exception as e: return f"Ошибка: {e}" ldap_sync_info.short_description = "Состояние LDAP синхронизации" add_fieldsets = ( ( None, { "classes": ("wide",), "fields": ( "email", "first_name", "last_name", "phone_number", "whatsapp", "position", "password1", "password2", "is_active", "email_verified", "is_staff", "is_superuser", "groups", "user_permissions", ), }, ), ) inlines = [EmployeeDepartmentInline] actions = [ "create_ldap_sync_records", "check_ldap_sync_status", "sync_from_django_to_ldap", "sync_from_ldap_to_django", "show_sync_diff", ] def create_ldap_sync_records(self, request, queryset): """Создать записи синхронизации для выбранных сотрудников.""" created_count = 0 already_exists = 0 for employee in queryset: if not employee.is_ldap_managed: continue sync_state, created = LdapSyncState.objects.get_or_create( model="employee", object_pk=str(employee.pk), defaults={ "last_sync_dir": "django", }, ) if created: created_count += 1 else: already_exists += 1 if created_count: self.message_user( request, f"Создано {created_count} записей синхронизации", level="success", ) if already_exists: self.message_user( request, (f"У {already_exists} сотрудников уже есть записи синхронизации"), level="info", ) if created_count == 0 and already_exists == 0: self.message_user( request, "Выбраны только локальные пользователи (не управляются LDAP)", level="warning", ) create_ldap_sync_records.short_description = "Создать записи LDAP синхронизации" def check_ldap_sync_status(self, request, queryset): """Проверить статус LDAP синхронизации для выбранных сотрудников.""" ldap_managed = queryset.filter(is_ldap_managed=True) ldap_count = ldap_managed.count() if ldap_count == 0: self.message_user( request, "Среди выбранных нет пользователей, управляемых через LDAP", level="warning", ) return # Проверяем наличие записей синхронизации sync_records = LdapSyncState.objects.filter( model="employee", object_pk__in=[str(e.pk) for e in ldap_managed] ) has_sync = sync_records.count() missing_sync = ldap_count - has_sync # Проверяем полноту данных complete = ( sync_records.filter(ldap_guid__isnull=False, ldap_dn__isnull=False) .exclude(ldap_dn="") .count() ) incomplete = has_sync - complete message = "Статус LDAP синхронизации:\n" message += f"• Управляются LDAP: {ldap_count}\n" message += f"• Есть запись синхронизации: {has_sync}\n" message += f"• Нет записи синхронизации: {missing_sync}\n" message += f"• Полные данные: {complete}\n" message += f"• Неполные данные: {incomplete}" self.message_user(request, message, level="info") check_ldap_sync_status.short_description = "Проверить статус LDAP синхронизации" @admin.action( description=( "🔄 Синхронизировать Django → LDAP (создать/обновить в Active Directory)" ) ) def sync_from_django_to_ldap(self, request, queryset): """Создаёт отсутствующих LDAP-пользователей и обновляет существующих.""" import logging from employees.ldap.services import UserService logger = logging.getLogger(__name__) service = UserService() counts = {"created": 0, "updated": 0, "current": 0, "error": 0} for employee in queryset: logger.info( "LDAP admin synchronization started", extra={ "event": "ldap.admin.employee_sync.started", "employee_id": employee.pk, "is_ldap_managed": employee.is_ldap_managed, }, ) try: ldap_user = _resolve_ldap_user_for_employee(employee) if ldap_user is None: result = self._create_ldap_employee(request, employee, service) else: result = self._update_ldap_employee( request, employee, ldap_user, service ) counts[result] += 1 except Exception as error: logger.exception( "LDAP admin employee synchronization failed", extra={ "event": "ldap.admin.employee_sync.failed", "employee_id": employee.pk, "error_type": type(error).__name__, }, ) counts["error"] += 1 self.message_user( request, f"Ошибка обработки {employee.email}: {error}", level=messages.ERROR, ) self._report_employee_ldap_sync(request, counts) def _update_ldap_employee(self, request, employee, ldap_user, service): import logging from django.utils import timezone from ldap3 import MODIFY_REPLACE from employees.ldap.infrastructure.connections import _ldap from employees.ldap.orm_models import LdapUser from employees.ldap.utils.ldap_utils import get_ldap_str logger = logging.getLogger(__name__) logger.info( "LDAP admin user update started", extra={ "event": "ldap.admin.employee_update.started", "employee_id": employee.pk, }, ) LdapSyncState.objects.update_or_create( model="employee", object_pk=str(employee.pk), defaults={ "ldap_dn": ldap_user.dn, "last_sync_dir": "django", "last_django_modify_ts": timezone.now(), }, ) if get_ldap_str(getattr(ldap_user, "employee_number", "")) != str(employee.pk): with _ldap() as connection: updated = connection.modify( ldap_user.dn, {"employeeNumber": [(MODIFY_REPLACE, [str(employee.pk)])]}, ) if updated: ldap_user.employee_number = str(employee.pk) else: logger.warning( "LDAP employee number update failed", extra={ "event": "ldap.admin.employee_number.failed", "employee_id": employee.pk, "ldap_result_code": (connection.result or {}).get("result"), }, ) if not employee.is_ldap_managed: employee.is_ldap_managed = True employee._skip_ldap_sync = True employee.save(update_fields=["is_ldap_managed"]) desired_is_active = employee.is_active and employee.is_actually_active changes = self._ldap_employee_changes( employee, ldap_user, desired_is_active, get_ldap_str ) if changes: service.update_user(employee, changes) expected_base = self._ldap_employee_expected_base(employee, desired_is_active) if expected_base: current_parts = ldap_user.dn.split(",", 1) current_base = current_parts[1] if len(current_parts) > 1 else "" if current_base.lower() != expected_base.lower(): self._move_ldap_employee( request, employee, ldap_user, expected_base, desired_is_active ) try: fresh_user = _resolve_ldap_user_for_employee(employee) if fresh_user is None: raise LdapUser.DoesNotExist fresh_dn = fresh_user.dn except LdapUser.DoesNotExist: fresh_dn = ldap_user.dn LdapSyncState.objects.update_or_create( model="employee", object_pk=str(employee.pk), defaults={"ldap_dn": fresh_dn, "last_sync_dir": "django"}, ) return "updated" if changes else "current" @staticmethod def _ldap_employee_changes(employee, ldap_user, desired_is_active, get_ldap_str): changes = {} field_pairs = ( ("first_name", "given_name", employee.first_name), ("last_name", "sn", employee.last_name), ("email", "mail", employee.email), ) for target_field, ldap_field, expected_value in field_pairs: if get_ldap_str(getattr(ldap_user, ldap_field)) != expected_value: changes[target_field] = expected_value ldap_phone = get_ldap_str(ldap_user.telephone_number or ldap_user.mobile) if employee.phone_number and ldap_phone != employee.phone_number: changes["phone_number"] = employee.phone_number try: from employees.ldap.utils.ldap_utils import _uac_is_active ldap_is_active = _uac_is_active(ldap_user.user_account_control) except Exception: ldap_is_active = None if ldap_is_active != desired_is_active: changes["is_active"] = desired_is_active return changes @staticmethod def _ldap_employee_expected_base(employee, desired_is_active): from django.conf import settings as ldap_settings if not desired_is_active: return getattr(ldap_settings, "LDAP_DISMISSED_BASE", None) department = employee.departments.first() if department is None: return None sync_state = LdapSyncState.objects.filter( model="department", object_pk=str(department.pk) ).first() if sync_state and sync_state.ldap_dn: return sync_state.ldap_dn departments_base = getattr(ldap_settings, "LDAP_DEPARTMENTS_BASE", "") return f"OU={department.name},{departments_base}" if departments_base else None def _move_ldap_employee( self, request, employee, ldap_user, expected_base, desired_is_active ): import logging logger = logging.getLogger(__name__) try: ldap_user.move_to(expected_base) logger.info( "LDAP employee moved", extra={ "event": "ldap.admin.employee.moved", "employee_id": employee.pk, }, ) self.message_user( request, f"📦 {employee.email}: перемещён в {expected_base}", level=messages.INFO, ) if desired_is_active: self._add_ldap_employee_to_department_group( employee, ldap_user, expected_base ) except Exception as error: logger.warning( "LDAP employee move failed", extra={ "event": "ldap.admin.employee_move.failed", "employee_id": employee.pk, "error_type": type(error).__name__, }, exc_info=True, ) self.message_user( request, f"⚠️ {employee.email}: не удалось переместить в " f"{expected_base}: {error}", level=messages.WARNING, ) @staticmethod def _add_ldap_employee_to_department_group(employee, ldap_user, expected_base): import logging from employees.ldap.orm_models import LdapOrganizationalUnit logger = logging.getLogger(__name__) try: organizational_unit = LdapOrganizationalUnit.objects.get(dn=expected_base) organizational_unit.ensure_department_group().add_member(ldap_user.dn) except Exception as error: logger.warning( "LDAP department group membership update failed", extra={ "event": "ldap.admin.department_group_membership.failed", "employee_id": employee.pk, "error_type": type(error).__name__, }, exc_info=True, ) def _create_ldap_employee(self, request, employee, service): import logging import secrets from employees.ldap.domain.dtos import DirectoryUserDTO from employees.ldap.utils.phone_utils import normalize_phone logger = logging.getLogger(__name__) logger.info( "LDAP admin user creation started", extra={ "event": "ldap.admin.employee_create.started", "employee_id": employee.pk, }, ) temporary_password = secrets.token_urlsafe(16) desired_is_active = employee.is_active and employee.is_actually_active department_dn = self._ldap_employee_expected_base(employee, desired_is_active) phone_e164 = ( normalize_phone(employee.phone_number) if employee.phone_number else None ) service.create_user( DirectoryUserDTO( first_name=employee.first_name, last_name=employee.last_name, email=employee.email, phone_e164=phone_e164, department_dn=department_dn, group_cns=[], initial_password=temporary_password, is_active=desired_is_active, ) ) logger.info( "LDAP admin user created", extra={ "event": "ldap.admin.employee_create.completed", "employee_id": employee.pk, }, ) employee.is_ldap_managed = True employee.save(update_fields=["is_ldap_managed"]) self.message_user( request, f"✅ Создан LDAP пользователь для {employee.email}. " f"Временный пароль: {temporary_password}", level=messages.SUCCESS, ) return "created" def _report_employee_ldap_sync(self, request, counts): summaries = ( ("created", "✅ Создано в LDAP", messages.SUCCESS), ("updated", "🔄 Обновлено в LDAP", messages.SUCCESS), ("current", "✓ Уже синхронизировано", messages.INFO), ("error", "❌ Ошибок", messages.WARNING), ) for key, label, level in summaries: if counts[key]: self.message_user( request, f"{label}: {counts[key]} пользователей", level=level ) @admin.action( description=("🔄 Синхронизировать LDAP → Django (обновить из Active Directory)") ) def sync_from_ldap_to_django(self, request, queryset): """Обновляет Django пользователей из LDAP. LDAP считается источником истины - данные из Active Directory перезапишут данные в Django Employee. """ from employees.ldap.orm_models import LdapUser from employees.ldap.utils.ldap_utils import get_ldap_str success_count = 0 error_count = 0 not_in_ldap = 0 for employee in queryset: if not employee.is_ldap_managed: not_in_ldap += 1 continue try: # Ищем LDAP пользователя ldap_user = _resolve_ldap_user_for_employee(employee) if not ldap_user: # Попробуем по email ldap_user = LdapUser.objects.filter(mail=employee.email).first() if not ldap_user: error_count += 1 self.message_user( request, f"❌ {employee.email}: не найден в LDAP", level=messages.WARNING, ) continue # Обновляем Django из LDAP updated_fields = [] ldap_first = get_ldap_str(ldap_user.given_name) if ldap_first and ldap_first != employee.first_name: employee.first_name = ldap_first updated_fields.append("first_name") ldap_last = get_ldap_str(ldap_user.sn) if ldap_last and ldap_last != employee.last_name: employee.last_name = ldap_last updated_fields.append("last_name") ldap_email = get_ldap_str(ldap_user.mail) if ldap_email and ldap_email != employee.email: employee.email = ldap_email updated_fields.append("email") ldap_phone = get_ldap_str( ldap_user.telephone_number or ldap_user.mobile ) if ldap_phone and ldap_phone != employee.phone_number: employee.phone_number = ldap_phone updated_fields.append("phone_number") if updated_fields: employee.save(update_fields=updated_fields) # Обновляем LdapSyncState LdapSyncState.objects.update_or_create( model="employee", object_pk=str(employee.pk), defaults={ "ldap_dn": ldap_user.dn, "ldap_guid": None, # objectGUID удален из модели "last_sync_dir": "ldap", }, ) success_count += 1 except Exception as e: error_count += 1 self.message_user( request, f"Ошибка синхронизации {employee.email}: {e}", level=messages.ERROR, ) if success_count: self.message_user( request, f"✅ Успешно синхронизировано: {success_count} " "пользователей (LDAP → Django)", level=messages.SUCCESS, ) if not_in_ldap: self.message_user( request, f"ℹ️ Пропущено локальных пользователей: {not_in_ldap}", level=messages.INFO, ) if error_count: self.message_user( request, f"❌ Ошибок: {error_count}", level=messages.WARNING ) @admin.action(description="🔍 Показать различия Django ↔ LDAP") def show_sync_diff(self, request, queryset): """Показывает различия между Django и LDAP. Сравнение выполняется для выбранных пользователей. """ from employees.ldap.orm_models import LdapUser from employees.ldap.utils.ldap_utils import get_ldap_str diffs = [] no_ldap_count = 0 for employee in queryset: if not employee.is_ldap_managed: continue try: # Ищем LDAP пользователя ldap_user = _resolve_ldap_user_for_employee(employee) if not ldap_user: ldap_user = LdapUser.objects.filter(mail=employee.email).first() if not ldap_user: no_ldap_count += 1 diffs.append(f"❌ {employee.email}: не найден в LDAP") continue user_diffs = [] # Сравниваем поля ldap_first = get_ldap_str(ldap_user.given_name) if ldap_first != employee.first_name: user_diffs.append( f"Имя: Django='{employee.first_name}' vs LDAP='{ldap_first}'" ) ldap_last = get_ldap_str(ldap_user.sn) if ldap_last != employee.last_name: user_diffs.append( f"Фамилия: Django='{employee.last_name}' vs LDAP='{ldap_last}'" ) ldap_email = get_ldap_str(ldap_user.mail) if ldap_email != employee.email: user_diffs.append( f"Email: Django='{employee.email}' vs LDAP='{ldap_email}'" ) ldap_phone = get_ldap_str( ldap_user.telephone_number or ldap_user.mobile ) if ( employee.phone_number and ldap_phone and ldap_phone != employee.phone_number ): user_diffs.append( f"Телефон: Django='{employee.phone_number}' vs LDAP='{ ldap_phone }'" ) if user_diffs: diffs.append(f"⚠️ {employee.email}: " + " | ".join(user_diffs)) else: diffs.append(f"✅ {employee.email}: данные совпадают") except Exception as e: diffs.append(f"❌ {employee.email}: ошибка сравнения - {e}") if diffs: message = "Результаты сравнения Django ↔ LDAP:\n\n" + "\n".join(diffs) self.message_user(request, message, level=messages.INFO) else: self.message_user( request, "Выбраны только локальные пользователи (не управляются LDAP)", level=messages.WARNING, ) @admin.register(Position) class PositionAdmin(SimpleHistoryAdmin): list_display = ("name", "description") search_fields = ("name", "description", "ldap_group_dn") filter_horizontal = ("groups",) fieldsets = ( (None, {"fields": ("name", "description")}), ("Права", {"fields": ("groups",)}), ( "LDAP", { "fields": ("ldap_group_dn",), "classes": ("collapse",), }, ), ) @admin.register(EmployeeAction) class EmployeeActionAdmin(SimpleHistoryAdmin): list_display = ("employee", "action", "date", "date_to") list_filter = ("action",) search_fields = ("employee__first_name", "employee__last_name", "comment") autocomplete_fields = ("employee",) date_hierarchy = "date" @admin.register(Department) class DepartmentAdmin(admin.ModelAdmin): list_display = ( "name", "head", "ldap_sync_status", "head_appointed_at", "created_at", ) list_filter = ("head",) search_fields = ("name", "description", "ldap_group_dn") autocomplete_fields = ("head",) readonly_fields = ("created_at", "head_appointed_at", "ldap_sync_info") inlines = [DepartmentMembershipInline] fieldsets = ( (None, {"fields": ("name", "description")}), ( "Руководство", { "fields": ("head", "head_appointed_at"), }, ), ( "LDAP", { "fields": ("ldap_group_dn", "ldap_sync_info"), "classes": ("collapse",), }, ), ("Служебное", {"fields": ("created_at",)}), ) actions = [ "sync_from_ldap_to_django", "sync_from_django_to_ldap", "show_sync_diff", ] def ldap_sync_status(self, obj): """Краткий статус LDAP синхронизации для списка.""" if not obj.pk: return "-" from django.urls import reverse from django.utils.html import format_html try: sync_state = LdapSyncState.objects.filter( model="department", object_pk=str(obj.pk) ).first() if not sync_state: # Нет записи синхронизации create_url = reverse("admin:employees_ldapsyncstate_add") return format_html( '' '❌ Нет записи' "", create_url, obj.pk, ) # Проверяем полноту данных has_dn = bool(sync_state.ldap_dn) has_updated = bool(sync_state.updated_at) # Полные данные if has_dn and has_updated: # Определяем направление последней синхронизации if sync_state.last_sync_dir == "ldap": icon = "⬇️" title_dir = "LDAP → Django" elif sync_state.last_sync_dir == "django": icon = "⬆️" title_dir = "Django → LDAP" else: icon = "✅" title_dir = "Синхронизирован" dn_short = ( sync_state.ldap_dn[:60] + "..." if len(sync_state.ldap_dn) > 60 else sync_state.ldap_dn ) return format_html( '{} {}', icon, dn_short, title_dir, sync_state.updated_at.strftime("%d.%m.%Y %H:%M"), ) # Неполные данные elif has_updated: return format_html( '⚠️ {}', sync_state.updated_at.strftime("%d.%m.%Y %H:%M"), ) else: return format_html( '' "⚠️ Неполные данные" ) except Exception as e: return format_html( '❌ Ошибка', str(e), ) ldap_sync_status.short_description = "LDAP статус" ldap_sync_status.admin_order_field = "id" def ldap_sync_info(self, obj): """Подробная информация о синхронизации для формы.""" if not obj.pk: return "-" from django.urls import reverse from django.utils.html import format_html try: sync_state = LdapSyncState.objects.filter( model="department", object_pk=str(obj.pk) ).first() if not sync_state: create_url = reverse("admin:employees_ldapsyncstate_add") return format_html( '
' "❌ Нет записи синхронизации
" "Создайте запись синхронизации или выполните " "синхронизацию.
" '➕ Создать запись' "
", create_url, obj.pk, ) # Формируем подробную информацию sync_url = reverse( "admin:employees_ldapsyncstate_change", args=[sync_state.pk] ) if sync_state.last_sync_dir == "ldap": direction = "⬇️ LDAP → Django" color = "#17a2b8" elif sync_state.last_sync_dir == "django": direction = "⬆️ Django → LDAP" color = "#0066cc" else: direction = "↔️ Неизвестно" color = "#6c757d" return format_html( '
' "Информация о синхронизации:
" '' "" '' "" "" "
LDAP DN:{}
Направление:{}
Обновлено:{}
" '' "📋 Открыть запись" "
", color, sync_state.ldap_dn or "—", direction, sync_state.updated_at.strftime("%Y-%m-%d %H:%M:%S") if sync_state.updated_at else "—", sync_url, ) except Exception as e: return format_html( '
' "❌ Ошибка: {}" "
", str(e), ) ldap_sync_info.short_description = "Статус синхронизации" @admin.action( description=("🔄 Синхронизировать LDAP → Django (LDAP как источник истины)") ) def sync_from_ldap_to_django(self, request, queryset): """Синхронизирует выбранные отделы из LDAP OU. Изменения переносятся в Django Department. """ from employees.ldap.orm_models import LdapOrganizationalUnit from employees.ldap.utils.ldap_utils import get_ldap_str success_count = 0 error_count = 0 warnings = [] for dept in queryset: # Ищем соответствующую OU в LDAP sync_state = LdapSyncState.objects.filter( model="department", object_pk=str(dept.pk), ).first() ldap_ou = None if sync_state and sync_state.ldap_dn: try: ldap_ou = LdapOrganizationalUnit.objects.get(dn=sync_state.ldap_dn) except LdapOrganizationalUnit.DoesNotExist: warnings.append( f"⚠️ {dept.name}: OU с DN {sync_state.ldap_dn} не найдена" ) if not ldap_ou: # Пытаемся найти по имени try: ldap_ou = LdapOrganizationalUnit.objects.get(ou=dept.name) except ( LdapOrganizationalUnit.DoesNotExist, LdapOrganizationalUnit.MultipleObjectsReturned, ): error_count += 1 warnings.append( f"⚠️ {dept.name}: не найдена соответствующая OU в LDAP" ) continue try: # Обновляем Department из LDAP OU dept.description = get_ldap_str(ldap_ou.description) or dept.description # Обновляем руководителя из managedBy mb = get_ldap_str(ldap_ou.managed_by) if mb: try: from employees.ldap.orm_models import LdapUser mgr = LdapUser.objects.get(dn=mb) if mgr.employee_number: try: dept.head = Employee.objects.get( pk=int(mgr.employee_number) ) except (Employee.DoesNotExist, ValueError): warnings.append( f"⚠️ {dept.name}: руководитель Employee #{ mgr.employee_number } не найден" ) except LdapUser.DoesNotExist: warnings.append(f"⚠️ {dept.name}: managedBy DN не найден в LDAP") dept._skip_ldap_sync = True dept.save() LdapSyncState.objects.update_or_create( model="department", object_pk=str(dept.pk), defaults={ "ldap_dn": ldap_ou.dn, "last_sync_dir": "ldap", }, ) success_count += 1 except Exception as e: error_count += 1 self.message_user( request, f"Ошибка синхронизации {dept.name}: {e}", level=messages.ERROR, ) for w in warnings[:10]: self.message_user(request, w, level=messages.WARNING) if len(warnings) > 10: self.message_user( request, f"...и ещё {len(warnings) - 10} предупреждений", level=messages.WARNING, ) if success_count: self.message_user( request, f"Успешно синхронизировано: {success_count} отделов (LDAP → Django)", level=messages.SUCCESS, ) if error_count: self.message_user( request, f"Ошибок: {error_count}", level=messages.WARNING, ) @admin.action( description=("🔄 Синхронизировать Django → LDAP (Django как источник истины)") ) def sync_from_django_to_ldap(self, request, queryset): """Синхронизирует выбранные отделы из Django в LDAP OU.""" from employees.ldap.orm_models import LdapOrganizationalUnit counts = {"success": 0, "error": 0, "created": 0} warnings = [] for department in queryset: if not department.name or not department.name.strip(): warnings.append(f"⚠️ Department #{department.pk}: пустое название") counts["error"] += 1 continue try: organizational_unit, created = self._find_or_create_department_ou( department ) counts["created"] += int(created) except LdapOrganizationalUnit.MultipleObjectsReturned: counts["error"] += 1 warnings.append( f"⚠️ {department.name}: найдено несколько OU с таким именем" ) continue except Exception as error: counts["error"] += 1 self.message_user( request, f"Ошибка создания OU для {department.name}: {error}", level=messages.ERROR, ) continue try: self._update_ldap_department( request, department, organizational_unit, warnings ) counts["success"] += 1 except Exception as error: counts["error"] += 1 self.message_user( request, f"Ошибка синхронизации {department.name}: {error}", level=messages.ERROR, ) self._report_department_ldap_sync(request, counts, warnings) @staticmethod def _find_or_create_department_ou(department): from employees.ldap.orm_models import LdapOrganizationalUnit sync_state = LdapSyncState.objects.filter( model="department", object_pk=str(department.pk) ).first() if sync_state and sync_state.ldap_dn: organizational_unit = LdapOrganizationalUnit.objects.filter( dn=sync_state.ldap_dn ).first() if organizational_unit is not None: return organizational_unit, False try: return LdapOrganizationalUnit.objects.get(ou=department.name), False except LdapOrganizationalUnit.DoesNotExist: return DepartmentAdmin._create_department_ou(department), True @staticmethod def _create_department_ou(department): from django.conf import settings from ldap3 import BASE from employees.ldap.infrastructure.connections import _ldap from employees.ldap.orm_models import LdapOrganizationalUnit base_dn = getattr(settings, "LDAP_DEPARTMENTS_BASE", "") if not base_dn: raise RuntimeError("LDAP_DEPARTMENTS_BASE not configured") organizational_unit_dn = f"OU={department.name},{base_dn}" with _ldap() as connection: exists = connection.search( organizational_unit_dn, "(objectClass=organizationalUnit)", search_scope=BASE, ) if not exists or not connection.entries: created = connection.add( organizational_unit_dn, ["top", "organizationalUnit"] ) if not created: raise RuntimeError(f"LDAP add OU failed: {connection.result}") try: return LdapOrganizationalUnit.objects.get(dn=organizational_unit_dn) except LdapOrganizationalUnit.DoesNotExist as error: raise RuntimeError( f"OU был создан, но не найден по DN: {organizational_unit_dn}" ) from error def _update_ldap_department( self, request, department, organizational_unit, warnings ): organizational_unit.description = department.description or "" if department.head: head_sync = LdapSyncState.objects.filter( model="employee", object_pk=str(department.head.pk) ).first() if head_sync and head_sync.ldap_dn: organizational_unit.managed_by = head_sync.ldap_dn else: warnings.append( f"⚠️ {department.name}: руководитель " f"{department.head} не имеет LDAP DN" ) else: organizational_unit.managed_by = "" organizational_unit.save() try: group = organizational_unit.ensure_department_group() result = group.sync_members(self._department_member_dns(department)) if result["added"] or result["removed"]: self.message_user( request, f"👥 {department.name}: группа DEP_* — " f"+{result['added']}/−{result['removed']}", level=messages.INFO, ) except Exception as error: warnings.append(f"⚠️ {department.name}: группа DEP_*: {error}") LdapSyncState.objects.update_or_create( model="department", object_pk=str(department.pk), defaults={ "ldap_dn": organizational_unit.dn, "last_sync_dir": "django", }, ) @staticmethod def _department_member_dns(department): from employees.models import EmployeeDepartment links = EmployeeDepartment.objects.filter( department=department, is_active=True ).select_related("employee") member_dns = [] for link in links: if not (link.employee.is_active and link.employee.is_actually_active): continue sync_state = LdapSyncState.objects.filter( model="employee", object_pk=str(link.employee_id) ).first() if sync_state and sync_state.ldap_dn: member_dns.append(sync_state.ldap_dn) return member_dns def _report_department_ldap_sync(self, request, counts, warnings): for warning in warnings[:10]: self.message_user(request, warning, level=messages.WARNING) if len(warnings) > 10: self.message_user( request, f"...и ещё {len(warnings) - 10} предупреждений", level=messages.WARNING, ) summaries = ( ("created", "Создано новых OU в LDAP", messages.SUCCESS), ( "success", "Успешно синхронизировано отделов (Django → LDAP)", messages.SUCCESS, ), ("error", "Ошибок", messages.WARNING), ) for key, label, level in summaries: if counts[key]: self.message_user(request, f"{label}: {counts[key]}", level=level) @admin.action(description="🔍 Показать различия Django ↔ LDAP") def show_sync_diff(self, request, queryset): """Показывает различия между Django Department и LDAP OU.""" from employees.ldap.orm_models import LdapOrganizationalUnit from employees.ldap.utils.ldap_utils import get_ldap_str diffs = [] for dept in queryset: # Ищем OU sync_state = LdapSyncState.objects.filter( model="department", object_pk=str(dept.pk), ).first() ldap_ou = None if sync_state and sync_state.ldap_dn: try: ldap_ou = LdapOrganizationalUnit.objects.get(dn=sync_state.ldap_dn) except LdapOrganizationalUnit.DoesNotExist: pass if not ldap_ou: try: ldap_ou = LdapOrganizationalUnit.objects.get(ou=dept.name) except ( LdapOrganizationalUnit.DoesNotExist, LdapOrganizationalUnit.MultipleObjectsReturned, ): diffs.append(f"{dept.name}: не найдена соответствующая OU в LDAP") continue dept_diffs = [] # Сравниваем описание ldap_desc = get_ldap_str(ldap_ou.description) or "" django_desc = dept.description or "" if ldap_desc != django_desc: dept_diffs.append( f"Описание: LDAP='{ldap_desc[:50]}' vs Django='{django_desc[:50]}'" ) # Сравниваем руководителя mb = get_ldap_str(ldap_ou.managed_by) if dept.head: head_sync = LdapSyncState.objects.filter( model="employee", object_pk=str(dept.head.pk), ).first() head_dn = head_sync.ldap_dn if head_sync else None if mb != head_dn: dept_diffs.append( f"Руководитель: LDAP='{mb or '—'}' vs Django='{dept.head}'" ) elif mb: dept_diffs.append(f"Руководитель: LDAP='{mb}' vs Django='не назначен'") if dept_diffs: diffs.append(f"{dept.name}: " + ", ".join(dept_diffs)) if diffs: self.message_user( request, "Найдены различия:\n" + "\n".join(diffs), level=messages.WARNING, ) else: self.message_user( request, "Различий не найдено. Все выбранные отделы синхронизированы.", level=messages.SUCCESS, ) @admin.register(DepartmentRole) class DepartmentRoleAdmin(admin.ModelAdmin): list_display = ("name", "department") list_filter = ("department",) search_fields = ("name", "department__name", "ldap_group_dn") autocomplete_fields = ("department",) filter_horizontal = ("scoped_permissions",) actions = ["sync_to_ldap"] fieldsets = ( (None, {"fields": ("department", "name")}), ("Права", {"fields": ("scoped_permissions",)}), ( "LDAP", { "fields": ("ldap_group_dn",), "classes": ("collapse",), }, ), ) @admin.action( description=( "🔄 Пересинхронизировать выбранные роли в LDAP (ROLE_* и вложение в DEP_*)" ) ) def sync_to_ldap(self, request, queryset): """Приводит LDAP-состояние выбранных ролей к текущему состоянию БД.""" from employees.ldap.services.department_service import ( DepartmentService, ) service = DepartmentService() success_count = 0 error_count = 0 for role in queryset.select_related("department"): try: service.sync_role_state(role) success_count += 1 except Exception as e: error_count += 1 self.message_user( request, f"Ошибка синхронизации роли " f"«{role.name}» ({role.department.name}): {e}", level=messages.ERROR, ) if success_count: self.message_user( request, f"Успешно пересинхронизировано ролей: {success_count}", level=messages.SUCCESS, ) if error_count: self.message_user( request, f"Ошибок синхронизации ролей: {error_count}", level=messages.WARNING, ) @admin.register(EmployeeDepartment) class EmployeeDepartmentAdmin(admin.ModelAdmin): list_display = ( "employee", "department", "role", "is_active", "date_from", "date_to", ) list_filter = ("is_active", "department", "role") search_fields = ( "employee__first_name", "employee__last_name", "department__name", ) autocomplete_fields = ("employee", "department", "role") date_hierarchy = "date_from" @admin.register(Skill) class SkillAdmin(admin.ModelAdmin): list_display = ("name", "is_secret", "progression_type", "system_code") list_filter = ("is_secret", "progression_type") search_fields = ("name",) @admin.register(EmployeeSkillProgress) class EmployeeSkillProgressAdmin(admin.ModelAdmin): list_display = ( "employee", "skill", "experience", "special_rank", "is_active", "updated_at", ) list_filter = ("is_active", "special_rank", "skill") search_fields = ( "employee__email", "employee__first_name", "employee__last_name", "skill__name", ) autocomplete_fields = ("employee", "skill", "created_by") readonly_fields = ("created_at", "updated_at") @admin.register(SkillExperienceEvent) class SkillExperienceEventAdmin(admin.ModelAdmin): list_display = ( "progress", "points", "source_type", "actor", "created_at", "revoked_at", ) list_filter = ("source_type", "revoked_at") search_fields = ( "progress__employee__email", "progress__skill__name", "reason", "idempotency_key", ) autocomplete_fields = ("progress", "actor", "revoked_by") readonly_fields = ("created_at",) # Полезно иметь Permission в админке @admin.register(Permission) class PermissionAdmin(admin.ModelAdmin): list_display = ("name", "codename", "content_type") list_filter = ("content_type",) search_fields = ("name", "codename") @admin.register(LdapSyncState) class LdapSyncStateAdmin(admin.ModelAdmin): """Админка для просмотра состояния синхронизации LDAP.""" list_display = ( "model", "object_pk", "ldap_dn_short", "ldap_guid", "last_sync_dir", "updated_at", ) list_filter = ("model", "last_sync_dir") search_fields = ("object_pk", "ldap_dn", "ldap_guid") readonly_fields = ("updated_at",) fields = ( "model", "object_pk", "ldap_dn", "ldap_guid", "last_ldap_modify_ts", "last_django_modify_ts", "last_sync_dir", "data_hash", "updated_at", ) def ldap_dn_short(self, obj): """Показываем сокращенную версию DN для удобства.""" if obj.ldap_dn: return obj.ldap_dn[:50] + "..." if len(obj.ldap_dn) > 50 else obj.ldap_dn return "-" ldap_dn_short.short_description = "LDAP DN" def has_add_permission(self, request): """Разрешаем создание записей для отсутствующих пользователей.""" return True def has_delete_permission(self, request, obj=None): """Разрешаем удаление для очистки.""" return True # ========================= # Кастомный GroupAdmin с LDAP синхронизацией # ========================= admin.site.unregister(Group) @admin.register(Group) class GroupAdmin(DefaultGroupAdmin): """Расширенный GroupAdmin с кнопками синхронизации LDAP.""" list_display = ("name", "ldap_dn_display") actions = ["sync_groups_from_ldap", "sync_groups_to_ldap"] def ldap_dn_display(self, obj): sync = LdapSyncState.objects.filter( model="group", object_pk=str(obj.pk), ).first() return sync.ldap_dn if sync else "—" ldap_dn_display.short_description = "LDAP DN" # ---------- LDAP → Django ---------- def sync_groups_from_ldap(self, request, queryset): """Синхронизирует ВСЕ группы из OU=Groups в Django Group.""" from employees.ldap.services.group_service import GroupService svc = GroupService() try: created = svc.sync_catalog(throttle_seconds=0) self.message_user( request, f"LDAP → Django: создано {created} новых Django Group", level=messages.SUCCESS, ) except Exception as e: self.message_user( request, f"Ошибка синхронизации: {e}", level=messages.ERROR, ) sync_groups_from_ldap.short_description = "⬇️ LDAP → Django (создать Django Group)" # ---------- Django → LDAP ---------- def sync_groups_to_ldap(self, request, queryset): """Создаёт в LDAP группы, которых там ещё нет. Перед созданием выполняется реальная проверка LDAP. """ from employees.ldap.orm_models import LdapGroup from employees.ldap.services.group_service import GroupService svc = GroupService() created = 0 already = 0 for dg in queryset: # Реальная проверка наличия группы в LDAP exists_in_ldap = LdapGroup.objects.filter(cn=dg.name).exists() if exists_in_ldap: ldap_grp = LdapGroup.objects.filter(cn=dg.name).first() # Обновляем/создаём LdapSyncState на случай рассинхрона LdapSyncState.objects.update_or_create( model="group", object_pk=str(dg.pk), defaults={ "ldap_dn": ldap_grp.dn, "last_sync_dir": "ldap", }, ) already += 1 continue try: dn = svc.create(cn=dg.name) LdapSyncState.objects.update_or_create( model="group", object_pk=str(dg.pk), defaults={ "ldap_dn": dn, "last_sync_dir": "django", }, ) created += 1 except Exception as e: self.message_user( request, f"⚠️ {dg.name}: {e}", level=messages.WARNING, ) self.message_user( request, f"Django → LDAP: создано {created}, уже существовало {already}", level=messages.SUCCESS, ) sync_groups_to_ldap.short_description = "⬆️ Django → LDAP (создать LDAP Group)"