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.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.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(): return RedirectResponse(url="/system-settings", status_code=303) 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/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: 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: 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: 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: return RedirectResponse(url='/services', 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_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_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_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_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_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_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_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_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_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()