diff --git a/app/modules/services/catalogue_export.py b/app/modules/services/catalogue_export.py new file mode 100644 index 0000000..66a53f5 --- /dev/null +++ b/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() diff --git a/app/modules/services/templates/services/catalogue_list.html b/app/modules/services/templates/services/catalogue_list.html index 987925e..c8d71da 100644 --- a/app/modules/services/templates/services/catalogue_list.html +++ b/app/modules/services/templates/services/catalogue_list.html @@ -12,6 +12,7 @@ Firm Task Templates {% if can_create %} System Default Tasks + Export Excel Add Catalogue Service {% endif %} diff --git a/app/modules/services/ui.py b/app/modules/services/ui.py index 9cf7c9a..f9ee3a3 100644 --- a/app/modules/services/ui.py +++ b/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()