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