Rakseeibra/arrr-erp
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Auto initialise firm tasks from system defaults

Browse Source
This commit is contained in:
A R R R Associates
2026-08-03 17:43:18 +05:30
parent 34c258e26e
commit 4274676513
2 changed files with 118 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/modules/services/ui.py
+90 -1
View File
@@ -849,6 +849,56 @@ def catalogue_edit_submit(request: Request, catalogue_id: int, service_name: str
db.close()
def _copy_system_defaults_if_firm_tasks_empty(
db,
*,
tenant_id: int,
catalogue_id: int,
user_id: int,
) -> tuple[int, bool]:
"""Initialise firm task templates once without overwriting custom firm tasks."""
existing_tasks = get_firm_task_templates(
db,
tenant_id=tenant_id,
catalogue_id=catalogue_id,
)
if existing_tasks:
return 0, True
defaults = get_default_task_templates(db, catalogue_id=catalogue_id)
copied_count = 0
for default in defaults:
db.add(FirmServiceTaskTemplate(
tenant_id=tenant_id,
service_catalogue_id=catalogue_id,
task_name=default.task_name,
description=default.description,
sequence_no=default.sequence_no,
default_role_name=default.default_role_name,
is_mandatory=default.is_mandatory,
requires_review=default.requires_review,
task_category=getattr(default, "task_category", None),
response_required=getattr(default, "response_required", False),
response_type=getattr(default, "response_type", "NONE") or "NONE",
evidence_required=getattr(default, "evidence_required", False),
remarks_required_if_no=getattr(default, "remarks_required_if_no", False),
is_aqmm_task=getattr(default, "is_aqmm_task", False),
aqmm_mandatory=getattr(default, "aqmm_mandatory", False),
aqmm_evidence_required=getattr(default, "aqmm_evidence_required", False),
aqmm_manager_review_required=getattr(default, "aqmm_manager_review_required", False),
aqmm_partner_review_required=getattr(default, "aqmm_partner_review_required", False),
aqmm_review_partner_required=getattr(default, "aqmm_review_partner_required", False),
aqmm_blocks_final_release=getattr(default, "aqmm_blocks_final_release", False),
aqmm_reference=getattr(default, "aqmm_reference", None),
is_active=default.is_active,
created_by_user_id=user_id,
updated_by_user_id=user_id,
))
copied_count += 1
return copied_count, False
@router.post('/catalogue/bulk-selection')
def bulk_update_firm_services(
request: Request,
@@ -900,6 +950,9 @@ def bulk_update_firm_services(
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(
@@ -949,6 +1002,18 @@ def bulk_update_firm_services(
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
@@ -966,6 +1031,9 @@ def bulk_update_firm_services(
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}',
]
@@ -1002,6 +1070,7 @@ def toggle_firm_service(request: Request, catalogue_id: int, default_branch_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,
@@ -1017,8 +1086,28 @@ def toggle_firm_service(request: Request, catalogue_id: int, default_branch_id:
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()
return RedirectResponse(url='/services', status_code=303)
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()