235 lines
7.9 KiB
Python
235 lines
7.9 KiB
Python
from django.db.models import Q
|
|
from attendance.models import AttendanceRecord
|
|
from documents.models import Document
|
|
from employees.models import Department, EmployeeDepartment, RoleAssignment
|
|
from feed.models import Post
|
|
from requests_app.enums import RequestStatus
|
|
from requests_app.models import Request as EmployeeRequest
|
|
from schedule.models import Event
|
|
from scheduling.services import user_can_view_calendar
|
|
|
|
from .models import TaskBoard, TaskBoardAccessScope, TaskLinkedObjectKind
|
|
|
|
|
|
def get_user_department_ids(user) -> list[int]:
|
|
if not user or not user.is_authenticated:
|
|
return []
|
|
|
|
department_ids = set(
|
|
EmployeeDepartment.objects.filter(
|
|
employee_id=user.id,
|
|
is_active=True,
|
|
).values_list("department_id", flat=True)
|
|
)
|
|
department_ids.update(
|
|
Department.objects.filter(head_id=user.id).values_list("id", flat=True)
|
|
)
|
|
department_ids.update(
|
|
RoleAssignment.objects.filter(
|
|
employee_id=user.id,
|
|
is_active=True,
|
|
).values_list("role__department_id", flat=True)
|
|
)
|
|
return sorted(department_ids)
|
|
|
|
|
|
def task_board_access_q(user):
|
|
department_ids = get_user_department_ids(user)
|
|
q = Q(created_by=user)
|
|
non_private_q = ~Q(access_scope=TaskBoardAccessScope.PRIVATE)
|
|
q |= non_private_q & Q(members=user)
|
|
if department_ids:
|
|
q |= non_private_q & Q(departments__in=department_ids)
|
|
q |= Q(
|
|
access_scope=TaskBoardAccessScope.ALL,
|
|
members__isnull=True,
|
|
departments__isnull=True,
|
|
)
|
|
return q
|
|
|
|
|
|
def task_board_group_access_q(user):
|
|
"""Groups owned by the user or shared through at least one accessible board."""
|
|
accessible_boards = TaskBoard.objects.filter(task_board_access_q(user))
|
|
return Q(user=user) | Q(
|
|
is_shared=True,
|
|
boards__in=accessible_boards,
|
|
)
|
|
|
|
|
|
def user_can_access_task_board(user, board: TaskBoard | None) -> bool:
|
|
if not user or not user.is_authenticated or board is None:
|
|
return False
|
|
if board.created_by_id == user.id:
|
|
return True
|
|
if board.access_scope == TaskBoardAccessScope.PRIVATE:
|
|
return False
|
|
if board.members.filter(id=user.id).exists():
|
|
return True
|
|
department_ids = get_user_department_ids(user)
|
|
if department_ids and board.departments.filter(id__in=department_ids).exists():
|
|
return True
|
|
return bool(
|
|
board.access_scope == TaskBoardAccessScope.ALL
|
|
and not board.members.exists()
|
|
and not board.departments.exists()
|
|
)
|
|
|
|
|
|
def user_can_access_calendar_event(user, event: Event | None) -> bool:
|
|
if not user or not user.is_authenticated or event is None:
|
|
return False
|
|
return user_can_view_calendar(user, event.calendar)
|
|
|
|
|
|
def user_can_access_document(user, document: Document | None) -> bool:
|
|
if not user or not user.is_authenticated or document is None:
|
|
return False
|
|
if user.is_staff or user.is_superuser:
|
|
return True
|
|
if (
|
|
user.has_perm("documents.view_document")
|
|
or user.has_perm("documents.add_document")
|
|
or user.has_perm("documents.change_document")
|
|
or user.has_perm("documents.delete_document")
|
|
):
|
|
return True
|
|
if document.sent_to_all and user.is_active:
|
|
return True
|
|
if document.uploaded_by_id == user.id:
|
|
return True
|
|
if document.recipients.filter(id=user.id, is_active=True).exists():
|
|
return True
|
|
return document.departments.filter(
|
|
employeedepartment__employee=user,
|
|
employeedepartment__is_active=True,
|
|
).exists()
|
|
|
|
|
|
def user_can_access_employee_request(
|
|
user,
|
|
request_obj: EmployeeRequest | None,
|
|
) -> bool:
|
|
if not user or not user.is_authenticated or request_obj is None:
|
|
return False
|
|
|
|
if request_obj.status == RequestStatus.DRAFT:
|
|
return request_obj.employee_id == user.id
|
|
|
|
if request_obj.employee_id == user.id:
|
|
return True
|
|
if request_obj.recipients.filter(id=user.id).exists():
|
|
return True
|
|
return request_obj.cc_users.filter(id=user.id).exists()
|
|
|
|
|
|
def user_can_access_procurement_request(user, procurement_request) -> bool:
|
|
if not user or not user.is_authenticated or procurement_request is None:
|
|
return False
|
|
|
|
from procurement.constants import ProcurementStatus
|
|
from procurement.services import ProcurementApprovalResolver
|
|
|
|
if user.is_staff or user.is_superuser:
|
|
return True
|
|
if (
|
|
user.has_perm("procurement.view_procurementrequest")
|
|
or user.has_perm("procurement.change_procurementrequest")
|
|
or user.has_perm("procurement.delete_procurementrequest")
|
|
or user.has_perm("procurement.execute_procurement")
|
|
):
|
|
return True
|
|
if procurement_request.requestor_id == user.id:
|
|
return True
|
|
if procurement_request.executor_id == user.id:
|
|
return True
|
|
if procurement_request.approvals.filter(approver=user).exists():
|
|
return True
|
|
|
|
participant_department_ids = (
|
|
ProcurementApprovalResolver.get_user_department_participant_ids(user)
|
|
)
|
|
if procurement_request.department_id in participant_department_ids:
|
|
return True
|
|
if procurement_request.processing_department_id in participant_department_ids:
|
|
return True
|
|
|
|
return (
|
|
procurement_request.status == ProcurementStatus.APPROVED
|
|
and procurement_request.processing_department_id is None
|
|
)
|
|
|
|
|
|
def user_can_access_employee(user, employee) -> bool:
|
|
return bool(user and user.is_authenticated and employee is not None)
|
|
|
|
|
|
def user_can_access_guest(user, guest) -> bool:
|
|
return bool(user and user.is_authenticated and guest is not None)
|
|
|
|
|
|
def user_can_access_guest_visit(user, visit) -> bool:
|
|
if not user or not user.is_authenticated or visit is None:
|
|
return False
|
|
|
|
from guests.permissions import is_guest_admin
|
|
|
|
return is_guest_admin(user) or visit.inviter_id == user.id
|
|
|
|
|
|
def user_can_access_attendance_record(
|
|
user,
|
|
record: AttendanceRecord | None,
|
|
) -> bool:
|
|
if not user or not user.is_authenticated or record is None:
|
|
return False
|
|
return bool(user.is_staff or user.is_superuser or record.employee_id == user.id)
|
|
|
|
|
|
def user_can_access_project(user, project) -> bool:
|
|
if not user or not user.is_authenticated or project is None:
|
|
return False
|
|
from projects.permissions import visible_projects
|
|
|
|
return visible_projects(user).filter(pk=project.pk).exists()
|
|
|
|
|
|
def user_can_access_feed_post(user, post: Post | None) -> bool:
|
|
return bool(user and user.is_authenticated and post is not None)
|
|
|
|
|
|
def user_can_access_task_link_source(user, link) -> bool:
|
|
"""Check whether a linked resource grants read-only access to its task."""
|
|
if not user or not user.is_authenticated or link is None:
|
|
return False
|
|
|
|
source = link.content_object
|
|
if source is None:
|
|
return False
|
|
|
|
checkers = {
|
|
TaskLinkedObjectKind.POST: user_can_access_feed_post,
|
|
TaskLinkedObjectKind.CALENDAR_EVENT: user_can_access_calendar_event,
|
|
TaskLinkedObjectKind.DOCUMENT: user_can_access_document,
|
|
TaskLinkedObjectKind.REQUEST: user_can_access_employee_request,
|
|
TaskLinkedObjectKind.PROCUREMENT_REQUEST: user_can_access_procurement_request,
|
|
TaskLinkedObjectKind.EMPLOYEE: user_can_access_employee,
|
|
TaskLinkedObjectKind.GUEST: user_can_access_guest,
|
|
TaskLinkedObjectKind.GUEST_VISIT: user_can_access_guest_visit,
|
|
TaskLinkedObjectKind.ATTENDANCE_RECORD: user_can_access_attendance_record,
|
|
TaskLinkedObjectKind.PROJECT: user_can_access_project,
|
|
}
|
|
if link.kind == TaskLinkedObjectKind.MESSAGE:
|
|
from communications.utils import user_can_access_chat
|
|
|
|
return user_can_access_chat(source.chat, user)
|
|
|
|
checker = checkers.get(link.kind)
|
|
if checker is None:
|
|
return False
|
|
try:
|
|
return bool(checker(user, source))
|
|
except (AttributeError, TypeError):
|
|
# A malformed content-type/kind pair must never grant access.
|
|
return False
|