customerTicket/apps/tickets/policies.py

from django.db.models import Q

from .models import TicketAccessGrant


class TicketPolicy:
    @classmethod
    def can_operate(cls, user, ticket):
        profile = getattr(user, "staff_profile", None)
        return bool(user.is_authenticated and user.is_active and getattr(user, "account_type", None) == "STAFF"
                    and (profile is None or profile.is_active) and cls.can_view(user, ticket))
    @staticmethod
    def is_customer(user): return getattr(user,"account_type",None)=="CUSTOMER"
    @staticmethod
    def unit_ids(user):
        if not user.is_authenticated:
            return set()
        if user.is_superuser:
            from apps.staff.models import OrganizationalUnit
            return set(OrganizationalUnit.objects.filter(is_active=True).values_list("id", flat=True))
        return set(
            user.unit_memberships.filter(is_active=True, unit__is_active=True)
            .values_list("unit_id", flat=True)
        )
    @staticmethod
    def primary_unit(user):
        profile = getattr(user, "staff_profile", None)
        if profile and profile.primary_unit_id:
            return profile.primary_unit
        membership = user.unit_memberships.filter(is_active=True, unit__is_active=True).select_related("unit").order_by("-is_primary", "id").first()
        return membership.unit if membership else None
    @classmethod
    def unit_scope(cls, user, unit):
        if not user.is_authenticated or unit is None:
            return None
        if user.is_superuser or user.has_perm("tickets.view_all_tickets"):
            return "GLOBAL"
        return user.unit_memberships.filter(is_active=True, unit=unit).values_list("ticket_scope", flat=True).first()
    @classmethod
    def has_unit_permission(cls, user, unit, permission):
        """Resolve a permission dynamically from unit and membership configuration."""
        if not user.is_authenticated or unit is None:
            return False
        app_label, codename = permission.split(".", 1)
        memberships = user.unit_memberships.filter(is_active=True, unit=unit)
        return memberships.filter(
            Q(unit__permissions__content_type__app_label=app_label, unit__permissions__codename=codename)
            | Q(permissions__content_type__app_label=app_label, permissions__codename=codename)
        ).exists()
    @classmethod
    def can_access_unit(cls, user, unit, permission="tickets.view_ticket"):
        return bool(
            user.is_authenticated
            and (
                user.is_superuser
                or user.has_perm("tickets.view_all_tickets")
                or cls.has_unit_permission(user, unit, permission)
            )
        )
    @classmethod
    def viewable_unit_ids(cls, user):
        if not user.is_authenticated:
            return set()
        if user.is_superuser or user.has_perm("tickets.view_all_tickets"):
            return set(user.unit_memberships.filter(is_active=True).values_list("unit_id", flat=True))
        return {
            membership.unit_id
            for membership in user.unit_memberships.filter(is_active=True, unit__is_active=True).select_related("unit")
            if cls.has_unit_permission(user, membership.unit, "tickets.view_ticket")
        }
    @classmethod
    def can_view(cls,user,ticket):
        if not user.is_authenticated: return False
        if cls.is_customer(user): return ticket.customer_id==user.id
        if user.is_superuser or user.has_perm("tickets.view_all_tickets"): return True
        if ticket.created_by_id == user.id or ticket.selected_expert_id == user.id or ticket.owner_id == user.id: return True
        if ticket.access_grants.filter(user=user,revoked_at__isnull=True).exists(): return True
        unit = ticket.assigned_unit or ticket.owning_unit
        if unit and cls.can_access_unit(user, unit):
            scope = cls.unit_scope(user, unit)
            if scope == "UNIT": return True
            return bool(ticket.origin_visibility == "PUBLIC" and ticket.owner_id is None and ticket.selected_expert_id is None)
        return False
    @classmethod
    def viewable_q(cls, user):
        """Database equivalent of can_view; use it for every staff ticket collection."""
        if user.is_superuser or user.has_perm("tickets.view_all_tickets"):
            return Q()
        direct = Q(created_by=user) | Q(selected_expert=user) | Q(owner=user) | Q(access_grants__user=user, access_grants__revoked_at__isnull=True)
        memberships = user.unit_memberships.filter(is_active=True, unit__is_active=True)
        permitted_ids = {
            item.unit_id for item in memberships.select_related("unit")
            if cls.has_unit_permission(user, item.unit, "tickets.view_ticket")
        }
        unit_wide_ids = set(memberships.filter(ticket_scope="UNIT", unit_id__in=permitted_ids).values_list("unit_id", flat=True))
        unit_wide = Q(assigned_unit_id__in=unit_wide_ids) | Q(
            assigned_unit__isnull=True,
            owning_unit_id__in=unit_wide_ids,
        )
        limited_ids = permitted_ids - unit_wide_ids
        public_unassigned = (
            (
                Q(assigned_unit_id__in=limited_ids)
                | Q(assigned_unit__isnull=True, owning_unit_id__in=limited_ids)
            )
            & Q(origin_visibility="PUBLIC", owner__isnull=True, selected_expert__isnull=True)
        )
        # There is deliberately no global/unitless public queue. Legacy tickets
        # must be assigned to a unit by the data migration before staff can see
        # them through a unit queue.
        return direct | unit_wide | public_unassigned
    @classmethod
    def can_reply(cls,user,ticket):
        if not cls.can_view(user,ticket): return False
        if cls.is_customer(user): return ticket.customer_id==user.id
        if cls.has_unit_permission(user,ticket.assigned_unit or ticket.owning_unit,"messaging.add_message"): return True
        if user.has_perm("tickets.view_all_tickets") or ticket.owner_id==user.id or ticket.selected_expert_id==user.id or ticket.created_by_id==user.id: return True
        if ticket.owner_id is None and ticket.selected_expert_id is None and user.has_perm("messaging.add_message"): return True
        return ticket.access_grants.filter(user=user,revoked_at__isnull=True,permission_level__in=[TicketAccessGrant.Level.REPLY,TicketAccessGrant.Level.FULL]).exists()
    @classmethod
    def can_claim(cls,user,ticket):
        permitted = user.has_perm("tickets.claim_ticket") or cls.has_unit_permission(user, ticket.assigned_unit or ticket.owning_unit, "tickets.claim_ticket")
        return permitted and cls.can_view(user, ticket) and ticket.owner_id is None and ticket.selected_expert_id is None
    @classmethod
    def can_transfer(cls,user,ticket):
        return cls.can_operate(user, ticket)
    @classmethod
    def can_share(cls,user,ticket):
        return cls.can_operate(user, ticket)
    @classmethod
    def can_delete(cls, user, ticket):
        if not user.is_authenticated or cls.is_customer(user):
            return False
        if user.is_superuser:
            return True
        if ticket.owning_unit_id:
            return cls.has_unit_permission(user,ticket.owning_unit,"tickets.delete_ticket") and cls.can_view(user,ticket)
        profile = getattr(user, "staff_profile", None)
        return bool(profile and profile.role != profile.Role.SALES_EXPERT and user.has_perm("tickets.delete_ticket") and cls.can_view(user,ticket))


class DispatchPolicy:
    CREATE_PERMISSIONS = {
        "BULK": "tickets.create_bulk_dispatch",
        "INQUIRY": "tickets.create_inquiry",
    }
    VIEW_ALL_PERMISSIONS = {
        "BULK": "tickets.view_all_bulk_dispatches",
        "INQUIRY": "tickets.view_all_inquiries",
    }

    @classmethod
    def can_create(cls, user, unit, operation_type):
        permission = cls.CREATE_PERMISSIONS.get(operation_type)
        if not permission or not user.is_authenticated or unit is None:
            return False
        allowed = user.is_superuser or user.has_perm(permission) or TicketPolicy.has_unit_permission(user, unit, permission)
        if operation_type == "BULK":
            allowed = allowed or user.has_perm("tickets.dispatch_group_tickets") or TicketPolicy.has_unit_permission(user, unit, "tickets.dispatch_group_tickets")
        return bool(allowed)

    @classmethod
    def can_view_all(cls, user, operation_type, unit=None):
        permission = cls.VIEW_ALL_PERMISSIONS.get(operation_type)
        if not permission or not user.is_authenticated:
            return False
        return bool(user.is_superuser or user.has_perm(permission) or (unit and TicketPolicy.has_unit_permission(user, unit, permission)))

    @classmethod
    def can_view(cls, user, dispatch):
        if dispatch.created_by_id == user.id:
            return cls.can_create(user, dispatch.unit, dispatch.operation_type)
        return cls.can_view_all(user, dispatch.operation_type, dispatch.unit)