from __future__ import annotations from dataclasses import dataclass from datetime import date, datetime, timezone from typing import Any from sqlalchemy import or_, select from sqlalchemy.orm import Session, selectinload from app.modules.clients import repository as client_repository from app.modules.clients.models import Client from app.modules.consultants.models import ClientConsultantLink, ConsultantProfile from app.modules.core.rbac.deps import get_user_roles from app.modules.documents.models import EngagementDocument, PermanentClientDocument from app.modules.services.execution import ( CLOSED_TASK_STATUSES, TASK_COMMENT_TYPES, TASK_COMMENT_VISIBILITIES, TASK_PRIORITIES, TASK_STATUSES, add_task_comment, ) from app.modules.services.models import ( ClientServiceSubscription, ClientServiceTaskInstance, ServiceTaskComment, ) MANAGEMENT_ROLES = {"System Admin", "Firm Admin"} PARTNER_ROLES = {"Partner"} MANAGER_ROLES = {"Manager", "Branch Manager"} STAFF_ROLES = {"Staff", "Employee"} CLIENT_ROLES = {"Client"} CONSULTANT_ROLES = {"Consultant"} @dataclass(frozen=True) class WorkAccess: role_context: str can_update_tasks: bool can_comment: bool allowed_comment_types: list[tuple[str, str]] allowed_visibilities: list[tuple[str, str]] back_url: str def _roles(db: Session, user) -> set[str]: return set(get_user_roles(db, user.id)) def _safe_int(value: Any) -> int | None: try: return int(value) if value not in (None, "", "None") else None except Exception: return None def _current_client_row(db: Session, user): try: return client_repository.get_portal_client_for_user(db, user=user) except Exception: return None def _current_consultant(db: Session, user) -> ConsultantProfile | None: return db.execute( select(ConsultantProfile).where( ConsultantProfile.user_id == user.id, ConsultantProfile.tenant_id == user.tenant_id, ConsultantProfile.is_active.is_(True), ) ).scalar_one_or_none() def _consultant_can_view_engagement(db: Session, *, consultant: ConsultantProfile, engagement: ClientServiceSubscription) -> bool: linked = db.execute( select(ClientConsultantLink.id).where( ClientConsultantLink.tenant_id == consultant.tenant_id, ClientConsultantLink.client_id == engagement.client_id, ClientConsultantLink.consultant_id == consultant.id, ClientConsultantLink.is_active.is_(True), ClientConsultantLink.can_view_communications.is_(True), ) ).first() if not linked: return False visible_comment = db.execute( select(ServiceTaskComment.id) .join(ClientServiceTaskInstance, ClientServiceTaskInstance.id == ServiceTaskComment.task_instance_id) .where( ServiceTaskComment.tenant_id == consultant.tenant_id, ServiceTaskComment.subscription_id == engagement.id, ServiceTaskComment.visibility == "consultant", ServiceTaskComment.is_deleted.is_(False), ClientServiceTaskInstance.is_active.is_(True), ) .limit(1) ).first() return bool(visible_comment) def _is_assigned_staff(engagement: ClientServiceSubscription, tasks: list[ClientServiceTaskInstance], user) -> bool: if _safe_int(getattr(engagement, "assigned_staff_user_id", None)) == user.id: return True return any(_safe_int(getattr(task, "assigned_to_user_id", None)) == user.id for task in tasks) def _is_manager_for_engagement(engagement: ClientServiceSubscription, user) -> bool: return _safe_int(getattr(engagement, "assigned_manager_user_id", None)) == user.id def _is_partner_for_engagement(engagement: ClientServiceSubscription, user) -> bool: return user.id in { _safe_int(getattr(engagement, "assigned_partner_user_id", None)), _safe_int(getattr(engagement, "review_partner_user_id", None)), } def _engagement_query(engagement_id: int): return ( select(ClientServiceSubscription) .options( selectinload(ClientServiceSubscription.client), selectinload(ClientServiceSubscription.catalogue), selectinload(ClientServiceSubscription.assigned_partner), selectinload(ClientServiceSubscription.assigned_manager), selectinload(ClientServiceSubscription.assigned_staff), selectinload(ClientServiceSubscription.review_partner), ) .where(ClientServiceSubscription.id == int(engagement_id), ClientServiceSubscription.is_active.is_(True)) ) def _load_tasks(db: Session, engagement: ClientServiceSubscription) -> list[ClientServiceTaskInstance]: return db.execute( select(ClientServiceTaskInstance) .options( selectinload(ClientServiceTaskInstance.assigned_to), selectinload(ClientServiceTaskInstance.catalogue), selectinload(ClientServiceTaskInstance.comments).selectinload(ServiceTaskComment.created_by), ) .where( ClientServiceTaskInstance.tenant_id == engagement.tenant_id, ClientServiceTaskInstance.subscription_id == engagement.id, ClientServiceTaskInstance.is_active.is_(True), ) .order_by(ClientServiceTaskInstance.sequence_no.asc(), ClientServiceTaskInstance.id.asc()) ).scalars().all() def _load_documents(db: Session, engagement: ClientServiceSubscription) -> tuple[list[EngagementDocument], list[PermanentClientDocument]]: engagement_documents = db.execute( select(EngagementDocument) .options(selectinload(EngagementDocument.versions)) .where( EngagementDocument.tenant_id == engagement.tenant_id, EngagementDocument.client_id == engagement.client_id, EngagementDocument.engagement_id == engagement.id, EngagementDocument.is_deleted.is_(False), ) .order_by(EngagementDocument.document_type.asc(), EngagementDocument.title.asc()) ).unique().scalars().all() permanent_documents = db.execute( select(PermanentClientDocument) .options(selectinload(PermanentClientDocument.versions)) .where( PermanentClientDocument.tenant_id == engagement.tenant_id, PermanentClientDocument.client_id == engagement.client_id, PermanentClientDocument.is_deleted.is_(False), ) .order_by(PermanentClientDocument.category.asc(), PermanentClientDocument.title.asc()) .limit(50) ).unique().scalars().all() return engagement_documents, permanent_documents def _load_timeline(db: Session, engagement: ClientServiceSubscription, access: WorkAccess) -> list[ServiceTaskComment]: stmt = ( select(ServiceTaskComment) .options(selectinload(ServiceTaskComment.created_by), selectinload(ServiceTaskComment.task)) .where( ServiceTaskComment.tenant_id == engagement.tenant_id, ServiceTaskComment.subscription_id == engagement.id, ServiceTaskComment.is_deleted.is_(False), ) ) if access.role_context == "client": stmt = stmt.where(ServiceTaskComment.visibility == "client") elif access.role_context == "consultant": stmt = stmt.where(ServiceTaskComment.visibility == "consultant") return db.execute(stmt.order_by(ServiceTaskComment.created_at_utc.asc(), ServiceTaskComment.id.asc())).scalars().all() def _build_access(db: Session, *, user, engagement: ClientServiceSubscription, tasks: list[ClientServiceTaskInstance]) -> WorkAccess | None: roles = _roles(db, user) if roles.intersection(MANAGEMENT_ROLES): return WorkAccess("admin", True, True, TASK_COMMENT_TYPES, TASK_COMMENT_VISIBILITIES, "/services/work-tracker") if roles.intersection(CLIENT_ROLES): client_row = _current_client_row(db, user) if client_row and int(client_row.get("id") or 0) == engagement.client_id and int(client_row.get("tenant_id") or 0) == engagement.tenant_id: return WorkAccess("client", False, True, [("client_clarification", "Client Clarification")], [("client", "Client")], "/client/compliance") return None if roles.intersection(CONSULTANT_ROLES): consultant = _current_consultant(db, user) if consultant and _consultant_can_view_engagement(db, consultant=consultant, engagement=engagement): return WorkAccess("consultant", False, True, [("consultant_clarification", "Consultant Clarification")], [("consultant", "Consultant")], "/consultant/work") return None if roles.intersection(PARTNER_ROLES) and _is_partner_for_engagement(engagement, user): return WorkAccess("partner", True, True, TASK_COMMENT_TYPES, TASK_COMMENT_VISIBILITIES, "/partner/reviews") if roles.intersection(MANAGER_ROLES): if _is_manager_for_engagement(engagement, user) or (engagement.tenant_id == user.tenant_id and (engagement.branch_id in (None, user.branch_id))): return WorkAccess("manager", True, True, TASK_COMMENT_TYPES, TASK_COMMENT_VISIBILITIES, "/manager/work") if roles.intersection(STAFF_ROLES) or roles.intersection({"Employee"}): if _is_assigned_staff(engagement, tasks, user): return WorkAccess("staff", True, True, [("internal_note", "Internal Note"), ("client_clarification", "Client Clarification")], [("internal", "Internal"), ("client", "Client")], "/employee/work") return None def load_unified_engagement_detail(db: Session, *, request, user, engagement_id: int) -> dict[str, Any] | None: engagement = db.execute(_engagement_query(engagement_id)).scalar_one_or_none() if not engagement: return None roles = _roles(db, user) if "System Admin" not in roles and engagement.tenant_id != user.tenant_id: return None tasks = _load_tasks(db, engagement) access = _build_access(db, user=user, engagement=engagement, tasks=tasks) if not access: return None engagement_documents, permanent_documents = _load_documents(db, engagement) timeline = _load_timeline(db, engagement, access) today = date.today() for task in tasks: status = (task.status or "pending").lower() task.status_label = dict(TASK_STATUSES).get(status, status.replace("_", " ").title()) task.priority_label = dict(TASK_PRIORITIES).get(task.priority or "normal", (task.priority or "normal").replace("_", " ").title()) task.is_closed_display = status in CLOSED_TASK_STATUSES task.is_overdue_display = bool(task.internal_target_date and task.internal_target_date < today and status not in CLOSED_TASK_STATUSES) task.comments_visible_count = len([c for c in getattr(task, "comments", []) if not getattr(c, "is_deleted", False)]) status_counts: dict[str, int] = {code: 0 for code, _ in TASK_STATUSES} for task in tasks: status_counts[(task.status or "pending").lower()] = status_counts.get((task.status or "pending").lower(), 0) + 1 return { "engagement": engagement, "tasks": tasks, "engagement_documents": engagement_documents, "permanent_documents": permanent_documents, "timeline": timeline, "access": access, "role_context": access.role_context, "task_statuses": TASK_STATUSES, "task_priorities": TASK_PRIORITIES, "status_counts": status_counts, "open_task_count": sum(1 for task in tasks if (task.status or "pending").lower() not in CLOSED_TASK_STATUSES), "completed_task_count": sum(1 for task in tasks if (task.status or "pending").lower() in CLOSED_TASK_STATUSES), } def get_task_for_action(db: Session, *, user, task_id: int) -> tuple[ClientServiceTaskInstance | None, WorkAccess | None]: task = db.execute( select(ClientServiceTaskInstance) .options(selectinload(ClientServiceTaskInstance.subscription), selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.client)) .where(ClientServiceTaskInstance.id == int(task_id), ClientServiceTaskInstance.is_active.is_(True)) ).scalar_one_or_none() if not task or not task.subscription: return None, None detail = load_unified_engagement_detail(db, request=None, user=user, engagement_id=task.subscription_id) if not detail: return None, None return task, detail["access"] def save_task_status(db: Session, *, task: ClientServiceTaskInstance, status: str, priority: str | None, user_id: int) -> None: allowed_statuses = {code for code, _ in TASK_STATUSES} allowed_priorities = {code for code, _ in TASK_PRIORITIES} clean_status = (status or task.status or "pending").strip().lower() clean_priority = (priority or task.priority or "normal").strip().lower() if clean_status in allowed_statuses: task.status = clean_status if clean_priority in allowed_priorities: task.priority = clean_priority if task.status == "in_progress" and not task.started_at_utc: task.started_at_utc = datetime.now(timezone.utc) if task.status in CLOSED_TASK_STATUSES and not task.completed_at_utc: task.completed_at_utc = datetime.now(timezone.utc) task.updated_by_user_id = user_id def save_task_comment(db: Session, *, task: ClientServiceTaskInstance, access: WorkAccess, comment_type: str, visibility: str, message: str, user_id: int) -> bool: allowed_comment_types = {code for code, _ in access.allowed_comment_types} allowed_visibilities = {code for code, _ in access.allowed_visibilities} clean_type = comment_type if comment_type in allowed_comment_types else next(iter(allowed_comment_types), "internal_note") clean_visibility = visibility if visibility in allowed_visibilities else next(iter(allowed_visibilities), "internal") row = add_task_comment(db, task=task, comment_type=clean_type, visibility=clean_visibility, message=message, user_id=user_id) return row is not None