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

Add task administration upgrade with duplicate merge and task tools

Browse Source
This commit is contained in:
A R R R Associates
2026-08-30 22:12:27 +05:30
parent ad0815c807
commit af4f6e8b28
17 changed files with 2072 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/modules/services/bulk_imports.py
+16 -3
View File
@@ -22,7 +22,8 @@ from app.modules.services.models import (
ServiceDueDateExtension,
ServiceDueDateRule,
)
from app.modules.services.services import normalize_code, normalize_engagement_type
from app.modules.services.task_tools import normalize_task_tool_code
from app.modules.services.services import ensure_task_category, normalize_code, normalize_engagement_type
from app.modules.services.client_services import (
assessment_year_from_financial_year,
attach_engagement_to_plan,
@@ -1014,7 +1015,7 @@ def import_due_date_extensions(
return {"created": created if not errors else 0, "updated": updated if not errors else 0, "skipped": skipped, "errors": errors}
def import_system_default_tasks(db: Session, *, current_user, file_bytes: bytes, update_existing: bool = True) -> dict:
def import_system_default_tasks(db: Session, *, current_user, file_bytes: bytes, update_existing: bool = True, expected_service_code: str | None = None) -> dict:
ws, headers = _load_sheet(file_bytes, "system_default_tasks")
missing = _validate_headers(headers, ["service_code", "sequence_no", "task_name"])
if missing:
@@ -1026,6 +1027,8 @@ def import_system_default_tasks(db: Session, *, current_user, file_bytes: bytes,
continue
try:
service_code = normalize_code(_cell(row, headers, "service_code"))
if expected_service_code and service_code != normalize_code(expected_service_code):
raise ValueError(f"This import accepts only service code {normalize_code(expected_service_code)}.")
sequence_no = _int(_cell(row, headers, "sequence_no"), None)
task_name = _clean(_cell(row, headers, "task_name"))
catalogue = find_service(db, service_code=service_code)
@@ -1054,12 +1057,16 @@ def import_system_default_tasks(db: Session, *, current_user, file_bytes: bytes,
task.is_mandatory = _bool(_cell(row, headers, "is_mandatory"), True)
task.requires_review = _bool(_cell(row, headers, "requires_review"), False)
task.task_category = _clean(_cell(row, headers, "task_category")) or None
category = ensure_task_category(db, catalogue_id=catalogue.id, tenant_id=None, name=task.task_category, user_id=current_user.id)
task.task_category_id = category.id if category else None
task.task_category = category.name if category else None
task.response_required = _bool(_cell(row, headers, "response_required"), False)
task.response_type = (_clean(_cell(row, headers, "response_type")) or "NONE").upper()
if task.response_type not in {"NONE", "YES_NO_NA", "YES_NO", "TEXT", "NUMBER", "DATE"}:
raise ValueError("response_type must be NONE, YES_NO_NA, YES_NO, TEXT, NUMBER or DATE.")
task.evidence_required = _bool(_cell(row, headers, "evidence_required"), False)
task.remarks_required_if_no = _bool(_cell(row, headers, "remarks_required_if_no"), False)
task.task_tool_code = normalize_task_tool_code(_clean(_cell(row, headers, "task_tool_code")) or "NONE")
task.is_aqmm_task = _bool(_cell(row, headers, "is_aqmm_task"), False)
task.aqmm_mandatory = _bool(_cell(row, headers, "aqmm_mandatory"), False)
task.aqmm_evidence_required = _bool(_cell(row, headers, "aqmm_evidence_required"), False)
@@ -1078,7 +1085,7 @@ def import_system_default_tasks(db: Session, *, current_user, file_bytes: bytes,
return {"created": created if not errors else 0, "updated": updated if not errors else 0, "skipped": skipped, "errors": errors}
def import_firm_task_templates(db: Session, *, current_user, tenant_id: int, file_bytes: bytes, update_existing: bool = True) -> dict:
def import_firm_task_templates(db: Session, *, current_user, tenant_id: int, file_bytes: bytes, update_existing: bool = True, expected_service_code: str | None = None) -> dict:
ws, headers = _load_sheet(file_bytes, "firm_task_templates")
missing = _validate_headers(headers, ["service_code", "sequence_no", "task_name"])
if missing:
@@ -1090,6 +1097,8 @@ def import_firm_task_templates(db: Session, *, current_user, tenant_id: int, fil
continue
try:
service_code = normalize_code(_cell(row, headers, "service_code"))
if expected_service_code and service_code != normalize_code(expected_service_code):
raise ValueError(f"This import accepts only service code {normalize_code(expected_service_code)}.")
sequence_no = _int(_cell(row, headers, "sequence_no"), None)
task_name = _clean(_cell(row, headers, "task_name"))
enabled = find_enabled_service(db, tenant_id=tenant_id, service_code=service_code)
@@ -1126,12 +1135,16 @@ def import_firm_task_templates(db: Session, *, current_user, tenant_id: int, fil
task.is_mandatory = _bool(_cell(row, headers, "is_mandatory"), True)
task.requires_review = _bool(_cell(row, headers, "requires_review"), False)
task.task_category = _clean(_cell(row, headers, "task_category")) or None
category = ensure_task_category(db, catalogue_id=catalogue.id, tenant_id=tenant_id, name=task.task_category, user_id=current_user.id)
task.task_category_id = category.id if category else None
task.task_category = category.name if category else None
task.response_required = _bool(_cell(row, headers, "response_required"), False)
task.response_type = (_clean(_cell(row, headers, "response_type")) or "NONE").upper()
if task.response_type not in {"NONE", "YES_NO_NA", "YES_NO", "TEXT", "NUMBER", "DATE"}:
raise ValueError("response_type must be NONE, YES_NO_NA, YES_NO, TEXT, NUMBER or DATE.")
task.evidence_required = _bool(_cell(row, headers, "evidence_required"), False)
task.remarks_required_if_no = _bool(_cell(row, headers, "remarks_required_if_no"), False)
task.task_tool_code = normalize_task_tool_code(_clean(_cell(row, headers, "task_tool_code")) or "NONE")
task.is_aqmm_task = _bool(_cell(row, headers, "is_aqmm_task"), False)
task.aqmm_mandatory = _bool(_cell(row, headers, "aqmm_mandatory"), False)
task.aqmm_evidence_required = _bool(_cell(row, headers, "aqmm_evidence_required"), False)