from __future__ import annotations from fastapi import APIRouter, Form, Request from fastapi.responses import RedirectResponse from sqlalchemy import select from app.core.db.common import CommonSessionLocal from app.core.security.csrf import get_or_create_csrf_token, validate_csrf from app.core.security.session_auth import get_current_user from app.core.templating import templates from app.modules.core.rbac.deps import get_user_permissions, get_user_roles from app.modules.core.iam.models import User from app.modules.core.rbac.permission_guard import require_permission from app.modules.services.execution import ( TASK_COMMENT_TYPES, TASK_COMMENT_VISIBILITIES, TASK_PRIORITIES, TASK_STATUSES, add_task_comment, apply_bulk_task_update, apply_task_review, apply_task_checklist_response, apply_task_update, dashboard_stats, generate_tasks_for_subscription, get_subscription_for_execution, get_task, get_tasks_for_bulk_update, list_assignees_for_execution, list_task_comments, list_subscription_execution_payload, list_tasks_payload, parse_date_value, recalculate_task_aqmm_status, submit_task_for_review, ) router = APIRouter(prefix="/services/work-tracker", tags=["services-work-tracker-ui"]) def _base_ctx(request: Request, user, db, **ctx): base = { "request": request, "current_user": user, "current_user_roles": get_user_roles(db, user.id), "current_user_permissions": get_user_permissions(db, user.id), "csrf_token": get_or_create_csrf_token(request), "task_statuses": TASK_STATUSES, "task_priorities": TASK_PRIORITIES, "task_comment_types": TASK_COMMENT_TYPES, "task_comment_visibilities": TASK_COMMENT_VISIBILITIES, } base.update(ctx) return base def _render(request: Request, template: str, db, user, **ctx): return templates.TemplateResponse(template, _base_ctx(request, user, db, **ctx)) def _redirect_denied(): from app.core.http_responses import ui_access_denied return ui_access_denied() def _has_perm(db, user, code: str) -> bool: try: require_permission(db, user, code) return True except Exception: return False def _role_names(db, user) -> set[str]: return set(get_user_roles(db, user.id)) def _is_firm_admin(db, user) -> bool: return "Firm Admin" in _role_names(db, user) def _is_partner(db, user) -> bool: return "Partner" in _role_names(db, user) def _is_staff(db, user) -> bool: return "Staff" in _role_names(db, user) def _can_bulk_manage_tasks(db, user) -> bool: # Operational bulk task management is intentionally limited to Firm Admin and Partner. # Firm Admin keeps the normal edit-permission gate. # Partner is allowed with service_tasks.view because partner visibility is enforced again # on every submitted task id in get_tasks_for_bulk_update(..., partner_user_id=user.id). if _is_firm_admin(db, user): return _has_perm(db, user, "service_tasks.edit") if _is_partner(db, user): return _has_perm(db, user, "service_tasks.view") return False def _can_assign_staff(db, user) -> bool: return _can_bulk_manage_tasks(db, user) def _partner_visibility_user_id(db, user) -> int | None: return int(user.id) if _is_partner(db, user) else None def _staff_own_task_user_id(db, user) -> int | None: return int(user.id) if _is_staff(db, user) else None def _can_staff_update_own_task(db, user, task) -> bool: return _is_staff(db, user) and int(task.assigned_to_user_id or 0) == int(user.id) and _has_perm(db, user, "service_tasks.view") def _can_update_task_status(db, user, task) -> bool: return _can_bulk_manage_tasks(db, user) or _can_staff_update_own_task(db, user, task) def _allowed_comment_type_codes(db, user) -> set[str]: if _can_bulk_manage_tasks(db, user): return {"internal_note", "client_clarification", "consultant_clarification", "manager_review_note", "partner_review_note", "review_partner_review_note", "rework_note"} if _is_staff(db, user): # Staff may communicate through consultants for clients where consultant is the communication channel. return {"internal_note", "staff_work_note", "client_clarification", "consultant_clarification"} return set() def _can_add_task_comment(db, user, task) -> bool: if getattr(task, "is_locked", False) or getattr(getattr(task, "subscription", None), "is_locked", False): return False return _can_update_task_status(db, user, task) def _is_manager_role(db, user) -> bool: roles = _role_names(db, user) return bool({"Manager", "Branch Manager", "Firm Admin"}.intersection(roles)) def _can_submit_task_for_review(db, user, task) -> bool: return _can_update_task_status(db, user, task) def _can_manager_review_task(db, user, task) -> bool: return _can_bulk_manage_tasks(db, user) or (_is_manager_role(db, user) and _has_perm(db, user, "service_tasks.edit")) def _can_partner_review_task(db, user, task) -> bool: return _can_bulk_manage_tasks(db, user) def _can_review_partner_review_task(db, user, task) -> bool: subscription = getattr(task, "subscription", None) if not subscription: return False if _is_firm_admin(db, user): return _has_perm(db, user, "service_tasks.edit") return _is_partner(db, user) and int(getattr(subscription, "review_partner_user_id", 0) or 0) == int(user.id) def _comment_type_options_for_user(db, user): allowed = _allowed_comment_type_codes(db, user) return [(code, label) for code, label in TASK_COMMENT_TYPES if code in allowed] def _resolve_allowed_assignee_id(db, *, tenant_id: int, branch_id: int | None, assigned_to_user_id: int | None) -> int | None: if assigned_to_user_id is None: return None query = select(User.id).where(User.id == assigned_to_user_id, User.tenant_id == tenant_id, User.is_active.is_(True)) if branch_id: query = query.where((User.branch_id == branch_id) | (User.branch_id.is_(None))) return assigned_to_user_id if db.execute(query).first() else None def _active_tenant_id(request: Request, user) -> int: return int( request.session.get("active_tenant_id") or request.session.get("selected_tenant_id") or request.session.get("tenant_id") or user.tenant_id ) def _active_branch_id(request: Request, user, db) -> int | None: value = request.session.get("active_branch_id") if value in (None, "", 0, "0"): if _has_perm(db, user, "clients.cross_branch"): return None return int(getattr(user, "branch_id", 0) or 0) or None return int(value) def _active_financial_year(request: Request) -> str | None: value = request.session.get("active_financial_year") or getattr(request.state, "year_code", None) value = (value or "").strip() return value or None def _assigned_user_filter(db, user) -> int | None: # Staff users see/update only their own assigned tasks. Partner visibility is handled separately. return _staff_own_task_user_id(db, user) def _can_manage_execution(db, user) -> bool: return _can_bulk_manage_tasks(db, user) def _safe_int_list(values: list[str] | None) -> list[int]: ids: list[int] = [] for value in values or []: try: ids.append(int(value)) except (TypeError, ValueError): continue return ids @router.get("") def execution_dashboard(request: Request, q: str = "", status: str = ""): db = CommonSessionLocal() try: user = get_current_user(request, db=db) if not user: return RedirectResponse(url="/login", status_code=303) try: require_permission(db, user, "service_tasks.view") except Exception: return _redirect_denied() tenant_id = _active_tenant_id(request, user) branch_id = _active_branch_id(request, user, db) assigned_to_user_id = _assigned_user_filter(db, user) partner_user_id = _partner_visibility_user_id(db, user) financial_year = _active_financial_year(request) stats = dashboard_stats( db, tenant_id=tenant_id, branch_id=branch_id, assigned_to_user_id=assigned_to_user_id, partner_user_id=partner_user_id, financial_year=financial_year, ) tasks = list_tasks_payload( db, tenant_id=tenant_id, branch_id=branch_id, assigned_to_user_id=assigned_to_user_id, partner_user_id=partner_user_id, status=status, q=q, financial_year=financial_year, ) assignees = list_assignees_for_execution(db, tenant_id=tenant_id, branch_id=branch_id) return _render( request, "modules/services/templates/services/work_tracker/dashboard.html", db, user, title="Work Tracker Dashboard", stats=stats, tasks=tasks, q=q, status=status, assignees=assignees, can_manage=_can_manage_execution(db, user), can_bulk_manage=_can_bulk_manage_tasks(db, user), can_assign_staff=_can_assign_staff(db, user), financial_year=financial_year, ) finally: db.close() @router.post("/tasks/bulk-update") def task_bulk_update_submit( request: Request, task_ids: list[str] = Form(default=[]), bulk_status: str = Form(""), bulk_assigned_to_user_id: str = Form("__no_change__"), bulk_internal_target_date: str = Form(""), update_internal_target_date: str | None = Form(None), csrf_token: str = Form(...), ): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user = get_current_user(request, db=db) if not user: return RedirectResponse(url="/login", status_code=303) if not _can_bulk_manage_tasks(db, user): return _redirect_denied() ids = _safe_int_list(task_ids) tenant_id = _active_tenant_id(request, user) branch_id = _active_branch_id(request, user, db) partner_user_id = _partner_visibility_user_id(db, user) financial_year = _active_financial_year(request) tasks = get_tasks_for_bulk_update( db, tenant_id=tenant_id, task_ids=ids, branch_id=branch_id, assigned_to_user_id=None, partner_user_id=partner_user_id, financial_year=financial_year, ) update_assignee = bulk_assigned_to_user_id != "__no_change__" assigned_to_user_id = None if update_assignee and not _can_assign_staff(db, user): return _redirect_denied() if update_assignee and bulk_assigned_to_user_id.strip(): try: requested_assignee_id = int(bulk_assigned_to_user_id) except ValueError: return _redirect_denied() assigned_to_user_id = _resolve_allowed_assignee_id( db, tenant_id=tenant_id, branch_id=branch_id, assigned_to_user_id=requested_assignee_id ) if assigned_to_user_id != requested_assignee_id: return _redirect_denied() target_date = parse_date_value(bulk_internal_target_date) if update_internal_target_date is not None else None apply_bulk_task_update( db, tasks, status=bulk_status.strip() or None, assigned_to_user_id=assigned_to_user_id, update_assignee=update_assignee, internal_target_date=target_date, update_internal_target_date=update_internal_target_date is not None, user_id=user.id, ) db.commit() return RedirectResponse(url="/services/work-tracker", status_code=303) finally: db.close() @router.get("/subscriptions") def subscription_execution_list(request: Request, q: str = ""): db = CommonSessionLocal() try: user = get_current_user(request, db=db) if not user: return RedirectResponse(url="/login", status_code=303) try: require_permission(db, user, "clients.view") except Exception: return _redirect_denied() tenant_id = _active_tenant_id(request, user) branch_id = _active_branch_id(request, user, db) financial_year = _active_financial_year(request) rows = list_subscription_execution_payload(db, tenant_id=tenant_id, branch_id=branch_id, financial_year=financial_year, q=q) return _render( request, "modules/services/templates/services/work_tracker/subscriptions.html", db, user, title="Generate Service Tasks", rows=rows, q=q, can_generate=_has_perm(db, user, "service_tasks.create"), financial_year=financial_year, ) finally: db.close() @router.post("/subscriptions/{subscription_id}/generate") def generate_subscription_tasks(request: Request, subscription_id: int, csrf_token: str = Form(...)): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user = get_current_user(request, db=db) if not user: return RedirectResponse(url="/login", status_code=303) try: require_permission(db, user, "service_tasks.create") except Exception: return _redirect_denied() tenant_id = _active_tenant_id(request, user) financial_year = _active_financial_year(request) subscription = get_subscription_for_execution(db, tenant_id=tenant_id, subscription_id=subscription_id) if subscription and financial_year and subscription.financial_year != financial_year: return RedirectResponse(url="/services/work-tracker/subscriptions", status_code=303) if not subscription or not subscription.is_active or subscription.status != "active": return RedirectResponse(url="/services/work-tracker/subscriptions", status_code=303) generate_tasks_for_subscription(db, subscription=subscription, user_id=user.id) db.commit() return RedirectResponse(url="/services/work-tracker", status_code=303) finally: db.close() @router.get("/tasks/{task_id}/edit") def task_edit_page(request: Request, task_id: int): db = CommonSessionLocal() try: user = get_current_user(request, db=db) if not user: return RedirectResponse(url="/login", status_code=303) try: require_permission(db, user, "service_tasks.view") except Exception: return _redirect_denied() tenant_id = _active_tenant_id(request, user) branch_id = _active_branch_id(request, user, db) task = get_task( db, tenant_id=tenant_id, task_id=task_id, branch_id=branch_id, assigned_to_user_id=_assigned_user_filter(db, user), partner_user_id=_partner_visibility_user_id(db, user), financial_year=_active_financial_year(request), ) if not task: return RedirectResponse(url="/services/work-tracker", status_code=303) assignees = list_assignees_for_execution(db, tenant_id=tenant_id, branch_id=branch_id) comments = list_task_comments(db, tenant_id=tenant_id, task_id=task.id) return _render( request, "modules/services/templates/services/work_tracker/task_form.html", db, user, title="Update Service Task", task=task, assignees=assignees, comments=comments, comment_type_options=_comment_type_options_for_user(db, user), can_add_comment=_can_add_task_comment(db, user, task), can_edit=_can_update_task_status(db, user, task), can_submit_review=_can_submit_task_for_review(db, user, task), can_manager_review=_can_manager_review_task(db, user, task), can_partner_review=_can_partner_review_task(db, user, task), can_review_partner_review=_can_review_partner_review_task(db, user, task), can_manage_fields=_can_bulk_manage_tasks(db, user), can_reassign=_can_assign_staff(db, user), ) finally: db.close() @router.post("/tasks/{task_id}/edit") def task_edit_submit( request: Request, task_id: int, status: str = Form("pending"), priority: str = Form("normal"), assigned_to_user_id: str = Form(""), internal_target_date: str = Form(""), remarks: str = Form(""), checklist_response: str = Form(""), checklist_text_response: str = Form(""), checklist_number_response: str = Form(""), checklist_date_response: str = Form(""), checklist_remarks: str = Form(""), is_active: str | None = Form(None), csrf_token: str = Form(...), ): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user = get_current_user(request, db=db) if not user: return RedirectResponse(url="/login", status_code=303) tenant_id = _active_tenant_id(request, user) branch_id = _active_branch_id(request, user, db) task = get_task( db, tenant_id=tenant_id, task_id=task_id, branch_id=branch_id, assigned_to_user_id=_assigned_user_filter(db, user), partner_user_id=_partner_visibility_user_id(db, user), financial_year=_active_financial_year(request), ) if not task: return RedirectResponse(url="/services/work-tracker", status_code=303) if not _can_update_task_status(db, user, task): return _redirect_denied() can_manage_fields = _can_bulk_manage_tasks(db, user) if can_manage_fields: if assigned_to_user_id.strip(): try: requested_assignee_id = int(assigned_to_user_id) except ValueError: return _redirect_denied() resolved_assignee = _resolve_allowed_assignee_id( db, tenant_id=tenant_id, branch_id=branch_id, assigned_to_user_id=requested_assignee_id ) if resolved_assignee != requested_assignee_id: return _redirect_denied() else: resolved_assignee = None resolved_priority = priority resolved_internal_target_date = parse_date_value(internal_target_date) resolved_is_active = is_active is not None else: # Staff can update only status and work note for their own task. resolved_assignee = task.assigned_to_user_id resolved_priority = task.priority resolved_internal_target_date = task.internal_target_date resolved_is_active = task.is_active try: apply_task_checklist_response( db, task, checklist_response=checklist_response, checklist_text_response=checklist_text_response, checklist_number_response=checklist_number_response, checklist_date_response=checklist_date_response, checklist_remarks=checklist_remarks, requested_status=status, ) except ValueError: db.rollback() return RedirectResponse(url=f"/services/work-tracker/tasks/{task_id}/edit?checklist_error=1", status_code=303) apply_task_update( task, status=status, priority=resolved_priority, assigned_to_user_id=resolved_assignee, internal_target_date=resolved_internal_target_date, remarks=remarks, is_active=resolved_is_active, user_id=user.id, ) recalculate_task_aqmm_status(db, task) db.commit() return RedirectResponse(url="/services/work-tracker", status_code=303) finally: db.close() @router.post("/tasks/{task_id}/submit-review") def task_submit_review( request: Request, task_id: int, note: str = Form(""), csrf_token: str = Form(...), ): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user = get_current_user(request, db=db) if not user: return RedirectResponse(url="/login", status_code=303) tenant_id = _active_tenant_id(request, user) branch_id = _active_branch_id(request, user, db) task = get_task( db, tenant_id=tenant_id, task_id=task_id, branch_id=branch_id, assigned_to_user_id=_assigned_user_filter(db, user), partner_user_id=_partner_visibility_user_id(db, user), financial_year=_active_financial_year(request), ) if not task: return RedirectResponse(url="/services/work-tracker", status_code=303) if not _can_submit_task_for_review(db, user, task): return _redirect_denied() try: submit_task_for_review(db, task=task, note=note, user_id=user.id) db.commit() return RedirectResponse(url=f"/services/work-tracker/tasks/{task_id}/edit?submitted=1", status_code=303) except ValueError: db.rollback() return RedirectResponse(url=f"/services/work-tracker/tasks/{task_id}/edit?review_error=1", status_code=303) finally: db.close() @router.post("/tasks/{task_id}/review/{review_level}") def task_review_submit( request: Request, task_id: int, review_level: str, decision: str = Form("reviewed"), note: str = Form(""), csrf_token: str = Form(...), ): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user = get_current_user(request, db=db) if not user: return RedirectResponse(url="/login", status_code=303) tenant_id = _active_tenant_id(request, user) branch_id = _active_branch_id(request, user, db) task = get_task( db, tenant_id=tenant_id, task_id=task_id, branch_id=branch_id, assigned_to_user_id=None, partner_user_id=_partner_visibility_user_id(db, user), financial_year=_active_financial_year(request), ) if not task: return RedirectResponse(url="/services/work-tracker", status_code=303) allowed = False if review_level == "manager": allowed = _can_manager_review_task(db, user, task) elif review_level == "partner": allowed = _can_partner_review_task(db, user, task) elif review_level == "review_partner": allowed = _can_review_partner_review_task(db, user, task) if not allowed: return _redirect_denied() try: apply_task_review(db, task=task, review_level=review_level, decision=decision, note=note, user_id=user.id) db.commit() return RedirectResponse(url=f"/services/work-tracker/tasks/{task_id}/edit?reviewed=1", status_code=303) except ValueError: db.rollback() return RedirectResponse(url=f"/services/work-tracker/tasks/{task_id}/edit?review_error=1", status_code=303) finally: db.close() @router.post("/tasks/{task_id}/comments") def task_comment_submit( request: Request, task_id: int, comment_type: str = Form("internal_note"), visibility: str = Form("internal"), message: str = Form(""), csrf_token: str = Form(...), ): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user = get_current_user(request, db=db) if not user: return RedirectResponse(url="/login", status_code=303) tenant_id = _active_tenant_id(request, user) branch_id = _active_branch_id(request, user, db) task = get_task( db, tenant_id=tenant_id, task_id=task_id, branch_id=branch_id, assigned_to_user_id=_assigned_user_filter(db, user), partner_user_id=_partner_visibility_user_id(db, user), financial_year=_active_financial_year(request), ) if not task: return RedirectResponse(url="/services/work-tracker", status_code=303) if not _can_add_task_comment(db, user, task): return _redirect_denied() if comment_type not in _allowed_comment_type_codes(db, user): return _redirect_denied() add_task_comment( db, task=task, comment_type=comment_type, visibility=visibility, message=message, user_id=user.id, ) db.commit() return RedirectResponse(url=f"/services/work-tracker/tasks/{task_id}/edit", status_code=303) finally: db.close()