Add task category master for service task templates
This commit is contained in:
A R R R Associates
+146
-7
@@ -15,7 +15,7 @@ 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.registrations.models import RegistrationType
|
||||
from app.modules.services.models import FirmServiceSelection, FirmServiceTaskTemplate, ServiceCatalogue, ServiceCategory, ServiceDefaultTaskTemplate, ServiceDueDateRule, FirmTaskDocumentRequirement, FirmTaskDocumentTemplate
|
||||
from app.modules.services.models import FirmServiceSelection, FirmServiceTaskTemplate, ServiceCatalogue, ServiceCategory, ServiceDefaultTaskTemplate, ServiceDueDateRule, FirmTaskDocumentRequirement, FirmTaskDocumentTemplate, ServiceTaskCategory
|
||||
from app.modules.services.bulk_imports import (
|
||||
build_template as build_bulk_import_template,
|
||||
import_client_service_assignments,
|
||||
@@ -68,6 +68,9 @@ from app.modules.services.services import (
|
||||
get_default_task_templates,
|
||||
get_firm_task_template,
|
||||
get_default_task_template,
|
||||
list_task_categories,
|
||||
get_task_category,
|
||||
ensure_task_category,
|
||||
)
|
||||
|
||||
router = APIRouter(prefix="/services", tags=["services-ui"])
|
||||
@@ -619,6 +622,7 @@ def catalogue_detail(request: Request, catalogue_id: int):
|
||||
can_manage_firm_services=_can_manage_firm_services(db, user),
|
||||
can_manage_firm_tasks=_can_manage_firm_tasks(db, user),
|
||||
is_system_admin=_is_system_admin(db, user),
|
||||
task_categories=list_task_categories(db, catalogue_id=catalogue_id, tenant_id=tenant_id, include_inactive=True),
|
||||
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),
|
||||
@@ -958,6 +962,12 @@ def _copy_system_defaults_if_firm_tasks_empty(
|
||||
is_mandatory=default.is_mandatory,
|
||||
requires_review=default.requires_review,
|
||||
normal_review_role=getattr(default, "normal_review_role", None),
|
||||
task_category_id=(firm_category.id if (firm_category := ensure_task_category(db, catalogue_id=catalogue_id, tenant_id=tenant_id, name=getattr(default, "task_category", None), user_id=user_id)) else None),
|
||||
task_category=(firm_category.name if firm_category else None),
|
||||
response_required=getattr(default, "response_required", False),
|
||||
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),
|
||||
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),
|
||||
@@ -1245,6 +1255,7 @@ def task_templates_detail(request: Request, catalogue_id: int):
|
||||
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),
|
||||
task_categories=list_task_categories(db, catalogue_id=catalogue_id, tenant_id=tenant_id, include_inactive=True),
|
||||
)
|
||||
finally:
|
||||
db.close()
|
||||
@@ -1278,7 +1289,8 @@ def task_template_create_submit(request: Request, catalogue_id: int, task_name:
|
||||
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=task_category.strip() or None,
|
||||
task_category_id=(category.id if (category := ensure_task_category(db, catalogue_id=catalogue_id, tenant_id=tenant_id, 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,
|
||||
@@ -1356,7 +1368,7 @@ def default_templates_detail(request: Request, catalogue_id: int):
|
||||
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)
|
||||
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, task_categories=list_task_categories(db, catalogue_id=catalogue_id, tenant_id=None, include_inactive=True))
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@@ -1375,7 +1387,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=task_category.strip() or 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"), 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:
|
||||
@@ -1415,7 +1427,8 @@ def copy_defaults_to_firm(request: Request, catalogue_id: int, csrf_token: str =
|
||||
is_mandatory=d.is_mandatory,
|
||||
requires_review=d.requires_review,
|
||||
normal_review_role=getattr(d, "normal_review_role", None),
|
||||
task_category=getattr(d, "task_category", None),
|
||||
task_category_id=(firm_category.id if (firm_category := ensure_task_category(db, catalogue_id=catalogue_id, tenant_id=tenant_id, name=getattr(d, "task_category", None), user_id=user.id)) else None),
|
||||
task_category=(firm_category.name if firm_category else None),
|
||||
response_required=getattr(d, "response_required", False),
|
||||
response_type=getattr(d, "response_type", "NONE") or "NONE",
|
||||
evidence_required=getattr(d, "evidence_required", False),
|
||||
@@ -1760,6 +1773,7 @@ def firm_task_template_edit_page(request: Request, catalogue_id: int, task_id: i
|
||||
title=f'Edit Firm Task - {catalogue.service_name}',
|
||||
service=catalogue,
|
||||
task=task,
|
||||
task_categories=list_task_categories(db, catalogue_id=catalogue_id, tenant_id=tenant_id, include_inactive=True),
|
||||
)
|
||||
finally:
|
||||
db.close()
|
||||
@@ -1825,7 +1839,9 @@ def firm_task_template_edit_submit(
|
||||
task.is_mandatory = is_mandatory is not None
|
||||
task.requires_review = requires_review is not None
|
||||
task.normal_review_role = _normal_review_role_value(normal_review_role, requires_review=task.requires_review)
|
||||
task.task_category = task_category.strip() or None
|
||||
category = ensure_task_category(db, catalogue_id=catalogue_id, tenant_id=tenant_id, name=task_category, user_id=user.id)
|
||||
task.task_category_id = category.id if category else None
|
||||
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.evidence_required = evidence_required is not None
|
||||
@@ -1904,6 +1920,7 @@ def default_task_template_edit_page(request: Request, catalogue_id: int, task_id
|
||||
title=f'Edit Default Task - {catalogue.service_name}',
|
||||
service=catalogue,
|
||||
task=task,
|
||||
task_categories=list_task_categories(db, catalogue_id=catalogue_id, tenant_id=None, include_inactive=True),
|
||||
)
|
||||
finally:
|
||||
db.close()
|
||||
@@ -1969,7 +1986,9 @@ def default_task_template_edit_submit(
|
||||
task.is_mandatory = is_mandatory is not None
|
||||
task.requires_review = requires_review is not None
|
||||
task.normal_review_role = _normal_review_role_value(normal_review_role, requires_review=task.requires_review)
|
||||
task.task_category = task_category.strip() or None
|
||||
category = ensure_task_category(db, catalogue_id=catalogue_id, tenant_id=None, name=task_category, user_id=user.id)
|
||||
task.task_category_id = category.id if category else None
|
||||
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.evidence_required = evidence_required is not None
|
||||
@@ -1989,6 +2008,126 @@ def default_task_template_edit_submit(
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# Service Task Category Master
|
||||
# -----------------------------------------------------------------------------
|
||||
|
||||
@router.get('/templates/{catalogue_id}/categories')
|
||||
def firm_task_categories_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 (_has_perm(db, user, 'service_tasks.view') or _can_manage_firm_tasks(db, user)):
|
||||
return _redirect_denied()
|
||||
tenant_id = _active_tenant_id(request, user)
|
||||
service = get_catalogue(db, catalogue_id)
|
||||
selection = get_firm_selection(db, tenant_id=tenant_id, catalogue_id=catalogue_id)
|
||||
if not service or not selection or not selection.is_enabled:
|
||||
return RedirectResponse(url='/services/templates', status_code=303)
|
||||
return _render(request, 'modules/services/templates/services/task_category_list.html', db, user,
|
||||
title=f'Task Categories - {service.service_name}', service=service,
|
||||
rows=list_task_categories(db, catalogue_id=catalogue_id, tenant_id=tenant_id, include_inactive=True),
|
||||
system_scope=False, can_manage=_can_manage_firm_tasks(db, user))
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
@router.post('/templates/{catalogue_id}/categories/new')
|
||||
def firm_task_category_create(request: Request, catalogue_id: int, name: str = Form(...), code: 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)
|
||||
service = get_catalogue(db, catalogue_id)
|
||||
if not service: return RedirectResponse(url='/services/templates', status_code=303)
|
||||
row = ensure_task_category(db, catalogue_id=catalogue_id, tenant_id=tenant_id, name=name, user_id=user.id)
|
||||
if row:
|
||||
requested_code = normalize_code(code)[:50] if code.strip() else row.code
|
||||
duplicate = db.execute(select(ServiceTaskCategory.id).where(ServiceTaskCategory.tenant_id == tenant_id, ServiceTaskCategory.service_catalogue_id == catalogue_id, ServiceTaskCategory.code == requested_code, ServiceTaskCategory.id != row.id)).scalar_one_or_none()
|
||||
if duplicate is not None: return RedirectResponse(url=f'/services/templates/{catalogue_id}/categories?error=duplicate_code', status_code=303)
|
||||
row.code = requested_code; row.sort_order = sort_order; row.is_active = True; row.updated_by_user_id = user.id
|
||||
db.commit()
|
||||
return RedirectResponse(url=f'/services/templates/{catalogue_id}/categories?created=1', status_code=303)
|
||||
finally: db.close()
|
||||
|
||||
|
||||
@router.post('/templates/{catalogue_id}/categories/{category_id}/update')
|
||||
def firm_task_category_update(request: Request, catalogue_id: int, category_id: int, name: str = Form(...), code: 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)
|
||||
if not _can_manage_firm_tasks(db,user): return _redirect_denied()
|
||||
tenant_id=_active_tenant_id(request,user); row=get_task_category(db, category_id=category_id, catalogue_id=catalogue_id, tenant_id=tenant_id)
|
||||
if not row: return RedirectResponse(url=f'/services/templates/{catalogue_id}/categories?error=not_found', status_code=303)
|
||||
clean_name=name.strip(); clean_code=normalize_code(code or clean_name)[:50]
|
||||
duplicate=db.execute(select(ServiceTaskCategory.id).where(ServiceTaskCategory.tenant_id==tenant_id, ServiceTaskCategory.service_catalogue_id==catalogue_id, ServiceTaskCategory.code==clean_code, ServiceTaskCategory.id!=row.id)).scalar_one_or_none()
|
||||
if duplicate is not None: return RedirectResponse(url=f'/services/templates/{catalogue_id}/categories?error=duplicate_code', status_code=303)
|
||||
old_name=row.name; row.name=clean_name; row.code=clean_code; row.sort_order=sort_order; row.is_active=is_active is not None; row.updated_by_user_id=user.id
|
||||
for task in db.execute(select(FirmServiceTaskTemplate).where(FirmServiceTaskTemplate.tenant_id==tenant_id, FirmServiceTaskTemplate.service_catalogue_id==catalogue_id, FirmServiceTaskTemplate.task_category_id==row.id)).scalars().all(): task.task_category=clean_name
|
||||
# Backward-compatible rows created before category IDs are also renamed when their text exactly matches.
|
||||
for task in db.execute(select(FirmServiceTaskTemplate).where(FirmServiceTaskTemplate.tenant_id==tenant_id, FirmServiceTaskTemplate.service_catalogue_id==catalogue_id, FirmServiceTaskTemplate.task_category_id.is_(None), FirmServiceTaskTemplate.task_category==old_name)).scalars().all(): task.task_category_id=row.id; task.task_category=clean_name
|
||||
db.commit(); return RedirectResponse(url=f'/services/templates/{catalogue_id}/categories?updated=1', status_code=303)
|
||||
finally: db.close()
|
||||
|
||||
|
||||
@router.get('/catalogue/{catalogue_id}/task-categories')
|
||||
def system_task_categories_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.view')
|
||||
if not _is_system_admin(db,user): return _redirect_denied()
|
||||
service=get_catalogue(db,catalogue_id)
|
||||
if not service: return RedirectResponse(url='/services/defaults', status_code=303)
|
||||
return _render(request,'modules/services/templates/services/task_category_list.html',db,user,title=f'System Task Categories - {service.service_name}',service=service,rows=list_task_categories(db,catalogue_id=catalogue_id,tenant_id=None,include_inactive=True),system_scope=True,can_manage=_has_perm(db,user,'service_tasks.edit') or _has_perm(db,user,'service_tasks.create'))
|
||||
finally: db.close()
|
||||
|
||||
|
||||
@router.post('/catalogue/{catalogue_id}/task-categories/new')
|
||||
def system_task_category_create(request: Request, catalogue_id: int, name: str = Form(...), code: 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)
|
||||
require_permission(db,user,'service_tasks.create')
|
||||
if not _is_system_admin(db,user): return _redirect_denied()
|
||||
row=ensure_task_category(db,catalogue_id=catalogue_id,tenant_id=None,name=name,user_id=user.id)
|
||||
if row:
|
||||
requested_code=normalize_code(code)[:50] if code.strip() else row.code
|
||||
duplicate=db.execute(select(ServiceTaskCategory.id).where(ServiceTaskCategory.tenant_id.is_(None), ServiceTaskCategory.service_catalogue_id==catalogue_id, ServiceTaskCategory.code==requested_code, ServiceTaskCategory.id!=row.id)).scalar_one_or_none()
|
||||
if duplicate is not None: return RedirectResponse(url=f'/services/catalogue/{catalogue_id}/task-categories?error=duplicate_code',status_code=303)
|
||||
row.code=requested_code; row.sort_order=sort_order; row.is_active=True; row.updated_by_user_id=user.id
|
||||
db.commit(); return RedirectResponse(url=f'/services/catalogue/{catalogue_id}/task-categories?created=1',status_code=303)
|
||||
finally: db.close()
|
||||
|
||||
|
||||
@router.post('/catalogue/{catalogue_id}/task-categories/{category_id}/update')
|
||||
def system_task_category_update(request: Request, catalogue_id: int, category_id: int, name: str = Form(...), code: 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,'service_tasks.edit')
|
||||
if not _is_system_admin(db,user): return _redirect_denied()
|
||||
row=get_task_category(db,category_id=category_id,catalogue_id=catalogue_id,tenant_id=None)
|
||||
if not row: return RedirectResponse(url=f'/services/catalogue/{catalogue_id}/task-categories?error=not_found',status_code=303)
|
||||
clean_name=name.strip(); clean_code=normalize_code(code or clean_name)[:50]
|
||||
duplicate=db.execute(select(ServiceTaskCategory.id).where(ServiceTaskCategory.tenant_id.is_(None), ServiceTaskCategory.service_catalogue_id==catalogue_id, ServiceTaskCategory.code==clean_code, ServiceTaskCategory.id!=row.id)).scalar_one_or_none()
|
||||
if duplicate is not None: return RedirectResponse(url=f'/services/catalogue/{catalogue_id}/task-categories?error=duplicate_code',status_code=303)
|
||||
old_name=row.name; row.name=clean_name; row.code=clean_code; row.sort_order=sort_order; row.is_active=is_active is not None; row.updated_by_user_id=user.id
|
||||
for task in db.execute(select(ServiceDefaultTaskTemplate).where(ServiceDefaultTaskTemplate.service_catalogue_id==catalogue_id, ServiceDefaultTaskTemplate.task_category_id==row.id)).scalars().all(): task.task_category=row.name
|
||||
for task in db.execute(select(ServiceDefaultTaskTemplate).where(ServiceDefaultTaskTemplate.service_catalogue_id==catalogue_id, ServiceDefaultTaskTemplate.task_category_id.is_(None), ServiceDefaultTaskTemplate.task_category==old_name)).scalars().all(): task.task_category_id=row.id; task.task_category=row.name
|
||||
db.commit(); return RedirectResponse(url=f'/services/catalogue/{catalogue_id}/task-categories?updated=1',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.
|
||||
|
||||
Reference in New Issue
Block a user