from __future__ import annotations from datetime import datetime, timezone 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.default_tasks_export import build_default_tasks_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: str = '', recurrence_type: str = '', engagement_type: str = '', page: int = 1, per_page: int = 20): category_id_value: int | None = None category_id_text = (category_id or '').strip() if category_id_text: try: parsed_category_id = int(category_id_text) if parsed_category_id > 0: category_id_value = parsed_category_id except (TypeError, ValueError): category_id_value = None 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_value, 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), 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(...)): 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, 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, 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('/defaults/export') def default_tasks_export(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() workbook_bytes = build_default_tasks_export(db) filename = f"system_default_tasks_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/{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), 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(...)): 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, 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) 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, task_category=getattr(d, "task_category", None), response_required=getattr(d, "response_required", False), 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), 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), 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(...), ): 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.task_category = task_category.strip() or 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 task.remarks_required_if_no = remarks_required_if_no 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), 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(...), ): 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.task_category = task_category.strip() or 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 task.remarks_required_if_no = remarks_required_if_no 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()