@@ -197,7 +211,7 @@
{{ task.default_role_name or '-' }} Eligible: {{ task.eligible_role_names or task.default_role_name or '-' }} |
- {% if task.task_category %} {{ task.task_category }}{% endif %}
+ {% if task.task_category %} {{ task.task_category }}{% endif %}{% if task.task_tool_code and task.task_tool_code != "NONE" %} Tool: {{ task.task_tool_code|replace("_"," ")|title }}{% endif %}
{% if task.is_mandatory %} Mandatory{% endif %}
{% if task.requires_review %} Review: {{ (task.normal_review_role or "manager").replace("_"," ").title() }}{% endif %}
{% if task.is_aqmm_task %} AQMM{% endif %}
diff --git a/app/modules/services/templates/services/work_tracker/task_form.html b/app/modules/services/templates/services/work_tracker/task_form.html
index b45d5ec..6c89b3e 100644
--- a/app/modules/services/templates/services/work_tracker/task_form.html
+++ b/app/modules/services/templates/services/work_tracker/task_form.html
@@ -86,6 +86,12 @@
{% if not can_reassign and task.assigned_to %} Assignee changes are restricted for this role. {% endif %}
{% if can_edit and not can_manage_fields %} You can update status and work note only for your own assigned task. {% endif %}
+ ERP Task Tool · {{ task_tool.name }} {{ task_tool.description }} {% if task_tool.launch_url %} Open Tool{% endif %}
+ Opening the tool does not automatically mark this task complete. Save the task response/evidence after completing the verification.
+
diff --git a/app/modules/services/ui.py b/app/modules/services/ui.py
index dc5e688..969e84e 100644
--- a/app/modules/services/ui.py
+++ b/app/modules/services/ui.py
@@ -26,6 +26,9 @@ from app.modules.services.bulk_imports import (
)
from app.modules.services.catalogue_export import build_service_catalogue_export
from app.modules.services.default_tasks_export import build_default_tasks_export
+from app.modules.services.firm_tasks_export import build_firm_tasks_export
+from app.modules.services.task_tools import list_task_tools, normalize_task_tool_code
+from app.modules.services.task_duplicates import find_duplicate_pairs, merge_default_tasks, merge_firm_tasks
from app.modules.services.due_dates import (
DUE_PERIOD_TYPES,
DUE_YEAR_BASIS_CHOICES,
@@ -88,6 +91,7 @@ def _base_ctx(request: Request, user, db, **ctx):
"due_period_types": DUE_PERIOD_TYPES,
"due_year_basis_choices": DUE_YEAR_BASIS_CHOICES,
"task_execution_roles": TASK_EXECUTION_ROLES,
+ "task_tools": list_task_tools(),
}
base.update(ctx)
return base
@@ -968,6 +972,7 @@ def _copy_system_defaults_if_firm_tasks_empty(
response_type=getattr(default, "response_type", "NONE") or "NONE",
evidence_required=getattr(default, "evidence_required", False),
remarks_required_if_no=getattr(default, "remarks_required_if_no", False),
+ task_tool_code=getattr(default, "task_tool_code", "NONE") or "NONE",
is_aqmm_task=getattr(default, "is_aqmm_task", False),
aqmm_mandatory=getattr(default, "aqmm_mandatory", False),
aqmm_evidence_required=getattr(default, "aqmm_evidence_required", False),
@@ -1262,7 +1267,7 @@ def task_templates_detail(request: Request, catalogue_id: int):
@router.post('/templates/{catalogue_id}/tasks/new')
-def task_template_create_submit(request: Request, catalogue_id: int, task_name: str = Form(...), description: str = Form(''), default_role_name: str = Form(''), eligible_roles: list[str] = Form([]), sequence_no: int | None = Form(None), is_mandatory: str | None = Form(None), requires_review: str | None = Form(None), normal_review_role: str = Form('manager'), task_category: str = Form(''), response_required: str | None = Form(None), response_type: str = Form('NONE'), evidence_required: str | None = Form(None), remarks_required_if_no: str | None = Form(None), is_aqmm_task: str | None = Form(None), aqmm_mandatory: str | None = Form(None), aqmm_evidence_required: str | None = Form(None), aqmm_manager_review_required: str | None = Form(None), aqmm_partner_review_required: str | None = Form(None), aqmm_review_partner_required: str | None = Form(None), aqmm_blocks_final_release: str | None = Form(None), aqmm_reference: str = Form(''), is_active: str | None = Form(None), csrf_token: str = Form(...)):
+def task_template_create_submit(request: Request, catalogue_id: int, task_name: str = Form(...), description: str = Form(''), default_role_name: str = Form(''), eligible_roles: list[str] = Form([]), sequence_no: int | None = Form(None), is_mandatory: str | None = Form(None), requires_review: str | None = Form(None), normal_review_role: str = Form('manager'), task_category: str = Form(''), response_required: str | None = Form(None), response_type: str = Form('NONE'), task_tool_code: str = Form('NONE'), evidence_required: str | None = Form(None), remarks_required_if_no: str | None = Form(None), is_aqmm_task: str | None = Form(None), aqmm_mandatory: str | None = Form(None), aqmm_evidence_required: str | None = Form(None), aqmm_manager_review_required: str | None = Form(None), aqmm_partner_review_required: str | None = Form(None), aqmm_review_partner_required: str | None = Form(None), aqmm_blocks_final_release: str | None = Form(None), aqmm_reference: str = Form(''), is_active: str | None = Form(None), csrf_token: str = Form(...)):
validate_csrf(request, csrf_token)
db = CommonSessionLocal()
try:
@@ -1293,6 +1298,7 @@ def task_template_create_submit(request: Request, catalogue_id: int, task_name:
task_category=(category.name if category else None),
response_required=response_required is not None,
response_type=(response_type.strip().upper() or "NONE"),
+ task_tool_code=normalize_task_tool_code(task_tool_code),
evidence_required=evidence_required is not None,
remarks_required_if_no=remarks_required_if_no is not None,
is_aqmm_task=is_aqmm_task is not None,
@@ -1374,7 +1380,7 @@ def default_templates_detail(request: Request, catalogue_id: int):
@router.post('/catalogue/{catalogue_id}/defaults/new')
-def default_template_create_submit(request: Request, catalogue_id: int, task_name: str = Form(...), description: str = Form(''), default_role_name: str = Form(''), sequence_no: int | None = Form(None), is_mandatory: str | None = Form(None), requires_review: str | None = Form(None), normal_review_role: str = Form('manager'), task_category: str = Form(''), response_required: str | None = Form(None), response_type: str = Form('NONE'), evidence_required: str | None = Form(None), remarks_required_if_no: str | None = Form(None), is_aqmm_task: str | None = Form(None), aqmm_mandatory: str | None = Form(None), aqmm_evidence_required: str | None = Form(None), aqmm_manager_review_required: str | None = Form(None), aqmm_partner_review_required: str | None = Form(None), aqmm_review_partner_required: str | None = Form(None), aqmm_blocks_final_release: str | None = Form(None), aqmm_reference: str = Form(''), is_active: str | None = Form(None), csrf_token: str = Form(...)):
+def default_template_create_submit(request: Request, catalogue_id: int, task_name: str = Form(...), description: str = Form(''), default_role_name: str = Form(''), sequence_no: int | None = Form(None), is_mandatory: str | None = Form(None), requires_review: str | None = Form(None), normal_review_role: str = Form('manager'), task_category: str = Form(''), response_required: str | None = Form(None), response_type: str = Form('NONE'), task_tool_code: str = Form('NONE'), evidence_required: str | None = Form(None), remarks_required_if_no: str | None = Form(None), is_aqmm_task: str | None = Form(None), aqmm_mandatory: str | None = Form(None), aqmm_evidence_required: str | None = Form(None), aqmm_manager_review_required: str | None = Form(None), aqmm_partner_review_required: str | None = Form(None), aqmm_review_partner_required: str | None = Form(None), aqmm_blocks_final_release: str | None = Form(None), aqmm_reference: str = Form(''), is_active: str | None = Form(None), csrf_token: str = Form(...)):
validate_csrf(request, csrf_token)
db = CommonSessionLocal()
try:
@@ -1387,7 +1393,7 @@ def default_template_create_submit(request: Request, catalogue_id: int, task_nam
catalogue = get_catalogue(db, catalogue_id)
if not catalogue:
return RedirectResponse(url='/services/defaults', status_code=303)
- row = ServiceDefaultTaskTemplate(service_catalogue_id=catalogue_id, task_name=task_name.strip(), description=description.strip() or None, default_role_name=default_role_name.strip() or None, sequence_no=sequence_no or next_default_task_sequence(db, catalogue_id=catalogue_id), is_mandatory=is_mandatory is not None, requires_review=requires_review is not None, normal_review_role=_normal_review_role_value(normal_review_role, requires_review=requires_review is not None), task_category_id=(category.id if (category := ensure_task_category(db, catalogue_id=catalogue_id, tenant_id=None, name=task_category, user_id=user.id)) else None), task_category=(category.name if category else None), response_required=response_required is not None, response_type=(response_type.strip().upper() or "NONE"), evidence_required=evidence_required is not None, remarks_required_if_no=remarks_required_if_no is not None, is_aqmm_task=is_aqmm_task is not None, aqmm_mandatory=aqmm_mandatory is not None, aqmm_evidence_required=aqmm_evidence_required is not None, aqmm_manager_review_required=aqmm_manager_review_required is not None, aqmm_partner_review_required=aqmm_partner_review_required is not None, aqmm_review_partner_required=aqmm_review_partner_required is not None, aqmm_blocks_final_release=aqmm_blocks_final_release is not None, aqmm_reference=aqmm_reference.strip() or None, is_active=is_active is not None)
+ row = ServiceDefaultTaskTemplate(service_catalogue_id=catalogue_id, task_name=task_name.strip(), description=description.strip() or None, default_role_name=default_role_name.strip() or None, sequence_no=sequence_no or next_default_task_sequence(db, catalogue_id=catalogue_id), is_mandatory=is_mandatory is not None, requires_review=requires_review is not None, normal_review_role=_normal_review_role_value(normal_review_role, requires_review=requires_review is not None), task_category_id=(category.id if (category := ensure_task_category(db, catalogue_id=catalogue_id, tenant_id=None, name=task_category, user_id=user.id)) else None), task_category=(category.name if category else None), response_required=response_required is not None, response_type=(response_type.strip().upper() or "NONE"), task_tool_code=normalize_task_tool_code(task_tool_code), evidence_required=evidence_required is not None, remarks_required_if_no=remarks_required_if_no is not None, is_aqmm_task=is_aqmm_task is not None, aqmm_mandatory=aqmm_mandatory is not None, aqmm_evidence_required=aqmm_evidence_required is not None, aqmm_manager_review_required=aqmm_manager_review_required is not None, aqmm_partner_review_required=aqmm_partner_review_required is not None, aqmm_review_partner_required=aqmm_review_partner_required is not None, aqmm_blocks_final_release=aqmm_blocks_final_release is not None, aqmm_reference=aqmm_reference.strip() or None, is_active=is_active is not None)
db.add(row); db.commit()
return RedirectResponse(url=f'/services/catalogue/{catalogue_id}/defaults', status_code=303)
finally:
@@ -1433,6 +1439,7 @@ def copy_defaults_to_firm(request: Request, catalogue_id: int, csrf_token: str =
response_type=getattr(d, "response_type", "NONE") or "NONE",
evidence_required=getattr(d, "evidence_required", False),
remarks_required_if_no=getattr(d, "remarks_required_if_no", False),
+ task_tool_code=getattr(d, "task_tool_code", "NONE") or "NONE",
is_aqmm_task=getattr(d, "is_aqmm_task", False),
aqmm_mandatory=getattr(d, "aqmm_mandatory", False),
aqmm_evidence_required=getattr(d, "aqmm_evidence_required", False),
@@ -1795,6 +1802,7 @@ def firm_task_template_edit_submit(
task_category: str = Form(''),
response_required: str | None = Form(None),
response_type: str = Form('NONE'),
+ task_tool_code: str = Form('NONE'),
evidence_required: str | None = Form(None),
remarks_required_if_no: str | None = Form(None),
is_aqmm_task: str | None = Form(None),
@@ -1844,6 +1852,7 @@ def firm_task_template_edit_submit(
task.task_category = category.name if category else None
task.response_required = response_required is not None
task.response_type = response_type.strip().upper() or "NONE"
+ task.task_tool_code = normalize_task_tool_code(task_tool_code)
task.evidence_required = evidence_required is not None
task.remarks_required_if_no = remarks_required_if_no is not None
task.is_aqmm_task = is_aqmm_task is not None
@@ -1942,6 +1951,7 @@ def default_task_template_edit_submit(
task_category: str = Form(''),
response_required: str | None = Form(None),
response_type: str = Form('NONE'),
+ task_tool_code: str = Form('NONE'),
evidence_required: str | None = Form(None),
remarks_required_if_no: str | None = Form(None),
is_aqmm_task: str | None = Form(None),
@@ -1991,6 +2001,7 @@ def default_task_template_edit_submit(
task.task_category = category.name if category else None
task.response_required = response_required is not None
task.response_type = response_type.strip().upper() or "NONE"
+ task.task_tool_code = normalize_task_tool_code(task_tool_code)
task.evidence_required = evidence_required is not None
task.remarks_required_if_no = remarks_required_if_no is not None
task.is_aqmm_task = is_aqmm_task is not None
@@ -2008,6 +2019,118 @@ def default_task_template_edit_submit(
finally:
db.close()
+# -----------------------------------------------------------------------------
+# Task administration: service export/import and safe duplicate consolidation
+# -----------------------------------------------------------------------------
+
+@router.get('/catalogue/{catalogue_id}/defaults/export')
+def default_tasks_service_export(request: Request, catalogue_id: int):
+ db = CommonSessionLocal()
+ try:
+ user = get_current_user(request, db=db)
+ if not user: return RedirectResponse(url='/login', status_code=303)
+ require_permission(db, user, 'service_tasks.view')
+ if not _is_system_admin(db, user): return _redirect_denied()
+ catalogue = get_catalogue(db, catalogue_id)
+ if not catalogue: return RedirectResponse(url='/services/defaults', status_code=303)
+ payload = build_default_tasks_export(db, catalogue_id=catalogue_id)
+ filename = f"{catalogue.service_code}_default_tasks_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}.xlsx"
+ return StreamingResponse(iter([payload]), media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', headers={'Content-Disposition': f'attachment; filename="{filename}"'})
+ finally: db.close()
+
+@router.post('/catalogue/{catalogue_id}/defaults/import')
+def default_tasks_service_import(request: Request, catalogue_id: int, file: UploadFile = File(...), 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)
+ require_permission(db,user,'service_tasks.edit')
+ if not _is_system_admin(db,user): return _redirect_denied()
+ catalogue=get_catalogue(db,catalogue_id)
+ if not catalogue: return RedirectResponse(url='/services/defaults',status_code=303)
+ result=import_system_default_tasks(db,current_user=user,file_bytes=file.file.read(),update_existing=True,expected_service_code=catalogue.service_code)
+ if result.get('errors'): return RedirectResponse(url=f'/services/catalogue/{catalogue_id}/defaults?import_error=1',status_code=303)
+ return RedirectResponse(url=f'/services/catalogue/{catalogue_id}/defaults?imported=1&created={result.get("created",0)}&updated={result.get("updated",0)}',status_code=303)
+ finally: db.close()
+
+@router.get('/templates/{catalogue_id}/export')
+def firm_tasks_service_export(request: Request, catalogue_id: int):
+ db=CommonSessionLocal()
+ try:
+ user=get_current_user(request,db=db)
+ if not user: return RedirectResponse(url='/login',status_code=303)
+ if not _can_manage_firm_tasks(db,user): return _redirect_denied()
+ tenant_id=_active_tenant_id(request,user); catalogue=get_catalogue(db,catalogue_id)
+ if not catalogue: return RedirectResponse(url='/services/templates',status_code=303)
+ payload=build_firm_tasks_export(db,tenant_id=tenant_id,catalogue_id=catalogue_id)
+ filename=f"{catalogue.service_code}_firm_tasks_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}.xlsx"
+ return StreamingResponse(iter([payload]),media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',headers={'Content-Disposition':f'attachment; filename="{filename}"'})
+ finally: db.close()
+
+@router.post('/templates/{catalogue_id}/import')
+def firm_tasks_service_import(request: Request, catalogue_id: int, file: UploadFile = File(...), 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_manage_firm_tasks(db,user): return _redirect_denied()
+ tenant_id=_active_tenant_id(request,user); catalogue=get_catalogue(db,catalogue_id)
+ if not catalogue: return RedirectResponse(url='/services/templates',status_code=303)
+ result=import_firm_task_templates(db,current_user=user,tenant_id=tenant_id,file_bytes=file.file.read(),update_existing=True,expected_service_code=catalogue.service_code)
+ if result.get('errors'): return RedirectResponse(url=f'/services/templates/{catalogue_id}?import_error=1',status_code=303)
+ return RedirectResponse(url=f'/services/templates/{catalogue_id}?imported=1&created={result.get("created",0)}&updated={result.get("updated",0)}',status_code=303)
+ finally: db.close()
+
+@router.get('/catalogue/{catalogue_id}/defaults/duplicates')
+def default_task_duplicates_page(request: Request, catalogue_id: int):
+ db=CommonSessionLocal()
+ try:
+ user=get_current_user(request,db=db)
+ if not user: return RedirectResponse(url='/login',status_code=303)
+ require_permission(db,user,'service_tasks.edit')
+ if not _is_system_admin(db,user): return _redirect_denied()
+ service=get_catalogue(db,catalogue_id); tasks=get_default_task_templates(db,catalogue_id=catalogue_id)
+ return _render(request,'modules/services/templates/services/task_duplicate_review.html',db,user,title=f'Duplicate Tasks - {service.service_name}',service=service,tasks=tasks,pairs=find_duplicate_pairs(tasks),scope='default')
+ finally: db.close()
+
+@router.post('/catalogue/{catalogue_id}/defaults/duplicates/merge')
+def default_task_duplicates_merge(request: Request,catalogue_id:int,master_task_id:int=Form(...),source_task_ids:list[int]=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)
+ require_permission(db,user,'service_tasks.edit')
+ if not _is_system_admin(db,user): return _redirect_denied()
+ count=merge_default_tasks(db,catalogue_id=catalogue_id,master_task_id=master_task_id,source_task_ids=source_task_ids); db.commit()
+ return RedirectResponse(url=f'/services/catalogue/{catalogue_id}/defaults/duplicates?merged={count}',status_code=303)
+ except ValueError:
+ db.rollback(); return RedirectResponse(url=f'/services/catalogue/{catalogue_id}/defaults/duplicates?merge_error=1',status_code=303)
+ finally: db.close()
+
+@router.get('/templates/{catalogue_id}/duplicates')
+def firm_task_duplicates_page(request: Request,catalogue_id:int):
+ db=CommonSessionLocal()
+ try:
+ user=get_current_user(request,db=db)
+ if not user: return RedirectResponse(url='/login',status_code=303)
+ if not _can_manage_firm_tasks(db,user): return _redirect_denied()
+ tenant_id=_active_tenant_id(request,user); service=get_catalogue(db,catalogue_id); tasks=get_firm_task_templates(db,tenant_id=tenant_id,catalogue_id=catalogue_id)
+ return _render(request,'modules/services/templates/services/task_duplicate_review.html',db,user,title=f'Duplicate Firm Tasks - {service.service_name}',service=service,tasks=tasks,pairs=find_duplicate_pairs(tasks),scope='firm')
+ finally: db.close()
+
+@router.post('/templates/{catalogue_id}/duplicates/merge')
+def firm_task_duplicates_merge(request: Request,catalogue_id:int,master_task_id:int=Form(...),source_task_ids:list[int]=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)
+ if not _can_manage_firm_tasks(db,user): return _redirect_denied()
+ tenant_id=_active_tenant_id(request,user); count=merge_firm_tasks(db,tenant_id=tenant_id,catalogue_id=catalogue_id,master_task_id=master_task_id,source_task_ids=source_task_ids,user_id=user.id); db.commit()
+ return RedirectResponse(url=f'/services/templates/{catalogue_id}/duplicates?merged={count}',status_code=303)
+ except ValueError:
+ db.rollback(); return RedirectResponse(url=f'/services/templates/{catalogue_id}/duplicates?merge_error=1',status_code=303)
+ finally: db.close()
+
# -----------------------------------------------------------------------------
# Service Task Category Master
# -----------------------------------------------------------------------------
diff --git a/app/modules/services/ui.py.bak_20260718_122052 b/app/modules/services/ui.py.bak_20260718_122052
new file mode 100644
index 0000000..29085f8
--- /dev/null
+++ b/app/modules/services/ui.py.bak_20260718_122052
@@ -0,0 +1,1622 @@
+from __future__ import annotations
+
+from fastapi import APIRouter, File, Form, Request, UploadFile
+from fastapi.responses import FileResponse, RedirectResponse, StreamingResponse
+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.audit.service import model_snapshot, pair_before_after, write_audit_log
+from app.modules.core.rbac.deps import get_user_permissions, get_user_roles
+from app.modules.core.rbac.permission_guard import require_permission
+from app.modules.core.tenancy.models import Branch
+from app.modules.services.models import FirmServiceSelection, FirmServiceTaskTemplate, ServiceCatalogue, ServiceCategory, ServiceDefaultTaskTemplate, ServiceDueDateRule, FirmTaskDocumentRequirement, FirmTaskDocumentTemplate
+from app.modules.services.bulk_imports import (
+ build_template as build_bulk_import_template,
+ import_client_service_assignments,
+ import_firm_task_templates,
+ import_service_master,
+ import_system_default_tasks,
+ import_due_date_extensions,
+)
+from app.modules.services.catalogue_export import build_service_catalogue_export
+from app.modules.services.due_dates import (
+ DUE_PERIOD_TYPES,
+ DUE_YEAR_BASIS_CHOICES,
+ create_due_date_extension,
+ get_due_rule,
+ list_due_extensions,
+ list_due_rules,
+ parse_optional_date,
+)
+from app.modules.services.catalogue_merge import build_merge_preview, merge_service_catalogues
+from app.modules.services.task_documents import (
+ create_task_document_requirement,
+ get_task_document_requirement,
+ list_task_document_requirements,
+ list_task_document_templates,
+ save_task_document_template,
+ template_absolute_path,
+ update_task_document_requirement,
+)
+from app.modules.services.services import (
+ RECURRENCE_CHOICES,
+ ENGAGEMENT_TYPE_CHOICES,
+ normalize_engagement_type,
+ get_catalogue,
+ get_category,
+ get_firm_selection,
+ get_firm_task_templates,
+ list_catalogue_payload,
+ list_categories,
+ list_disabled_catalogues,
+ list_firm_services_payload,
+ next_task_sequence,
+ next_default_task_sequence,
+ normalize_code,
+ get_default_task_templates,
+ get_firm_task_template,
+ get_default_task_template,
+)
+
+router = APIRouter(prefix="/services", tags=["services-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),
+ "recurrence_choices": RECURRENCE_CHOICES,
+ "engagement_type_choices": ENGAGEMENT_TYPE_CHOICES,
+ "due_period_types": DUE_PERIOD_TYPES,
+ "due_year_basis_choices": DUE_YEAR_BASIS_CHOICES,
+ }
+ 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]:
+ try:
+ return set(get_user_roles(db, user.id))
+ except Exception:
+ return set()
+
+
+def _has_role(db, user, *names: str) -> bool:
+ roles = _role_names(db, user)
+ return any(name in roles for name in names)
+
+
+def _can_view_services(db, user) -> bool:
+ return _has_perm(db, user, "services.view") or _has_role(db, user, "System Admin", "Firm Admin", "Partner", "Branch Manager")
+
+
+def _is_system_admin(db, user) -> bool:
+ return _has_perm(db, user, "services.cross_tenant")
+
+
+def _can_manage_firm_services(db, user) -> bool:
+ return _has_perm(db, user, "services.edit") or _has_role(db, user, "Firm Admin", "Partner")
+
+
+def _can_manage_firm_tasks(db, user) -> bool:
+ return _has_perm(db, user, "service_tasks.create") or _has_perm(db, user, "service_tasks.edit") or _has_role(db, user, "Firm Admin", "Partner")
+
+
+def _active_tenant_id(request: Request, user) -> int:
+ value = (
+ request.session.get("active_tenant_id")
+ or request.session.get("selected_tenant_id")
+ or request.session.get("tenant_id")
+ or user.tenant_id
+ )
+ return int(value)
+
+
+@router.get("")
+def services_home(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, "services.view")
+ except Exception:
+ return _redirect_denied()
+
+ tenant_id = _active_tenant_id(request, user)
+
+ enabled_rows = list_firm_services_payload(db, tenant_id=tenant_id, q=q)
+ disabled_rows = list_disabled_catalogues(db, tenant_id=tenant_id, q=q)
+ branches = db.execute(
+ select(Branch).where(Branch.tenant_id == tenant_id, Branch.is_active.is_(True)).order_by(Branch.name.asc())
+ ).scalars().all()
+ return _render(
+ request,
+ "modules/services/templates/services/list.html",
+ db,
+ user,
+ title="Services",
+ q=q,
+ enabled_rows=enabled_rows,
+ disabled_rows=disabled_rows,
+ branches=branches,
+ can_manage_catalogue=_is_system_admin(db, user),
+ can_manage_firm_services=_can_manage_firm_services(db, user),
+ can_manage_firm_tasks=_can_manage_firm_tasks(db, user),
+ )
+ finally:
+ db.close()
+
+
+@router.get('/categories')
+def category_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)
+ if not _can_view_services(db, user):
+ return _redirect_denied()
+ rows = list_categories(db, q=q)
+ return _render(request, 'modules/services/templates/services/category_list.html', db, user, title='Service Categories', q=q, rows=rows, can_create=(_has_perm(db, user, 'services.create') and _is_system_admin(db, user)))
+ except Exception:
+ return _redirect_denied()
+ finally:
+ db.close()
+
+
+@router.get('/categories/new')
+def category_create_page(request: Request):
+ db = CommonSessionLocal()
+ try:
+ user = get_current_user(request, db=db)
+ if not user:
+ return RedirectResponse(url='/login', status_code=303)
+ require_permission(db, user, 'services.create')
+ if not _is_system_admin(db, user):
+ return _redirect_denied()
+ return _render(request, 'modules/services/templates/services/category_form.html', db, user, title='Create Service Category', mode='create', category=None)
+ except Exception:
+ return _redirect_denied()
+ finally:
+ db.close()
+
+
+@router.post('/categories/new')
+def category_create_submit(
+ request: Request,
+ code: str = Form(...),
+ name: str = Form(...),
+ sort_order: int = Form(100),
+ 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)
+
+ require_permission(db, user, 'services.create')
+
+ if not _is_system_admin(db, user):
+ return _redirect_denied()
+
+ row = ServiceCategory(
+ code=normalize_code(code),
+ name=name.strip(),
+ sort_order=sort_order,
+ is_active=is_active is not None,
+ )
+
+ db.add(row)
+ db.commit()
+ db.refresh(row)
+
+ return RedirectResponse(url='/services/categories', status_code=303)
+ finally:
+ db.close()
+
+@router.get('/categories/{category_id}/edit')
+def category_edit_page(request: Request, category_id: int):
+ db = CommonSessionLocal()
+ try:
+ user = get_current_user(request, db=db)
+ if not user:
+ return RedirectResponse(url='/login', status_code=303)
+ require_permission(db, user, 'services.edit')
+ if not _is_system_admin(db, user):
+ return _redirect_denied()
+ row = get_category(db, category_id)
+ if not row:
+ return RedirectResponse(url='/services/categories', status_code=303)
+ return _render(request, 'modules/services/templates/services/category_form.html', db, user, title='Edit Service Category', mode='edit', category=row)
+ except Exception:
+ return _redirect_denied()
+ finally:
+ db.close()
+
+
+@router.post('/categories/{category_id}/edit')
+def category_edit_submit(request: Request, category_id: int, name: str = Form(...), sort_order: int = Form(100), 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)
+ require_permission(db, user, 'services.edit')
+ if not _is_system_admin(db, user):
+ return _redirect_denied()
+ row = get_category(db, category_id)
+ if not row:
+ return RedirectResponse(url='/services/categories', status_code=303)
+ row.name = name.strip()
+ row.sort_order = sort_order
+ row.is_active = is_active is not None
+ db.commit()
+ return RedirectResponse(url='/services/categories', status_code=303)
+ finally:
+ db.close()
+
+
+@router.get('/catalogue')
+def catalogue_list(request: Request, q: str = '', category_id: int | None = None, recurrence_type: str = '', engagement_type: str = '', page: int = 1, per_page: int = 20):
+ db = CommonSessionLocal()
+ try:
+ user = get_current_user(request, db=db)
+ if not user:
+ return RedirectResponse(url='/login', status_code=303)
+ if not _can_view_services(db, user):
+ return _redirect_denied()
+ payload = list_catalogue_payload(db, q=q, category_id=category_id, recurrence_type=recurrence_type, engagement_type=engagement_type, page=page, per_page=per_page)
+ tenant_id = _active_tenant_id(request, user)
+ can_create_catalogue = _is_system_admin(db, user) and _has_perm(db, user, 'services.create')
+ can_manage_firm_services = _can_manage_firm_services(db, user)
+
+ catalogue_ids = [row.id for row in payload.get('rows', [])]
+ firm_selection_by_catalogue = {}
+ if catalogue_ids:
+ selections = db.execute(
+ select(FirmServiceSelection).where(
+ FirmServiceSelection.tenant_id == tenant_id,
+ FirmServiceSelection.service_catalogue_id.in_(catalogue_ids),
+ )
+ ).scalars().all()
+ firm_selection_by_catalogue = {selection.service_catalogue_id: selection for selection in selections}
+
+ branches = db.execute(
+ select(Branch)
+ .where(Branch.tenant_id == tenant_id, Branch.is_active.is_(True))
+ .order_by(Branch.name.asc())
+ ).scalars().all()
+
+ return _render(
+ request,
+ 'modules/services/templates/services/catalogue_list.html',
+ db,
+ user,
+ title='Service Catalogue',
+ can_create=can_create_catalogue,
+ can_manage_firm_services=can_manage_firm_services,
+ categories=list_categories(db),
+ branches=branches,
+ tenant_id=tenant_id,
+ firm_selection_by_catalogue=firm_selection_by_catalogue,
+ **payload,
+ )
+ finally:
+ db.close()
+
+
+@router.get('/catalogue/export')
+def catalogue_export(request: Request):
+ db = CommonSessionLocal()
+ try:
+ user = get_current_user(request, db=db)
+ if not user:
+ return RedirectResponse(url='/login', status_code=303)
+ if not _is_system_admin(db, user):
+ return _redirect_denied()
+ require_permission(db, user, 'services.view')
+
+ workbook_bytes = build_service_catalogue_export(db)
+ filename = f"service_catalogue_export_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}.xlsx"
+ return StreamingResponse(
+ iter([workbook_bytes]),
+ media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
+ headers={'Content-Disposition': f'attachment; filename="{filename}"'},
+ )
+ finally:
+ db.close()
+
+
+@router.get('/catalogue/merge')
+def catalogue_merge_page(request: Request, source_id: int | None = None, target_id: int | None = None):
+ db = CommonSessionLocal()
+ try:
+ user = get_current_user(request, db=db)
+ if not user:
+ return RedirectResponse(url='/login', status_code=303)
+ if not _is_system_admin(db, user):
+ return _redirect_denied()
+ require_permission(db, user, 'services.edit')
+
+ services = db.execute(
+ select(ServiceCatalogue)
+ .where(ServiceCatalogue.is_active.is_(True))
+ .order_by(ServiceCatalogue.service_code.asc())
+ ).scalars().all()
+ preview = None
+ error = None
+ if source_id and target_id:
+ try:
+ preview = build_merge_preview(db, source_id=source_id, target_id=target_id)
+ except ValueError as exc:
+ error = str(exc)
+
+ return _render(
+ request,
+ 'modules/services/templates/services/catalogue_merge.html',
+ db,
+ user,
+ title='Merge Service Catalogue',
+ services=services,
+ source_id=source_id,
+ target_id=target_id,
+ preview=preview,
+ error=error,
+ )
+ except Exception:
+ return _redirect_denied()
+ finally:
+ db.close()
+
+
+@router.post('/catalogue/merge')
+def catalogue_merge_submit(
+ request: Request,
+ source_id: int = Form(...),
+ target_id: int = Form(...),
+ confirmation: 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)
+ if not _is_system_admin(db, user):
+ return _redirect_denied()
+ require_permission(db, user, 'services.edit')
+
+ preview = build_merge_preview(db, source_id=source_id, target_id=target_id)
+ expected = f"MERGE {preview.source.service_code} INTO {preview.target.service_code}"
+ if confirmation.strip().upper() != expected.upper():
+ return RedirectResponse(
+ url=f'/services/catalogue/merge?source_id={source_id}&target_id={target_id}&error=confirmation',
+ status_code=303,
+ )
+
+ result = merge_service_catalogues(
+ db,
+ source_id=source_id,
+ target_id=target_id,
+ actor_user_id=user.id,
+ )
+ db.commit()
+
+ write_audit_log(
+ db,
+ action='services.catalogue.merged',
+ entity_type='service_catalogue',
+ entity_id=target_id,
+ entity_name=result['target_code'],
+ actor=user,
+ request=request,
+ details=result,
+ )
+ return RedirectResponse(url=f'/services/catalogue/{target_id}?merged=1', status_code=303)
+ except ValueError:
+ db.rollback()
+ return RedirectResponse(url='/services/catalogue/merge?error=invalid', status_code=303)
+ except Exception:
+ db.rollback()
+ raise
+ finally:
+ db.close()
+
+
+@router.get('/catalogue/new')
+def catalogue_create_page(request: Request):
+ db = CommonSessionLocal()
+ try:
+ user = get_current_user(request, db=db)
+ if not user:
+ return RedirectResponse(url='/login', status_code=303)
+ if not _is_system_admin(db, user):
+ return _redirect_denied()
+ require_permission(db, user, 'services.create')
+ return _render(request, 'modules/services/templates/services/catalogue_form.html', db, user, title='Create Service Catalogue', mode='create', catalogue=None, categories=list_categories(db))
+ except Exception:
+ return _redirect_denied()
+ finally:
+ db.close()
+
+
+@router.post('/catalogue/new')
+def catalogue_create_submit(request: Request, service_code: str = Form(...), service_name: str = Form(...), category_id: str = Form(''), recurrence_type: str = Form(''), engagement_type: str = Form('non_assurance'), sort_order: int = Form(100), description: str = Form(''), applicable_individual: str | None = Form(None), applicable_proprietorship: str | None = Form(None), applicable_partnership: str | None = Form(None), applicable_llp: str | None = Form(None), applicable_company: str | None = Form(None), applicable_trust: str | None = Form(None), applicable_society: str | None = Form(None), is_active: str | None = Form(None), is_client_requestable: str | None = Form(None), is_consultant_requestable: 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 _is_system_admin(db, user):
+ return _redirect_denied()
+ require_permission(db, user, 'services.create')
+ selected_category = get_category(db, int(category_id)) if str(category_id).strip() else None
+ row = ServiceCatalogue(
+ service_code=normalize_code(service_code),
+ service_name=service_name.strip(),
+ category_id=selected_category.id if selected_category else None,
+ category=selected_category.name if selected_category else None,
+ recurrence_type=recurrence_type.strip() or None,
+ engagement_type=normalize_engagement_type(engagement_type),
+ sort_order=sort_order,
+ description=description.strip() or None,
+ applicable_individual=applicable_individual is not None,
+ applicable_proprietorship=applicable_proprietorship is not None,
+ applicable_partnership=applicable_partnership is not None,
+ applicable_llp=applicable_llp is not None,
+ applicable_company=applicable_company is not None,
+ applicable_trust=applicable_trust is not None,
+ applicable_society=applicable_society is not None,
+ is_active=is_active is not None,
+ is_client_requestable=is_client_requestable is not None,
+ is_consultant_requestable=is_consultant_requestable is not None,
+ created_by_user_id=user.id,
+ updated_by_user_id=user.id,
+ )
+ db.add(row); db.commit(); db.refresh(row)
+ return RedirectResponse(url=f'/services/catalogue/{row.id}', status_code=303)
+ finally:
+ db.close()
+
+
+@router.get('/catalogue/{catalogue_id}')
+def catalogue_detail(request: Request, catalogue_id: int):
+ db = CommonSessionLocal()
+ try:
+ user = get_current_user(request, db=db)
+ if not user:
+ return RedirectResponse(url='/login', status_code=303)
+ if not _can_view_services(db, user):
+ return _redirect_denied()
+ row = get_catalogue(db, catalogue_id)
+ if not row or not row.is_active:
+ return RedirectResponse(url='/services/catalogue', status_code=303)
+ tenant_id = _active_tenant_id(request, user)
+ can_edit = _is_system_admin(db, user) and _has_perm(db, user, 'services.edit')
+ branches = db.execute(
+ select(Branch)
+ .where(Branch.tenant_id == tenant_id, Branch.is_active.is_(True))
+ .order_by(Branch.name.asc())
+ ).scalars().all()
+ return _render(
+ request,
+ 'modules/services/templates/services/catalogue_detail.html',
+ db,
+ user,
+ title=f'Service Catalogue - {row.service_name}',
+ catalogue=row,
+ can_edit=can_edit,
+ can_manage_firm_services=_can_manage_firm_services(db, user),
+ branches=branches,
+ current_selection=get_firm_selection(db, tenant_id=tenant_id, catalogue_id=row.id),
+ current_templates=get_firm_task_templates(db, tenant_id=tenant_id, catalogue_id=row.id),
+ default_templates=get_default_task_templates(db, catalogue_id=row.id),
+ due_rules=list_due_rules(db, catalogue_id=row.id),
+ due_extensions=list_due_extensions(db, catalogue_id=row.id, tenant_id=tenant_id),
+ tenant_id=tenant_id,
+ )
+ finally:
+ db.close()
+
+
+@router.get('/catalogue/{catalogue_id}/due-rules/new')
+def due_rule_create_page(request: Request, catalogue_id: int):
+ db = CommonSessionLocal()
+ try:
+ user = get_current_user(request, db=db)
+ if not user:
+ return RedirectResponse(url='/login', status_code=303)
+ if not _is_system_admin(db, user):
+ return _redirect_denied()
+ require_permission(db, user, 'services.edit')
+ catalogue = get_catalogue(db, catalogue_id)
+ if not catalogue:
+ return RedirectResponse(url='/services/catalogue', status_code=303)
+ return _render(
+ request,
+ 'modules/services/templates/services/due_dates/rule_form.html',
+ db,
+ user,
+ title='Add Due Date Rule',
+ mode='create',
+ catalogue=catalogue,
+ rule=None,
+ )
+ finally:
+ db.close()
+
+
+@router.post('/catalogue/{catalogue_id}/due-rules/new')
+def due_rule_create_submit(
+ request: Request,
+ catalogue_id: int,
+ rule_name: str = Form(...),
+ period_type: str = Form('yearly'),
+ due_year_basis: str = Form('assessment_year_start'),
+ due_day: str = Form(''),
+ due_month: str = Form(''),
+ due_month_offset: int = Form(0),
+ days_offset_after_event: str = Form(''),
+ renewal_days_before_expiry: str = Form(''),
+ sort_order: int = Form(100),
+ is_active: str | None = Form(None),
+ remarks: 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)
+ if not _is_system_admin(db, user):
+ return _redirect_denied()
+ require_permission(db, user, 'services.edit')
+ catalogue = get_catalogue(db, catalogue_id)
+ if not catalogue:
+ return RedirectResponse(url='/services/catalogue', status_code=303)
+ row = ServiceDueDateRule(
+ service_catalogue_id=catalogue_id,
+ rule_name=rule_name.strip(),
+ period_type=(period_type or 'yearly').strip(),
+ due_year_basis=(due_year_basis or 'assessment_year_start').strip(),
+ due_day=int(due_day) if str(due_day).strip() else None,
+ due_month=int(due_month) if str(due_month).strip() else None,
+ due_month_offset=int(due_month_offset or 0),
+ days_offset_after_event=int(days_offset_after_event) if str(days_offset_after_event).strip() else None,
+ renewal_days_before_expiry=int(renewal_days_before_expiry) if str(renewal_days_before_expiry).strip() else None,
+ sort_order=sort_order,
+ is_active=is_active is not None,
+ remarks=remarks.strip() or None,
+ created_by_user_id=user.id,
+ updated_by_user_id=user.id,
+ )
+ db.add(row)
+ db.commit()
+ return RedirectResponse(url=f'/services/catalogue/{catalogue_id}', status_code=303)
+ finally:
+ db.close()
+
+
+@router.get('/catalogue/{catalogue_id}/due-rules/{rule_id}/edit')
+def due_rule_edit_page(request: Request, catalogue_id: int, rule_id: int):
+ db = CommonSessionLocal()
+ try:
+ user = get_current_user(request, db=db)
+ if not user:
+ return RedirectResponse(url='/login', status_code=303)
+ if not _is_system_admin(db, user):
+ return _redirect_denied()
+ require_permission(db, user, 'services.edit')
+ catalogue = get_catalogue(db, catalogue_id)
+ rule = get_due_rule(db, rule_id=rule_id, catalogue_id=catalogue_id)
+ if not catalogue or not rule:
+ return RedirectResponse(url=f'/services/catalogue/{catalogue_id}', status_code=303)
+ return _render(
+ request,
+ 'modules/services/templates/services/due_dates/rule_form.html',
+ db,
+ user,
+ title='Edit Due Date Rule',
+ mode='edit',
+ catalogue=catalogue,
+ rule=rule,
+ )
+ finally:
+ db.close()
+
+
+@router.post('/catalogue/{catalogue_id}/due-rules/{rule_id}/edit')
+def due_rule_edit_submit(
+ request: Request,
+ catalogue_id: int,
+ rule_id: int,
+ rule_name: str = Form(...),
+ period_type: str = Form('yearly'),
+ due_year_basis: str = Form('assessment_year_start'),
+ due_day: str = Form(''),
+ due_month: str = Form(''),
+ due_month_offset: int = Form(0),
+ days_offset_after_event: str = Form(''),
+ renewal_days_before_expiry: str = Form(''),
+ sort_order: int = Form(100),
+ is_active: str | None = Form(None),
+ remarks: 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)
+ if not _is_system_admin(db, user):
+ return _redirect_denied()
+ require_permission(db, user, 'services.edit')
+ rule = get_due_rule(db, rule_id=rule_id, catalogue_id=catalogue_id)
+ if not rule:
+ return RedirectResponse(url=f'/services/catalogue/{catalogue_id}', status_code=303)
+ rule.rule_name = rule_name.strip()
+ rule.period_type = (period_type or 'yearly').strip()
+ rule.due_year_basis = (due_year_basis or 'assessment_year_start').strip()
+ rule.due_day = int(due_day) if str(due_day).strip() else None
+ rule.due_month = int(due_month) if str(due_month).strip() else None
+ rule.due_month_offset = int(due_month_offset or 0)
+ rule.days_offset_after_event = int(days_offset_after_event) if str(days_offset_after_event).strip() else None
+ rule.renewal_days_before_expiry = int(renewal_days_before_expiry) if str(renewal_days_before_expiry).strip() else None
+ rule.sort_order = sort_order
+ rule.is_active = is_active is not None
+ rule.remarks = remarks.strip() or None
+ rule.updated_by_user_id = user.id
+ db.commit()
+ return RedirectResponse(url=f'/services/catalogue/{catalogue_id}', status_code=303)
+ finally:
+ db.close()
+
+
+@router.get('/catalogue/{catalogue_id}/due-extensions/new')
+def due_extension_create_page(request: Request, catalogue_id: int):
+ db = CommonSessionLocal()
+ try:
+ user = get_current_user(request, db=db)
+ if not user:
+ return RedirectResponse(url='/login', status_code=303)
+ if not _is_system_admin(db, user):
+ return _redirect_denied()
+ require_permission(db, user, 'services.edit')
+ catalogue = get_catalogue(db, catalogue_id)
+ if not catalogue:
+ return RedirectResponse(url='/services/catalogue', status_code=303)
+ return _render(
+ request,
+ 'modules/services/templates/services/due_dates/extension_form.html',
+ db,
+ user,
+ title='Add Due Date Extension',
+ catalogue=catalogue,
+ rules=list_due_rules(db, catalogue_id=catalogue_id, include_inactive=False),
+ )
+ finally:
+ db.close()
+
+
+@router.post('/catalogue/{catalogue_id}/due-extensions/new')
+def due_extension_create_submit(
+ request: Request,
+ catalogue_id: int,
+ due_date_rule_id: str = Form(''),
+ financial_year: str = Form(...),
+ assessment_year: str = Form(''),
+ period_label: str = Form(''),
+ extended_due_date: str = Form(...),
+ notification_reference: str = Form(''),
+ notification_date: str = Form(''),
+ remarks: 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)
+ if not _is_system_admin(db, user):
+ return _redirect_denied()
+ require_permission(db, user, 'services.edit')
+ catalogue = get_catalogue(db, catalogue_id)
+ if not catalogue:
+ return RedirectResponse(url='/services/catalogue', status_code=303)
+ tenant_id = _active_tenant_id(request, user)
+ create_due_date_extension(
+ db,
+ tenant_id=tenant_id,
+ catalogue_id=catalogue_id,
+ due_date_rule_id=int(due_date_rule_id) if str(due_date_rule_id).strip() else None,
+ financial_year=financial_year.strip(),
+ assessment_year=assessment_year.strip() or None,
+ period_label=period_label.strip() or None,
+ extended_due_date=parse_optional_date(extended_due_date),
+ notification_reference=notification_reference,
+ notification_date=parse_optional_date(notification_date),
+ remarks=remarks,
+ user_id=user.id,
+ )
+ db.commit()
+ return RedirectResponse(url=f'/services/catalogue/{catalogue_id}', status_code=303)
+ finally:
+ db.close()
+
+
+@router.get('/catalogue/{catalogue_id}/edit')
+def catalogue_edit_page(request: Request, catalogue_id: int):
+ db = CommonSessionLocal()
+ try:
+ user = get_current_user(request, db=db)
+ if not user:
+ return RedirectResponse(url='/login', status_code=303)
+ if not _is_system_admin(db, user):
+ return _redirect_denied()
+ require_permission(db, user, 'services.edit')
+ row = get_catalogue(db, catalogue_id)
+ if not row or not row.is_active:
+ return RedirectResponse(url='/services/catalogue', status_code=303)
+ return _render(request, 'modules/services/templates/services/catalogue_form.html', db, user, title='Edit Service Catalogue', mode='edit', catalogue=row, categories=list_categories(db))
+ except Exception:
+ return _redirect_denied()
+ finally:
+ db.close()
+
+
+@router.post('/catalogue/{catalogue_id}/edit')
+def catalogue_edit_submit(request: Request, catalogue_id: int, service_name: str = Form(...), category_id: str = Form(''), recurrence_type: str = Form(''), engagement_type: str = Form('non_assurance'), sort_order: int = Form(100), description: str = Form(''), applicable_individual: str | None = Form(None), applicable_proprietorship: str | None = Form(None), applicable_partnership: str | None = Form(None), applicable_llp: str | None = Form(None), applicable_company: str | None = Form(None), applicable_trust: str | None = Form(None), applicable_society: str | None = Form(None), is_active: str | None = Form(None), is_client_requestable: str | None = Form(None), is_consultant_requestable: 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 _is_system_admin(db, user):
+ return _redirect_denied()
+ require_permission(db, user, 'services.edit')
+ row = get_catalogue(db, catalogue_id)
+ if not row or not row.is_active:
+ return RedirectResponse(url='/services/catalogue', status_code=303)
+ selected_category = get_category(db, int(category_id)) if str(category_id).strip() else None
+ row.service_name = service_name.strip()
+ row.category_id = selected_category.id if selected_category else None
+ row.category = selected_category.name if selected_category else None
+ row.recurrence_type = recurrence_type.strip() or None
+ row.engagement_type = normalize_engagement_type(engagement_type)
+ row.sort_order = sort_order
+ row.applicable_individual = applicable_individual is not None
+ row.applicable_proprietorship = applicable_proprietorship is not None
+ row.applicable_partnership = applicable_partnership is not None
+ row.applicable_llp = applicable_llp is not None
+ row.applicable_company = applicable_company is not None
+ row.applicable_trust = applicable_trust is not None
+ row.applicable_society = applicable_society is not None
+ row.is_active = is_active is not None
+ row.is_client_requestable = is_client_requestable is not None
+ row.is_consultant_requestable = is_consultant_requestable is not None
+ row.updated_by_user_id = user.id
+ db.commit()
+ return RedirectResponse(url=f'/services/catalogue/{row.id}', status_code=303)
+ finally:
+ db.close()
+
+
+@router.post('/catalogue/{catalogue_id}/toggle')
+def toggle_firm_service(request: Request, catalogue_id: int, default_branch_id: 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)
+ if not _can_manage_firm_services(db, user):
+ return _redirect_denied()
+ tenant_id = _active_tenant_id(request, user)
+ row = get_catalogue(db, catalogue_id)
+ if not row or not row.is_active:
+ return RedirectResponse(url='/services/catalogue', status_code=303)
+ selection = get_firm_selection(db, tenant_id=tenant_id, catalogue_id=catalogue_id)
+ if not selection:
+ selection = FirmServiceSelection(
+ tenant_id=tenant_id,
+ service_catalogue_id=catalogue_id,
+ is_enabled=True,
+ default_branch_id=int(default_branch_id) if str(default_branch_id).strip() else None,
+ activated_by_user_id=user.id,
+ updated_by_user_id=user.id,
+ )
+ db.add(selection)
+ else:
+ selection.is_enabled = not bool(selection.is_enabled)
+ if str(default_branch_id).strip():
+ selection.default_branch_id = int(default_branch_id)
+ selection.updated_by_user_id = user.id
+ db.commit()
+ return RedirectResponse(url='/services', status_code=303)
+ finally:
+ db.close()
+
+
+@router.get('/templates')
+def task_templates_list(request: Request):
+ db = CommonSessionLocal()
+ try:
+ user = get_current_user(request, db=db)
+ if not user:
+ return RedirectResponse(url='/login', status_code=303)
+ if not (_has_perm(db, user, 'service_tasks.view') or _can_manage_firm_tasks(db, user)):
+ return _redirect_denied()
+ tenant_id = _active_tenant_id(request, user)
+ enabled_rows = list_firm_services_payload(db, tenant_id=tenant_id)
+ for row in enabled_rows:
+ row['task_count'] = len(get_firm_task_templates(db, tenant_id=tenant_id, catalogue_id=row['catalogue'].id))
+ return _render(request, 'modules/services/templates/services/task_templates_list.html', db, user, title='Firm Service Task Templates', rows=enabled_rows, can_manage_tasks=_can_manage_firm_tasks(db, user))
+ finally:
+ db.close()
+
+
+@router.get('/templates/{catalogue_id}')
+def task_templates_detail(request: Request, catalogue_id: int):
+ db = CommonSessionLocal()
+ try:
+ user = get_current_user(request, db=db)
+ if not user:
+ return RedirectResponse(url='/login', status_code=303)
+ if not (_has_perm(db, user, 'service_tasks.view') or _can_manage_firm_tasks(db, user)):
+ return _redirect_denied()
+ catalogue = get_catalogue(db, catalogue_id)
+ if not catalogue:
+ return RedirectResponse(url='/services/templates', status_code=303)
+ tenant_id = _active_tenant_id(request, user)
+ selection = get_firm_selection(db, tenant_id=tenant_id, catalogue_id=catalogue_id)
+ if not selection or not selection.is_enabled:
+ return RedirectResponse(url='/services', status_code=303)
+ tasks = get_firm_task_templates(db, tenant_id=tenant_id, catalogue_id=catalogue_id)
+ task_requirement_map = {
+ task.id: list_task_document_requirements(db, tenant_id=tenant_id, firm_task_template_id=task.id)
+ for task in tasks
+ }
+ task_template_file_map = {
+ task.id: list_task_document_templates(db, tenant_id=tenant_id, firm_task_template_id=task.id)
+ for task in tasks
+ }
+ return _render(
+ request,
+ 'modules/services/templates/services/task_template_detail.html',
+ db,
+ user,
+ title=f'Firm Task Templates - {catalogue.service_name}',
+ service=catalogue,
+ selection=selection,
+ task_templates=tasks,
+ task_requirement_map=task_requirement_map,
+ task_template_file_map=task_template_file_map,
+ can_manage_tasks=_can_manage_firm_tasks(db, user),
+ default_tasks=get_default_task_templates(db, catalogue_id=catalogue_id),
+ is_system_admin=_is_system_admin(db, user),
+ )
+ finally:
+ db.close()
+
+
+@router.post('/templates/{catalogue_id}/tasks/new')
+def task_template_create_submit(request: Request, catalogue_id: int, task_name: str = Form(...), description: str = Form(''), default_role_name: str = Form(''), sequence_no: int | None = Form(None), is_mandatory: str | None = Form(None), requires_review: str | None = Form(None), is_aqmm_task: str | None = Form(None), aqmm_mandatory: str | None = Form(None), aqmm_evidence_required: str | None = Form(None), aqmm_manager_review_required: str | None = Form(None), aqmm_partner_review_required: str | None = Form(None), aqmm_review_partner_required: str | None = Form(None), aqmm_blocks_final_release: str | None = Form(None), aqmm_reference: 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)
+ if not _can_manage_firm_tasks(db, user):
+ return _redirect_denied()
+ catalogue = get_catalogue(db, catalogue_id)
+ if not catalogue:
+ return RedirectResponse(url='/services/templates', status_code=303)
+ tenant_id = _active_tenant_id(request, user)
+ selection = get_firm_selection(db, tenant_id=tenant_id, catalogue_id=catalogue_id)
+ if not selection or not selection.is_enabled:
+ return RedirectResponse(url='/services', status_code=303)
+ row = FirmServiceTaskTemplate(
+ tenant_id=tenant_id,
+ service_catalogue_id=catalogue_id,
+ task_name=task_name.strip(),
+ description=description.strip() or None,
+ default_role_name=default_role_name.strip() or None,
+ sequence_no=sequence_no or next_task_sequence(db, tenant_id=tenant_id, catalogue_id=catalogue_id),
+ is_mandatory=is_mandatory is not None,
+ requires_review=requires_review is not None,
+ is_aqmm_task=is_aqmm_task is not None,
+ aqmm_mandatory=aqmm_mandatory is not None,
+ aqmm_evidence_required=aqmm_evidence_required is not None,
+ aqmm_manager_review_required=aqmm_manager_review_required is not None,
+ aqmm_partner_review_required=aqmm_partner_review_required is not None,
+ aqmm_review_partner_required=aqmm_review_partner_required is not None,
+ aqmm_blocks_final_release=aqmm_blocks_final_release is not None,
+ aqmm_reference=aqmm_reference.strip() or None,
+ is_active=is_active is not None,
+ created_by_user_id=user.id,
+ updated_by_user_id=user.id,
+ )
+ db.add(row); db.commit()
+ return RedirectResponse(url=f'/services/templates/{catalogue_id}', status_code=303)
+ finally:
+ db.close()
+
+
+@router.get('/defaults')
+def default_templates_catalogue_list(request: Request):
+ db = CommonSessionLocal()
+ try:
+ user = get_current_user(request, db=db)
+ if not user:
+ return RedirectResponse(url='/login', status_code=303)
+ require_permission(db, user, 'service_tasks.view')
+ if not _is_system_admin(db, user):
+ return _redirect_denied()
+ payload = list_catalogue_payload(db, page=1, per_page=500)
+ rows = payload['rows']
+ return _render(request, 'modules/services/templates/services/default_templates_list.html', db, user, title='Default Task Templates', rows=rows, can_manage_defaults=_has_perm(db, user, 'service_tasks.create'))
+ except Exception:
+ return _redirect_denied()
+ finally:
+ db.close()
+
+
+@router.get('/catalogue/{catalogue_id}/defaults')
+def default_templates_detail(request: Request, catalogue_id: int):
+ db = CommonSessionLocal()
+ try:
+ user = get_current_user(request, db=db)
+ if not user:
+ return RedirectResponse(url='/login', status_code=303)
+ require_permission(db, user, 'service_tasks.view')
+ if not _is_system_admin(db, user):
+ return _redirect_denied()
+ catalogue = get_catalogue(db, catalogue_id)
+ if not catalogue:
+ return RedirectResponse(url='/services/defaults', status_code=303)
+ defaults = get_default_task_templates(db, catalogue_id=catalogue_id)
+ return _render(request, 'modules/services/templates/services/default_templates_detail.html', db, user, title=f'Default Tasks - {catalogue.service_name}', service=catalogue, default_tasks=defaults)
+ finally:
+ db.close()
+
+
+@router.post('/catalogue/{catalogue_id}/defaults/new')
+def default_template_create_submit(request: Request, catalogue_id: int, task_name: str = Form(...), description: str = Form(''), default_role_name: str = Form(''), sequence_no: int | None = Form(None), is_mandatory: str | None = Form(None), requires_review: str | None = Form(None), is_aqmm_task: str | None = Form(None), aqmm_mandatory: str | None = Form(None), aqmm_evidence_required: str | None = Form(None), aqmm_manager_review_required: str | None = Form(None), aqmm_partner_review_required: str | None = Form(None), aqmm_review_partner_required: str | None = Form(None), aqmm_blocks_final_release: str | None = Form(None), aqmm_reference: 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)
+ require_permission(db, user, 'service_tasks.create')
+ if not _is_system_admin(db, user):
+ return _redirect_denied()
+ catalogue = get_catalogue(db, catalogue_id)
+ if not catalogue:
+ return RedirectResponse(url='/services/defaults', status_code=303)
+ row = ServiceDefaultTaskTemplate(service_catalogue_id=catalogue_id, task_name=task_name.strip(), description=description.strip() or None, default_role_name=default_role_name.strip() or None, sequence_no=sequence_no or next_default_task_sequence(db, catalogue_id=catalogue_id), is_mandatory=is_mandatory is not None, requires_review=requires_review is not None, is_aqmm_task=is_aqmm_task is not None, aqmm_mandatory=aqmm_mandatory is not None, aqmm_evidence_required=aqmm_evidence_required is not None, aqmm_manager_review_required=aqmm_manager_review_required is not None, aqmm_partner_review_required=aqmm_partner_review_required is not None, aqmm_review_partner_required=aqmm_review_partner_required is not None, aqmm_blocks_final_release=aqmm_blocks_final_release is not None, aqmm_reference=aqmm_reference.strip() or None, is_active=is_active is not None)
+ db.add(row); db.commit()
+ return RedirectResponse(url=f'/services/catalogue/{catalogue_id}/defaults', status_code=303)
+ finally:
+ db.close()
+
+
+@router.post('/catalogue/{catalogue_id}/defaults/copy-to-firm')
+def copy_defaults_to_firm(request: Request, catalogue_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)
+ if not _can_manage_firm_tasks(db, user):
+ return _redirect_denied()
+ tenant_id = _active_tenant_id(request, user)
+ selection = get_firm_selection(db, tenant_id=tenant_id, catalogue_id=catalogue_id)
+ if not selection or not selection.is_enabled:
+ return RedirectResponse(url='/services', status_code=303)
+ existing_sequences = {t.sequence_no for t in get_firm_task_templates(db, tenant_id=tenant_id, catalogue_id=catalogue_id)}
+ defaults = get_default_task_templates(db, catalogue_id=catalogue_id)
+ next_seq = next_task_sequence(db, tenant_id=tenant_id, catalogue_id=catalogue_id)
+ for d in defaults:
+ seq = d.sequence_no
+ if seq in existing_sequences:
+ seq = next_seq
+ next_seq += 1
+ existing_sequences.add(seq)
+ db.add(FirmServiceTaskTemplate(
+ tenant_id=tenant_id,
+ service_catalogue_id=catalogue_id,
+ task_name=d.task_name,
+ description=d.description,
+ sequence_no=seq,
+ default_role_name=d.default_role_name,
+ is_mandatory=d.is_mandatory,
+ requires_review=d.requires_review,
+ is_aqmm_task=getattr(d, "is_aqmm_task", False),
+ aqmm_mandatory=getattr(d, "aqmm_mandatory", False),
+ aqmm_evidence_required=getattr(d, "aqmm_evidence_required", False),
+ aqmm_manager_review_required=getattr(d, "aqmm_manager_review_required", False),
+ aqmm_partner_review_required=getattr(d, "aqmm_partner_review_required", False),
+ aqmm_review_partner_required=getattr(d, "aqmm_review_partner_required", False),
+ aqmm_blocks_final_release=getattr(d, "aqmm_blocks_final_release", False),
+ aqmm_reference=getattr(d, "aqmm_reference", None),
+ is_active=d.is_active,
+ created_by_user_id=user.id,
+ updated_by_user_id=user.id,
+ ))
+ db.commit()
+ return RedirectResponse(url=f'/services/templates/{catalogue_id}', status_code=303)
+ finally:
+ db.close()
+
+
+@router.post('/templates/{catalogue_id}/tasks/{task_id}/document-requirements/new')
+def firm_task_document_requirement_create(
+ request: Request,
+ catalogue_id: int,
+ task_id: int,
+ document_name: str = Form(...),
+ document_type: str = Form('GENERAL'),
+ is_mandatory: str | None = Form(None),
+ allowed_file_types: str = Form(''),
+ instructions: str = Form(''),
+ sort_order: int = Form(100),
+ 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_manage_firm_tasks(db, user):
+ return _redirect_denied()
+ tenant_id = _active_tenant_id(request, user)
+ task = get_firm_task_template(db, task_id=task_id, tenant_id=tenant_id, catalogue_id=catalogue_id)
+ if not task:
+ return RedirectResponse(url=f'/services/templates/{catalogue_id}?error=task_missing', status_code=303)
+ create_task_document_requirement(
+ db,
+ task_template=task,
+ document_name=document_name,
+ document_type=document_type,
+ is_mandatory=is_mandatory is not None,
+ allowed_file_types=allowed_file_types,
+ instructions=instructions,
+ sort_order=sort_order,
+ user=user,
+ )
+ db.commit()
+ return RedirectResponse(url=f'/services/templates/{catalogue_id}?requirement_added=1', status_code=303)
+ finally:
+ db.close()
+
+
+@router.post('/templates/{catalogue_id}/tasks/{task_id}/document-requirements/{requirement_id}/toggle')
+def firm_task_document_requirement_toggle(
+ request: Request,
+ catalogue_id: int,
+ task_id: int,
+ requirement_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)
+ if not _can_manage_firm_tasks(db, user):
+ return _redirect_denied()
+ tenant_id = _active_tenant_id(request, user)
+ task = get_firm_task_template(db, task_id=task_id, tenant_id=tenant_id, catalogue_id=catalogue_id)
+ requirement = get_task_document_requirement(db, requirement_id=requirement_id, tenant_id=tenant_id)
+ if not task or not requirement or requirement.firm_task_template_id != task.id:
+ return RedirectResponse(url=f'/services/templates/{catalogue_id}?error=requirement_missing', status_code=303)
+ requirement.is_active = not bool(requirement.is_active)
+ requirement.updated_by_user_id = user.id
+ db.commit()
+ return RedirectResponse(url=f'/services/templates/{catalogue_id}?requirement_updated=1', status_code=303)
+ finally:
+ db.close()
+
+
+@router.post('/templates/{catalogue_id}/tasks/{task_id}/document-templates/upload')
+def firm_task_document_template_upload(
+ request: Request,
+ catalogue_id: int,
+ task_id: int,
+ template_name: str = Form(''),
+ template_category: str = Form(''),
+ description: str = Form(''),
+ file: UploadFile = File(...),
+ 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_manage_firm_tasks(db, user):
+ return _redirect_denied()
+ tenant_id = _active_tenant_id(request, user)
+ task = get_firm_task_template(db, task_id=task_id, tenant_id=tenant_id, catalogue_id=catalogue_id)
+ if not task:
+ return RedirectResponse(url=f'/services/templates/{catalogue_id}?error=task_missing', status_code=303)
+ if not file or not file.filename:
+ return RedirectResponse(url=f'/services/templates/{catalogue_id}?error=missing_file', status_code=303)
+ save_task_document_template(
+ db,
+ task_template=task,
+ template_name=template_name or file.filename,
+ template_category=template_category,
+ description=description,
+ upload_file=file,
+ user=user,
+ )
+ db.commit()
+ return RedirectResponse(url=f'/services/templates/{catalogue_id}?template_uploaded=1', status_code=303)
+ finally:
+ db.close()
+
+
+@router.get('/document-templates/{template_id}/download')
+def firm_task_document_template_download(request: Request, template_id: int):
+ db = CommonSessionLocal()
+ try:
+ user = get_current_user(request, db=db)
+ if not user:
+ return RedirectResponse(url='/login', status_code=303)
+ if not (_has_perm(db, user, 'service_tasks.view') or _can_manage_firm_tasks(db, user)):
+ return _redirect_denied()
+ tenant_id = _active_tenant_id(request, user)
+ template = db.get(FirmTaskDocumentTemplate, template_id)
+ if not template or template.tenant_id != tenant_id or not template.is_active:
+ return _redirect_denied()
+ path = template_absolute_path(template)
+ if not path.exists():
+ return RedirectResponse(url=f'/services/templates/{template.service_catalogue_id}?error=template_file_missing', status_code=303)
+ return FileResponse(path, filename=template.original_filename, media_type=template.content_type or 'application/octet-stream')
+ finally:
+ db.close()
+
+
+@router.get('/templates/{catalogue_id}/tasks/{task_id}/edit')
+def firm_task_template_edit_page(request: Request, catalogue_id: int, task_id: int):
+ db = CommonSessionLocal()
+ try:
+ user = get_current_user(request, db=db)
+ if not user:
+ return RedirectResponse(url='/login', status_code=303)
+
+ if not _can_manage_firm_tasks(db, user):
+ return _redirect_denied()
+
+ tenant_id = _active_tenant_id(request, user)
+
+ catalogue = get_catalogue(db, catalogue_id)
+ if not catalogue:
+ return RedirectResponse(url='/services/templates', status_code=303)
+
+ selection = get_firm_selection(db, tenant_id=tenant_id, catalogue_id=catalogue_id)
+ if not selection or not selection.is_enabled:
+ return RedirectResponse(url='/services', status_code=303)
+
+ task = get_firm_task_template(
+ db,
+ tenant_id=tenant_id,
+ catalogue_id=catalogue_id,
+ task_id=task_id,
+ )
+ if not task:
+ return RedirectResponse(url=f'/services/templates/{catalogue_id}', status_code=303)
+
+ return _render(
+ request,
+ 'modules/services/templates/services/firm_task_form.html',
+ db,
+ user,
+ title=f'Edit Firm Task - {catalogue.service_name}',
+ service=catalogue,
+ task=task,
+ )
+ finally:
+ db.close()
+
+
+@router.post('/templates/{catalogue_id}/tasks/{task_id}/edit')
+def firm_task_template_edit_submit(
+ request: Request,
+ catalogue_id: int,
+ task_id: int,
+ task_name: str = Form(...),
+ description: str = Form(''),
+ default_role_name: str = Form(''),
+ sequence_no: int = Form(1),
+ is_mandatory: str | None = Form(None),
+ requires_review: str | None = Form(None),
+ is_aqmm_task: str | None = Form(None),
+ aqmm_mandatory: str | None = Form(None),
+ aqmm_evidence_required: str | None = Form(None),
+ aqmm_manager_review_required: str | None = Form(None),
+ aqmm_partner_review_required: str | None = Form(None),
+ aqmm_review_partner_required: str | None = Form(None),
+ aqmm_blocks_final_release: str | None = Form(None),
+ aqmm_reference: 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)
+
+ if not _can_manage_firm_tasks(db, user):
+ return _redirect_denied()
+
+ tenant_id = _active_tenant_id(request, user)
+
+ task = get_firm_task_template(
+ db,
+ tenant_id=tenant_id,
+ catalogue_id=catalogue_id,
+ task_id=task_id,
+ )
+ if not task:
+ return RedirectResponse(url=f'/services/templates/{catalogue_id}', status_code=303)
+
+ task.task_name = task_name.strip()
+ task.description = description.strip() or None
+ task.default_role_name = default_role_name.strip() or None
+ task.sequence_no = sequence_no
+ task.is_mandatory = is_mandatory is not None
+ task.requires_review = requires_review is not None
+ task.is_aqmm_task = is_aqmm_task is not None
+ task.aqmm_mandatory = aqmm_mandatory is not None
+ task.aqmm_evidence_required = aqmm_evidence_required is not None
+ task.aqmm_manager_review_required = aqmm_manager_review_required is not None
+ task.aqmm_partner_review_required = aqmm_partner_review_required is not None
+ task.aqmm_review_partner_required = aqmm_review_partner_required is not None
+ task.aqmm_blocks_final_release = aqmm_blocks_final_release is not None
+ task.aqmm_reference = aqmm_reference.strip() or None
+ task.is_active = is_active is not None
+ task.updated_by_user_id = user.id
+
+ db.commit()
+ return RedirectResponse(url=f'/services/templates/{catalogue_id}', status_code=303)
+ finally:
+ db.close()
+
+
+@router.get('/catalogue/{catalogue_id}/defaults/{task_id}/edit')
+def default_task_template_edit_page(request: Request, catalogue_id: int, 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.edit')
+ except Exception:
+ return _redirect_denied()
+
+ if not _is_system_admin(db, user):
+ return _redirect_denied()
+
+ catalogue = get_catalogue(db, catalogue_id)
+ if not catalogue:
+ return RedirectResponse(url='/services/defaults', status_code=303)
+
+ task = get_default_task_template(
+ db,
+ catalogue_id=catalogue_id,
+ task_id=task_id,
+ )
+ if not task:
+ return RedirectResponse(url=f'/services/catalogue/{catalogue_id}/defaults', status_code=303)
+
+ return _render(
+ request,
+ 'modules/services/templates/services/default_task_form.html',
+ db,
+ user,
+ title=f'Edit Default Task - {catalogue.service_name}',
+ service=catalogue,
+ task=task,
+ )
+ finally:
+ db.close()
+
+
+@router.post('/catalogue/{catalogue_id}/defaults/{task_id}/edit')
+def default_task_template_edit_submit(
+ request: Request,
+ catalogue_id: int,
+ task_id: int,
+ task_name: str = Form(...),
+ description: str = Form(''),
+ default_role_name: str = Form(''),
+ sequence_no: int = Form(1),
+ is_mandatory: str | None = Form(None),
+ requires_review: str | None = Form(None),
+ is_aqmm_task: str | None = Form(None),
+ aqmm_mandatory: str | None = Form(None),
+ aqmm_evidence_required: str | None = Form(None),
+ aqmm_manager_review_required: str | None = Form(None),
+ aqmm_partner_review_required: str | None = Form(None),
+ aqmm_review_partner_required: str | None = Form(None),
+ aqmm_blocks_final_release: str | None = Form(None),
+ aqmm_reference: 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)
+
+ try:
+ require_permission(db, user, 'service_tasks.edit')
+ except Exception:
+ return _redirect_denied()
+
+ if not _is_system_admin(db, user):
+ return _redirect_denied()
+
+ task = get_default_task_template(
+ db,
+ catalogue_id=catalogue_id,
+ task_id=task_id,
+ )
+ if not task:
+ return RedirectResponse(url=f'/services/catalogue/{catalogue_id}/defaults', status_code=303)
+
+ task.task_name = task_name.strip()
+ task.description = description.strip() or None
+ task.default_role_name = default_role_name.strip() or None
+ task.sequence_no = sequence_no
+ task.is_mandatory = is_mandatory is not None
+ task.requires_review = requires_review is not None
+ task.is_aqmm_task = is_aqmm_task is not None
+ task.aqmm_mandatory = aqmm_mandatory is not None
+ task.aqmm_evidence_required = aqmm_evidence_required is not None
+ task.aqmm_manager_review_required = aqmm_manager_review_required is not None
+ task.aqmm_partner_review_required = aqmm_partner_review_required is not None
+ task.aqmm_review_partner_required = aqmm_review_partner_required is not None
+ task.aqmm_blocks_final_release = aqmm_blocks_final_release is not None
+ task.aqmm_reference = aqmm_reference.strip() or None
+ task.is_active = is_active is not None
+
+ db.commit()
+ return RedirectResponse(url=f'/services/catalogue/{catalogue_id}/defaults', status_code=303)
+ finally:
+ db.close()
+
+# -----------------------------------------------------------------------------
+# S4.5 Bulk import and bulk service assignment routes
+# Kept inside existing Services module intentionally; no separate router required.
+# -----------------------------------------------------------------------------
+
+def _locked_partner_id(db, user) -> int | None:
+ return int(user.id) if _has_perm(db, user, "clients.view.own_only") else None
+
+
+def _bulk_template_response(template_type: str, filename: str):
+ data = build_bulk_import_template(template_type)
+ return StreamingResponse(
+ iter([data]),
+ media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+ headers={"Content-Disposition": f'attachment; filename="{filename}"'},
+ )
+
+
+@router.get("/bulk-imports")
+def services_bulk_import_home(request: Request):
+ db = CommonSessionLocal()
+ try:
+ user = get_current_user(request, db=db)
+ if not user:
+ return RedirectResponse(url="/login", status_code=303)
+ can_client_assignment = _has_perm(db, user, "clients.edit")
+ can_firm_tasks = _has_perm(db, user, "service_tasks.create")
+ can_system_import = _is_system_admin(db, user)
+ can_due_date_extensions = _can_manage_firm_services(db, user) or can_system_import
+ if not (can_client_assignment or can_firm_tasks or can_system_import or can_due_date_extensions):
+ return _redirect_denied()
+ return _render(
+ request,
+ "modules/services/templates/services/bulk_imports/index.html",
+ db,
+ user,
+ title="Services Bulk Import",
+ can_client_assignment=can_client_assignment,
+ can_firm_tasks=can_firm_tasks,
+ can_system_import=can_system_import,
+ can_due_date_extensions=can_due_date_extensions,
+ )
+ finally:
+ db.close()
+
+
+@router.get("/bulk-imports/templates/engagement-assignments.xlsx")
+def download_client_service_assignment_template(request: Request):
+ return _bulk_template_response("client_service_assignments", "client_service_assignments_template.xlsx")
+
+
+@router.get("/bulk-imports/templates/service-master.xlsx")
+def download_service_master_template(request: Request):
+ return _bulk_template_response("service_master", "service_master_template.xlsx")
+
+
+@router.get("/bulk-imports/templates/system-default-tasks.xlsx")
+def download_system_default_tasks_template(request: Request):
+ return _bulk_template_response("system_default_tasks", "system_default_tasks_template.xlsx")
+
+
+@router.get("/bulk-imports/templates/due-date-extensions.xlsx")
+def download_due_date_extensions_template(request: Request):
+ return _bulk_template_response("due_date_extensions", "due_date_extensions_template.xlsx")
+
+
+@router.get("/bulk-imports/templates/firm-task-templates.xlsx")
+def download_firm_task_templates_template(request: Request):
+ return _bulk_template_response("firm_task_templates", "firm_task_templates_template.xlsx")
+
+
+@router.post("/bulk-imports/engagement-assignments")
+def upload_client_service_assignments(
+ request: Request,
+ file: UploadFile = File(...),
+ update_existing: 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)
+ try:
+ require_permission(db, user, "clients.edit")
+ except Exception:
+ return _redirect_denied()
+ result = import_client_service_assignments(
+ db,
+ current_user=user,
+ tenant_id=_active_tenant_id(request, user),
+ locked_partner_id=_locked_partner_id(db, user),
+ file_bytes=file.file.read(),
+ update_existing=update_existing is not None,
+ )
+ return _render(request, "modules/services/templates/services/bulk_imports/result.html", db, user, title="Engagement Assignment Import Result", result=result, back_url="/services/bulk-imports")
+ finally:
+ db.close()
+
+
+@router.post("/bulk-imports/service-master")
+def upload_service_master(request: Request, file: UploadFile = File(...), update_existing: 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 _is_system_admin(db, user):
+ return _redirect_denied()
+ try:
+ require_permission(db, user, "services.create")
+ except Exception:
+ return _redirect_denied()
+ result = import_service_master(
+ db,
+ current_user=user,
+ tenant_id=_active_tenant_id(request, user),
+ file_bytes=file.file.read(),
+ update_existing=update_existing is not None,
+ )
+ return _render(request, "modules/services/templates/services/bulk_imports/result.html", db, user, title="Service Master Import Result", result=result, back_url="/services/bulk-imports")
+ finally:
+ db.close()
+
+
+@router.post("/bulk-imports/system-default-tasks")
+def upload_system_default_tasks(request: Request, file: UploadFile = File(...), update_existing: 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 _is_system_admin(db, user):
+ return _redirect_denied()
+ try:
+ require_permission(db, user, "service_tasks.create")
+ except Exception:
+ return _redirect_denied()
+ result = import_system_default_tasks(db, current_user=user, file_bytes=file.file.read(), update_existing=update_existing is not None)
+ return _render(request, "modules/services/templates/services/bulk_imports/result.html", db, user, title="System Default Tasks Import Result", result=result, back_url="/services/bulk-imports")
+ finally:
+ db.close()
+
+
+@router.post("/bulk-imports/due-date-extensions")
+def upload_due_date_extensions(request: Request, file: UploadFile = File(...), update_existing: 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_manage_firm_services(db, user) or _is_system_admin(db, user)):
+ return _redirect_denied()
+ result = import_due_date_extensions(
+ db,
+ current_user=user,
+ tenant_id=_active_tenant_id(request, user),
+ file_bytes=file.file.read(),
+ update_existing=update_existing is not None,
+ )
+ return _render(request, "modules/services/templates/services/bulk_imports/result.html", db, user, title="Due Date Extension Import Result", result=result, back_url="/services/bulk-imports")
+ finally:
+ db.close()
+
+
+@router.post("/bulk-imports/firm-task-templates")
+def upload_firm_task_templates(request: Request, file: UploadFile = File(...), update_existing: 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)
+ try:
+ require_permission(db, user, "service_tasks.create")
+ except Exception:
+ return _redirect_denied()
+ result = import_firm_task_templates(db, current_user=user, tenant_id=_active_tenant_id(request, user), file_bytes=file.file.read(), update_existing=update_existing is not None)
+ return _render(request, "modules/services/templates/services/bulk_imports/result.html", db, user, title="Firm Task Templates Import Result", result=result, back_url="/services/bulk-imports")
+ finally:
+ db.close()
diff --git a/app/modules/services/work_tracker_ui.py b/app/modules/services/work_tracker_ui.py
index 4c652a7..43ceb62 100644
--- a/app/modules/services/work_tracker_ui.py
+++ b/app/modules/services/work_tracker_ui.py
@@ -11,6 +11,7 @@ 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.task_tools import get_task_tool
from app.modules.services.execution import (
TASK_COMMENT_TYPES,
TASK_COMMENT_VISIBILITIES,
@@ -443,6 +444,7 @@ def task_edit_page(request: Request, task_id: int):
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),
+ task_tool=get_task_tool(getattr(task, "task_tool_code", "NONE")),
)
finally:
db.close()
|