Add service catalogue Excel export and duplicate review

app/modules/services/catalogue_export.py

@@ -0,0 +1,202 @@
+from __future__ import annotations
+
+from collections import defaultdict
+from datetime import datetime, timezone
+from io import BytesIO
+import re
+
+from openpyxl import Workbook
+from openpyxl.styles import Alignment, Font, PatternFill
+from openpyxl.utils import get_column_letter
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session, selectinload
+
+from app.modules.consultants.models import ClientConsultantLink
+from app.modules.services.models import (
+    ClientServiceSubscription,
+    ClientServiceTaskInstance,
+    FirmServiceSelection,
+    FirmServiceTaskTemplate,
+    FirmTaskDocumentRequirement,
+    FirmTaskDocumentTemplate,
+    ServiceCatalogue,
+    ServiceDefaultTaskTemplate,
+    ServiceDueDateRule,
+)
+
+
+_HEADER_FILL = PatternFill("solid", fgColor="1F4E78")
+_HEADER_FONT = Font(color="FFFFFF", bold=True)
+_WARNING_FILL = PatternFill("solid", fgColor="FFF2CC")
+_DANGER_FILL = PatternFill("solid", fgColor="FCE4D6")
+_SAFE_FILL = PatternFill("solid", fgColor="E2F0D9")
+
+
+def _normalise_name(value: str | None) -> str:
+    return " ".join((value or "").strip().lower().split())
+
+
+def _compact_name(value: str | None) -> str:
+    return re.sub(r"[^a-z0-9]+", "", _normalise_name(value))
+
+
+def _count_by_catalogue(db: Session, model) -> dict[int, int]:
+    rows = db.execute(
+        select(model.service_catalogue_id, func.count(model.id))
+        .group_by(model.service_catalogue_id)
+    ).all()
+    return {int(catalogue_id): int(count) for catalogue_id, count in rows}
+
+
+def _style_sheet(ws, freeze: str = "A2") -> None:
+    ws.freeze_panes = freeze
+    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)
+    ws.row_dimensions[1].height = 32
+    for row in ws.iter_rows(min_row=2):
+        for cell in row:
+            cell.alignment = Alignment(vertical="top", wrap_text=True)
+    for column_cells in ws.columns:
+        max_length = max(len(str(cell.value or "")) for cell in column_cells)
+        ws.column_dimensions[get_column_letter(column_cells[0].column)].width = min(max(max_length + 2, 11), 42)
+
+
+def build_service_catalogue_export(db: Session) -> bytes:
+    catalogues = db.execute(
+        select(ServiceCatalogue)
+        .options(selectinload(ServiceCatalogue.service_category))
+        .order_by(ServiceCatalogue.service_code.asc())
+    ).scalars().all()
+
+    dependency_maps = {
+        "firm_selection_count": _count_by_catalogue(db, FirmServiceSelection),
+        "system_default_task_count": _count_by_catalogue(db, ServiceDefaultTaskTemplate),
+        "firm_task_template_count": _count_by_catalogue(db, FirmServiceTaskTemplate),
+        "due_date_rule_count": _count_by_catalogue(db, ServiceDueDateRule),
+        "document_requirement_count": _count_by_catalogue(db, FirmTaskDocumentRequirement),
+        "document_template_count": _count_by_catalogue(db, FirmTaskDocumentTemplate),
+        "client_subscription_count": _count_by_catalogue(db, ClientServiceSubscription),
+        "execution_task_count": _count_by_catalogue(db, ClientServiceTaskInstance),
+        "consultant_link_count": _count_by_catalogue(db, ClientConsultantLink),
+    }
+
+    exact_groups: dict[str, list[ServiceCatalogue]] = defaultdict(list)
+    compact_groups: dict[str, list[ServiceCatalogue]] = defaultdict(list)
+    for row in catalogues:
+        exact_groups[_normalise_name(row.service_name)].append(row)
+        compact_groups[_compact_name(row.service_name)].append(row)
+
+    duplicate_ids: set[int] = set()
+    for group in list(exact_groups.values()) + list(compact_groups.values()):
+        if len(group) > 1:
+            duplicate_ids.update(item.id for item in group)
+
+    wb = Workbook()
+    ws = wb.active
+    ws.title = "Service Catalogue"
+    headers = [
+        "ID", "Service Code", "Service Name", "Category Code", "Category Name",
+        "Recurrence Type", "Engagement Type", "Sort Order", "Description",
+        "Individual", "Proprietorship", "Partnership", "LLP", "Company", "Trust", "Society",
+        "Active", "Client Requestable", "Consultant Requestable",
+        "Possible Duplicate", "Created At UTC", "Updated At UTC",
+    ]
+    ws.append(headers)
+    for row in catalogues:
+        ws.append([
+            row.id,
+            row.service_code,
+            row.service_name,
+            row.service_category.code if row.service_category else "",
+            row.service_category.name if row.service_category else (row.category or ""),
+            row.recurrence_type or "",
+            row.engagement_type,
+            row.sort_order,
+            row.description or "",
+            row.applicable_individual,
+            row.applicable_proprietorship,
+            row.applicable_partnership,
+            row.applicable_llp,
+            row.applicable_company,
+            row.applicable_trust,
+            row.applicable_society,
+            row.is_active,
+            row.is_client_requestable,
+            row.is_consultant_requestable,
+            "YES" if row.id in duplicate_ids else "NO",
+            row.created_at_utc.isoformat() if row.created_at_utc else "",
+            row.updated_at_utc.isoformat() if row.updated_at_utc else "",
+        ])
+    _style_sheet(ws)
+
+    dup_ws = wb.create_sheet("Duplicate Candidates")
+    dup_ws.append([
+        "Match Type", "Normalised Value", "Record Count", "IDs", "Service Codes", "Service Names", "Suggested Action"
+    ])
+    emitted: set[tuple[str, tuple[int, ...]]] = set()
+    for match_type, groups in (("Exact normalised name", exact_groups), ("Ignoring spaces/punctuation", compact_groups)):
+        for normalised, group in sorted(groups.items()):
+            if not normalised or len(group) < 2:
+                continue
+            ids = tuple(sorted(item.id for item in group))
+            key = (match_type, ids)
+            if key in emitted:
+                continue
+            emitted.add(key)
+            dup_ws.append([
+                match_type,
+                normalised,
+                len(group),
+                ", ".join(str(item.id) for item in group),
+                "\n".join(item.service_code for item in group),
+                "\n".join(item.service_name for item in group),
+                "Review dependencies before disabling or merging",
+            ])
+    if dup_ws.max_row == 1:
+        dup_ws.append(["No duplicate candidates", "", 0, "", "", "", "No action required"])
+    _style_sheet(dup_ws)
+    for row in dup_ws.iter_rows(min_row=2):
+        for cell in row:
+            cell.fill = _WARNING_FILL
+
+    dep_ws = wb.create_sheet("Dependency Summary")
+    dep_headers = [
+        "ID", "Service Code", "Service Name",
+        "Firm Selections", "System Default Tasks", "Firm Task Templates",
+        "Due Date Rules", "Document Requirements", "Document Templates",
+        "Client Subscriptions", "Execution Tasks", "Consultant Links",
+        "Total Dependencies", "Safe To Hard Delete", "Recommended Action",
+    ]
+    dep_ws.append(dep_headers)
+    for row in catalogues:
+        counts = [mapping.get(row.id, 0) for mapping in dependency_maps.values()]
+        total = sum(counts)
+        safe = total == 0
+        dep_ws.append([
+            row.id,
+            row.service_code,
+            row.service_name,
+            *counts,
+            total,
+            "YES" if safe else "NO",
+            "Can be reviewed for deletion" if safe else "Disable or merge; do not hard delete",
+        ])
+        fill = _SAFE_FILL if safe else _DANGER_FILL
+        for cell in dep_ws[dep_ws.max_row]:
+            cell.fill = fill
+    _style_sheet(dep_ws)
+
+    info_ws = wb.create_sheet("Export Information")
+    info_ws.append(["Field", "Value"])
+    info_ws.append(["Generated At UTC", datetime.now(timezone.utc).isoformat()])
+    info_ws.append(["Catalogue Records", len(catalogues)])
+    info_ws.append(["Purpose", "Review service catalogue, identify duplicate candidates, and inspect dependencies before disabling or merging services."])
+    info_ws.append(["Important", "A service with dependencies should not be hard deleted. Disable it or perform a controlled merge instead."])
+    _style_sheet(info_ws)
+
+    output = BytesIO()
+    wb.save(output)
+    return output.getvalue()

app/modules/services/templates/services/catalogue_list.html

@@ -12,6 +12,7 @@
       <a href="/services/templates" class="rounded-xl border border-slate-300 px-4 py-2 text-sm font-medium text-slate-700 hover:bg-slate-50">Firm Task Templates</a>
       {% if can_create %}
       <a href="/services/defaults" class="rounded-xl border border-slate-300 px-4 py-2 text-sm font-medium text-slate-700 hover:bg-slate-50">System Default Tasks</a>
+      <a href="/services/catalogue/export" class="rounded-xl border border-emerald-300 bg-emerald-50 px-4 py-2 text-sm font-medium text-emerald-800 hover:bg-emerald-100">Export Excel</a>
       <a href="/services/catalogue/new" class="rounded-xl bg-brand-600 px-4 py-2 text-sm font-medium text-white hover:bg-brand-700">Add Catalogue Service</a>
       {% endif %}
     </div>

app/modules/services/ui.py

@@ -1,5 +1,7 @@
 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
@@ -30,6 +32,7 @@ from app.modules.services.due_dates import (
     list_due_rules,
     parse_optional_date,
 )
+from app.modules.services.catalogue_export import build_service_catalogue_export
 from app.modules.services.task_documents import (
     create_task_document_requirement,
     get_task_document_requirement,
@@ -332,6 +335,28 @@ def catalogue_list(request: Request, q: str = '', category_id: int | None = None
         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/new')
 def catalogue_create_page(request: Request):
     db = CommonSessionLocal()