Add Excel import export and code linking for task categories

This commit is contained in:
A R R R Associates
2026-08-31 12:52:50 +05:30
parent af4f6e8b28
commit fbb8607842
6 changed files with 551 additions and 14 deletions
+60 -4
View File
@@ -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)
+12 -1
View File
@@ -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",
+7 -3
View File
@@ -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),
@@ -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": []}
@@ -1,11 +1,105 @@
{% extends "ui/templates/base/layout.html" %}
{% block content %}
<div class="space-y-6">
<div class="flex flex-col gap-3 sm:flex-row sm:items-start sm:justify-between"><div><h2 class="text-xl font-semibold text-slate-900">Task Categories</h2><p class="mt-1 text-sm text-slate-500">{{ service.service_code }} · {{ service.service_name }} · {{ 'System defaults' if system_scope else 'Current firm' }}</p></div><a href="{{ '/services/catalogue/' ~ service.id ~ '/defaults' if system_scope else '/services/templates/' ~ service.id }}" class="rounded-xl border border-slate-300 px-4 py-2 text-sm font-medium text-slate-700 hover:bg-slate-50">Back to Tasks</a></div>
{% if request.query_params.get('created') %}<div class="rounded-xl border border-emerald-200 bg-emerald-50 p-3 text-sm text-emerald-800">Task category created.</div>{% endif %}
{% if request.query_params.get('updated') %}<div class="rounded-xl border border-emerald-200 bg-emerald-50 p-3 text-sm text-emerald-800">Task category updated. Linked task templates keep the same grouping name.</div>{% endif %}
{% if request.query_params.get('error') %}<div class="rounded-xl border border-rose-200 bg-rose-50 p-3 text-sm text-rose-800">Unable to save this category. Check the code/name and try again.</div>{% endif %}
{% if can_manage %}<section class="rounded-2xl bg-white p-5 shadow-soft"><h3 class="text-sm font-semibold text-slate-900">Add Task Category</h3><form method="post" action="{{ '/services/catalogue/' ~ service.id ~ '/task-categories/new' if system_scope else '/services/templates/' ~ service.id ~ '/categories/new' }}" class="mt-4 grid gap-3 md:grid-cols-4"><input type="hidden" name="csrf_token" value="{{ csrf_token }}"><input name="name" required placeholder="Category name" class="rounded-xl border border-slate-300 px-3 py-2 text-sm"><input name="code" placeholder="Code (auto if blank)" class="rounded-xl border border-slate-300 px-3 py-2 text-sm"><input name="sort_order" type="number" value="100" class="rounded-xl border border-slate-300 px-3 py-2 text-sm"><button class="rounded-xl bg-brand-600 px-4 py-2 text-sm font-medium text-white hover:bg-brand-700">Add Category</button></form></section>{% endif %}
<section class="overflow-hidden rounded-2xl bg-white shadow-soft"><table class="min-w-full divide-y divide-slate-200"><thead class="bg-slate-50"><tr><th class="px-4 py-3 text-left text-xs font-semibold uppercase text-slate-500">Order</th><th class="px-4 py-3 text-left text-xs font-semibold uppercase text-slate-500">Code</th><th class="px-4 py-3 text-left text-xs font-semibold uppercase text-slate-500">Category</th><th class="px-4 py-3 text-left text-xs font-semibold uppercase text-slate-500">Status</th><th class="px-4 py-3"></th></tr></thead><tbody class="divide-y divide-slate-100">{% for row in rows %}<tr>{% if can_manage %}<form method="post" action="{{ '/services/catalogue/' ~ service.id ~ '/task-categories/' ~ row.id ~ '/update' if system_scope else '/services/templates/' ~ service.id ~ '/categories/' ~ row.id ~ '/update' }}"><input type="hidden" name="csrf_token" value="{{ csrf_token }}"><td class="px-4 py-3"><input name="sort_order" type="number" value="{{ row.sort_order }}" class="w-20 rounded-lg border border-slate-300 px-2 py-1 text-sm"></td><td class="px-4 py-3"><input name="code" value="{{ row.code }}" required class="w-36 rounded-lg border border-slate-300 px-2 py-1 text-sm"></td><td class="px-4 py-3"><input name="name" value="{{ row.name }}" required class="w-full rounded-lg border border-slate-300 px-2 py-1 text-sm"></td><td class="px-4 py-3 text-sm"><label class="inline-flex items-center gap-2"><input type="checkbox" name="is_active" {% if row.is_active %}checked{% endif %}> Active</label></td><td class="px-4 py-3 text-right"><button class="rounded-lg border border-brand-200 bg-brand-50 px-3 py-1.5 text-xs font-semibold text-brand-700">Save</button></td></form>{% else %}<td class="px-4 py-3 text-sm">{{ row.sort_order }}</td><td class="px-4 py-3 text-sm">{{ row.code }}</td><td class="px-4 py-3 text-sm font-medium">{{ row.name }}</td><td class="px-4 py-3 text-sm">{{ 'Active' if row.is_active else 'Inactive' }}</td><td></td>{% endif %}</tr>{% else %}<tr><td colspan="5" class="px-4 py-8 text-center text-sm text-slate-500">No task categories yet. Existing uncategorised tasks continue to work normally.</td></tr>{% endfor %}</tbody></table></section>
<div class="flex flex-col gap-3 sm:flex-row sm:items-start sm:justify-between">
<div>
<h2 class="text-xl font-semibold text-slate-900">Task Categories</h2>
<p class="mt-1 text-sm text-slate-500">{{ service.service_code }} · {{ service.service_name }} · {{ 'System defaults' if system_scope else 'Current firm' }}</p>
</div>
<div class="flex flex-wrap gap-2">
<a href="{{ '/services/catalogue/' ~ service.id ~ '/task-categories/export' if system_scope else '/services/templates/' ~ service.id ~ '/categories/export' }}" class="rounded-xl border border-slate-300 bg-white px-4 py-2 text-sm font-medium text-slate-700 hover:bg-slate-50">Export Excel</a>
<a href="{{ '/services/catalogue/' ~ service.id ~ '/defaults' if system_scope else '/services/templates/' ~ service.id }}" class="rounded-xl border border-slate-300 px-4 py-2 text-sm font-medium text-slate-700 hover:bg-slate-50">Back to Tasks</a>
</div>
</div>
{% if request.query_params.get('created') %}
<div class="rounded-xl border border-emerald-200 bg-emerald-50 p-3 text-sm text-emerald-800">Task category created.</div>
{% endif %}
{% if request.query_params.get('updated') %}
<div class="rounded-xl border border-emerald-200 bg-emerald-50 p-3 text-sm text-emerald-800">Task category updated. Linked task templates keep the same grouping name.</div>
{% endif %}
{% if request.query_params.get('imported') %}
<div class="rounded-xl border border-emerald-200 bg-emerald-50 p-3 text-sm text-emerald-800">
Task Category Excel imported successfully. Created {{ request.query_params.get('created', '0') }} and updated {{ request.query_params.get('updated', '0') }} categories.
</div>
{% endif %}
{% if request.query_params.get('import_error') %}
<div class="rounded-xl border border-rose-200 bg-rose-50 p-3 text-sm text-rose-800">
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.
</div>
{% endif %}
{% if request.query_params.get('error') %}
<div class="rounded-xl border border-rose-200 bg-rose-50 p-3 text-sm text-rose-800">Unable to save this category. Check the code/name and try again.</div>
{% endif %}
{% if can_manage %}
<section class="rounded-2xl bg-white p-5 shadow-soft">
<div class="flex flex-col gap-4 lg:flex-row lg:items-end lg:justify-between">
<div>
<h3 class="text-sm font-semibold text-slate-900">Excel Import / Export</h3>
<p class="mt-1 max-w-3xl text-sm text-slate-500">
Export the current category master, edit it in Excel and import it back. Categories are matched by <strong>service_code + category_code</strong>; the stable category code can then be used in Task Excel through <strong>task_category_code</strong>. Existing name-only task imports remain supported.
</p>
</div>
<form method="post" enctype="multipart/form-data" action="{{ '/services/catalogue/' ~ service.id ~ '/task-categories/import' if system_scope else '/services/templates/' ~ service.id ~ '/categories/import' }}" class="flex flex-col gap-2 sm:flex-row sm:items-center">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<input type="file" name="file" accept=".xlsx" required class="block max-w-sm rounded-xl border border-slate-300 bg-white px-3 py-2 text-sm text-slate-700 file:mr-3 file:rounded-lg file:border-0 file:bg-slate-100 file:px-3 file:py-1.5 file:text-xs file:font-semibold file:text-slate-700">
<button class="whitespace-nowrap rounded-xl bg-brand-600 px-4 py-2 text-sm font-medium text-white hover:bg-brand-700">Import Categories</button>
</form>
</div>
<div class="mt-4 rounded-xl border border-blue-200 bg-blue-50 p-3 text-xs text-blue-800">
Excel columns: <strong>service_code</strong>, <strong>category_code</strong>, <strong>category_name</strong>, <strong>sort_order</strong>, <strong>is_active</strong>. Import never deletes categories omitted from the workbook.
</div>
</section>
<section class="rounded-2xl bg-white p-5 shadow-soft">
<h3 class="text-sm font-semibold text-slate-900">Add Task Category</h3>
<form method="post" action="{{ '/services/catalogue/' ~ service.id ~ '/task-categories/new' if system_scope else '/services/templates/' ~ service.id ~ '/categories/new' }}" class="mt-4 grid gap-3 md:grid-cols-4">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<input name="name" required placeholder="Category name" class="rounded-xl border border-slate-300 px-3 py-2 text-sm">
<input name="code" placeholder="Code (auto if blank)" class="rounded-xl border border-slate-300 px-3 py-2 text-sm">
<input name="sort_order" type="number" value="100" class="rounded-xl border border-slate-300 px-3 py-2 text-sm">
<button class="rounded-xl bg-brand-600 px-4 py-2 text-sm font-medium text-white hover:bg-brand-700">Add Category</button>
</form>
</section>
{% endif %}
<section class="overflow-hidden rounded-2xl bg-white shadow-soft">
<table class="min-w-full divide-y divide-slate-200">
<thead class="bg-slate-50">
<tr>
<th class="px-4 py-3 text-left text-xs font-semibold uppercase text-slate-500">Order</th>
<th class="px-4 py-3 text-left text-xs font-semibold uppercase text-slate-500">Code</th>
<th class="px-4 py-3 text-left text-xs font-semibold uppercase text-slate-500">Category</th>
<th class="px-4 py-3 text-left text-xs font-semibold uppercase text-slate-500">Status</th>
<th class="px-4 py-3"></th>
</tr>
</thead>
<tbody class="divide-y divide-slate-100">
{% for row in rows %}
<tr>
{% if can_manage %}
<form method="post" action="{{ '/services/catalogue/' ~ service.id ~ '/task-categories/' ~ row.id ~ '/update' if system_scope else '/services/templates/' ~ service.id ~ '/categories/' ~ row.id ~ '/update' }}">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<td class="px-4 py-3"><input name="sort_order" type="number" value="{{ row.sort_order }}" class="w-20 rounded-lg border border-slate-300 px-2 py-1 text-sm"></td>
<td class="px-4 py-3"><input name="code" value="{{ row.code }}" required class="w-36 rounded-lg border border-slate-300 px-2 py-1 text-sm"></td>
<td class="px-4 py-3"><input name="name" value="{{ row.name }}" required class="w-full rounded-lg border border-slate-300 px-2 py-1 text-sm"></td>
<td class="px-4 py-3 text-sm"><label class="inline-flex items-center gap-2"><input type="checkbox" name="is_active" {% if row.is_active %}checked{% endif %}> Active</label></td>
<td class="px-4 py-3 text-right"><button class="rounded-lg border border-brand-200 bg-brand-50 px-3 py-1.5 text-xs font-semibold text-brand-700">Save</button></td>
</form>
{% else %}
<td class="px-4 py-3 text-sm">{{ row.sort_order }}</td>
<td class="px-4 py-3 text-sm">{{ row.code }}</td>
<td class="px-4 py-3 text-sm font-medium">{{ row.name }}</td>
<td class="px-4 py-3 text-sm">{{ 'Active' if row.is_active else 'Inactive' }}</td>
<td></td>
{% endif %}
</tr>
{% else %}
<tr><td colspan="5" class="px-4 py-8 text-center text-sm text-slate-500">No task categories yet. Export Excel to get the correct column format, fill the categories and import it here, or add them manually above.</td></tr>
{% endfor %}
</tbody>
</table>
</section>
</div>
{% endblock %}
+87
View File
@@ -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()