A R R R Associates
297 lines
10 KiB
Python
297 lines
10 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, ServiceTaskCategory
|
|
|
|
|
|
_HEADERS = [
|
|
"system_task_id",
|
|
"service_code",
|
|
"service_name",
|
|
"category_code",
|
|
"category_name",
|
|
"sequence_no",
|
|
"task_name",
|
|
"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",
|
|
"aqmm_manager_review_required",
|
|
"aqmm_partner_review_required",
|
|
"aqmm_review_partner_required",
|
|
"aqmm_blocks_final_release",
|
|
"aqmm_reference",
|
|
"is_active",
|
|
"description",
|
|
]
|
|
|
|
_IMPORT_HEADERS = [
|
|
"system_task_id",
|
|
"service_code",
|
|
"sequence_no",
|
|
"task_name",
|
|
"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",
|
|
"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, *, catalogue_id: int | None = None) -> list[ServiceCatalogue]:
|
|
stmt = select(ServiceCatalogue).where(ServiceCatalogue.is_active.is_(True))
|
|
if catalogue_id is not None:
|
|
stmt = stmt.where(ServiceCatalogue.id == catalogue_id)
|
|
return list(
|
|
db.execute(
|
|
stmt
|
|
.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, *, catalogue_id: int | None = None) -> bytes:
|
|
services = _query_active_services(db, catalogue_id=catalogue_id)
|
|
|
|
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"])
|
|
|
|
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
|
|
|
|
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([
|
|
task.id,
|
|
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),
|
|
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",
|
|
_yes_no(task.evidence_required),
|
|
_yes_no(task.remarks_required_if_no),
|
|
getattr(task, "task_tool_code", "NONE") or "NONE",
|
|
_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([
|
|
task.id,
|
|
service.service_code,
|
|
task.sequence_no,
|
|
task.task_name,
|
|
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",
|
|
_yes_no(task.evidence_required),
|
|
_yes_no(task.remarks_required_if_no),
|
|
getattr(task, "task_tool_code", "NONE") or "NONE",
|
|
_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", "Includes stable system_task_id so renaming/resequencing can update the correct task safely"),
|
|
("Historical inactive services", "Not included, so merged or inactive services do not create confusion"),
|
|
("Important", "Keep system_task_id unchanged for existing tasks. New rows may leave it blank. Full Synchronization can retire existing tasks omitted from the workbook."),
|
|
]
|
|
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: 24,
|
|
11: 18, 12: 18, 13: 18, 14: 20, 15: 14, 16: 16, 17: 20, 18: 22, 19: 22, 20: 23,
|
|
21: 22, 22: 30, 23: 12, 24: 72},
|
|
)
|
|
_style_sheet(
|
|
import_ws,
|
|
{1: 32, 2: 12, 3: 44, 4: 20, 5: 14, 6: 14, 7: 24, 8: 18, 9: 18, 10: 18,
|
|
11: 20, 12: 14, 13: 16, 14: 20, 15: 22, 16: 22, 17: 23, 18: 22, 19: 30, 20: 12, 21: 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()
|