lokses
81dee5cba2
- ad_connector: страницы управления OU и импорта из AD, поле ad_ou_dn в Directory, ADSyncLog; расширение sync-сервиса и клиента - mattermost_connector: personal access token шифруется (Fernet, token_service), бот-сервис, команды создания бота и синхронизации; публикация постов в каналы. Тесты (19) - core: ConnectorConfig, Directory (ad_ou_dn), обновления админки Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
288 lines
12 KiB
Python
288 lines
12 KiB
Python
import functools
|
||
|
||
from django.db import models
|
||
from treenode.models import TreeNodeModel
|
||
from core.models import BaseModel
|
||
from pytils.translit import slugify
|
||
from django.core.exceptions import ValidationError
|
||
from django.utils.translation import gettext_lazy as _
|
||
|
||
|
||
def _resolve_directory_default(type_code, code):
|
||
return Directory.objects.filter(
|
||
directory_type__code=type_code, code=code,
|
||
).values_list('id', flat=True).first()
|
||
|
||
|
||
def directory_default(type_code, code):
|
||
"""Ледачий default= для ForeignKey('core.Directory') на місці колишнього
|
||
choices=CharField: резолвиться при збереженні, а не при імпорті моделі
|
||
(щоб не залежати від порядку застосування seed-міграцій). functools.partial,
|
||
а не замикання, — щоб Django міг серіалізувати default у міграціях."""
|
||
return functools.partial(_resolve_directory_default, type_code, code)
|
||
|
||
|
||
class DirectoryType(BaseModel):
|
||
code = models.CharField(max_length=50, unique=True, verbose_name=_('Код типа'))
|
||
name = models.CharField(max_length=100, verbose_name=_('Название типа'))
|
||
description = models.TextField(blank=True, verbose_name=_('Описание'))
|
||
is_active = models.BooleanField(default=True, verbose_name=_('Активный'))
|
||
|
||
# Новые поля для управления иерархией
|
||
HIERARCHY_TYPES = [
|
||
('strict', _('Строгая иерархия (одинаковые типы)')),
|
||
('flexible', _('Гибкая иерархия (разные типы)')),
|
||
('custom', _('Кастомная иерархия')),
|
||
]
|
||
|
||
hierarchy_type = models.CharField(
|
||
max_length=10,
|
||
choices=HIERARCHY_TYPES,
|
||
default='strict',
|
||
verbose_name=_('Тип иерархии')
|
||
)
|
||
|
||
allowed_child_types = models.ManyToManyField(
|
||
'self',
|
||
symmetrical=False,
|
||
blank=True,
|
||
verbose_name=_('Разрешенные типы потомков'),
|
||
help_text=_('Только для гибкой и кастомной иерархии')
|
||
)
|
||
|
||
class Meta:
|
||
verbose_name = _('Тип справочника')
|
||
verbose_name_plural = _('Типы справочников')
|
||
ordering = ['name']
|
||
|
||
def can_have_child_type(self, child_type):
|
||
"""Проверить, может ли данный тип быть потомком (для admin и валидации)."""
|
||
if self.hierarchy_type == 'strict':
|
||
return child_type == self
|
||
if self.hierarchy_type == 'flexible':
|
||
return True
|
||
if self.hierarchy_type == 'custom':
|
||
return child_type in self.allowed_child_types.all()
|
||
return False
|
||
|
||
def __str__(self):
|
||
return self.name
|
||
|
||
class DirectoryManager(models.Manager):
|
||
def get_by_type(self, type_code):
|
||
"""Получить активные записи по типу справочника"""
|
||
return self.filter(
|
||
directory_type__code=type_code,
|
||
is_active=True
|
||
).order_by('name',)
|
||
|
||
def get_choices(self, type_code):
|
||
"""Получить choices для форм"""
|
||
return self.get_by_type(type_code).values_list('id', 'name')
|
||
|
||
def get_by_code(self, type_code, item_code):
|
||
"""Получить элемент по коду типа и коду элемента. Raises DoesNotExist если не найден."""
|
||
return self.get(
|
||
directory_type__code=type_code,
|
||
code=item_code,
|
||
is_active=True
|
||
)
|
||
def with_related(self):
|
||
"""Оптимизированный queryset с select_related"""
|
||
return self.select_related('directory_type', 'parent')
|
||
|
||
def ordered(self):
|
||
"""Упорядоченный queryset"""
|
||
return self.order_by('_path',)
|
||
|
||
def for_admin(self, request):
|
||
"""Queryset для админки"""
|
||
return self.with_related().ordered()
|
||
|
||
|
||
|
||
class Directory(TreeNodeModel, BaseModel):
|
||
"""Универсальная модель для иерархических справочников"""
|
||
|
||
# Поле для отображения в дереве (treenode admin)
|
||
display_field = "name"
|
||
|
||
|
||
|
||
|
||
directory_type = models.ForeignKey(
|
||
DirectoryType,
|
||
on_delete=models.CASCADE,
|
||
verbose_name=_('Тип справочника')
|
||
)
|
||
code = models.CharField(max_length=100, verbose_name=_('Код'))
|
||
name = models.CharField(max_length=255, verbose_name=_('Название'))
|
||
description = models.TextField(blank=True, verbose_name=_('Описание'))
|
||
is_active = models.BooleanField(default=True, verbose_name=_('Активный'))
|
||
|
||
extra_data = models.JSONField(default=dict, blank=True, verbose_name=_('Дополнительные данные'))
|
||
|
||
location_types = models.ManyToManyField(
|
||
'self',
|
||
symmetrical=False,
|
||
blank=True,
|
||
limit_choices_to={'directory_type__code': 'location_type', 'is_active': True},
|
||
related_name='available_posts',
|
||
verbose_name=_('Типы локаций'),
|
||
help_text=_('Для должностей: в каких типах локаций доступна эта должность')
|
||
)
|
||
|
||
# Active Directory интеграция (для отделов)
|
||
ad_group_dn = models.CharField(
|
||
'AD Group DN',
|
||
max_length=500,
|
||
blank=True,
|
||
help_text='Distinguished Name группы в Active Directory (например: CN=IT_Department,OU=Groups,DC=company,DC=local). Только для отделов.'
|
||
)
|
||
ad_ou_dn = models.CharField(
|
||
'AD OU DN',
|
||
max_length=500,
|
||
blank=True,
|
||
help_text='Distinguished Name организационного юнита (OU) в Active Directory, куда переносятся '
|
||
'учётные записи сотрудников этого отдела (например: OU=IT,OU=Users,DC=company,DC=local). '
|
||
'Только для отделов. Если не задано — используется корневая OU коннектора AD.'
|
||
)
|
||
|
||
objects = DirectoryManager()
|
||
|
||
class Meta:
|
||
verbose_name = _('Элемент справочника')
|
||
verbose_name_plural = _('Элементы справочников')
|
||
unique_together = ['directory_type', 'code']
|
||
indexes = [
|
||
models.Index(fields=['directory_type', 'is_active']),
|
||
models.Index(fields=['parent', 'directory_type']),
|
||
models.Index(fields=['is_active']),
|
||
]
|
||
|
||
def __str__(self):
|
||
return self.name
|
||
|
||
def save(self, *args, **kwargs):
|
||
if not self.code:
|
||
self.code = slugify(self.name)
|
||
|
||
# Убедиться, что код уникален в пределах типа
|
||
if self.pk is None: # Только для новых объектов
|
||
base_code = self.code
|
||
counter = 1
|
||
while Directory.objects.filter(
|
||
directory_type=self.directory_type,
|
||
code=self.code
|
||
).exists():
|
||
self.code = f"{base_code}_{counter}"
|
||
counter += 1
|
||
|
||
return super().save(*args, **kwargs)
|
||
|
||
def clean(self):
|
||
super().clean()
|
||
|
||
if not self.parent and self.directory_type.hierarchy_type == 'custom':
|
||
if self.directory_type.allowed_child_types.exists():
|
||
raise ValidationError(
|
||
f'Тип "{self.directory_type}" с кастомной иерархией и заданными '
|
||
f'разрешёнными типами потомков не допускает корневых элементов.'
|
||
)
|
||
|
||
if self.parent:
|
||
parent_type = self.parent.directory_type
|
||
current_type = self.directory_type
|
||
|
||
# Проверка в зависимости от типа иерархии родителя
|
||
if parent_type.hierarchy_type == 'strict':
|
||
if current_type != parent_type:
|
||
raise ValidationError(
|
||
f'Для типа "{parent_type}" разрешена только строгая иерархия. '
|
||
f'Тип потомка должен быть "{parent_type}", а не "{current_type}"'
|
||
)
|
||
|
||
elif parent_type.hierarchy_type == 'custom':
|
||
if current_type not in parent_type.allowed_child_types.all():
|
||
raise ValidationError(
|
||
f'Для типа "{parent_type}" разрешены только следующие типы потомков: '
|
||
f'{", ".join(str(t) for t in parent_type.allowed_child_types.all())}'
|
||
)
|
||
|
||
# def get_queryset(self, request):
|
||
# qs = super().get_queryset(request).select_related('directory_type', 'parent')
|
||
|
||
# return qs.order_by('sort_order', '_path')
|
||
|
||
def get_full_name(self):
|
||
"""Полное название с учетом иерархии"""
|
||
return " -> ".join([node.name for node in self.get_ancestors()] + [self.name])
|
||
|
||
@property
|
||
def all_descendants_count(self):
|
||
"""Количество всех потомков"""
|
||
return self.get_descendants().count()
|
||
|
||
def get_children_count(self):
|
||
"""Безопасное получение количества дочерних элементов"""
|
||
return Directory.objects.filter(parent=self).count()
|
||
|
||
def get_siblings_same_type(self):
|
||
"""Получить соседей того же типа"""
|
||
return self.__class__.objects.filter(
|
||
directory_type=self.directory_type,
|
||
parent=self.parent
|
||
).exclude(pk=self.pk)
|
||
|
||
|
||
def get_extra_value(self, key, default=None):
|
||
"""Получить значение из extra_data по ключу"""
|
||
return self.extra_data.get(key, default)
|
||
|
||
def set_extra_value(self, key, value, save=False):
|
||
"""Установить значение в extra_data. При save=True сохраняет в БД."""
|
||
if self.extra_data is None:
|
||
self.extra_data = {}
|
||
self.extra_data[key] = value
|
||
if save:
|
||
self.save(update_fields=['extra_data', 'updated_at'])
|
||
|
||
def get_email_key(self):
|
||
"""Получить ключ для формирования email"""
|
||
return self.get_extra_value('email_key', '')
|
||
|
||
def get_short_name(self):
|
||
"""Получить короткое название"""
|
||
return self.get_extra_value('short_name', self.name)
|
||
def get_ancestors_same_type(self):
|
||
"""Получить предков того же типа (для строгой иерархии)"""
|
||
if self.directory_type.hierarchy_type != 'strict':
|
||
return self.get_ancestors()
|
||
return self.get_ancestors().filter(directory_type=self.directory_type)
|
||
|
||
def get_descendants_same_type(self):
|
||
"""Получить потомков того же типа (для строгой иерархии)"""
|
||
if self.directory_type.hierarchy_type != 'strict':
|
||
return self.get_descendants()
|
||
return self.get_descendants().filter(directory_type=self.directory_type)
|
||
|
||
def can_have_child_type(self, child_type):
|
||
"""Проверить, может ли данный тип быть потомком"""
|
||
if self.directory_type.hierarchy_type == 'strict':
|
||
return child_type == self.directory_type
|
||
elif self.directory_type.hierarchy_type == 'flexible':
|
||
return True
|
||
elif self.directory_type.hierarchy_type == 'custom':
|
||
return child_type in self.directory_type.allowed_child_types.all()
|
||
return False
|
||
|
||
@property
|
||
def allowed_child_types_list(self):
|
||
"""Список разрешенных типов потомков"""
|
||
if self.directory_type.hierarchy_type == 'strict':
|
||
return [self.directory_type]
|
||
elif self.directory_type.hierarchy_type == 'flexible':
|
||
return DirectoryType.objects.all()
|
||
elif self.directory_type.hierarchy_type == 'custom':
|
||
return self.directory_type.allowed_child_types.all()
|
||
return [] |