from __future__ import annotations from django.contrib.auth import get_user_model from django.contrib.contenttypes.models import ContentType from django.db import transaction from django.db.models import Q, QuerySet from schedule.models import Calendar, CalendarRelation, Event, EventRelation from employees.models import Department, EmployeeDepartment, RoleAssignment from .models import CalendarBinding User = get_user_model() def get_calendar_binding(calendar: Calendar | None) -> CalendarBinding | None: if calendar is None: return None binding = getattr(calendar, "binding", None) if binding is not None: return binding return ( CalendarBinding.objects.select_related("context_content_type") .filter(calendar=calendar) .first() ) def is_department_calendar(calendar: Calendar | None) -> bool: binding = get_calendar_binding(calendar) return bool(binding and binding.type == CalendarBinding.BindingType.DEPARTMENT) def is_company_wide_calendar(calendar: Calendar | None) -> bool: binding = get_calendar_binding(calendar) return bool( binding and binding.type == CalendarBinding.BindingType.DEFAULT and binding.access_scope == CalendarBinding.AccessScope.ALL ) def get_department_calendar_binding( department: Department, ) -> CalendarBinding | None: dept_ct = ContentType.objects.get_for_model(Department) return ( CalendarBinding.objects.select_related("calendar") .filter( type=CalendarBinding.BindingType.DEPARTMENT, context_content_type=dept_ct, context_object_id=department.id, ) .first() ) def department_calendar_name(department: Department) -> str: return department.name def department_calendar_slug(department: Department) -> str: return f"department-{department.id}" def _unique_calendar_slug(base_slug: str) -> str: if not Calendar.objects.filter(slug=base_slug).exists(): return base_slug counter = 1 while Calendar.objects.filter(slug=f"{base_slug}-{counter}").exists(): counter += 1 return f"{base_slug}-{counter}" def _find_existing_department_calendar( department: Department, ) -> Calendar | None: slug = department_calendar_slug(department) return ( Calendar.objects.filter(slug=slug) .exclude(binding__isnull=False) .order_by("-id") .first() ) @transaction.atomic def get_or_create_department_calendar( department: Department, ) -> tuple[Calendar, CalendarBinding]: binding = get_department_calendar_binding(department) if binding: sync_department_calendar_binding(department, binding=binding) return binding.calendar, binding calendar = _find_existing_department_calendar(department) if calendar is None: calendar = Calendar.objects.create( name=department_calendar_name(department), slug=_unique_calendar_slug(department_calendar_slug(department)), ) dept_ct = ContentType.objects.get_for_model(Department) binding, _ = CalendarBinding.objects.get_or_create( calendar=calendar, defaults={ "type": CalendarBinding.BindingType.DEPARTMENT, "context_content_type": dept_ct, "context_object_id": department.id, }, ) if ( binding.type != CalendarBinding.BindingType.DEPARTMENT or binding.context_content_type_id != dept_ct.id or binding.context_object_id != department.id ): binding.type = CalendarBinding.BindingType.DEPARTMENT binding.context_content_type = dept_ct binding.context_object_id = department.id binding.save( update_fields=[ "type", "context_content_type", "context_object_id", "updated_at", ] ) sync_department_calendar_binding(department, binding=binding) return binding.calendar, binding def _department_participant_ids(department: Department) -> set[int]: ids: set[int] = set( EmployeeDepartment.objects.filter( department=department, is_active=True, employee__is_active=True, ).values_list("employee_id", flat=True) ) role_only_ids = RoleAssignment.objects.filter( role__department=department, is_active=True, employee__is_active=True, ).values_list("employee_id", flat=True) ids.update(role_only_ids) if department.head_id: ids.add(department.head_id) return ids def _set_relation_role(calendar: Calendar, user_id: int, distinction: str) -> None: user_ct = ContentType.objects.get_for_model(User) relations = list( CalendarRelation.objects.filter( calendar=calendar, content_type=user_ct, object_id=user_id, ).order_by("id") ) if relations: primary = relations[0] changed = False if primary.distinction != distinction: primary.distinction = distinction changed = True if primary.inheritable is not True: primary.inheritable = True changed = True if changed: primary.save(update_fields=["distinction", "inheritable"]) if len(relations) > 1: CalendarRelation.objects.filter( id__in=[relation.id for relation in relations[1:]] ).delete() return calendar.create_relation( User.objects.get(pk=user_id), distinction=distinction, inheritable=True, ) def sync_department_calendar_owner( department: Department, *, binding: CalendarBinding | None = None ) -> Calendar | None: binding = binding or get_department_calendar_binding(department) if not binding: return None calendar = binding.calendar desired_owner_id = department.head_id user_ct = ContentType.objects.get_for_model(User) owner_relations = CalendarRelation.objects.filter( calendar=calendar, content_type=user_ct, distinction="owner", ).order_by("id") if desired_owner_id is None: owner_relations.delete() return calendar _set_relation_role(calendar, desired_owner_id, "owner") owner_relations.exclude(object_id=desired_owner_id).delete() return calendar def sync_department_calendar_members( department: Department, *, binding: CalendarBinding | None = None ) -> Calendar | None: binding = binding or get_department_calendar_binding(department) if not binding: return None calendar = binding.calendar desired_ids = _department_participant_ids(department) owner_id = department.head_id user_ct = ContentType.objects.get_for_model(User) existing_relations = list( CalendarRelation.objects.filter( calendar=calendar, content_type=user_ct, ).order_by("id") ) if owner_id: _set_relation_role(calendar, owner_id, "owner") for user_id in sorted(desired_ids): if user_id == owner_id: continue _set_relation_role(calendar, user_id, "editor") stale_relation_ids = [ relation.id for relation in existing_relations if relation.object_id not in desired_ids ] if stale_relation_ids: CalendarRelation.objects.filter(id__in=stale_relation_ids).delete() return calendar def sync_department_calendar_binding( department: Department, *, binding: CalendarBinding | None = None ) -> Calendar | None: binding = binding or get_department_calendar_binding(department) if not binding: return None calendar = binding.calendar desired_name = department_calendar_name(department) if calendar.name != desired_name: calendar.name = desired_name calendar.save(update_fields=["name"]) sync_department_calendar_owner(department, binding=binding) sync_department_calendar_members(department, binding=binding) return calendar def sync_department_calendar_if_bound(department: Department) -> None: binding = get_department_calendar_binding(department) if not binding: return sync_department_calendar_binding(department, binding=binding) def sync_department_calendar_binding_by_department_id(department_id: int) -> None: try: department = Department.objects.get(pk=department_id) except Department.DoesNotExist: return sync_department_calendar_if_bound(department) def delete_department_calendar(department_id: int) -> None: try: department = Department.objects.get(pk=department_id) except Department.DoesNotExist: department = None if department: binding = get_department_calendar_binding(department) else: dept_ct = ContentType.objects.get_for_model(Department) binding = ( CalendarBinding.objects.filter( type=CalendarBinding.BindingType.DEPARTMENT, context_content_type=dept_ct, context_object_id=department_id, ) .select_related("calendar") .first() ) if binding: binding.calendar.delete() def user_relation_distinction(calendar: Calendar, user) -> str | None: if not getattr(user, "is_authenticated", False): return None if user.is_staff or user.is_superuser: return "owner" user_ct = ContentType.objects.get_for_model(User) relation = ( CalendarRelation.objects.filter( calendar=calendar, content_type=user_ct, object_id=user.id, ) .order_by("id") .first() ) return relation.distinction if relation else None def user_can_view_calendar(user, calendar: Calendar) -> bool: if not getattr(user, "is_authenticated", False): return False if user.is_staff or user.is_superuser: return True if is_department_calendar(calendar): return True if is_company_wide_calendar(calendar): return True return user_relation_distinction(calendar, user) is not None def filter_events_for_user( queryset: QuerySet, user, *, allow_admin_all: bool = False, ) -> QuerySet: """Ограничить события календарями и персональными приглашениями пользователя.""" if not getattr(user, "is_authenticated", False): return queryset.none() if allow_admin_all and (user.is_staff or user.is_superuser): return queryset user_content_type = ContentType.objects.get_for_model(User) calendar_ids = CalendarRelation.objects.filter( content_type=user_content_type, object_id=user.id, ).values_list("calendar_id", flat=True) participant_event_ids = EventRelation.objects.filter( content_type=user_content_type, object_id=user.id, ).values_list("event_id", flat=True) return queryset.filter( Q(calendar_id__in=calendar_ids) | Q(id__in=participant_event_ids) | Q( calendar__binding__type=CalendarBinding.BindingType.DEFAULT, calendar__binding__access_scope=CalendarBinding.AccessScope.ALL, ) ).distinct() def user_can_view_event(user, event: Event) -> bool: if not event: return False if user_can_view_calendar(user, event.calendar): return True user_content_type = ContentType.objects.get_for_model(User) return EventRelation.objects.filter( event=event, content_type=user_content_type, object_id=user.id, ).exists() def user_can_create_event(user, calendar: Calendar) -> bool: if not getattr(user, "is_authenticated", False): return False if user.is_staff or user.is_superuser: return True if is_department_calendar(calendar): return True return user_relation_distinction(calendar, user) in {"owner", "editor"} def user_can_edit_calendar(user, calendar: Calendar) -> bool: if not getattr(user, "is_authenticated", False): return False if user.is_staff or user.is_superuser: return True return user_relation_distinction(calendar, user) == "owner" def user_can_manage_participants(user, calendar: Calendar) -> bool: if not getattr(user, "is_authenticated", False): return False if user.is_staff or user.is_superuser: return True if is_department_calendar(calendar): return False return user_relation_distinction(calendar, user) == "owner" def user_can_manage_calendar_notifications(user, calendar: Calendar) -> bool: """Проверить право менять общие правила уведомлений календаря.""" if not getattr(user, "is_authenticated", False): return False if user.is_staff or user.is_superuser: return True return user_relation_distinction(calendar, user) in {"owner", "editor"} def calendar_notification_enabled(calendar: Calendar | None, field: str) -> bool: """Вернуть настройку уведомления, сохраняя прежнее поведение без binding.""" binding = get_calendar_binding(calendar) if binding is None: return True return bool(getattr(binding, field, True)) def user_can_edit_event(user, event: Event) -> bool: if not getattr(user, "is_authenticated", False): return False if user.is_staff or user.is_superuser: return True calendar = event.calendar if is_department_calendar(calendar): relation = user_relation_distinction(calendar, user) if relation in {"owner", "editor"}: return True return event.creator_id == user.id return user_relation_distinction(calendar, user) in {"owner", "editor"} def user_can_delete_event(user, event: Event) -> bool: return user_can_edit_event(user, event)