diff --git a/app/modules/services/bulk_imports.py b/app/modules/services/bulk_imports.py index 9c4f365..b59ebd7 100644 --- a/app/modules/services/bulk_imports.py +++ b/app/modules/services/bulk_imports.py @@ -19,6 +19,7 @@ from app.modules.services.models import ( ServiceCatalogue, ServiceCategory, ServiceDefaultTaskTemplate, + ServiceTaskCategory, ServiceDueDateExtension, ServiceDueDateRule, ) @@ -84,6 +85,13 @@ DEFAULT_TASK_COLUMNS = [ "default_role_name", "is_mandatory", "requires_review", + "task_category_code", + "task_category", + "response_required", + "response_type", + "evidence_required", + "remarks_required_if_no", + "task_tool_code", "is_aqmm_task", "aqmm_mandatory", "aqmm_evidence_required", @@ -103,6 +111,13 @@ FIRM_TASK_COLUMNS = [ "default_role_name", "is_mandatory", "requires_review", + "task_category_code", + "task_category", + "response_required", + "response_type", + "evidence_required", + "remarks_required_if_no", + "task_tool_code", "is_aqmm_task", "aqmm_mandatory", "aqmm_evidence_required", @@ -1015,6 +1030,35 @@ 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 _resolve_import_task_category(db: Session, *, catalogue_id: int, tenant_id: int | None, category_code: str, category_name: str, user_id: int | None): + """Resolve category by stable code first; preserve legacy name-based imports. + + When task_category_code is supplied it must already exist in the category + master for the same service/scope. This prevents spelling mistakes from + silently creating unintended category records. Older workbooks that only + contain task_category continue to work exactly as before. + """ + code = normalize_code(category_code) if category_code else "" + name = _clean(category_name) + scope_filter = ServiceTaskCategory.tenant_id.is_(None) if tenant_id is None else ServiceTaskCategory.tenant_id == tenant_id + if code: + category = db.execute( + select(ServiceTaskCategory).where( + ServiceTaskCategory.service_catalogue_id == catalogue_id, + scope_filter, + ServiceTaskCategory.code == code, + ) + ).scalar_one_or_none() + if category is None: + raise ValueError(f"Task category code {code} was not found for this service. Import/create the Task Category master first.") + if name and category.name.strip().lower() != name.strip().lower(): + # Code is the stable key. The master name wins so renamed categories + # do not break task imports. + pass + return category + return ensure_task_category(db, catalogue_id=catalogue_id, tenant_id=tenant_id, name=name or None, user_id=user_id) + + 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"]) @@ -1056,8 +1100,14 @@ def import_system_default_tasks(db: Session, *, current_user, file_bytes: bytes, task.default_role_name = _clean(_cell(row, headers, "default_role_name")) or None 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) + category = _resolve_import_task_category( + db, + catalogue_id=catalogue.id, + tenant_id=None, + category_code=_clean(_cell(row, headers, "task_category_code")), + category_name=_clean(_cell(row, headers, "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) @@ -1134,8 +1184,14 @@ def import_firm_task_templates(db: Session, *, current_user, tenant_id: int, fil task.default_role_name = _clean(_cell(row, headers, "default_role_name")) or None 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) + category = _resolve_import_task_category( + db, + catalogue_id=catalogue.id, + tenant_id=tenant_id, + category_code=_clean(_cell(row, headers, "task_category_code")), + category_name=_clean(_cell(row, headers, "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) diff --git a/app/modules/services/default_tasks_export.py b/app/modules/services/default_tasks_export.py index 99ef3e6..0759dcf 100644 --- a/app/modules/services/default_tasks_export.py +++ b/app/modules/services/default_tasks_export.py @@ -8,7 +8,7 @@ from openpyxl.utils import get_column_letter from sqlalchemy import select from sqlalchemy.orm import Session, selectinload -from app.modules.services.models import ServiceCatalogue, ServiceDefaultTaskTemplate +from app.modules.services.models import ServiceCatalogue, ServiceDefaultTaskTemplate, ServiceTaskCategory _HEADERS = [ @@ -21,6 +21,7 @@ _HEADERS = [ "default_role_name", "is_mandatory", "requires_review", + "task_category_code", "task_category", "response_required", "response_type", @@ -46,6 +47,7 @@ _IMPORT_HEADERS = [ "default_role_name", "is_mandatory", "requires_review", + "task_category_code", "task_category", "response_required", "response_type", @@ -144,6 +146,13 @@ def build_default_tasks_export(db: Session, *, catalogue_id: int | None = None) ]) missing_ws.append(["service_code", "service_name", "category_code", "category_name", "recurrence_type", "engagement_type"]) + category_code_by_id = { + row.id: row.code + for row in db.execute( + select(ServiceTaskCategory).where(ServiceTaskCategory.tenant_id.is_(None)) + ).scalars().all() + } + total_tasks = 0 active_tasks = 0 services_without_tasks = 0 @@ -168,6 +177,7 @@ def build_default_tasks_export(db: Session, *, catalogue_id: int | None = None) task.default_role_name or "", _yes_no(task.is_mandatory), _yes_no(task.requires_review), + category_code_by_id.get(getattr(task, "task_category_id", None), ""), task.task_category or "", _yes_no(task.response_required), task.response_type or "NONE", @@ -192,6 +202,7 @@ def build_default_tasks_export(db: Session, *, catalogue_id: int | None = None) task.default_role_name or "", _yes_no(task.is_mandatory), _yes_no(task.requires_review), + category_code_by_id.get(getattr(task, "task_category_id", None), ""), task.task_category or "", _yes_no(task.response_required), task.response_type or "NONE", diff --git a/app/modules/services/firm_tasks_export.py b/app/modules/services/firm_tasks_export.py index 448a02d..38ff3a8 100644 --- a/app/modules/services/firm_tasks_export.py +++ b/app/modules/services/firm_tasks_export.py @@ -7,11 +7,11 @@ from openpyxl.utils import get_column_letter from sqlalchemy import select from sqlalchemy.orm import Session -from app.modules.services.models import FirmServiceTaskTemplate, ServiceCatalogue +from app.modules.services.models import FirmServiceTaskTemplate, ServiceCatalogue, ServiceTaskCategory HEADERS = [ "service_code", "sequence_no", "task_name", "default_role_name", "eligible_role_names", - "is_mandatory", "requires_review", "normal_review_role", "task_category", "response_required", + "is_mandatory", "requires_review", "normal_review_role", "task_category_code", "task_category", "response_required", "response_type", "evidence_required", "remarks_required_if_no", "task_tool_code", "is_aqmm_task", "aqmm_mandatory", "aqmm_evidence_required", "aqmm_manager_review_required", "aqmm_partner_review_required", "aqmm_review_partner_required", "aqmm_blocks_final_release", @@ -26,10 +26,14 @@ def build_firm_tasks_export(db: Session, *, tenant_id: int, catalogue_id: int) - FirmServiceTaskTemplate.tenant_id == tenant_id, FirmServiceTaskTemplate.service_catalogue_id == catalogue_id, ).order_by(FirmServiceTaskTemplate.sequence_no, FirmServiceTaskTemplate.id)).scalars().all()) + category_code_by_id = {row.id: row.code for row in db.execute(select(ServiceTaskCategory).where( + ServiceTaskCategory.tenant_id == tenant_id, + ServiceTaskCategory.service_catalogue_id == catalogue_id, + )).scalars().all()} wb = Workbook(); ws = wb.active; ws.title = "Firm Task Templates"; ws.append(HEADERS) for task in tasks: ws.append([service.service_code, task.sequence_no, task.task_name, task.default_role_name or "", task.eligible_role_names or "", - _yn(task.is_mandatory), _yn(task.requires_review), task.normal_review_role or "", task.task_category or "", + _yn(task.is_mandatory), _yn(task.requires_review), task.normal_review_role or "", category_code_by_id.get(getattr(task, "task_category_id", None), ""), task.task_category or "", _yn(task.response_required), task.response_type or "NONE", _yn(task.evidence_required), _yn(task.remarks_required_if_no), getattr(task, "task_tool_code", "NONE") or "NONE", _yn(task.is_aqmm_task), _yn(task.aqmm_mandatory), _yn(task.aqmm_evidence_required), _yn(task.aqmm_manager_review_required), _yn(task.aqmm_partner_review_required), diff --git a/app/modules/services/task_categories_excel.py b/app/modules/services/task_categories_excel.py new file mode 100644 index 0000000..a48fec3 --- /dev/null +++ b/app/modules/services/task_categories_excel.py @@ -0,0 +1,285 @@ +from __future__ import annotations + +from io import BytesIO + +from openpyxl import Workbook, load_workbook +from openpyxl.styles import Alignment, Font, PatternFill +from openpyxl.utils import get_column_letter +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.modules.services.models import ServiceCatalogue, ServiceTaskCategory +from app.modules.services.services import normalize_code + +CATEGORY_HEADERS = [ + "service_code", + "category_code", + "category_name", + "sort_order", + "is_active", +] + +_TRUE_VALUES = {"1", "true", "yes", "y", "on"} +_FALSE_VALUES = {"0", "false", "no", "n", "off"} + + +def _clean(value) -> str: + if value is None: + return "" + return str(value).strip() + + +def _bool(value, default: bool = True) -> bool: + if value is None or value == "": + return default + if isinstance(value, bool): + return value + clean = _clean(value).lower() + if clean in _TRUE_VALUES: + return True + if clean in _FALSE_VALUES: + return False + raise ValueError("is_active must be TRUE/FALSE, YES/NO or 1/0.") + + +def _int(value, default: int = 100) -> int: + if value is None or value == "": + return default + try: + return int(float(value)) + except (TypeError, ValueError) as exc: + raise ValueError("sort_order must be a whole number.") from exc + + +def _style_sheet(ws) -> None: + header_fill = PatternFill("solid", fgColor="1F4E78") + header_font = Font(color="FFFFFF", bold=True) + ws.freeze_panes = "A2" + ws.auto_filter.ref = ws.dimensions + for cell in ws[1]: + cell.fill = header_fill + cell.font = header_font + cell.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True) + widths = {1: 22, 2: 22, 3: 40, 4: 14, 5: 14} + for idx, width in widths.items(): + ws.column_dimensions[get_column_letter(idx)].width = width + for row in ws.iter_rows(min_row=2): + for cell in row: + cell.alignment = Alignment(vertical="top", wrap_text=True) + + +def build_task_categories_export( + db: Session, + *, + catalogue_id: int, + tenant_id: int | None, +) -> bytes: + service = db.execute( + select(ServiceCatalogue).where(ServiceCatalogue.id == catalogue_id) + ).scalar_one() + scope_filter = ( + ServiceTaskCategory.tenant_id.is_(None) + if tenant_id is None + else ServiceTaskCategory.tenant_id == tenant_id + ) + rows = list( + db.execute( + select(ServiceTaskCategory) + .where( + ServiceTaskCategory.service_catalogue_id == catalogue_id, + scope_filter, + ) + .order_by( + ServiceTaskCategory.sort_order.asc(), + ServiceTaskCategory.name.asc(), + ServiceTaskCategory.id.asc(), + ) + ).scalars().all() + ) + + wb = Workbook() + ws = wb.active + ws.title = "Task Categories" + ws.append(CATEGORY_HEADERS) + for row in rows: + ws.append( + [ + service.service_code, + row.code, + row.name, + row.sort_order, + "TRUE" if row.is_active else "FALSE", + ] + ) + _style_sheet(ws) + + info = wb.create_sheet("Instructions") + info.append(["Field", "Instruction"]) + instructions = [ + ("service_code", "Required. Existing Service Catalogue code. Do not change when using a per-service import."), + ("category_code", "Required stable code used to link task Excel rows to this category. Keep it unchanged when renaming a category."), + ("category_name", "Required display name shown in task forms and Work Tracker grouping."), + ("sort_order", "Optional whole number. Lower values appear first. Default 100."), + ("is_active", "TRUE/FALSE. Inactive categories remain preserved for historical/template references."), + ("Task linking", "In task Excel exports/imports use task_category_code. Existing task_category name-only files remain supported for backward compatibility."), + ("Safety", "Import updates or creates categories by service_code + category_code. It does not delete categories omitted from Excel."), + ] + for item in instructions: + info.append(item) + info.column_dimensions["A"].width = 24 + info.column_dimensions["B"].width = 105 + for cell in info[1]: + cell.fill = PatternFill("solid", fgColor="D9EAF7") + cell.font = Font(bold=True) + for row in info.iter_rows(): + for cell in row: + cell.alignment = Alignment(vertical="top", wrap_text=True) + + out = BytesIO() + wb.save(out) + return out.getvalue() + + +def _load_category_sheet(file_bytes: bytes): + try: + wb = load_workbook(BytesIO(file_bytes), data_only=True) + except Exception as exc: + raise ValueError("Unable to read Excel workbook. Please upload a valid .xlsx file.") from exc + if "Task Categories" in wb.sheetnames: + ws = wb["Task Categories"] + else: + ws = wb[wb.sheetnames[0]] + headers = { + _clean(cell.value).lower(): idx + for idx, cell in enumerate(ws[1]) + if _clean(cell.value) + } + missing = [header for header in ("service_code", "category_code", "category_name") if header not in headers] + if missing: + raise ValueError(f"Missing columns: {', '.join(missing)}") + return ws, headers + + +def _cell(row, headers: dict[str, int], name: str): + idx = headers.get(name) + if idx is None or idx >= len(row): + return None + return row[idx] + + +def import_task_categories( + db: Session, + *, + current_user, + tenant_id: int | None, + file_bytes: bytes, + expected_service_code: str | None = None, +) -> dict: + try: + ws, headers = _load_category_sheet(file_bytes) + except Exception as exc: + return {"created": 0, "updated": 0, "errors": [{"row": 1, "message": str(exc)}]} + + created = 0 + updated = 0 + errors: list[dict] = [] + expected = normalize_code(expected_service_code) if expected_service_code else "" + + for row_no, row in enumerate(ws.iter_rows(min_row=2, values_only=True), start=2): + if not any(value not in (None, "") for value in row): + continue + try: + service_code = normalize_code(_cell(row, headers, "service_code")) + category_code = normalize_code(_cell(row, headers, "category_code"))[:50] + category_name = _clean(_cell(row, headers, "category_name"))[:100] + sort_order = _int(_cell(row, headers, "sort_order"), 100) + is_active = _bool(_cell(row, headers, "is_active"), True) + + if not service_code or not category_code or not category_name: + raise ValueError("service_code, category_code and category_name are required.") + if expected and service_code != expected: + raise ValueError(f"This import accepts only service code {expected}.") + + service = db.execute( + select(ServiceCatalogue).where(ServiceCatalogue.service_code == service_code) + ).scalar_one_or_none() + if service is None: + raise ValueError("Service code not found in Service Catalogue.") + + scope_filter = ( + ServiceTaskCategory.tenant_id.is_(None) + if tenant_id is None + else ServiceTaskCategory.tenant_id == tenant_id + ) + category = db.execute( + select(ServiceTaskCategory).where( + ServiceTaskCategory.service_catalogue_id == service.id, + scope_filter, + ServiceTaskCategory.code == category_code, + ) + ).scalar_one_or_none() + + duplicate_name = db.execute( + select(ServiceTaskCategory).where( + ServiceTaskCategory.service_catalogue_id == service.id, + scope_filter, + ServiceTaskCategory.name.ilike(category_name), + ServiceTaskCategory.code != category_code, + ) + ).scalar_one_or_none() + if duplicate_name is not None: + raise ValueError( + f"Category name '{category_name}' already exists with code {duplicate_name.code}. " + "Use the existing stable category code instead of creating another category." + ) + + if category is None: + category = ServiceTaskCategory( + tenant_id=tenant_id, + service_catalogue_id=service.id, + code=category_code, + name=category_name, + sort_order=sort_order, + is_active=is_active, + created_by_user_id=current_user.id, + updated_by_user_id=current_user.id, + ) + db.add(category) + created += 1 + else: + old_name = category.name + category.name = category_name + category.sort_order = sort_order + category.is_active = is_active + category.updated_by_user_id = current_user.id + updated += 1 + + # Keep the existing text snapshot in linked task templates aligned + # with a renamed master category. The category ID is not changed. + from app.modules.services.models import FirmServiceTaskTemplate, ServiceDefaultTaskTemplate + if tenant_id is None: + linked = db.execute( + select(ServiceDefaultTaskTemplate).where( + ServiceDefaultTaskTemplate.service_catalogue_id == service.id, + ServiceDefaultTaskTemplate.task_category_id == category.id, + ) + ).scalars().all() + else: + linked = db.execute( + select(FirmServiceTaskTemplate).where( + FirmServiceTaskTemplate.tenant_id == tenant_id, + FirmServiceTaskTemplate.service_catalogue_id == service.id, + FirmServiceTaskTemplate.task_category_id == category.id, + ) + ).scalars().all() + if old_name != category_name: + for task in linked: + task.task_category = category_name + except Exception as exc: + errors.append({"row": row_no, "message": str(exc)}) + + if errors: + db.rollback() + return {"created": 0, "updated": 0, "errors": errors} + db.commit() + return {"created": created, "updated": updated, "errors": []} diff --git a/app/modules/services/templates/services/task_category_list.html b/app/modules/services/templates/services/task_category_list.html index 689c70c..70bff94 100644 --- a/app/modules/services/templates/services/task_category_list.html +++ b/app/modules/services/templates/services/task_category_list.html @@ -1,11 +1,105 @@ {% extends "ui/templates/base/layout.html" %} {% block content %}
-

Task Categories

{{ service.service_code }} · {{ service.service_name }} · {{ 'System defaults' if system_scope else 'Current firm' }}

Back to Tasks
- {% if request.query_params.get('created') %}
Task category created.
{% endif %} - {% if request.query_params.get('updated') %}
Task category updated. Linked task templates keep the same grouping name.
{% endif %} - {% if request.query_params.get('error') %}
Unable to save this category. Check the code/name and try again.
{% endif %} - {% if can_manage %}

Add Task Category

{% endif %} -
{% for row in rows %}{% if can_manage %}{% else %}{% endif %}{% else %}{% endfor %}
OrderCodeCategoryStatus
{{ row.sort_order }}{{ row.code }}{{ row.name }}{{ 'Active' if row.is_active else 'Inactive' }}
No task categories yet. Existing uncategorised tasks continue to work normally.
+
+
+

Task Categories

+

{{ service.service_code }} · {{ service.service_name }} · {{ 'System defaults' if system_scope else 'Current firm' }}

+
+
+ Export Excel + Back to Tasks +
+
+ + {% if request.query_params.get('created') %} +
Task category created.
+ {% endif %} + {% if request.query_params.get('updated') %} +
Task category updated. Linked task templates keep the same grouping name.
+ {% endif %} + {% if request.query_params.get('imported') %} +
+ Task Category Excel imported successfully. Created {{ request.query_params.get('created', '0') }} and updated {{ request.query_params.get('updated', '0') }} categories. +
+ {% endif %} + {% if request.query_params.get('import_error') %} +
+ Task Category import was not applied because {{ request.query_params.get('import_error') }} row(s) failed validation. Correct the workbook and import it again; no partial category changes were committed. +
+ {% endif %} + {% if request.query_params.get('error') %} +
Unable to save this category. Check the code/name and try again.
+ {% endif %} + + {% if can_manage %} +
+
+
+

Excel Import / Export

+

+ Export the current category master, edit it in Excel and import it back. Categories are matched by service_code + category_code; the stable category code can then be used in Task Excel through task_category_code. Existing name-only task imports remain supported. +

+
+
+ + + +
+
+
+ Excel columns: service_code, category_code, category_name, sort_order, is_active. Import never deletes categories omitted from the workbook. +
+
+ +
+

Add Task Category

+
+ + + + + +
+
+ {% endif %} + +
+ + + + + + + + + + + + {% for row in rows %} + + {% if can_manage %} + + + + + + + + + {% else %} + + + + + + {% endif %} + + {% else %} + + {% endfor %} + +
OrderCodeCategoryStatus
{{ row.sort_order }}{{ row.code }}{{ row.name }}{{ 'Active' if row.is_active else 'Inactive' }}
No task categories yet. Export Excel to get the correct column format, fill the categories and import it here, or add them manually above.
+
{% endblock %} diff --git a/app/modules/services/ui.py b/app/modules/services/ui.py index 969e84e..1585f18 100644 --- a/app/modules/services/ui.py +++ b/app/modules/services/ui.py @@ -27,6 +27,7 @@ from app.modules.services.bulk_imports import ( 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.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 ( @@ -2157,6 +2158,50 @@ def firm_task_categories_page(request: Request, catalogue_id: int): 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) @@ -2231,6 +2276,48 @@ def system_task_category_create(request: Request, catalogue_id: int, name: str = 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()