A R R R Associates
2651 lines
121 KiB
Python
2651 lines
121 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from fastapi import APIRouter, File, Form, Request, UploadFile
|
|
from fastapi.responses import FileResponse, RedirectResponse, StreamingResponse
|
|
from sqlalchemy import select
|
|
|
|
from app.core.db.common import CommonSessionLocal
|
|
from app.core.security.csrf import get_or_create_csrf_token, validate_csrf
|
|
from app.core.security.session_auth import get_current_user
|
|
from app.core.templating import templates
|
|
from app.modules.core.audit.service import model_snapshot, pair_before_after, write_audit_log
|
|
from app.modules.core.rbac.deps import get_user_permissions, get_user_roles
|
|
from app.modules.core.rbac.permission_guard import require_permission
|
|
from app.modules.core.tenancy.models import Branch
|
|
from app.modules.registrations.models import RegistrationType
|
|
from app.modules.services.models import FirmServiceSelection, FirmServiceTaskTemplate, ServiceCatalogue, ServiceCategory, ServiceDefaultTaskTemplate, ServiceDueDateRule, FirmTaskDocumentRequirement, FirmTaskDocumentTemplate, ServiceTaskCategory
|
|
from app.modules.services.bulk_imports import (
|
|
build_template as build_bulk_import_template,
|
|
import_client_service_assignments,
|
|
import_firm_task_templates,
|
|
import_service_master,
|
|
import_system_default_tasks,
|
|
import_due_date_extensions,
|
|
)
|
|
from app.modules.services.catalogue_export import build_service_catalogue_export
|
|
from app.modules.services.default_tasks_export import build_default_tasks_export
|
|
from app.modules.services.firm_tasks_export import build_firm_tasks_export
|
|
from app.modules.services.default_task_sync import (
|
|
accept_system_update_for_firm_task,
|
|
keep_firm_customization_for_system_revision,
|
|
mark_firm_task_customized,
|
|
sync_firm_tasks_from_system_defaults,
|
|
sync_system_defaults_to_all_firms,
|
|
system_task_diff,
|
|
)
|
|
from app.modules.services.task_categories_excel import build_task_categories_export, import_task_categories
|
|
from app.modules.services.task_tools import list_task_tools, normalize_task_tool_code
|
|
from app.modules.services.task_duplicates import find_duplicate_pairs, merge_default_tasks, merge_firm_tasks
|
|
from app.modules.services.due_dates import (
|
|
DUE_PERIOD_TYPES,
|
|
DUE_YEAR_BASIS_CHOICES,
|
|
create_due_date_extension,
|
|
get_due_rule,
|
|
list_due_extensions,
|
|
list_due_rules,
|
|
parse_optional_date,
|
|
)
|
|
from app.modules.services.catalogue_merge import build_merge_preview, merge_service_catalogues
|
|
from app.modules.services.execution import (
|
|
TASK_EXECUTION_ROLES,
|
|
serialise_eligible_task_roles,
|
|
sync_open_engagement_tasks_for_service,
|
|
)
|
|
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,
|
|
list_task_categories,
|
|
get_task_category,
|
|
ensure_task_category,
|
|
)
|
|
|
|
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,
|
|
"task_execution_roles": TASK_EXECUTION_ROLES,
|
|
"task_tools": list_task_tools(),
|
|
}
|
|
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 _active_registration_types(db):
|
|
"""Return the active Registration Type master rows used by Service Catalogue forms."""
|
|
return db.execute(
|
|
select(RegistrationType)
|
|
.where(RegistrationType.is_active.is_(True))
|
|
.order_by(RegistrationType.sort_order, RegistrationType.name, RegistrationType.code)
|
|
).scalars().all()
|
|
|
|
|
|
def _registration_type_form_context(db):
|
|
registration_types = _active_registration_types(db)
|
|
return {
|
|
"registration_types": registration_types,
|
|
"registration_type_codes": [row.code.upper() for row in registration_types],
|
|
}
|
|
|
|
|
|
def _validated_registration_type_code(db, value: str, *, existing_value: str | None = None) -> str | None:
|
|
"""Validate a submitted catalogue registration code against Registration Type master.
|
|
|
|
Existing non-master legacy values are allowed only when left unchanged so editing an
|
|
unrelated field never destroys historical catalogue data. New/changed values must be
|
|
active Registration Type master codes.
|
|
"""
|
|
code = (value or "").strip().upper()
|
|
if not code:
|
|
return None
|
|
|
|
exists = db.execute(
|
|
select(RegistrationType.id).where(
|
|
RegistrationType.code == code,
|
|
RegistrationType.is_active.is_(True),
|
|
)
|
|
).scalar_one_or_none()
|
|
if exists is not None:
|
|
return code
|
|
|
|
current = (existing_value or "").strip().upper()
|
|
if current and code == current:
|
|
return code
|
|
|
|
raise ValueError("registration_type")
|
|
|
|
|
|
def _normal_review_role_value(value: str, *, requires_review: bool) -> str | None:
|
|
if not requires_review:
|
|
return None
|
|
clean = (value or "").strip().lower()
|
|
if clean in {"manager", "partner", "manager_or_partner"}:
|
|
return clean
|
|
return "manager"
|
|
|
|
|
|
def _redirect_denied():
|
|
from app.core.http_responses import ui_access_denied
|
|
return ui_access_denied()
|
|
|
|
|
|
def _has_perm(db, user, code: str) -> bool:
|
|
try:
|
|
require_permission(db, user, code)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _role_names(db, user) -> set[str]:
|
|
try:
|
|
return set(get_user_roles(db, user.id))
|
|
except Exception:
|
|
return set()
|
|
|
|
|
|
def _has_role(db, user, *names: str) -> bool:
|
|
roles = _role_names(db, user)
|
|
return any(name in roles for name in names)
|
|
|
|
|
|
def _can_view_services(db, user) -> bool:
|
|
return _has_perm(db, user, "services.view") or _has_role(db, user, "System Admin", "Firm Admin", "Partner", "Branch Manager")
|
|
|
|
|
|
def _is_system_admin(db, user) -> bool:
|
|
return _has_perm(db, user, "services.cross_tenant")
|
|
|
|
|
|
def _can_manage_firm_services(db, user) -> bool:
|
|
return _has_perm(db, user, "services.edit") or _has_role(db, user, "Firm Admin", "Partner")
|
|
|
|
|
|
def _can_manage_firm_tasks(db, user) -> bool:
|
|
return _has_perm(db, user, "service_tasks.create") or _has_perm(db, user, "service_tasks.edit") or _has_role(db, user, "Firm Admin", "Partner")
|
|
|
|
|
|
def _active_tenant_id(request: Request, user) -> int:
|
|
value = (
|
|
request.session.get("active_tenant_id")
|
|
or request.session.get("selected_tenant_id")
|
|
or request.session.get("tenant_id")
|
|
or user.tenant_id
|
|
)
|
|
return int(value)
|
|
|
|
|
|
@router.get("")
|
|
def services_home(request: Request, q: str = ""):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
try:
|
|
require_permission(db, user, "services.view")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
|
|
tenant_id = _active_tenant_id(request, user)
|
|
|
|
enabled_rows = list_firm_services_payload(db, tenant_id=tenant_id, q=q)
|
|
disabled_rows = list_disabled_catalogues(db, tenant_id=tenant_id, q=q)
|
|
branches = db.execute(
|
|
select(Branch).where(Branch.tenant_id == tenant_id, Branch.is_active.is_(True)).order_by(Branch.name.asc())
|
|
).scalars().all()
|
|
return _render(
|
|
request,
|
|
"modules/services/templates/services/list.html",
|
|
db,
|
|
user,
|
|
title="Services",
|
|
q=q,
|
|
enabled_rows=enabled_rows,
|
|
disabled_rows=disabled_rows,
|
|
branches=branches,
|
|
can_manage_catalogue=_is_system_admin(db, user),
|
|
can_manage_firm_services=_can_manage_firm_services(db, user),
|
|
can_manage_firm_tasks=_can_manage_firm_tasks(db, user),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get('/categories')
|
|
def category_list(request: Request, q: str = ''):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
if not _can_view_services(db, user):
|
|
return _redirect_denied()
|
|
rows = list_categories(db, q=q)
|
|
return _render(request, 'modules/services/templates/services/category_list.html', db, user, title='Service Categories', q=q, rows=rows, can_create=(_has_perm(db, user, 'services.create') and _is_system_admin(db, user)))
|
|
except Exception:
|
|
return _redirect_denied()
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get('/categories/new')
|
|
def category_create_page(request: Request):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
require_permission(db, user, 'services.create')
|
|
if not _is_system_admin(db, user):
|
|
return _redirect_denied()
|
|
return _render(request, 'modules/services/templates/services/category_form.html', db, user, title='Create Service Category', mode='create', category=None)
|
|
except Exception:
|
|
return _redirect_denied()
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post('/categories/new')
|
|
def category_create_submit(
|
|
request: Request,
|
|
code: str = Form(...),
|
|
name: str = Form(...),
|
|
sort_order: int = Form(100),
|
|
is_active: str | None = Form(None),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
|
|
require_permission(db, user, 'services.create')
|
|
|
|
if not _is_system_admin(db, user):
|
|
return _redirect_denied()
|
|
|
|
row = ServiceCategory(
|
|
code=normalize_code(code),
|
|
name=name.strip(),
|
|
sort_order=sort_order,
|
|
is_active=is_active is not None,
|
|
)
|
|
|
|
db.add(row)
|
|
db.commit()
|
|
db.refresh(row)
|
|
|
|
return RedirectResponse(url='/services/categories', status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
@router.get('/categories/{category_id}/edit')
|
|
def category_edit_page(request: Request, category_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
require_permission(db, user, 'services.edit')
|
|
if not _is_system_admin(db, user):
|
|
return _redirect_denied()
|
|
row = get_category(db, category_id)
|
|
if not row:
|
|
return RedirectResponse(url='/services/categories', status_code=303)
|
|
return _render(request, 'modules/services/templates/services/category_form.html', db, user, title='Edit Service Category', mode='edit', category=row)
|
|
except Exception:
|
|
return _redirect_denied()
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post('/categories/{category_id}/edit')
|
|
def category_edit_submit(request: Request, category_id: int, name: str = Form(...), sort_order: int = Form(100), is_active: str | None = Form(None), csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
require_permission(db, user, 'services.edit')
|
|
if not _is_system_admin(db, user):
|
|
return _redirect_denied()
|
|
row = get_category(db, category_id)
|
|
if not row:
|
|
return RedirectResponse(url='/services/categories', status_code=303)
|
|
row.name = name.strip()
|
|
row.sort_order = sort_order
|
|
row.is_active = is_active is not None
|
|
db.commit()
|
|
return RedirectResponse(url='/services/categories', status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get('/catalogue')
|
|
def catalogue_list(request: Request, q: str = '', category_id: str = '', recurrence_type: str = '', engagement_type: str = '', page: int = 1, per_page: int = 20):
|
|
category_id_value: int | None = None
|
|
category_id_text = (category_id or '').strip()
|
|
if category_id_text:
|
|
try:
|
|
parsed_category_id = int(category_id_text)
|
|
if parsed_category_id > 0:
|
|
category_id_value = parsed_category_id
|
|
except (TypeError, ValueError):
|
|
category_id_value = None
|
|
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
if not _can_view_services(db, user):
|
|
return _redirect_denied()
|
|
payload = list_catalogue_payload(db, q=q, category_id=category_id_value, recurrence_type=recurrence_type, engagement_type=engagement_type, page=page, per_page=per_page)
|
|
tenant_id = _active_tenant_id(request, user)
|
|
can_create_catalogue = _is_system_admin(db, user) and _has_perm(db, user, 'services.create')
|
|
can_manage_firm_services = _can_manage_firm_services(db, user)
|
|
|
|
catalogue_ids = [row.id for row in payload.get('rows', [])]
|
|
firm_selection_by_catalogue = {}
|
|
if catalogue_ids:
|
|
selections = db.execute(
|
|
select(FirmServiceSelection).where(
|
|
FirmServiceSelection.tenant_id == tenant_id,
|
|
FirmServiceSelection.service_catalogue_id.in_(catalogue_ids),
|
|
)
|
|
).scalars().all()
|
|
firm_selection_by_catalogue = {selection.service_catalogue_id: selection for selection in selections}
|
|
|
|
branches = db.execute(
|
|
select(Branch)
|
|
.where(Branch.tenant_id == tenant_id, Branch.is_active.is_(True))
|
|
.order_by(Branch.name.asc())
|
|
).scalars().all()
|
|
|
|
return _render(
|
|
request,
|
|
'modules/services/templates/services/catalogue_list.html',
|
|
db,
|
|
user,
|
|
title='Service Catalogue',
|
|
can_create=can_create_catalogue,
|
|
can_manage_firm_services=can_manage_firm_services,
|
|
categories=list_categories(db),
|
|
branches=branches,
|
|
tenant_id=tenant_id,
|
|
firm_selection_by_catalogue=firm_selection_by_catalogue,
|
|
**payload,
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get('/catalogue/export')
|
|
def catalogue_export(request: Request):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
if not _is_system_admin(db, user):
|
|
return _redirect_denied()
|
|
require_permission(db, user, 'services.view')
|
|
|
|
workbook_bytes = build_service_catalogue_export(db)
|
|
filename = f"service_catalogue_export_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}.xlsx"
|
|
return StreamingResponse(
|
|
iter([workbook_bytes]),
|
|
media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
|
|
headers={'Content-Disposition': f'attachment; filename="{filename}"'},
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get('/catalogue/merge')
|
|
def catalogue_merge_page(request: Request, source_id: int | None = None, target_id: int | None = None):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
if not _is_system_admin(db, user):
|
|
return _redirect_denied()
|
|
require_permission(db, user, 'services.edit')
|
|
|
|
services = db.execute(
|
|
select(ServiceCatalogue)
|
|
.where(ServiceCatalogue.is_active.is_(True))
|
|
.order_by(ServiceCatalogue.service_code.asc())
|
|
).scalars().all()
|
|
preview = None
|
|
error = None
|
|
if source_id and target_id:
|
|
try:
|
|
preview = build_merge_preview(db, source_id=source_id, target_id=target_id)
|
|
except ValueError as exc:
|
|
error = str(exc)
|
|
|
|
return _render(
|
|
request,
|
|
'modules/services/templates/services/catalogue_merge.html',
|
|
db,
|
|
user,
|
|
title='Merge Service Catalogue',
|
|
services=services,
|
|
source_id=source_id,
|
|
target_id=target_id,
|
|
preview=preview,
|
|
error=error,
|
|
)
|
|
except Exception:
|
|
return _redirect_denied()
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post('/catalogue/merge')
|
|
def catalogue_merge_submit(
|
|
request: Request,
|
|
source_id: int = Form(...),
|
|
target_id: int = Form(...),
|
|
confirmation: str = Form(...),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
if not _is_system_admin(db, user):
|
|
return _redirect_denied()
|
|
require_permission(db, user, 'services.edit')
|
|
|
|
preview = build_merge_preview(db, source_id=source_id, target_id=target_id)
|
|
expected = f"MERGE {preview.source.service_code} INTO {preview.target.service_code}"
|
|
if confirmation.strip().upper() != expected.upper():
|
|
return RedirectResponse(
|
|
url=f'/services/catalogue/merge?source_id={source_id}&target_id={target_id}&error=confirmation',
|
|
status_code=303,
|
|
)
|
|
|
|
result = merge_service_catalogues(
|
|
db,
|
|
source_id=source_id,
|
|
target_id=target_id,
|
|
actor_user_id=user.id,
|
|
)
|
|
db.commit()
|
|
|
|
write_audit_log(
|
|
db,
|
|
action='services.catalogue.merged',
|
|
entity_type='service_catalogue',
|
|
entity_id=target_id,
|
|
entity_name=result['target_code'],
|
|
actor=user,
|
|
request=request,
|
|
details=result,
|
|
)
|
|
return RedirectResponse(url=f'/services/catalogue/{target_id}?merged=1', status_code=303)
|
|
except ValueError:
|
|
db.rollback()
|
|
return RedirectResponse(url='/services/catalogue/merge?error=invalid', status_code=303)
|
|
except Exception:
|
|
db.rollback()
|
|
raise
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get('/catalogue/new')
|
|
def catalogue_create_page(request: Request):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
if not _is_system_admin(db, user):
|
|
return _redirect_denied()
|
|
require_permission(db, user, 'services.create')
|
|
return _render(request, 'modules/services/templates/services/catalogue_form.html', db, user, title='Create Service Catalogue', mode='create', catalogue=None, categories=list_categories(db), **_registration_type_form_context(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'), service_scope_type: str = Form('client'), required_registration_type: str = Form(''), 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
|
|
try:
|
|
registration_type_code = _validated_registration_type_code(db, required_registration_type)
|
|
except ValueError:
|
|
return RedirectResponse(url='/services/catalogue/new?error=registration_type', status_code=303)
|
|
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),
|
|
service_scope_type=(service_scope_type or 'client').strip().lower(),
|
|
required_registration_type=registration_type_code,
|
|
sort_order=sort_order,
|
|
description=description.strip() or None,
|
|
applicable_individual=applicable_individual is not None,
|
|
applicable_proprietorship=applicable_proprietorship is not None,
|
|
applicable_partnership=applicable_partnership is not None,
|
|
applicable_llp=applicable_llp is not None,
|
|
applicable_company=applicable_company is not None,
|
|
applicable_trust=applicable_trust is not None,
|
|
applicable_society=applicable_society is not None,
|
|
is_active=is_active is not None,
|
|
is_client_requestable=is_client_requestable is not None,
|
|
is_consultant_requestable=is_consultant_requestable is not None,
|
|
created_by_user_id=user.id,
|
|
updated_by_user_id=user.id,
|
|
)
|
|
db.add(row); db.commit(); db.refresh(row)
|
|
return RedirectResponse(url=f'/services/catalogue/{row.id}', status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get('/catalogue/{catalogue_id}')
|
|
def catalogue_detail(request: Request, catalogue_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
if not _can_view_services(db, user):
|
|
return _redirect_denied()
|
|
row = get_catalogue(db, catalogue_id)
|
|
if not row or not row.is_active:
|
|
return RedirectResponse(url='/services/catalogue', status_code=303)
|
|
tenant_id = _active_tenant_id(request, user)
|
|
can_edit = _is_system_admin(db, user) and _has_perm(db, user, 'services.edit')
|
|
branches = db.execute(
|
|
select(Branch)
|
|
.where(Branch.tenant_id == tenant_id, Branch.is_active.is_(True))
|
|
.order_by(Branch.name.asc())
|
|
).scalars().all()
|
|
return _render(
|
|
request,
|
|
'modules/services/templates/services/catalogue_detail.html',
|
|
db,
|
|
user,
|
|
title=f'Service Catalogue - {row.service_name}',
|
|
catalogue=row,
|
|
can_edit=can_edit,
|
|
can_manage_firm_services=_can_manage_firm_services(db, user),
|
|
can_manage_firm_tasks=_can_manage_firm_tasks(db, user),
|
|
is_system_admin=_is_system_admin(db, user),
|
|
task_categories=list_task_categories(db, catalogue_id=catalogue_id, tenant_id=tenant_id, include_inactive=True),
|
|
branches=branches,
|
|
current_selection=get_firm_selection(db, tenant_id=tenant_id, catalogue_id=row.id),
|
|
current_templates=get_firm_task_templates(db, tenant_id=tenant_id, catalogue_id=row.id),
|
|
default_templates=get_default_task_templates(db, catalogue_id=row.id),
|
|
due_rules=list_due_rules(db, catalogue_id=row.id),
|
|
due_extensions=list_due_extensions(db, catalogue_id=row.id, tenant_id=tenant_id),
|
|
tenant_id=tenant_id,
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get('/catalogue/{catalogue_id}/due-rules/new')
|
|
def due_rule_create_page(request: Request, catalogue_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
if not _is_system_admin(db, user):
|
|
return _redirect_denied()
|
|
require_permission(db, user, 'services.edit')
|
|
catalogue = get_catalogue(db, catalogue_id)
|
|
if not catalogue:
|
|
return RedirectResponse(url='/services/catalogue', status_code=303)
|
|
return _render(
|
|
request,
|
|
'modules/services/templates/services/due_dates/rule_form.html',
|
|
db,
|
|
user,
|
|
title='Add Due Date Rule',
|
|
mode='create',
|
|
catalogue=catalogue,
|
|
rule=None,
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post('/catalogue/{catalogue_id}/due-rules/new')
|
|
def due_rule_create_submit(
|
|
request: Request,
|
|
catalogue_id: int,
|
|
rule_name: str = Form(...),
|
|
period_type: str = Form('yearly'),
|
|
due_year_basis: str = Form('assessment_year_start'),
|
|
due_day: str = Form(''),
|
|
due_month: str = Form(''),
|
|
due_month_offset: int = Form(0),
|
|
days_offset_after_event: str = Form(''),
|
|
renewal_days_before_expiry: str = Form(''),
|
|
sort_order: int = Form(100),
|
|
is_active: str | None = Form(None),
|
|
remarks: str = Form(''),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
if not _is_system_admin(db, user):
|
|
return _redirect_denied()
|
|
require_permission(db, user, 'services.edit')
|
|
catalogue = get_catalogue(db, catalogue_id)
|
|
if not catalogue:
|
|
return RedirectResponse(url='/services/catalogue', status_code=303)
|
|
row = ServiceDueDateRule(
|
|
service_catalogue_id=catalogue_id,
|
|
rule_name=rule_name.strip(),
|
|
period_type=(period_type or 'yearly').strip(),
|
|
due_year_basis=(due_year_basis or 'assessment_year_start').strip(),
|
|
due_day=int(due_day) if str(due_day).strip() else None,
|
|
due_month=int(due_month) if str(due_month).strip() else None,
|
|
due_month_offset=int(due_month_offset or 0),
|
|
days_offset_after_event=int(days_offset_after_event) if str(days_offset_after_event).strip() else None,
|
|
renewal_days_before_expiry=int(renewal_days_before_expiry) if str(renewal_days_before_expiry).strip() else None,
|
|
sort_order=sort_order,
|
|
is_active=is_active is not None,
|
|
remarks=remarks.strip() or None,
|
|
created_by_user_id=user.id,
|
|
updated_by_user_id=user.id,
|
|
)
|
|
db.add(row)
|
|
db.commit()
|
|
return RedirectResponse(url=f'/services/catalogue/{catalogue_id}', status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get('/catalogue/{catalogue_id}/due-rules/{rule_id}/edit')
|
|
def due_rule_edit_page(request: Request, catalogue_id: int, rule_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
if not _is_system_admin(db, user):
|
|
return _redirect_denied()
|
|
require_permission(db, user, 'services.edit')
|
|
catalogue = get_catalogue(db, catalogue_id)
|
|
rule = get_due_rule(db, rule_id=rule_id, catalogue_id=catalogue_id)
|
|
if not catalogue or not rule:
|
|
return RedirectResponse(url=f'/services/catalogue/{catalogue_id}', status_code=303)
|
|
return _render(
|
|
request,
|
|
'modules/services/templates/services/due_dates/rule_form.html',
|
|
db,
|
|
user,
|
|
title='Edit Due Date Rule',
|
|
mode='edit',
|
|
catalogue=catalogue,
|
|
rule=rule,
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post('/catalogue/{catalogue_id}/due-rules/{rule_id}/edit')
|
|
def due_rule_edit_submit(
|
|
request: Request,
|
|
catalogue_id: int,
|
|
rule_id: int,
|
|
rule_name: str = Form(...),
|
|
period_type: str = Form('yearly'),
|
|
due_year_basis: str = Form('assessment_year_start'),
|
|
due_day: str = Form(''),
|
|
due_month: str = Form(''),
|
|
due_month_offset: int = Form(0),
|
|
days_offset_after_event: str = Form(''),
|
|
renewal_days_before_expiry: str = Form(''),
|
|
sort_order: int = Form(100),
|
|
is_active: str | None = Form(None),
|
|
remarks: str = Form(''),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
if not _is_system_admin(db, user):
|
|
return _redirect_denied()
|
|
require_permission(db, user, 'services.edit')
|
|
rule = get_due_rule(db, rule_id=rule_id, catalogue_id=catalogue_id)
|
|
if not rule:
|
|
return RedirectResponse(url=f'/services/catalogue/{catalogue_id}', status_code=303)
|
|
rule.rule_name = rule_name.strip()
|
|
rule.period_type = (period_type or 'yearly').strip()
|
|
rule.due_year_basis = (due_year_basis or 'assessment_year_start').strip()
|
|
rule.due_day = int(due_day) if str(due_day).strip() else None
|
|
rule.due_month = int(due_month) if str(due_month).strip() else None
|
|
rule.due_month_offset = int(due_month_offset or 0)
|
|
rule.days_offset_after_event = int(days_offset_after_event) if str(days_offset_after_event).strip() else None
|
|
rule.renewal_days_before_expiry = int(renewal_days_before_expiry) if str(renewal_days_before_expiry).strip() else None
|
|
rule.sort_order = sort_order
|
|
rule.is_active = is_active is not None
|
|
rule.remarks = remarks.strip() or None
|
|
rule.updated_by_user_id = user.id
|
|
db.commit()
|
|
return RedirectResponse(url=f'/services/catalogue/{catalogue_id}', status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get('/catalogue/{catalogue_id}/due-extensions/new')
|
|
def due_extension_create_page(request: Request, catalogue_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
if not _is_system_admin(db, user):
|
|
return _redirect_denied()
|
|
require_permission(db, user, 'services.edit')
|
|
catalogue = get_catalogue(db, catalogue_id)
|
|
if not catalogue:
|
|
return RedirectResponse(url='/services/catalogue', status_code=303)
|
|
return _render(
|
|
request,
|
|
'modules/services/templates/services/due_dates/extension_form.html',
|
|
db,
|
|
user,
|
|
title='Add Due Date Extension',
|
|
catalogue=catalogue,
|
|
rules=list_due_rules(db, catalogue_id=catalogue_id, include_inactive=False),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post('/catalogue/{catalogue_id}/due-extensions/new')
|
|
def due_extension_create_submit(
|
|
request: Request,
|
|
catalogue_id: int,
|
|
due_date_rule_id: str = Form(''),
|
|
financial_year: str = Form(...),
|
|
assessment_year: str = Form(''),
|
|
period_label: str = Form(''),
|
|
extended_due_date: str = Form(...),
|
|
notification_reference: str = Form(''),
|
|
notification_date: str = Form(''),
|
|
remarks: str = Form(''),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
if not _is_system_admin(db, user):
|
|
return _redirect_denied()
|
|
require_permission(db, user, 'services.edit')
|
|
catalogue = get_catalogue(db, catalogue_id)
|
|
if not catalogue:
|
|
return RedirectResponse(url='/services/catalogue', status_code=303)
|
|
tenant_id = _active_tenant_id(request, user)
|
|
create_due_date_extension(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
catalogue_id=catalogue_id,
|
|
due_date_rule_id=int(due_date_rule_id) if str(due_date_rule_id).strip() else None,
|
|
financial_year=financial_year.strip(),
|
|
assessment_year=assessment_year.strip() or None,
|
|
period_label=period_label.strip() or None,
|
|
extended_due_date=parse_optional_date(extended_due_date),
|
|
notification_reference=notification_reference,
|
|
notification_date=parse_optional_date(notification_date),
|
|
remarks=remarks,
|
|
user_id=user.id,
|
|
)
|
|
db.commit()
|
|
return RedirectResponse(url=f'/services/catalogue/{catalogue_id}', status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get('/catalogue/{catalogue_id}/edit')
|
|
def catalogue_edit_page(request: Request, catalogue_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
if not _is_system_admin(db, user):
|
|
return _redirect_denied()
|
|
require_permission(db, user, 'services.edit')
|
|
row = get_catalogue(db, catalogue_id)
|
|
if not row or not row.is_active:
|
|
return RedirectResponse(url='/services/catalogue', status_code=303)
|
|
return _render(request, 'modules/services/templates/services/catalogue_form.html', db, user, title='Edit Service Catalogue', mode='edit', catalogue=row, categories=list_categories(db), **_registration_type_form_context(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'), service_scope_type: str = Form('client'), required_registration_type: str = Form(''), sort_order: int = Form(100), description: str = Form(''), applicable_individual: str | None = Form(None), applicable_proprietorship: str | None = Form(None), applicable_partnership: str | None = Form(None), applicable_llp: str | None = Form(None), applicable_company: str | None = Form(None), applicable_trust: str | None = Form(None), applicable_society: str | None = Form(None), is_active: str | None = Form(None), is_client_requestable: str | None = Form(None), is_consultant_requestable: str | None = Form(None), csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
if not _is_system_admin(db, user):
|
|
return _redirect_denied()
|
|
require_permission(db, user, 'services.edit')
|
|
row = get_catalogue(db, catalogue_id)
|
|
if not row or not row.is_active:
|
|
return RedirectResponse(url='/services/catalogue', status_code=303)
|
|
selected_category = get_category(db, int(category_id)) if str(category_id).strip() else None
|
|
try:
|
|
registration_type_code = _validated_registration_type_code(
|
|
db, required_registration_type, existing_value=row.required_registration_type
|
|
)
|
|
except ValueError:
|
|
return RedirectResponse(
|
|
url=f'/services/catalogue/{catalogue_id}/edit?error=registration_type',
|
|
status_code=303,
|
|
)
|
|
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.service_scope_type = (service_scope_type or 'client').strip().lower()
|
|
row.required_registration_type = registration_type_code
|
|
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()
|
|
|
|
|
|
def _copy_system_defaults_if_firm_tasks_empty(
|
|
db,
|
|
*,
|
|
tenant_id: int,
|
|
catalogue_id: int,
|
|
user_id: int,
|
|
) -> tuple[int, bool]:
|
|
"""Initialise/synchronise inherited firm tasks from the current system defaults."""
|
|
existing_tasks = get_firm_task_templates(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
catalogue_id=catalogue_id,
|
|
)
|
|
had_existing = bool(existing_tasks)
|
|
result = sync_firm_tasks_from_system_defaults(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
service_catalogue_id=catalogue_id,
|
|
updated_by_user_id=user_id,
|
|
sync_open_engagements=False,
|
|
)
|
|
return result.created, had_existing
|
|
|
|
|
|
@router.post('/catalogue/bulk-selection')
|
|
def bulk_update_firm_services(
|
|
request: Request,
|
|
service_ids: list[int] = Form([]),
|
|
bulk_action: str = Form(...),
|
|
default_branch_id: str = Form(''),
|
|
return_q: str = Form(''),
|
|
return_category_id: str = Form(''),
|
|
return_recurrence_type: str = Form(''),
|
|
return_engagement_type: str = Form(''),
|
|
return_page: int = Form(1),
|
|
return_per_page: int = Form(20),
|
|
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)
|
|
action = (bulk_action or '').strip().lower()
|
|
if action not in {'enable', 'disable'}:
|
|
action = 'invalid'
|
|
|
|
unique_service_ids = list(dict.fromkeys(service_id for service_id in service_ids if service_id > 0))
|
|
branch_id: int | None = None
|
|
branch_value = str(default_branch_id or '').strip()
|
|
if branch_value:
|
|
try:
|
|
parsed_branch_id = int(branch_value)
|
|
except (TypeError, ValueError):
|
|
parsed_branch_id = 0
|
|
branch = db.execute(
|
|
select(Branch).where(
|
|
Branch.id == parsed_branch_id,
|
|
Branch.tenant_id == tenant_id,
|
|
Branch.is_active.is_(True),
|
|
)
|
|
).scalar_one_or_none()
|
|
if not branch:
|
|
action = 'invalid_branch'
|
|
else:
|
|
branch_id = branch.id
|
|
|
|
updated_count = 0
|
|
unchanged_count = 0
|
|
skipped_count = 0
|
|
tasks_copied_count = 0
|
|
services_with_existing_tasks = 0
|
|
services_without_defaults = 0
|
|
|
|
if action in {'enable', 'disable'} and unique_service_ids:
|
|
catalogue_rows = db.execute(
|
|
select(ServiceCatalogue).where(
|
|
ServiceCatalogue.id.in_(unique_service_ids),
|
|
ServiceCatalogue.is_active.is_(True),
|
|
)
|
|
).scalars().all()
|
|
catalogue_by_id = {row.id: row for row in catalogue_rows}
|
|
|
|
selections = db.execute(
|
|
select(FirmServiceSelection).where(
|
|
FirmServiceSelection.tenant_id == tenant_id,
|
|
FirmServiceSelection.service_catalogue_id.in_(unique_service_ids),
|
|
)
|
|
).scalars().all()
|
|
selection_by_catalogue = {
|
|
selection.service_catalogue_id: selection for selection in selections
|
|
}
|
|
|
|
for catalogue_id in unique_service_ids:
|
|
if catalogue_id not in catalogue_by_id:
|
|
skipped_count += 1
|
|
continue
|
|
|
|
selection = selection_by_catalogue.get(catalogue_id)
|
|
if action == 'enable':
|
|
if selection is None:
|
|
selection = FirmServiceSelection(
|
|
tenant_id=tenant_id,
|
|
service_catalogue_id=catalogue_id,
|
|
is_enabled=True,
|
|
default_branch_id=branch_id,
|
|
activated_by_user_id=user.id,
|
|
updated_by_user_id=user.id,
|
|
)
|
|
db.add(selection)
|
|
updated_count += 1
|
|
else:
|
|
changed = not bool(selection.is_enabled)
|
|
selection.is_enabled = True
|
|
if branch_id is not None and selection.default_branch_id != branch_id:
|
|
selection.default_branch_id = branch_id
|
|
changed = True
|
|
selection.updated_by_user_id = user.id
|
|
if changed:
|
|
updated_count += 1
|
|
else:
|
|
unchanged_count += 1
|
|
|
|
copied_count, had_existing_tasks = _copy_system_defaults_if_firm_tasks_empty(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
catalogue_id=catalogue_id,
|
|
user_id=user.id,
|
|
)
|
|
tasks_copied_count += copied_count
|
|
if had_existing_tasks:
|
|
services_with_existing_tasks += 1
|
|
elif copied_count == 0:
|
|
services_without_defaults += 1
|
|
else:
|
|
if selection is not None and selection.is_enabled:
|
|
selection.is_enabled = False
|
|
selection.updated_by_user_id = user.id
|
|
updated_count += 1
|
|
else:
|
|
unchanged_count += 1
|
|
|
|
db.commit()
|
|
elif not unique_service_ids:
|
|
action = 'no_selection'
|
|
|
|
query_parts = [
|
|
f'bulk_action={action}',
|
|
f'bulk_updated={updated_count}',
|
|
f'bulk_unchanged={unchanged_count}',
|
|
f'bulk_skipped={skipped_count}',
|
|
f'tasks_copied={tasks_copied_count}',
|
|
f'services_existing_tasks={services_with_existing_tasks}',
|
|
f'services_without_defaults={services_without_defaults}',
|
|
f'page={max(1, return_page)}',
|
|
f'per_page={return_per_page if return_per_page in (20, 50, 100) else 20}',
|
|
]
|
|
if return_q:
|
|
from urllib.parse import quote_plus
|
|
query_parts.append(f'q={quote_plus(return_q)}')
|
|
if return_category_id:
|
|
query_parts.append(f'category_id={return_category_id}')
|
|
if return_recurrence_type:
|
|
query_parts.append(f'recurrence_type={return_recurrence_type}')
|
|
if return_engagement_type:
|
|
query_parts.append(f'engagement_type={return_engagement_type}')
|
|
|
|
return RedirectResponse(
|
|
url='/services/catalogue?' + '&'.join(query_parts),
|
|
status_code=303,
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post('/catalogue/{catalogue_id}/toggle')
|
|
def toggle_firm_service(request: Request, catalogue_id: int, default_branch_id: str = Form(''), csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
if not _can_manage_firm_services(db, user):
|
|
return _redirect_denied()
|
|
tenant_id = _active_tenant_id(request, user)
|
|
row = get_catalogue(db, catalogue_id)
|
|
if not row or not row.is_active:
|
|
return RedirectResponse(url='/services/catalogue', status_code=303)
|
|
selection = get_firm_selection(db, tenant_id=tenant_id, catalogue_id=catalogue_id)
|
|
enabling = selection is None or not bool(selection.is_enabled)
|
|
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
|
|
|
|
copied_count = 0
|
|
had_existing_tasks = False
|
|
if enabling:
|
|
copied_count, had_existing_tasks = _copy_system_defaults_if_firm_tasks_empty(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
catalogue_id=catalogue_id,
|
|
user_id=user.id,
|
|
)
|
|
|
|
db.commit()
|
|
if enabling:
|
|
return RedirectResponse(
|
|
url=(
|
|
f'/services/catalogue?service_action=enabled'
|
|
f'&tasks_copied={copied_count}'
|
|
f'&existing_tasks={1 if had_existing_tasks else 0}'
|
|
),
|
|
status_code=303,
|
|
)
|
|
return RedirectResponse(url='/services/catalogue?service_action=disabled', 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)
|
|
system_update_diff_map = {}
|
|
for firm_task in tasks:
|
|
if getattr(firm_task, "system_update_available", False) and getattr(firm_task, "source_system_task_id", None):
|
|
system_task = db.get(ServiceDefaultTaskTemplate, int(firm_task.source_system_task_id))
|
|
if system_task is not None:
|
|
system_update_diff_map[firm_task.id] = system_task_diff(system_task, firm_task)
|
|
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,
|
|
system_update_diff_map=system_update_diff_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),
|
|
task_categories=list_task_categories(db, catalogue_id=catalogue_id, tenant_id=tenant_id, include_inactive=True),
|
|
)
|
|
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(''), eligible_roles: list[str] = Form([]), sequence_no: int | None = Form(None), is_mandatory: str | None = Form(None), requires_review: str | None = Form(None), normal_review_role: str = Form('manager'), task_category: str = Form(''), response_required: str | None = Form(None), response_type: str = Form('NONE'), task_tool_code: str = Form('NONE'), evidence_required: str | None = Form(None), remarks_required_if_no: str | None = Form(None), is_aqmm_task: str | None = Form(None), aqmm_mandatory: str | None = Form(None), aqmm_evidence_required: str | None = Form(None), aqmm_manager_review_required: str | None = Form(None), aqmm_partner_review_required: str | None = Form(None), aqmm_review_partner_required: str | None = Form(None), aqmm_blocks_final_release: str | None = Form(None), aqmm_reference: str = Form(''), is_active: str | None = Form(None), csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
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,
|
|
eligible_role_names=serialise_eligible_task_roles(eligible_roles, default_role_name),
|
|
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,
|
|
normal_review_role=_normal_review_role_value(normal_review_role, requires_review=requires_review is not None),
|
|
task_category_id=(category.id if (category := ensure_task_category(db, catalogue_id=catalogue_id, tenant_id=tenant_id, name=task_category, user_id=user.id)) else None),
|
|
task_category=(category.name if category else None),
|
|
response_required=response_required is not None,
|
|
response_type=(response_type.strip().upper() or "NONE"),
|
|
task_tool_code=normalize_task_tool_code(task_tool_code),
|
|
evidence_required=evidence_required is not None,
|
|
remarks_required_if_no=remarks_required_if_no is not None,
|
|
is_aqmm_task=is_aqmm_task is not None,
|
|
aqmm_mandatory=aqmm_mandatory is not None,
|
|
aqmm_evidence_required=aqmm_evidence_required is not None,
|
|
aqmm_manager_review_required=aqmm_manager_review_required is not None,
|
|
aqmm_partner_review_required=aqmm_partner_review_required is not None,
|
|
aqmm_review_partner_required=aqmm_review_partner_required is not None,
|
|
aqmm_blocks_final_release=aqmm_blocks_final_release is not None,
|
|
aqmm_reference=aqmm_reference.strip() or None,
|
|
is_active=is_active is not None,
|
|
is_customized=True,
|
|
created_by_user_id=user.id,
|
|
updated_by_user_id=user.id,
|
|
)
|
|
db.add(row); db.commit()
|
|
return RedirectResponse(url=f'/services/templates/{catalogue_id}', status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get('/defaults')
|
|
def default_templates_catalogue_list(request: Request):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
require_permission(db, user, 'service_tasks.view')
|
|
if not _is_system_admin(db, user):
|
|
return _redirect_denied()
|
|
payload = list_catalogue_payload(db, page=1, per_page=500)
|
|
rows = payload['rows']
|
|
return _render(request, 'modules/services/templates/services/default_templates_list.html', db, user, title='Default Task Templates', rows=rows, can_manage_defaults=_has_perm(db, user, 'service_tasks.create'))
|
|
except Exception:
|
|
return _redirect_denied()
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get('/defaults/export')
|
|
def default_tasks_export(request: Request):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
require_permission(db, user, 'service_tasks.view')
|
|
if not _is_system_admin(db, user):
|
|
return _redirect_denied()
|
|
|
|
workbook_bytes = build_default_tasks_export(db)
|
|
filename = f"system_default_tasks_export_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}.xlsx"
|
|
return StreamingResponse(
|
|
iter([workbook_bytes]),
|
|
media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
|
|
headers={'Content-Disposition': f'attachment; filename="{filename}"'},
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get('/catalogue/{catalogue_id}/defaults')
|
|
def default_templates_detail(request: Request, catalogue_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
require_permission(db, user, 'service_tasks.view')
|
|
if not _is_system_admin(db, user):
|
|
return _redirect_denied()
|
|
catalogue = get_catalogue(db, catalogue_id)
|
|
if not catalogue:
|
|
return RedirectResponse(url='/services/defaults', status_code=303)
|
|
defaults = get_default_task_templates(db, catalogue_id=catalogue_id)
|
|
return _render(request, 'modules/services/templates/services/default_templates_detail.html', db, user, title=f'Default Tasks - {catalogue.service_name}', service=catalogue, default_tasks=defaults, task_categories=list_task_categories(db, catalogue_id=catalogue_id, tenant_id=None, include_inactive=True))
|
|
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), normal_review_role: str = Form('manager'), task_category: str = Form(''), response_required: str | None = Form(None), response_type: str = Form('NONE'), task_tool_code: str = Form('NONE'), evidence_required: str | None = Form(None), remarks_required_if_no: str | None = Form(None), is_aqmm_task: str | None = Form(None), aqmm_mandatory: str | None = Form(None), aqmm_evidence_required: str | None = Form(None), aqmm_manager_review_required: str | None = Form(None), aqmm_partner_review_required: str | None = Form(None), aqmm_review_partner_required: str | None = Form(None), aqmm_blocks_final_release: str | None = Form(None), aqmm_reference: str = Form(''), is_active: str | None = Form(None), csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
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, normal_review_role=_normal_review_role_value(normal_review_role, requires_review=requires_review is not None), task_category_id=(category.id if (category := ensure_task_category(db, catalogue_id=catalogue_id, tenant_id=None, name=task_category, user_id=user.id)) else None), task_category=(category.name if category else None), response_required=response_required is not None, response_type=(response_type.strip().upper() or "NONE"), task_tool_code=normalize_task_tool_code(task_tool_code), evidence_required=evidence_required is not None, remarks_required_if_no=remarks_required_if_no is not None, is_aqmm_task=is_aqmm_task is not None, aqmm_mandatory=aqmm_mandatory is not None, aqmm_evidence_required=aqmm_evidence_required is not None, aqmm_manager_review_required=aqmm_manager_review_required is not None, aqmm_partner_review_required=aqmm_partner_review_required is not None, aqmm_review_partner_required=aqmm_review_partner_required is not None, aqmm_blocks_final_release=aqmm_blocks_final_release is not None, aqmm_reference=aqmm_reference.strip() or None, is_active=is_active is not None)
|
|
db.add(row)
|
|
db.flush()
|
|
sync_system_defaults_to_all_firms(
|
|
db,
|
|
service_catalogue_id=catalogue_id,
|
|
updated_by_user_id=user.id,
|
|
sync_open_engagements=True,
|
|
)
|
|
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)
|
|
result = sync_firm_tasks_from_system_defaults(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
service_catalogue_id=catalogue_id,
|
|
updated_by_user_id=user.id,
|
|
sync_open_engagements=True,
|
|
)
|
|
db.commit()
|
|
return RedirectResponse(
|
|
url=(
|
|
f'/services/templates/{catalogue_id}?sync_success=1'
|
|
f'&created={result.created}&updated={result.updated}'
|
|
f'&custom_updates={result.custom_updates_available}'
|
|
),
|
|
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.post('/templates/{catalogue_id}/tasks/{task_id}/toggle-active')
|
|
def firm_task_template_toggle_active(
|
|
request: Request,
|
|
catalogue_id: int,
|
|
task_id: int,
|
|
csrf_token: str = Form(...),
|
|
):
|
|
"""Enable/disable a firm task template without changing the system default.
|
|
|
|
Existing engagement task instances are snapshots and are intentionally not
|
|
modified. This affects task generation for future engagements only.
|
|
"""
|
|
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)
|
|
|
|
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}?error=task_missing',
|
|
status_code=303,
|
|
)
|
|
|
|
task.is_active = not bool(task.is_active)
|
|
mark_firm_task_customized(task)
|
|
task.updated_by_user_id = user.id
|
|
db.commit()
|
|
|
|
state = 'enabled' if task.is_active else 'disabled'
|
|
return RedirectResponse(
|
|
url=f'/services/templates/{catalogue_id}?task_{state}=1',
|
|
status_code=303,
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post('/templates/{catalogue_id}/tasks/{task_id}/system-update/upgrade')
|
|
def firm_task_accept_system_update(
|
|
request: Request,
|
|
catalogue_id: int,
|
|
task_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, tenant_id=tenant_id, catalogue_id=catalogue_id, task_id=task_id)
|
|
if not task or not getattr(task, 'system_update_available', False):
|
|
return RedirectResponse(url=f'/services/templates/{catalogue_id}?error=system_update_missing', status_code=303)
|
|
result = accept_system_update_for_firm_task(db, task=task, user_id=user.id, sync_open_engagements=True)
|
|
db.commit()
|
|
return RedirectResponse(
|
|
url=(f'/services/templates/{catalogue_id}?system_upgraded=1'
|
|
f'&engagement_created={result.engagement_created}'
|
|
f'&engagement_updated={result.engagement_updated_pending}'
|
|
f'&engagement_deactivated={result.engagement_deactivated_pending}'),
|
|
status_code=303,
|
|
)
|
|
except ValueError:
|
|
db.rollback()
|
|
return RedirectResponse(url=f'/services/templates/{catalogue_id}?error=system_update_missing', status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post('/templates/{catalogue_id}/tasks/{task_id}/system-update/keep-firm')
|
|
def firm_task_keep_customization(
|
|
request: Request,
|
|
catalogue_id: int,
|
|
task_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, tenant_id=tenant_id, catalogue_id=catalogue_id, task_id=task_id)
|
|
if not task or not getattr(task, 'system_update_available', False):
|
|
return RedirectResponse(url=f'/services/templates/{catalogue_id}?error=system_update_missing', status_code=303)
|
|
keep_firm_customization_for_system_revision(db, task=task, user_id=user.id)
|
|
db.commit()
|
|
return RedirectResponse(url=f'/services/templates/{catalogue_id}?system_kept=1', status_code=303)
|
|
except ValueError:
|
|
db.rollback()
|
|
return RedirectResponse(url=f'/services/templates/{catalogue_id}?error=system_update_missing', status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post('/templates/{catalogue_id}/sync-open-engagements')
|
|
def sync_firm_tasks_to_open_engagements(
|
|
request: Request,
|
|
catalogue_id: int,
|
|
include_started_open_tasks: 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)
|
|
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)
|
|
|
|
result = sync_open_engagement_tasks_for_service(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
catalogue_id=catalogue_id,
|
|
user_id=user.id,
|
|
include_started_open_tasks=include_started_open_tasks is not None,
|
|
)
|
|
db.commit()
|
|
return RedirectResponse(
|
|
url=(
|
|
f'/services/templates/{catalogue_id}'
|
|
f'?synced=1&created={result["created"]}'
|
|
f'&updated_pending={result["updated_pending"]}'
|
|
f'&updated_started={result["updated_started"]}'
|
|
f'&preserved={result["preserved_history"]}'
|
|
),
|
|
status_code=303,
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post('/templates/{catalogue_id}/tasks/{task_id}/move')
|
|
def firm_task_template_move(
|
|
request: Request,
|
|
catalogue_id: int,
|
|
task_id: int,
|
|
direction: str = Form(...),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
"""Move one firm task one position up/down without changing task identity."""
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
if not _can_manage_firm_tasks(db, user):
|
|
return _redirect_denied()
|
|
|
|
tenant_id = _active_tenant_id(request, user)
|
|
catalogue = get_catalogue(db, catalogue_id)
|
|
if not catalogue:
|
|
return RedirectResponse(url='/services/templates', status_code=303)
|
|
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)
|
|
index = next((i for i, row in enumerate(tasks) if int(row.id) == int(task_id)), None)
|
|
direction_value = (direction or '').strip().lower()
|
|
if index is None or direction_value not in {'up', 'down'}:
|
|
return RedirectResponse(url=f'/services/templates/{catalogue_id}?move_error=1', status_code=303)
|
|
|
|
other_index = index - 1 if direction_value == 'up' else index + 1
|
|
if other_index < 0 or other_index >= len(tasks):
|
|
return RedirectResponse(url=f'/services/templates/{catalogue_id}', status_code=303)
|
|
|
|
current = tasks[index]
|
|
other = tasks[other_index]
|
|
current_seq = int(current.sequence_no)
|
|
other_seq = int(other.sequence_no)
|
|
|
|
# Swap safely across the existing unique tenant/service/sequence constraint.
|
|
min_seq = min(int(row.sequence_no) for row in tasks)
|
|
mark_firm_task_customized(current)
|
|
mark_firm_task_customized(other)
|
|
current.sequence_no = min_seq - 1000000 - int(current.id)
|
|
current.updated_by_user_id = user.id
|
|
db.flush()
|
|
|
|
other.sequence_no = current_seq
|
|
other.updated_by_user_id = user.id
|
|
db.flush()
|
|
|
|
current.sequence_no = other_seq
|
|
db.commit()
|
|
return RedirectResponse(url=f'/services/templates/{catalogue_id}?moved=1', status_code=303)
|
|
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,
|
|
task_categories=list_task_categories(db, catalogue_id=catalogue_id, tenant_id=tenant_id, include_inactive=True),
|
|
)
|
|
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(''),
|
|
eligible_roles: list[str] = Form([]),
|
|
sequence_no: int = Form(1),
|
|
is_mandatory: str | None = Form(None),
|
|
requires_review: str | None = Form(None),
|
|
normal_review_role: str = Form('manager'),
|
|
task_category: str = Form(''),
|
|
response_required: str | None = Form(None),
|
|
response_type: str = Form('NONE'),
|
|
task_tool_code: str = Form('NONE'),
|
|
evidence_required: str | None = Form(None),
|
|
remarks_required_if_no: str | None = Form(None),
|
|
is_aqmm_task: str | None = Form(None),
|
|
aqmm_mandatory: str | None = Form(None),
|
|
aqmm_evidence_required: str | None = Form(None),
|
|
aqmm_manager_review_required: str | None = Form(None),
|
|
aqmm_partner_review_required: str | None = Form(None),
|
|
aqmm_review_partner_required: str | None = Form(None),
|
|
aqmm_blocks_final_release: str | None = Form(None),
|
|
aqmm_reference: str = Form(''),
|
|
is_active: str | None = Form(None),
|
|
sync_open_engagements: str | None = Form(None),
|
|
sync_started_open_tasks: 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.eligible_role_names = serialise_eligible_task_roles(eligible_roles, default_role_name)
|
|
task.sequence_no = sequence_no
|
|
task.is_mandatory = is_mandatory is not None
|
|
task.requires_review = requires_review is not None
|
|
task.normal_review_role = _normal_review_role_value(normal_review_role, requires_review=task.requires_review)
|
|
category = ensure_task_category(db, catalogue_id=catalogue_id, tenant_id=tenant_id, name=task_category, user_id=user.id)
|
|
task.task_category_id = category.id if category else None
|
|
task.task_category = category.name if category else None
|
|
task.response_required = response_required is not None
|
|
task.response_type = response_type.strip().upper() or "NONE"
|
|
task.task_tool_code = normalize_task_tool_code(task_tool_code)
|
|
task.evidence_required = evidence_required is not None
|
|
task.remarks_required_if_no = remarks_required_if_no is not None
|
|
task.is_aqmm_task = is_aqmm_task is not None
|
|
task.aqmm_mandatory = aqmm_mandatory is not None
|
|
task.aqmm_evidence_required = aqmm_evidence_required is not None
|
|
task.aqmm_manager_review_required = aqmm_manager_review_required is not None
|
|
task.aqmm_partner_review_required = aqmm_partner_review_required is not None
|
|
task.aqmm_review_partner_required = aqmm_review_partner_required is not None
|
|
task.aqmm_blocks_final_release = aqmm_blocks_final_release is not None
|
|
task.aqmm_reference = aqmm_reference.strip() or None
|
|
task.is_active = is_active is not None
|
|
mark_firm_task_customized(task)
|
|
task.updated_by_user_id = user.id
|
|
|
|
sync_result = None
|
|
if sync_open_engagements is not None:
|
|
sync_result = sync_open_engagement_tasks_for_service(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
catalogue_id=catalogue_id,
|
|
user_id=user.id,
|
|
include_started_open_tasks=sync_started_open_tasks is not None,
|
|
)
|
|
|
|
db.commit()
|
|
if sync_result is not None:
|
|
return RedirectResponse(
|
|
url=(
|
|
f'/services/templates/{catalogue_id}'
|
|
f'?synced=1&created={sync_result["created"]}'
|
|
f'&updated_pending={sync_result["updated_pending"]}'
|
|
f'&updated_started={sync_result["updated_started"]}'
|
|
f'&preserved={sync_result["preserved_history"]}'
|
|
),
|
|
status_code=303,
|
|
)
|
|
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,
|
|
task_categories=list_task_categories(db, catalogue_id=catalogue_id, tenant_id=None, include_inactive=True),
|
|
)
|
|
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(''),
|
|
eligible_roles: list[str] = Form([]),
|
|
sequence_no: int = Form(1),
|
|
is_mandatory: str | None = Form(None),
|
|
requires_review: str | None = Form(None),
|
|
normal_review_role: str = Form('manager'),
|
|
task_category: str = Form(''),
|
|
response_required: str | None = Form(None),
|
|
response_type: str = Form('NONE'),
|
|
task_tool_code: str = Form('NONE'),
|
|
evidence_required: str | None = Form(None),
|
|
remarks_required_if_no: str | None = Form(None),
|
|
is_aqmm_task: str | None = Form(None),
|
|
aqmm_mandatory: str | None = Form(None),
|
|
aqmm_evidence_required: str | None = Form(None),
|
|
aqmm_manager_review_required: str | None = Form(None),
|
|
aqmm_partner_review_required: str | None = Form(None),
|
|
aqmm_review_partner_required: str | None = Form(None),
|
|
aqmm_blocks_final_release: str | None = Form(None),
|
|
aqmm_reference: str = Form(''),
|
|
is_active: str | None = Form(None),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
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.eligible_role_names = serialise_eligible_task_roles(eligible_roles, default_role_name)
|
|
task.sequence_no = sequence_no
|
|
task.is_mandatory = is_mandatory is not None
|
|
task.requires_review = requires_review is not None
|
|
task.normal_review_role = _normal_review_role_value(normal_review_role, requires_review=task.requires_review)
|
|
category = ensure_task_category(db, catalogue_id=catalogue_id, tenant_id=None, name=task_category, user_id=user.id)
|
|
task.task_category_id = category.id if category else None
|
|
task.task_category = category.name if category else None
|
|
task.response_required = response_required is not None
|
|
task.response_type = response_type.strip().upper() or "NONE"
|
|
task.task_tool_code = normalize_task_tool_code(task_tool_code)
|
|
task.evidence_required = evidence_required is not None
|
|
task.remarks_required_if_no = remarks_required_if_no is not None
|
|
task.is_aqmm_task = is_aqmm_task is not None
|
|
task.aqmm_mandatory = aqmm_mandatory is not None
|
|
task.aqmm_evidence_required = aqmm_evidence_required is not None
|
|
task.aqmm_manager_review_required = aqmm_manager_review_required is not None
|
|
task.aqmm_partner_review_required = aqmm_partner_review_required is not None
|
|
task.aqmm_review_partner_required = aqmm_review_partner_required is not None
|
|
task.aqmm_blocks_final_release = aqmm_blocks_final_release is not None
|
|
task.aqmm_reference = aqmm_reference.strip() or None
|
|
task.is_active = is_active is not None
|
|
|
|
db.flush()
|
|
sync_system_defaults_to_all_firms(
|
|
db,
|
|
service_catalogue_id=catalogue_id,
|
|
updated_by_user_id=user.id,
|
|
sync_open_engagements=True,
|
|
)
|
|
db.commit()
|
|
return RedirectResponse(url=f'/services/catalogue/{catalogue_id}/defaults', status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Task administration: service export/import and safe duplicate consolidation
|
|
# -----------------------------------------------------------------------------
|
|
|
|
@router.get('/catalogue/{catalogue_id}/defaults/export')
|
|
def default_tasks_service_export(request: Request, catalogue_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user: return RedirectResponse(url='/login', status_code=303)
|
|
require_permission(db, user, 'service_tasks.view')
|
|
if not _is_system_admin(db, user): return _redirect_denied()
|
|
catalogue = get_catalogue(db, catalogue_id)
|
|
if not catalogue: return RedirectResponse(url='/services/defaults', status_code=303)
|
|
payload = build_default_tasks_export(db, catalogue_id=catalogue_id)
|
|
filename = f"{catalogue.service_code}_default_tasks_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}.xlsx"
|
|
return StreamingResponse(iter([payload]), media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', headers={'Content-Disposition': f'attachment; filename="{filename}"'})
|
|
finally: db.close()
|
|
|
|
@router.post('/catalogue/{catalogue_id}/defaults/import')
|
|
def default_tasks_service_import(request: Request, catalogue_id: int, file: UploadFile = File(...), full_sync: str | None = Form(None), csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token); db = CommonSessionLocal()
|
|
try:
|
|
user=get_current_user(request, db=db)
|
|
if not user: return RedirectResponse(url='/login',status_code=303)
|
|
require_permission(db,user,'service_tasks.edit')
|
|
if not _is_system_admin(db,user): return _redirect_denied()
|
|
catalogue=get_catalogue(db,catalogue_id)
|
|
if not catalogue: return RedirectResponse(url='/services/defaults',status_code=303)
|
|
result=import_system_default_tasks(db,current_user=user,file_bytes=file.file.read(),update_existing=True,expected_service_code=catalogue.service_code,full_sync=full_sync is not None)
|
|
if result.get('errors'): return RedirectResponse(url=f'/services/catalogue/{catalogue_id}/defaults?import_error=1',status_code=303)
|
|
return RedirectResponse(url=f'/services/catalogue/{catalogue_id}/defaults?imported=1&created={result.get("created",0)}&updated={result.get("updated",0)}&retired={result.get("retired",0)}&full_sync={1 if result.get("full_sync") else 0}',status_code=303)
|
|
finally: db.close()
|
|
|
|
@router.get('/templates/{catalogue_id}/export')
|
|
def firm_tasks_service_export(request: Request, catalogue_id: int):
|
|
db=CommonSessionLocal()
|
|
try:
|
|
user=get_current_user(request,db=db)
|
|
if not user: return RedirectResponse(url='/login',status_code=303)
|
|
if not _can_manage_firm_tasks(db,user): return _redirect_denied()
|
|
tenant_id=_active_tenant_id(request,user); catalogue=get_catalogue(db,catalogue_id)
|
|
if not catalogue: return RedirectResponse(url='/services/templates',status_code=303)
|
|
payload=build_firm_tasks_export(db,tenant_id=tenant_id,catalogue_id=catalogue_id)
|
|
filename=f"{catalogue.service_code}_firm_tasks_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}.xlsx"
|
|
return StreamingResponse(iter([payload]),media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',headers={'Content-Disposition':f'attachment; filename="{filename}"'})
|
|
finally: db.close()
|
|
|
|
@router.post('/templates/{catalogue_id}/import')
|
|
def firm_tasks_service_import(request: Request, catalogue_id: int, file: UploadFile = File(...), csrf_token: str = Form(...)):
|
|
validate_csrf(request,csrf_token); db=CommonSessionLocal()
|
|
try:
|
|
user=get_current_user(request,db=db)
|
|
if not user: return RedirectResponse(url='/login',status_code=303)
|
|
if not _can_manage_firm_tasks(db,user): return _redirect_denied()
|
|
tenant_id=_active_tenant_id(request,user); catalogue=get_catalogue(db,catalogue_id)
|
|
if not catalogue: return RedirectResponse(url='/services/templates',status_code=303)
|
|
result=import_firm_task_templates(db,current_user=user,tenant_id=tenant_id,file_bytes=file.file.read(),update_existing=True,expected_service_code=catalogue.service_code)
|
|
if result.get('errors'): return RedirectResponse(url=f'/services/templates/{catalogue_id}?import_error=1',status_code=303)
|
|
return RedirectResponse(url=f'/services/templates/{catalogue_id}?imported=1&created={result.get("created",0)}&updated={result.get("updated",0)}',status_code=303)
|
|
finally: db.close()
|
|
|
|
@router.get('/catalogue/{catalogue_id}/defaults/duplicates')
|
|
def default_task_duplicates_page(request: Request, catalogue_id: int):
|
|
db=CommonSessionLocal()
|
|
try:
|
|
user=get_current_user(request,db=db)
|
|
if not user: return RedirectResponse(url='/login',status_code=303)
|
|
require_permission(db,user,'service_tasks.edit')
|
|
if not _is_system_admin(db,user): return _redirect_denied()
|
|
service=get_catalogue(db,catalogue_id); tasks=get_default_task_templates(db,catalogue_id=catalogue_id)
|
|
return _render(request,'modules/services/templates/services/task_duplicate_review.html',db,user,title=f'Duplicate Tasks - {service.service_name}',service=service,tasks=tasks,pairs=find_duplicate_pairs(tasks),scope='default')
|
|
finally: db.close()
|
|
|
|
@router.post('/catalogue/{catalogue_id}/defaults/duplicates/merge')
|
|
def default_task_duplicates_merge(request: Request,catalogue_id:int,master_task_id:int=Form(...),source_task_ids:list[int]=Form([]),csrf_token:str=Form(...)):
|
|
validate_csrf(request,csrf_token); db=CommonSessionLocal()
|
|
try:
|
|
user=get_current_user(request,db=db)
|
|
if not user: return RedirectResponse(url='/login',status_code=303)
|
|
require_permission(db,user,'service_tasks.edit')
|
|
if not _is_system_admin(db,user): return _redirect_denied()
|
|
count=merge_default_tasks(db,catalogue_id=catalogue_id,master_task_id=master_task_id,source_task_ids=source_task_ids)
|
|
sync_system_defaults_to_all_firms(db,service_catalogue_id=catalogue_id,updated_by_user_id=user.id,sync_open_engagements=True)
|
|
db.commit()
|
|
return RedirectResponse(url=f'/services/catalogue/{catalogue_id}/defaults/duplicates?merged={count}',status_code=303)
|
|
except ValueError:
|
|
db.rollback(); return RedirectResponse(url=f'/services/catalogue/{catalogue_id}/defaults/duplicates?merge_error=1',status_code=303)
|
|
finally: db.close()
|
|
|
|
@router.get('/templates/{catalogue_id}/duplicates')
|
|
def firm_task_duplicates_page(request: Request,catalogue_id:int):
|
|
db=CommonSessionLocal()
|
|
try:
|
|
user=get_current_user(request,db=db)
|
|
if not user: return RedirectResponse(url='/login',status_code=303)
|
|
if not _can_manage_firm_tasks(db,user): return _redirect_denied()
|
|
tenant_id=_active_tenant_id(request,user); service=get_catalogue(db,catalogue_id); tasks=get_firm_task_templates(db,tenant_id=tenant_id,catalogue_id=catalogue_id)
|
|
return _render(request,'modules/services/templates/services/task_duplicate_review.html',db,user,title=f'Duplicate Firm Tasks - {service.service_name}',service=service,tasks=tasks,pairs=find_duplicate_pairs(tasks),scope='firm')
|
|
finally: db.close()
|
|
|
|
@router.post('/templates/{catalogue_id}/duplicates/merge')
|
|
def firm_task_duplicates_merge(request: Request,catalogue_id:int,master_task_id:int=Form(...),source_task_ids:list[int]=Form([]),csrf_token:str=Form(...)):
|
|
validate_csrf(request,csrf_token); db=CommonSessionLocal()
|
|
try:
|
|
user=get_current_user(request,db=db)
|
|
if not user: return RedirectResponse(url='/login',status_code=303)
|
|
if not _can_manage_firm_tasks(db,user): return _redirect_denied()
|
|
tenant_id=_active_tenant_id(request,user); count=merge_firm_tasks(db,tenant_id=tenant_id,catalogue_id=catalogue_id,master_task_id=master_task_id,source_task_ids=source_task_ids,user_id=user.id); db.commit()
|
|
return RedirectResponse(url=f'/services/templates/{catalogue_id}/duplicates?merged={count}',status_code=303)
|
|
except ValueError:
|
|
db.rollback(); return RedirectResponse(url=f'/services/templates/{catalogue_id}/duplicates?merge_error=1',status_code=303)
|
|
finally: db.close()
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Service Task Category Master
|
|
# -----------------------------------------------------------------------------
|
|
|
|
@router.get('/templates/{catalogue_id}/categories')
|
|
def firm_task_categories_page(request: Request, catalogue_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
if not (_has_perm(db, user, 'service_tasks.view') or _can_manage_firm_tasks(db, user)):
|
|
return _redirect_denied()
|
|
tenant_id = _active_tenant_id(request, user)
|
|
service = get_catalogue(db, catalogue_id)
|
|
selection = get_firm_selection(db, tenant_id=tenant_id, catalogue_id=catalogue_id)
|
|
if not service or not selection or not selection.is_enabled:
|
|
return RedirectResponse(url='/services/templates', status_code=303)
|
|
return _render(request, 'modules/services/templates/services/task_category_list.html', db, user,
|
|
title=f'Task Categories - {service.service_name}', service=service,
|
|
rows=list_task_categories(db, catalogue_id=catalogue_id, tenant_id=tenant_id, include_inactive=True),
|
|
system_scope=False, can_manage=_can_manage_firm_tasks(db, user))
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get('/templates/{catalogue_id}/categories/export')
|
|
def firm_task_categories_export(request: Request, catalogue_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
if not (_has_perm(db, user, 'service_tasks.view') or _can_manage_firm_tasks(db, user)):
|
|
return _redirect_denied()
|
|
tenant_id = _active_tenant_id(request, user)
|
|
service = get_catalogue(db, catalogue_id)
|
|
selection = get_firm_selection(db, tenant_id=tenant_id, catalogue_id=catalogue_id)
|
|
if not service or not selection or not selection.is_enabled:
|
|
return RedirectResponse(url='/services/templates', status_code=303)
|
|
payload = build_task_categories_export(db, catalogue_id=catalogue_id, tenant_id=tenant_id)
|
|
filename = f"{service.service_code}_firm_task_categories_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}.xlsx"
|
|
return StreamingResponse(iter([payload]), media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', headers={'Content-Disposition': f'attachment; filename="{filename}"'})
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post('/templates/{catalogue_id}/categories/import')
|
|
def firm_task_categories_import(request: Request, catalogue_id: int, file: UploadFile = File(...), csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
if not _can_manage_firm_tasks(db, user):
|
|
return _redirect_denied()
|
|
tenant_id = _active_tenant_id(request, user)
|
|
service = get_catalogue(db, catalogue_id)
|
|
selection = get_firm_selection(db, tenant_id=tenant_id, catalogue_id=catalogue_id)
|
|
if not service or not selection or not selection.is_enabled:
|
|
return RedirectResponse(url='/services/templates', status_code=303)
|
|
result = import_task_categories(db, current_user=user, tenant_id=tenant_id, file_bytes=file.file.read(), expected_service_code=service.service_code)
|
|
if result.get('errors'):
|
|
return RedirectResponse(url=f'/services/templates/{catalogue_id}/categories?import_error={len(result["errors"])}', status_code=303)
|
|
return RedirectResponse(url=f'/services/templates/{catalogue_id}/categories?imported=1&created={result.get("created", 0)}&updated={result.get("updated", 0)}', status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post('/templates/{catalogue_id}/categories/new')
|
|
def firm_task_category_create(request: Request, catalogue_id: int, name: str = Form(...), code: str = Form(''), sort_order: int = Form(100), csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user: return RedirectResponse(url='/login', status_code=303)
|
|
if not _can_manage_firm_tasks(db, user): return _redirect_denied()
|
|
tenant_id = _active_tenant_id(request, user)
|
|
service = get_catalogue(db, catalogue_id)
|
|
if not service: return RedirectResponse(url='/services/templates', status_code=303)
|
|
row = ensure_task_category(db, catalogue_id=catalogue_id, tenant_id=tenant_id, name=name, user_id=user.id)
|
|
if row:
|
|
requested_code = normalize_code(code)[:50] if code.strip() else row.code
|
|
duplicate = db.execute(select(ServiceTaskCategory.id).where(ServiceTaskCategory.tenant_id == tenant_id, ServiceTaskCategory.service_catalogue_id == catalogue_id, ServiceTaskCategory.code == requested_code, ServiceTaskCategory.id != row.id)).scalar_one_or_none()
|
|
if duplicate is not None: return RedirectResponse(url=f'/services/templates/{catalogue_id}/categories?error=duplicate_code', status_code=303)
|
|
row.code = requested_code; row.sort_order = sort_order; row.is_active = True; row.updated_by_user_id = user.id
|
|
db.commit()
|
|
return RedirectResponse(url=f'/services/templates/{catalogue_id}/categories?created=1', status_code=303)
|
|
finally: db.close()
|
|
|
|
|
|
@router.post('/templates/{catalogue_id}/categories/{category_id}/update')
|
|
def firm_task_category_update(request: Request, catalogue_id: int, category_id: int, name: str = Form(...), code: str = Form(''), sort_order: int = Form(100), is_active: str | None = Form(None), csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token); db = CommonSessionLocal()
|
|
try:
|
|
user=get_current_user(request, db=db)
|
|
if not user: return RedirectResponse(url='/login', status_code=303)
|
|
if not _can_manage_firm_tasks(db,user): return _redirect_denied()
|
|
tenant_id=_active_tenant_id(request,user); row=get_task_category(db, category_id=category_id, catalogue_id=catalogue_id, tenant_id=tenant_id)
|
|
if not row: return RedirectResponse(url=f'/services/templates/{catalogue_id}/categories?error=not_found', status_code=303)
|
|
clean_name=name.strip(); clean_code=normalize_code(code or clean_name)[:50]
|
|
duplicate=db.execute(select(ServiceTaskCategory.id).where(ServiceTaskCategory.tenant_id==tenant_id, ServiceTaskCategory.service_catalogue_id==catalogue_id, ServiceTaskCategory.code==clean_code, ServiceTaskCategory.id!=row.id)).scalar_one_or_none()
|
|
if duplicate is not None: return RedirectResponse(url=f'/services/templates/{catalogue_id}/categories?error=duplicate_code', status_code=303)
|
|
old_name=row.name; row.name=clean_name; row.code=clean_code; row.sort_order=sort_order; row.is_active=is_active is not None; row.updated_by_user_id=user.id
|
|
for task in db.execute(select(FirmServiceTaskTemplate).where(FirmServiceTaskTemplate.tenant_id==tenant_id, FirmServiceTaskTemplate.service_catalogue_id==catalogue_id, FirmServiceTaskTemplate.task_category_id==row.id)).scalars().all():
|
|
task.task_category=clean_name; mark_firm_task_customized(task)
|
|
# Backward-compatible rows created before category IDs are also renamed when their text exactly matches.
|
|
for task in db.execute(select(FirmServiceTaskTemplate).where(FirmServiceTaskTemplate.tenant_id==tenant_id, FirmServiceTaskTemplate.service_catalogue_id==catalogue_id, FirmServiceTaskTemplate.task_category_id.is_(None), FirmServiceTaskTemplate.task_category==old_name)).scalars().all():
|
|
task.task_category_id=row.id; task.task_category=clean_name; mark_firm_task_customized(task)
|
|
db.commit(); return RedirectResponse(url=f'/services/templates/{catalogue_id}/categories?updated=1', status_code=303)
|
|
finally: db.close()
|
|
|
|
|
|
@router.get('/catalogue/{catalogue_id}/task-categories')
|
|
def system_task_categories_page(request: Request, catalogue_id: int):
|
|
db=CommonSessionLocal()
|
|
try:
|
|
user=get_current_user(request, db=db)
|
|
if not user: return RedirectResponse(url='/login', status_code=303)
|
|
require_permission(db,user,'service_tasks.view')
|
|
if not _is_system_admin(db,user): return _redirect_denied()
|
|
service=get_catalogue(db,catalogue_id)
|
|
if not service: return RedirectResponse(url='/services/defaults', status_code=303)
|
|
return _render(request,'modules/services/templates/services/task_category_list.html',db,user,title=f'System Task Categories - {service.service_name}',service=service,rows=list_task_categories(db,catalogue_id=catalogue_id,tenant_id=None,include_inactive=True),system_scope=True,can_manage=_has_perm(db,user,'service_tasks.edit') or _has_perm(db,user,'service_tasks.create'))
|
|
finally: db.close()
|
|
|
|
|
|
@router.post('/catalogue/{catalogue_id}/task-categories/new')
|
|
def system_task_category_create(request: Request, catalogue_id: int, name: str = Form(...), code: str = Form(''), sort_order: int = Form(100), csrf_token: str = Form(...)):
|
|
validate_csrf(request,csrf_token); db=CommonSessionLocal()
|
|
try:
|
|
user=get_current_user(request,db=db)
|
|
if not user: return RedirectResponse(url='/login',status_code=303)
|
|
require_permission(db,user,'service_tasks.create')
|
|
if not _is_system_admin(db,user): return _redirect_denied()
|
|
row=ensure_task_category(db,catalogue_id=catalogue_id,tenant_id=None,name=name,user_id=user.id)
|
|
if row:
|
|
requested_code=normalize_code(code)[:50] if code.strip() else row.code
|
|
duplicate=db.execute(select(ServiceTaskCategory.id).where(ServiceTaskCategory.tenant_id.is_(None), ServiceTaskCategory.service_catalogue_id==catalogue_id, ServiceTaskCategory.code==requested_code, ServiceTaskCategory.id!=row.id)).scalar_one_or_none()
|
|
if duplicate is not None: return RedirectResponse(url=f'/services/catalogue/{catalogue_id}/task-categories?error=duplicate_code',status_code=303)
|
|
row.code=requested_code; row.sort_order=sort_order; row.is_active=True; row.updated_by_user_id=user.id
|
|
db.commit(); return RedirectResponse(url=f'/services/catalogue/{catalogue_id}/task-categories?created=1',status_code=303)
|
|
finally: db.close()
|
|
|
|
|
|
@router.get('/catalogue/{catalogue_id}/task-categories/export')
|
|
def system_task_categories_export(request: Request, catalogue_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
require_permission(db, user, 'service_tasks.view')
|
|
if not _is_system_admin(db, user):
|
|
return _redirect_denied()
|
|
service = get_catalogue(db, catalogue_id)
|
|
if not service:
|
|
return RedirectResponse(url='/services/defaults', status_code=303)
|
|
payload = build_task_categories_export(db, catalogue_id=catalogue_id, tenant_id=None)
|
|
filename = f"{service.service_code}_default_task_categories_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}.xlsx"
|
|
return StreamingResponse(iter([payload]), media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', headers={'Content-Disposition': f'attachment; filename="{filename}"'})
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post('/catalogue/{catalogue_id}/task-categories/import')
|
|
def system_task_categories_import(request: Request, catalogue_id: int, file: UploadFile = File(...), csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url='/login', status_code=303)
|
|
require_permission(db, user, 'service_tasks.edit')
|
|
if not _is_system_admin(db, user):
|
|
return _redirect_denied()
|
|
service = get_catalogue(db, catalogue_id)
|
|
if not service:
|
|
return RedirectResponse(url='/services/defaults', status_code=303)
|
|
result = import_task_categories(db, current_user=user, tenant_id=None, file_bytes=file.file.read(), expected_service_code=service.service_code)
|
|
if result.get('errors'):
|
|
return RedirectResponse(url=f'/services/catalogue/{catalogue_id}/task-categories?import_error={len(result["errors"])}', status_code=303)
|
|
return RedirectResponse(url=f'/services/catalogue/{catalogue_id}/task-categories?imported=1&created={result.get("created", 0)}&updated={result.get("updated", 0)}', status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post('/catalogue/{catalogue_id}/task-categories/{category_id}/update')
|
|
def system_task_category_update(request: Request, catalogue_id: int, category_id: int, name: str = Form(...), code: str = Form(''), sort_order: int = Form(100), is_active: str | None = Form(None), csrf_token: str = Form(...)):
|
|
validate_csrf(request,csrf_token); db=CommonSessionLocal()
|
|
try:
|
|
user=get_current_user(request,db=db)
|
|
if not user: return RedirectResponse(url='/login',status_code=303)
|
|
require_permission(db,user,'service_tasks.edit')
|
|
if not _is_system_admin(db,user): return _redirect_denied()
|
|
row=get_task_category(db,category_id=category_id,catalogue_id=catalogue_id,tenant_id=None)
|
|
if not row: return RedirectResponse(url=f'/services/catalogue/{catalogue_id}/task-categories?error=not_found',status_code=303)
|
|
clean_name=name.strip(); clean_code=normalize_code(code or clean_name)[:50]
|
|
duplicate=db.execute(select(ServiceTaskCategory.id).where(ServiceTaskCategory.tenant_id.is_(None), ServiceTaskCategory.service_catalogue_id==catalogue_id, ServiceTaskCategory.code==clean_code, ServiceTaskCategory.id!=row.id)).scalar_one_or_none()
|
|
if duplicate is not None: return RedirectResponse(url=f'/services/catalogue/{catalogue_id}/task-categories?error=duplicate_code',status_code=303)
|
|
old_name=row.name; row.name=clean_name; row.code=clean_code; row.sort_order=sort_order; row.is_active=is_active is not None; row.updated_by_user_id=user.id
|
|
for task in db.execute(select(ServiceDefaultTaskTemplate).where(ServiceDefaultTaskTemplate.service_catalogue_id==catalogue_id, ServiceDefaultTaskTemplate.task_category_id==row.id)).scalars().all(): task.task_category=row.name
|
|
for task in db.execute(select(ServiceDefaultTaskTemplate).where(ServiceDefaultTaskTemplate.service_catalogue_id==catalogue_id, ServiceDefaultTaskTemplate.task_category_id.is_(None), ServiceDefaultTaskTemplate.task_category==old_name)).scalars().all(): task.task_category_id=row.id; task.task_category=row.name
|
|
sync_system_defaults_to_all_firms(db,service_catalogue_id=catalogue_id,updated_by_user_id=user.id,sync_open_engagements=True)
|
|
db.commit(); return RedirectResponse(url=f'/services/catalogue/{catalogue_id}/task-categories?updated=1',status_code=303)
|
|
finally: db.close()
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# S4.5 Bulk import and bulk service assignment routes
|
|
# Kept inside existing Services module intentionally; no separate router required.
|
|
# -----------------------------------------------------------------------------
|
|
|
|
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), full_sync: 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, full_sync=full_sync 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()
|
|
|
|
# BEGIN Firm Default Task Idempotent Sync Fix v1
|
|
@router.post("/catalogue/{catalogue_id}/firm-tasks/sync-system-defaults")
|
|
def sync_system_default_tasks_for_firm(request: Request, catalogue_id: int):
|
|
"""Idempotent replacement for the old insert-only 'Copy System Default Tasks' action."""
|
|
db = CommonSessionLocal()
|
|
try:
|
|
current_user = get_current_user(request, db=db)
|
|
if not current_user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
|
|
roles = set(get_user_roles(db, current_user.id))
|
|
if not roles.intersection({"System Admin", "Firm Admin"}):
|
|
return RedirectResponse(url="/services", status_code=303)
|
|
|
|
tenant_id = (
|
|
getattr(request.state, "active_tenant_id", None)
|
|
or getattr(request.state, "tenant_id", None)
|
|
or getattr(current_user, "tenant_id", None)
|
|
)
|
|
if not tenant_id:
|
|
raise RuntimeError(
|
|
"Active tenant could not be resolved for system-default task sync."
|
|
)
|
|
result = sync_firm_tasks_from_system_defaults(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
service_catalogue_id=catalogue_id,
|
|
updated_by_user_id=current_user.id,
|
|
)
|
|
db.commit()
|
|
|
|
# Return to the actual page that submitted the sync action instead of
|
|
# assuming a hard-coded firm-task-list URL. This preserves the current
|
|
# production routing structure.
|
|
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
|
|
|
|
referer = request.headers.get("referer") or f"/services/catalogue/{catalogue_id}"
|
|
parsed = urlsplit(referer)
|
|
|
|
# Prevent an external/open redirect. For a normal browser form POST the
|
|
# Referer is same-origin; if it is not, fall back to the Services page.
|
|
request_host = request.headers.get("host", "")
|
|
if parsed.netloc and parsed.netloc != request_host:
|
|
referer = f"/services/catalogue/{catalogue_id}"
|
|
parsed = urlsplit(referer)
|
|
|
|
query = dict(parse_qsl(parsed.query, keep_blank_values=True))
|
|
query.update(
|
|
{
|
|
"sync_success": "1",
|
|
"created": str(result.created),
|
|
"updated": str(result.updated),
|
|
"unchanged": str(result.unchanged),
|
|
"duplicates_disabled": str(result.duplicates_disabled),
|
|
}
|
|
)
|
|
|
|
redirect_url = urlunsplit(
|
|
(
|
|
parsed.scheme,
|
|
parsed.netloc,
|
|
parsed.path,
|
|
urlencode(query),
|
|
parsed.fragment,
|
|
)
|
|
)
|
|
|
|
return RedirectResponse(url=redirect_url, status_code=303)
|
|
except Exception:
|
|
db.rollback()
|
|
raise
|
|
finally:
|
|
db.close()
|
|
# END Firm Default Task Idempotent Sync Fix v1
|