572 lines
23 KiB
Python
572 lines
23 KiB
Python
from django import forms
|
||
from django.contrib import admin, messages
|
||
from django.contrib.auth import password_validation
|
||
from django.contrib.auth.models import Permission
|
||
from django.contrib.contenttypes.models import ContentType
|
||
from django.core.exceptions import ValidationError as DjangoValidationError
|
||
from django.utils.html import format_html
|
||
from django.utils.safestring import mark_safe
|
||
from django.utils.translation import gettext_lazy as _
|
||
from unfold.admin import ModelAdmin, TabularInline
|
||
|
||
from core.models import Directory, PortalAccess
|
||
from core.security import filter_portal_permissions_for_user, log_security_event
|
||
from .models import Employee, PermissionGroup, EmployeeDefaults, EmployeePermissionOverride
|
||
from .models import PasswordSetupInvite
|
||
from .services.password_invite import can_offer_ad_password_sync
|
||
from .models import EmployeeHrRequest, LeaveEntitlement
|
||
from apps.visit_manager.models import WorkSchedule, AttendanceLog
|
||
|
||
|
||
def _portal_permissions_qs():
|
||
ct = ContentType.objects.get_for_model(PortalAccess, for_concrete_model=False)
|
||
access = Permission.objects.filter(content_type=ct, codename__startswith='access_')
|
||
extra = Permission.objects.filter(
|
||
content_type__app_label='reports',
|
||
codename='view_all_departments',
|
||
)
|
||
return (access | extra).order_by('content_type__app_label', 'codename')
|
||
|
||
|
||
class HierarchicalDepartmentChoiceField(forms.ModelChoiceField):
|
||
"""Поле выбора отдела с отступами по иерархии в выпадающем списке."""
|
||
|
||
def __init__(self, *args, **kwargs):
|
||
queryset = kwargs.get('queryset') or Directory.objects.filter(
|
||
directory_type__code='department',
|
||
is_active=True,
|
||
).select_related('directory_type').order_by('_path')
|
||
kwargs['queryset'] = queryset
|
||
super().__init__(*args, **kwargs)
|
||
|
||
def label_from_instance(self, obj):
|
||
depth = (getattr(obj, '_path', '') or '').count('.')
|
||
indent = '---' * depth + ('--- ' if depth else '')
|
||
return f'{indent}{obj.name}'
|
||
|
||
class OfficeWarehouseChoiceField(forms.ModelChoiceField):
|
||
def label_from_instance(self, obj):
|
||
type_label = obj.get_location_type_display()
|
||
addr = obj.get_full_address() or obj.address or ''
|
||
return f"{type_label} — {addr}" if addr else type_label
|
||
|
||
|
||
class EmployeeAdminForm(forms.ModelForm):
|
||
department = HierarchicalDepartmentChoiceField(
|
||
label=_('Отдел'),
|
||
required=False,
|
||
empty_label=_('— Выберите отдел —'),
|
||
widget=forms.Select(attrs={
|
||
'class': 'hierarchy-department-select',
|
||
'data-placeholder': 'Департамент → Отдел',
|
||
}),
|
||
help_text=_('Выберите из списка. Уровни вложенности отображаются отступами.'),
|
||
)
|
||
user_location = OfficeWarehouseChoiceField(
|
||
queryset=None,
|
||
required=False,
|
||
empty_label=_('— Выберите офис или склад —'),
|
||
label=_('Локация (офис/склад)'),
|
||
help_text=_('Заполняется для location_type=office или warehouse'),
|
||
)
|
||
new_password = forms.CharField(
|
||
label=_('Новый пароль'),
|
||
required=False,
|
||
strip=False,
|
||
widget=forms.PasswordInput(attrs={'autocomplete': 'new-password'}, render_value=False),
|
||
help_text=_('Оставьте пустым, чтобы не менять текущий пароль портала.'),
|
||
)
|
||
new_password_confirm = forms.CharField(
|
||
label=_('Подтверждение пароля'),
|
||
required=False,
|
||
strip=False,
|
||
widget=forms.PasswordInput(attrs={'autocomplete': 'new-password'}, render_value=False),
|
||
)
|
||
sync_ad_password = forms.BooleanField(
|
||
label=_('Также сменить пароль в Active Directory'),
|
||
required=False,
|
||
initial=False,
|
||
help_text=_('Доступно для офисных сотрудников с привязкой к AD.'),
|
||
)
|
||
|
||
def __init__(self, *args, **kwargs):
|
||
super().__init__(*args, **kwargs)
|
||
from apps.locations.models import Location
|
||
self.fields['user_location'].queryset = (
|
||
Location.objects
|
||
.filter(location_type__in=['office', 'warehouse'], is_active=True)
|
||
.select_related('location_region')
|
||
.order_by('location_type', 'address')
|
||
)
|
||
offer_ad = bool(self.instance and self.instance.pk and can_offer_ad_password_sync(self.instance))
|
||
self.fields['sync_ad_password'].disabled = not offer_ad
|
||
if not offer_ad:
|
||
self.fields['sync_ad_password'].help_text = _(
|
||
'Синхронизация с AD недоступна: нужен офисный сотрудник с заполненным DN.'
|
||
)
|
||
|
||
def clean(self):
|
||
cleaned = super().clean()
|
||
password = (cleaned.get('new_password') or '')
|
||
confirm = (cleaned.get('new_password_confirm') or '')
|
||
if password or confirm:
|
||
if password != confirm:
|
||
self.add_error('new_password_confirm', _('Пароли не совпадают.'))
|
||
elif password:
|
||
try:
|
||
password_validation.validate_password(password, self.instance)
|
||
except DjangoValidationError as exc:
|
||
self.add_error('new_password', exc)
|
||
if cleaned.get('sync_ad_password') and password:
|
||
if not (self.instance and self.instance.pk and can_offer_ad_password_sync(self.instance)):
|
||
self.add_error(
|
||
'sync_ad_password',
|
||
_('Смена пароля в AD недоступна для этого сотрудника.'),
|
||
)
|
||
return cleaned
|
||
|
||
def save(self, commit=True):
|
||
obj = super().save(commit=False)
|
||
password = self.cleaned_data.get('new_password') or ''
|
||
if password:
|
||
obj.set_password(password)
|
||
if commit:
|
||
obj.save()
|
||
self.save_m2m()
|
||
return obj
|
||
|
||
class Meta:
|
||
model = Employee
|
||
exclude = ('password',)
|
||
|
||
class WorkScheduleInline(TabularInline):
|
||
model = WorkSchedule
|
||
fk_name = 'employee'
|
||
extra = 0
|
||
fields = ('schedule_template', 'location', 'start_date', 'end_date', 'is_primary', 'is_active')
|
||
readonly_fields = ('is_active',)
|
||
ordering = ('-start_date',)
|
||
show_change_link = True
|
||
|
||
def has_add_permission(self, request, obj=None):
|
||
return False
|
||
|
||
|
||
class EmployeePermissionOverrideInline(TabularInline):
|
||
model = EmployeePermissionOverride
|
||
extra = 0
|
||
fields = ('permission', 'override_type', 'valid_from', 'valid_until', 'is_active', 'comment')
|
||
autocomplete_fields = []
|
||
verbose_name = _('Доп. право на раздел портала')
|
||
verbose_name_plural = _('Доп. права на разделы портала')
|
||
|
||
def formfield_for_foreignkey(self, db_field, request, **kwargs):
|
||
if db_field.name == 'permission':
|
||
kwargs['queryset'] = filter_portal_permissions_for_user(
|
||
_portal_permissions_qs(), request.user,
|
||
)
|
||
return super().formfield_for_foreignkey(db_field, request, **kwargs)
|
||
|
||
def has_add_permission(self, request, obj=None):
|
||
return request.user.is_superuser
|
||
|
||
def has_change_permission(self, request, obj=None):
|
||
return request.user.is_superuser
|
||
|
||
def has_delete_permission(self, request, obj=None):
|
||
return request.user.is_superuser
|
||
|
||
|
||
class AttendanceLogInline(TabularInline):
|
||
model = AttendanceLog
|
||
fk_name = 'employee'
|
||
extra = 0
|
||
max_num = 0
|
||
fields = ('timestamp', 'direction', 'location', 'device_id', 'synced')
|
||
readonly_fields = ('timestamp', 'direction', 'location', 'device_id', 'synced')
|
||
ordering = ('-timestamp',)
|
||
show_change_link = True
|
||
|
||
def has_add_permission(self, request, obj=None):
|
||
return False
|
||
|
||
def has_delete_permission(self, request, obj=None):
|
||
return False
|
||
|
||
|
||
@admin.register(Employee)
|
||
class EmployeeAdmin(ModelAdmin):
|
||
form = EmployeeAdminForm
|
||
inlines = [EmployeePermissionOverrideInline, WorkScheduleInline, AttendanceLogInline]
|
||
|
||
class Media:
|
||
css = {'all': ('users/css/employee_admin.css',)}
|
||
js = ('assets/js/admin_phone_format.js',)
|
||
|
||
list_display = [
|
||
'photo_thumbnail', 'username', 'last_name', 'first_name', 'groups_display',
|
||
'user_location_type', 'user_pharmacy', 'department_display', 'post',
|
||
'is_active', 'twofa_status', 'hire_date'
|
||
]
|
||
list_filter = ['groups', 'is_active', 'is_staff', 'user_location_type', 'department', 'totp_enabled']
|
||
search_fields = ['username', 'email', 'last_name', 'first_name', 'middle_name']
|
||
list_select_related = ['user_pharmacy', 'department', 'post']
|
||
readonly_fields = ('photo_preview',)
|
||
actions = ['reset_2fa']
|
||
|
||
@admin.display(description='2FA', boolean=True)
|
||
def twofa_status(self, obj):
|
||
return obj.totp_enabled
|
||
|
||
@admin.action(description='Сбросить 2FA (отключить у выбранных)')
|
||
def reset_2fa(self, request, queryset):
|
||
from apps.employees.services import twofa
|
||
n = 0
|
||
for emp in queryset:
|
||
if emp.totp_enabled or emp.totp_secret:
|
||
twofa.disable(emp)
|
||
n += 1
|
||
self.message_user(request, f'2FA сброшена у {n} сотрудник(ов).')
|
||
|
||
fieldsets = (
|
||
(_('Аутентификация'), {
|
||
'fields': (
|
||
'username', 'email', 'groups', 'is_active',
|
||
'new_password', 'new_password_confirm', 'sync_ad_password',
|
||
),
|
||
'classes': ('tab',),
|
||
}),
|
||
(_('ФИО'), {
|
||
'fields': ('last_name', 'first_name', 'middle_name'),
|
||
'classes': ('tab',),
|
||
}),
|
||
(_('Рабочая информация'), {
|
||
'fields': (
|
||
'user_location_type', 'user_organization',
|
||
'user_pharmacy', 'user_location',
|
||
'department', 'post', 'manager',
|
||
),
|
||
'classes': ('tab',),
|
||
}),
|
||
(_('Даты'), {
|
||
'fields': ('hire_date', 'termination_date', 'probation_end_date', 'date_joined'),
|
||
'classes': ('tab',),
|
||
}),
|
||
(_('Личные данные'), {
|
||
'fields': ('birth_date', 'inn_code', 'photo', 'photo_preview'),
|
||
'classes': ('tab',),
|
||
}),
|
||
(_('Контакты'), {
|
||
'fields': (
|
||
'phone', 'internal_number', 'personal_phone', 'personal_email',
|
||
'emergency_contact_name', 'emergency_contact_phone', 'emergency_contact_relation',
|
||
),
|
||
'classes': ('tab',),
|
||
}),
|
||
('Mattermost', {
|
||
'fields': ('mattermost_user_id', 'mattermost_username', 'mattermost_sync_status',
|
||
'mattermost_last_sync', 'mattermost_sync_enabled'),
|
||
'classes': ('tab', 'collapse'),
|
||
}),
|
||
('Active Directory', {
|
||
'fields': ('ad_user_dn', 'ad_sync_status', 'ad_sync_error', 'ad_last_sync', 'ad_sync_enabled'),
|
||
'classes': ('tab', 'collapse'),
|
||
}),
|
||
(_('Графік та відвідуваність'), {
|
||
'fields': (),
|
||
'classes': ('tab',),
|
||
'description': _('Графіки роботи та записи відвідуваності відображаються нижче.'),
|
||
}),
|
||
)
|
||
|
||
def get_fieldsets(self, request, obj=None):
|
||
fieldsets = list(super().get_fieldsets(request, obj))
|
||
if request.user.is_superuser:
|
||
fieldsets.append((
|
||
_('Системный доступ'),
|
||
{
|
||
'fields': ('is_staff', 'user_permissions'),
|
||
'classes': ('tab', 'collapse'),
|
||
'description': _('Только для IT-администраторов. is_staff — вход в Django admin.'),
|
||
},
|
||
))
|
||
return fieldsets
|
||
|
||
def get_readonly_fields(self, request, obj=None):
|
||
readonly = list(super().get_readonly_fields(request, obj))
|
||
if not request.user.is_superuser:
|
||
readonly.extend([
|
||
'groups',
|
||
'ad_user_dn', 'ad_sync_status', 'ad_sync_error',
|
||
'ad_last_sync', 'ad_sync_enabled',
|
||
])
|
||
return readonly
|
||
|
||
def save_model(self, request, obj, form, change):
|
||
if not request.user.is_superuser and change:
|
||
previous = Employee.objects.filter(pk=obj.pk).prefetch_related('groups').first()
|
||
if previous:
|
||
obj.groups.set(previous.groups.all())
|
||
super().save_model(request, obj, form, change)
|
||
password = (form.cleaned_data or {}).get('new_password') or ''
|
||
if not password:
|
||
return
|
||
from django.utils import timezone
|
||
PasswordSetupInvite.objects.filter(
|
||
employee=obj, used_at__isnull=True,
|
||
).update(used_at=timezone.now())
|
||
log_security_event(
|
||
'employee.password_set_via_admin',
|
||
actor=request.user,
|
||
target=str(obj.pk),
|
||
details=f'username={obj.username}',
|
||
)
|
||
self.message_user(request, _('Пароль портала обновлён.'), messages.SUCCESS)
|
||
if form.cleaned_data.get('sync_ad_password') and can_offer_ad_password_sync(obj):
|
||
from apps.ad_connector.services.sync_service import sync_ad_password
|
||
try:
|
||
ok = sync_ad_password(obj, password)
|
||
except Exception:
|
||
ok = False
|
||
if ok:
|
||
self.message_user(request, _('Пароль в Active Directory обновлён.'), messages.SUCCESS)
|
||
else:
|
||
self.message_user(
|
||
request,
|
||
_('Пароль портала сохранён, но сменить пароль в AD не удалось.'),
|
||
messages.WARNING,
|
||
)
|
||
|
||
# def permissions_summary(self, obj):
|
||
# perms = obj.effective_permissions()
|
||
# group = obj.permission_group.name if obj.permission_group_id else None
|
||
|
||
# if not perms and not group:
|
||
# return mark_safe('<span style="color:#999;">—</span>')
|
||
|
||
# parts = []
|
||
# if group:
|
||
# parts.append(format_html(
|
||
# '<span style="background:#1a7f37;color:#fff;'
|
||
# 'padding:2px 6px;border-radius:4px;font-size:11px;">📁 {}</span>',
|
||
# group,
|
||
# ))
|
||
# for p in perms:
|
||
# parts.append(format_html(
|
||
# '<span style="background:#6c47ff;color:#fff;'
|
||
# 'padding:2px 6px;border-radius:4px;font-size:11px;">{}</span>',
|
||
# p,
|
||
# ))
|
||
# return mark_safe(' '.join(parts))
|
||
# permissions_summary.short_description = _('Группа / Разрешения')
|
||
|
||
def photo_thumbnail(self, obj):
|
||
if obj.photo:
|
||
return format_html(
|
||
'<img src="{}" style="width:36px;height:36px;border-radius:50%;object-fit:cover;">',
|
||
obj.photo.url
|
||
)
|
||
return mark_safe('<span style="color:#999;">—</span>')
|
||
photo_thumbnail.short_description = ''
|
||
|
||
def photo_preview(self, obj):
|
||
if obj.photo:
|
||
return format_html(
|
||
'<img src="{}" style="max-height:200px;border-radius:8px;">',
|
||
obj.photo.url
|
||
)
|
||
return mark_safe('<span style="color:#999;">Фото не загружено</span>')
|
||
photo_preview.short_description = _('Предпросмотр')
|
||
|
||
def groups_display(self, obj):
|
||
names = list(obj.groups.values_list('name', flat=True))
|
||
if not names:
|
||
return mark_safe('<span style="color:#999;">—</span>')
|
||
return ', '.join(names)
|
||
groups_display.short_description = _('Группы')
|
||
|
||
def department_display(self, obj): # noqa: E303
|
||
"""Отдел с полным путём иерархии без дубликатов (Офис → HR → Кадры)."""
|
||
if not obj.department:
|
||
return mark_safe('<span style="color:#999;">—</span>')
|
||
try:
|
||
parts = [n.name for n in obj.department.get_ancestors()] + [obj.department.name]
|
||
# Убираем подряд идущие дубликаты (родитель и ребёнок с одинаковым именем)
|
||
unique = []
|
||
for p in parts:
|
||
if not unique or unique[-1] != p:
|
||
unique.append(p)
|
||
path = ' → '.join(unique)
|
||
return format_html(
|
||
'<span class="department-path" title="{}">{}</span>',
|
||
path, path
|
||
)
|
||
except Exception:
|
||
return obj.department.name or '—'
|
||
department_display.short_description = _('Отдел')
|
||
|
||
@admin.register(EmployeeDefaults)
|
||
class EmployeeDefaultsAdmin(ModelAdmin):
|
||
"""Единственная запись настроек по умолчанию для сотрудников."""
|
||
|
||
fieldsets = (
|
||
(_('Офисные сотрудники'), {
|
||
'fields': ('office_email_domain',),
|
||
'description': _(
|
||
'При утверждении заявки на найм офисному сотруднику '
|
||
'назначается рабочий email: логин@домен.'
|
||
),
|
||
}),
|
||
)
|
||
|
||
def has_add_permission(self, request):
|
||
return not EmployeeDefaults.objects.exists()
|
||
|
||
def has_delete_permission(self, request, obj=None):
|
||
return False
|
||
|
||
|
||
class PermissionGroupAdminForm(forms.ModelForm):
|
||
portal_permissions = forms.ModelMultipleChoiceField(
|
||
label=_('Разделы портала'),
|
||
queryset=Permission.objects.none(),
|
||
required=False,
|
||
widget=admin.widgets.FilteredSelectMultiple(_('разделы'), is_stacked=False),
|
||
help_text=_('Базовый набор доступов к разделам для этой роли. Доп. права сотруднику — через overrides.'),
|
||
)
|
||
|
||
class Meta:
|
||
model = PermissionGroup
|
||
fields = ('group', 'description', 'is_active')
|
||
|
||
def __init__(self, *args, **kwargs):
|
||
super().__init__(*args, **kwargs)
|
||
self.fields['portal_permissions'].queryset = _portal_permissions_qs()
|
||
if self.instance and self.instance.pk and self.instance.group_id:
|
||
portal_pks = set(_portal_permissions_qs().values_list('pk', flat=True))
|
||
self.fields['portal_permissions'].initial = [
|
||
p for p in self.instance.group.permissions.all() if p.pk in portal_pks
|
||
]
|
||
|
||
def save_portal_permissions(self, obj):
|
||
portal_qs = list(_portal_permissions_qs())
|
||
selected = self.cleaned_data.get('portal_permissions')
|
||
if selected is None:
|
||
return
|
||
group = obj.group
|
||
group.permissions.remove(*portal_qs)
|
||
group.permissions.add(*selected)
|
||
|
||
|
||
@admin.register(PermissionGroup)
|
||
class PermissionGroupAdmin(ModelAdmin):
|
||
form = PermissionGroupAdminForm
|
||
list_display = ('group', 'is_active', 'description', 'portal_matrix_link')
|
||
list_filter = ('is_active',)
|
||
search_fields = ('group__name', 'description')
|
||
|
||
fieldsets = (
|
||
(None, {
|
||
'fields': ('group', 'description', 'is_active', 'portal_permissions'),
|
||
}),
|
||
)
|
||
|
||
@admin.display(description=_('Матрица в портале'))
|
||
def portal_matrix_link(self, obj):
|
||
if not obj or not obj.group_id:
|
||
return '—'
|
||
from django.urls import reverse
|
||
from django.utils.html import format_html
|
||
url = reverse('user_group_access', kwargs={'group_id': obj.group_id})
|
||
return format_html('<a href="{}" target="_blank">{}</a>', url, _('Открыть'))
|
||
|
||
def get_readonly_fields(self, request, obj=None):
|
||
if request.user.is_superuser:
|
||
return ()
|
||
return ('group', 'description', 'is_active', 'portal_permissions')
|
||
|
||
def has_add_permission(self, request):
|
||
return request.user.is_superuser
|
||
|
||
def has_delete_permission(self, request, obj=None):
|
||
return request.user.is_superuser
|
||
|
||
def save_model(self, request, obj, form, change):
|
||
super().save_model(request, obj, form, change)
|
||
if request.user.is_superuser:
|
||
form.save_portal_permissions(obj)
|
||
|
||
@admin.register(EmployeePermissionOverride)
|
||
class EmployeePermissionOverrideAdmin(ModelAdmin):
|
||
list_display = (
|
||
'employee', 'permission', 'override_type',
|
||
'valid_from', 'valid_until', 'is_active',
|
||
)
|
||
list_filter = ('override_type', 'is_active')
|
||
search_fields = (
|
||
'employee__username', 'employee__last_name', 'employee__first_name',
|
||
'permission__codename', 'comment',
|
||
)
|
||
list_select_related = ('employee', 'permission')
|
||
autocomplete_fields = ('employee',)
|
||
readonly_fields = ('created_at', 'updated_at', 'uuid')
|
||
|
||
fieldsets = (
|
||
(None, {
|
||
'fields': ('employee', 'permission', 'override_type', 'is_active'),
|
||
'description': _(
|
||
'grant — выдать раздел сверх группы; '
|
||
'revoke — запретить раздел даже если право есть в группе.'
|
||
),
|
||
}),
|
||
(_('Срок действия'), {
|
||
'fields': ('valid_from', 'valid_until', 'comment'),
|
||
}),
|
||
(_('Служебное'), {
|
||
'classes': ('collapse',),
|
||
'fields': ('uuid', 'created_at', 'updated_at'),
|
||
}),
|
||
)
|
||
|
||
def formfield_for_foreignkey(self, db_field, request, **kwargs):
|
||
if db_field.name == 'permission':
|
||
kwargs['queryset'] = filter_portal_permissions_for_user(
|
||
_portal_permissions_qs(), request.user,
|
||
)
|
||
return super().formfield_for_foreignkey(db_field, request, **kwargs)
|
||
|
||
def has_add_permission(self, request):
|
||
return request.user.is_superuser
|
||
|
||
def has_change_permission(self, request, obj=None):
|
||
return request.user.is_superuser
|
||
|
||
def has_delete_permission(self, request, obj=None):
|
||
return request.user.is_superuser
|
||
|
||
|
||
@admin.register(LeaveEntitlement)
|
||
class LeaveEntitlementAdmin(ModelAdmin):
|
||
list_display = ('employee', 'year', 'annual_days', 'additional_days', 'sick_days')
|
||
list_filter = ('year',)
|
||
search_fields = ('employee__username', 'employee__last_name', 'employee__first_name')
|
||
autocomplete_fields = ('employee',)
|
||
readonly_fields = ('uuid', 'created_at', 'updated_at')
|
||
|
||
|
||
@admin.register(EmployeeHrRequest)
|
||
class EmployeeHrRequestAdmin(ModelAdmin):
|
||
list_display = (
|
||
'employee', 'request_type', 'substitute', 'date_from', 'date_to',
|
||
'status', 'created_at', 'decided_by',
|
||
)
|
||
list_filter = ('status', 'request_type', 'created_at')
|
||
search_fields = (
|
||
'employee__username', 'employee__last_name', 'employee__first_name',
|
||
'note', 'decision_note',
|
||
)
|
||
list_select_related = ('employee', 'decided_by', 'substitute')
|
||
autocomplete_fields = ('employee', 'decided_by', 'substitute')
|
||
readonly_fields = ('uuid', 'created_at', 'updated_at', 'decided_at')
|
||
date_hierarchy = 'created_at'
|