A R R R Associates
677 lines
25 KiB
Python
677 lines
25 KiB
Python
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()
|