Files
arrr-erp/app/modules/services/default_tasks_export.py
T
2026-07-19 11:53:26 +05:30

255 lines
8.7 KiB
Python

from __future__ import annotations
from io import BytesIO
from openpyxl import Workbook
from openpyxl.styles import Alignment, Font, PatternFill
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
_HEADERS = [
"service_code",
"service_name",
"category_code",
"category_name",
"sequence_no",
"task_name",
"default_role_name",
"is_mandatory",
"requires_review",
"is_aqmm_task",
"aqmm_mandatory",
"aqmm_evidence_required",
"aqmm_manager_review_required",
"aqmm_partner_review_required",
"aqmm_review_partner_required",
"aqmm_blocks_final_release",
"aqmm_reference",
"is_active",
"description",
]
_IMPORT_HEADERS = [
"service_code",
"sequence_no",
"task_name",
"default_role_name",
"is_mandatory",
"requires_review",
"is_aqmm_task",
"aqmm_mandatory",
"aqmm_evidence_required",
"aqmm_manager_review_required",
"aqmm_partner_review_required",
"aqmm_review_partner_required",
"aqmm_blocks_final_release",
"aqmm_reference",
"is_active",
"description",
]
_HEADER_FILL = PatternFill("solid", fgColor="1F4E78")
_HEADER_FONT = Font(color="FFFFFF", bold=True)
_SUBHEADER_FILL = PatternFill("solid", fgColor="D9EAF7")
def _yes_no(value: bool) -> str:
return "TRUE" if bool(value) else "FALSE"
def _category_values(service: ServiceCatalogue) -> tuple[str, str]:
category = service.service_category
if category is not None:
return category.code or "", category.name or ""
fallback = service.category or ""
return "", fallback
def _style_sheet(ws, widths: dict[int, int]) -> None:
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)
ws.row_dimensions[1].height = 34
for column_index, width in widths.items():
ws.column_dimensions[get_column_letter(column_index)].width = width
for row in ws.iter_rows(min_row=2):
for cell in row:
cell.alignment = Alignment(vertical="top", wrap_text=True)
def _query_active_services(db: Session) -> list[ServiceCatalogue]:
return list(
db.execute(
select(ServiceCatalogue)
.where(ServiceCatalogue.is_active.is_(True))
.options(
selectinload(ServiceCatalogue.service_category),
selectinload(ServiceCatalogue.default_task_templates),
)
.order_by(
ServiceCatalogue.sort_order.asc(),
ServiceCatalogue.service_code.asc(),
ServiceCatalogue.id.asc(),
)
).scalars().unique().all()
)
def build_default_tasks_export(db: Session) -> bytes:
services = _query_active_services(db)
workbook = Workbook()
all_tasks_ws = workbook.active
all_tasks_ws.title = "All Default Tasks"
import_ws = workbook.create_sheet("Import Ready")
summary_ws = workbook.create_sheet("Service Summary")
missing_ws = workbook.create_sheet("Missing Task Templates")
info_ws = workbook.create_sheet("Export Information")
all_tasks_ws.append(_HEADERS)
import_ws.append(_IMPORT_HEADERS)
summary_ws.append([
"service_code",
"service_name",
"category_name",
"total_default_tasks",
"active_default_tasks",
"mandatory_tasks",
"review_tasks",
"aqmm_tasks",
"partner_review_tasks",
"blocks_final_release_tasks",
"status",
])
missing_ws.append(["service_code", "service_name", "category_code", "category_name", "recurrence_type", "engagement_type"])
total_tasks = 0
active_tasks = 0
services_without_tasks = 0
for service in services:
category_code, category_name = _category_values(service)
tasks = sorted(
service.default_task_templates,
key=lambda row: (row.sequence_no, row.id),
)
total_tasks += len(tasks)
active_tasks += sum(1 for task in tasks if task.is_active)
for task in tasks:
all_tasks_ws.append([
service.service_code,
service.service_name,
category_code,
category_name,
task.sequence_no,
task.task_name,
task.default_role_name or "",
_yes_no(task.is_mandatory),
_yes_no(task.requires_review),
_yes_no(task.is_aqmm_task),
_yes_no(task.aqmm_mandatory),
_yes_no(task.aqmm_evidence_required),
_yes_no(task.aqmm_manager_review_required),
_yes_no(task.aqmm_partner_review_required),
_yes_no(task.aqmm_review_partner_required),
_yes_no(task.aqmm_blocks_final_release),
task.aqmm_reference or "",
_yes_no(task.is_active),
task.description or "",
])
import_ws.append([
service.service_code,
task.sequence_no,
task.task_name,
task.default_role_name or "",
_yes_no(task.is_mandatory),
_yes_no(task.requires_review),
_yes_no(task.is_aqmm_task),
_yes_no(task.aqmm_mandatory),
_yes_no(task.aqmm_evidence_required),
_yes_no(task.aqmm_manager_review_required),
_yes_no(task.aqmm_partner_review_required),
_yes_no(task.aqmm_review_partner_required),
_yes_no(task.aqmm_blocks_final_release),
task.aqmm_reference or "",
_yes_no(task.is_active),
task.description or "",
])
if not tasks:
services_without_tasks += 1
missing_ws.append([
service.service_code,
service.service_name,
category_code,
category_name,
service.recurrence_type or "",
service.engagement_type or "",
])
summary_ws.append([
service.service_code,
service.service_name,
category_name,
len(tasks),
sum(1 for task in tasks if task.is_active),
sum(1 for task in tasks if task.is_mandatory),
sum(1 for task in tasks if task.requires_review),
sum(1 for task in tasks if task.is_aqmm_task),
sum(1 for task in tasks if task.aqmm_partner_review_required or task.aqmm_review_partner_required),
sum(1 for task in tasks if task.aqmm_blocks_final_release),
"Configured" if tasks else "No Default Tasks",
])
info_rows = [
("Scope", "Active Service Catalogue records only"),
("Active services exported", len(services)),
("Default task rows exported", total_tasks),
("Active default task rows", active_tasks),
("Services without default tasks", services_without_tasks),
("Import Ready sheet", "Matches the existing System Default Tasks import column order"),
("Historical inactive services", "Not included, so merged or inactive services do not create confusion"),
("Important", "Review edited rows before re-importing. Service codes and task sequence numbers control matching and uniqueness."),
]
info_ws.append(["Item", "Value"])
for item, value in info_rows:
info_ws.append([item, value])
_style_sheet(
all_tasks_ws,
{1: 32, 2: 42, 3: 18, 4: 26, 5: 12, 6: 44, 7: 20, 8: 14, 9: 14, 10: 14,
11: 16, 12: 20, 13: 22, 14: 22, 15: 23, 16: 22, 17: 30, 18: 12, 19: 72},
)
_style_sheet(
import_ws,
{1: 32, 2: 12, 3: 44, 4: 20, 5: 14, 6: 14, 7: 14, 8: 16, 9: 20, 10: 22,
11: 22, 12: 23, 13: 22, 14: 30, 15: 12, 16: 72},
)
_style_sheet(
summary_ws,
{1: 32, 2: 42, 3: 26, 4: 18, 5: 18, 6: 18, 7: 16, 8: 14, 9: 22, 10: 24, 11: 20},
)
_style_sheet(
missing_ws,
{1: 32, 2: 42, 3: 18, 4: 26, 5: 18, 6: 20},
)
_style_sheet(info_ws, {1: 32, 2: 95})
if services_without_tasks == 0:
missing_ws.append(["", "All active services currently have at least one default task template.", "", "", "", ""])
missing_ws.merge_cells(start_row=2, start_column=2, end_row=2, end_column=6)
missing_ws["B2"].fill = _SUBHEADER_FILL
missing_ws["B2"].font = Font(bold=True)
output = BytesIO()
workbook.save(output)
return output.getvalue()