A R R R Associates
306 lines
11 KiB
Python
306 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
from io import BytesIO
|
|
from typing import Any
|
|
|
|
from openpyxl import Workbook, load_workbook
|
|
from openpyxl.styles import Font
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.modules.services.models import FirmServiceTaskTemplate, ServiceCatalogue
|
|
|
|
|
|
def build_import_template_workbook() -> bytes:
|
|
wb = Workbook()
|
|
|
|
ws_services = wb.active
|
|
ws_services.title = "service_catalogue"
|
|
service_headers = [
|
|
"service_code",
|
|
"service_name",
|
|
"category",
|
|
"description",
|
|
"recurrence_type",
|
|
"is_active",
|
|
"is_client_requestable",
|
|
"is_consultant_requestable",
|
|
]
|
|
ws_services.append(service_headers)
|
|
ws_services.append([
|
|
"GST-MONTHLY",
|
|
"GST Monthly Return Filing",
|
|
"GST",
|
|
"Monthly GST compliance service",
|
|
"MONTHLY",
|
|
"TRUE",
|
|
"TRUE",
|
|
"FALSE",
|
|
])
|
|
|
|
ws_tasks = wb.create_sheet("firm_task_templates")
|
|
task_headers = [
|
|
"service_code",
|
|
"sequence_no",
|
|
"task_name",
|
|
"description",
|
|
"default_role_name",
|
|
"sla_days",
|
|
"is_mandatory",
|
|
"requires_review",
|
|
"is_active",
|
|
]
|
|
ws_tasks.append(task_headers)
|
|
ws_tasks.append([
|
|
"GST-MONTHLY",
|
|
1,
|
|
"Collect Purchase and Sales Data",
|
|
"Collect source data from client",
|
|
"Staff",
|
|
3,
|
|
"TRUE",
|
|
"FALSE",
|
|
"TRUE",
|
|
])
|
|
ws_tasks.append([
|
|
"GST-MONTHLY",
|
|
2,
|
|
"Review and File Return",
|
|
"Manager review and final filing",
|
|
"Partner",
|
|
2,
|
|
"TRUE",
|
|
"TRUE",
|
|
"TRUE",
|
|
])
|
|
|
|
for ws in [ws_services, ws_tasks]:
|
|
for cell in ws[1]:
|
|
cell.font = Font(bold=True)
|
|
for col in ws.columns:
|
|
max_len = 0
|
|
col_letter = col[0].column_letter
|
|
for cell in col:
|
|
val = "" if cell.value is None else str(cell.value)
|
|
max_len = max(max_len, len(val))
|
|
ws.column_dimensions[col_letter].width = min(max(max_len + 2, 14), 40)
|
|
|
|
out = BytesIO()
|
|
wb.save(out)
|
|
return out.getvalue()
|
|
|
|
|
|
def _norm_text(value: Any) -> str:
|
|
if value is None:
|
|
return ""
|
|
return str(value).strip()
|
|
|
|
|
|
def _norm_upper(value: Any) -> str:
|
|
return _norm_text(value).upper()
|
|
|
|
|
|
def _norm_bool(value: Any, default: bool = False) -> bool:
|
|
if value is None or value == "":
|
|
return default
|
|
if isinstance(value, bool):
|
|
return value
|
|
return str(value).strip().lower() in {"1", "true", "yes", "y", "on"}
|
|
|
|
|
|
def _norm_int(value: Any, default: int | None = None) -> int | None:
|
|
if value is None or value == "":
|
|
return default
|
|
try:
|
|
return int(value)
|
|
except Exception:
|
|
return default
|
|
|
|
|
|
def _sheet_headers(ws) -> dict[str, int]:
|
|
headers = {}
|
|
first_row = next(ws.iter_rows(min_row=1, max_row=1, values_only=True), [])
|
|
for idx, val in enumerate(first_row):
|
|
key = _norm_text(val).lower()
|
|
if key:
|
|
headers[key] = idx
|
|
return headers
|
|
|
|
|
|
def _cell(row: tuple, headers: dict[str, int], key: str) -> Any:
|
|
idx = headers.get(key.lower())
|
|
if idx is None or idx >= len(row):
|
|
return None
|
|
return row[idx]
|
|
|
|
|
|
def parse_import_workbook(file_bytes: bytes) -> dict:
|
|
wb = load_workbook(BytesIO(file_bytes), data_only=True)
|
|
errors: list[str] = []
|
|
catalogue_rows: list[dict] = []
|
|
task_rows: list[dict] = []
|
|
|
|
if "service_catalogue" not in wb.sheetnames:
|
|
return {"ok": False, "errors": ["Workbook must contain a sheet named 'service_catalogue'."], "catalogue_rows": [], "task_rows": []}
|
|
|
|
ws_services = wb["service_catalogue"]
|
|
headers = _sheet_headers(ws_services)
|
|
for h in ["service_code", "service_name"]:
|
|
if h not in headers:
|
|
errors.append(f"Service catalogue sheet missing required column: {h}")
|
|
|
|
for row_no, row in enumerate(ws_services.iter_rows(min_row=2, values_only=True), start=2):
|
|
service_code = _norm_upper(_cell(row, headers, "service_code"))
|
|
service_name = _norm_text(_cell(row, headers, "service_name"))
|
|
if not service_code and not service_name:
|
|
continue
|
|
if not service_code:
|
|
errors.append(f"Service catalogue row {row_no}: service_code is required.")
|
|
continue
|
|
if not service_name:
|
|
errors.append(f"Service catalogue row {row_no}: service_name is required.")
|
|
continue
|
|
catalogue_rows.append({
|
|
"service_code": service_code,
|
|
"service_name": service_name,
|
|
"category": _norm_text(_cell(row, headers, "category")) or None,
|
|
"description": _norm_text(_cell(row, headers, "description")) or None,
|
|
"recurrence_type": _norm_text(_cell(row, headers, "recurrence_type")) or None,
|
|
"is_active": _norm_bool(_cell(row, headers, "is_active"), True),
|
|
"is_client_requestable": _norm_bool(_cell(row, headers, "is_client_requestable"), False),
|
|
"is_consultant_requestable": _norm_bool(_cell(row, headers, "is_consultant_requestable"), False),
|
|
})
|
|
|
|
if "firm_task_templates" in wb.sheetnames:
|
|
ws_tasks = wb["firm_task_templates"]
|
|
task_headers = _sheet_headers(ws_tasks)
|
|
for h in ["service_code", "sequence_no", "task_name"]:
|
|
if h not in task_headers:
|
|
errors.append(f"Firm task templates sheet missing required column: {h}")
|
|
for row_no, row in enumerate(ws_tasks.iter_rows(min_row=2, values_only=True), start=2):
|
|
service_code = _norm_upper(_cell(row, task_headers, "service_code"))
|
|
task_name = _norm_text(_cell(row, task_headers, "task_name"))
|
|
sequence_no = _norm_int(_cell(row, task_headers, "sequence_no"))
|
|
if not service_code and not task_name:
|
|
continue
|
|
if not service_code:
|
|
errors.append(f"Firm task templates row {row_no}: service_code is required.")
|
|
continue
|
|
if not task_name:
|
|
errors.append(f"Firm task templates row {row_no}: task_name is required.")
|
|
continue
|
|
if sequence_no is None:
|
|
errors.append(f"Firm task templates row {row_no}: sequence_no must be numeric.")
|
|
continue
|
|
task_rows.append({
|
|
"service_code": service_code,
|
|
"sequence_no": sequence_no,
|
|
"task_name": task_name,
|
|
"description": _norm_text(_cell(row, task_headers, "description")) or None,
|
|
"default_role_name": _norm_text(_cell(row, task_headers, "default_role_name")) or None,
|
|
"sla_days": _norm_int(_cell(row, task_headers, "sla_days")),
|
|
"is_mandatory": _norm_bool(_cell(row, task_headers, "is_mandatory"), True),
|
|
"requires_review": _norm_bool(_cell(row, task_headers, "requires_review"), False),
|
|
"is_active": _norm_bool(_cell(row, task_headers, "is_active"), True),
|
|
})
|
|
|
|
service_codes = {row["service_code"] for row in catalogue_rows}
|
|
for row in task_rows:
|
|
if row["service_code"] not in service_codes:
|
|
errors.append(f"Task row for service_code '{row['service_code']}' does not match any service in service_catalogue sheet.")
|
|
|
|
return {"ok": len(errors) == 0, "errors": errors, "catalogue_rows": catalogue_rows, "task_rows": task_rows}
|
|
|
|
|
|
def apply_import_payload(
|
|
db: Session,
|
|
*,
|
|
tenant_id: int,
|
|
branch_id: int | None,
|
|
actor_user_id: int,
|
|
catalogue_rows: list[dict],
|
|
task_rows: list[dict],
|
|
) -> dict:
|
|
created_catalogue = 0
|
|
updated_catalogue = 0
|
|
created_tasks = 0
|
|
updated_tasks = 0
|
|
catalogue_map: dict[str, ServiceCatalogue] = {}
|
|
|
|
for row in catalogue_rows:
|
|
service = db.execute(select(ServiceCatalogue).where(ServiceCatalogue.service_code == row["service_code"])).scalar_one_or_none()
|
|
if service is None:
|
|
service = ServiceCatalogue(
|
|
service_code=row["service_code"],
|
|
service_name=row["service_name"],
|
|
category=row["category"],
|
|
description=row["description"],
|
|
recurrence_type=row["recurrence_type"],
|
|
is_active=row["is_active"],
|
|
is_client_requestable=row["is_client_requestable"],
|
|
is_consultant_requestable=row["is_consultant_requestable"],
|
|
created_by_user_id=actor_user_id,
|
|
updated_by_user_id=actor_user_id,
|
|
)
|
|
db.add(service)
|
|
db.flush()
|
|
created_catalogue += 1
|
|
else:
|
|
service.service_name = row["service_name"]
|
|
service.category = row["category"]
|
|
service.description = row["description"]
|
|
service.recurrence_type = row["recurrence_type"]
|
|
service.is_active = row["is_active"]
|
|
service.is_client_requestable = row["is_client_requestable"]
|
|
service.is_consultant_requestable = row["is_consultant_requestable"]
|
|
service.updated_by_user_id = actor_user_id
|
|
updated_catalogue += 1
|
|
catalogue_map[row["service_code"]] = service
|
|
|
|
for row in task_rows:
|
|
service = catalogue_map[row["service_code"]]
|
|
task = db.execute(
|
|
select(FirmServiceTaskTemplate).where(
|
|
FirmServiceTaskTemplate.tenant_id == tenant_id,
|
|
FirmServiceTaskTemplate.service_catalogue_id == service.id,
|
|
FirmServiceTaskTemplate.sequence_no == row["sequence_no"],
|
|
)
|
|
).scalar_one_or_none()
|
|
if task is None:
|
|
task = FirmServiceTaskTemplate(
|
|
tenant_id=tenant_id,
|
|
branch_id=branch_id,
|
|
service_catalogue_id=service.id,
|
|
sequence_no=row["sequence_no"],
|
|
task_name=row["task_name"],
|
|
description=row["description"],
|
|
default_role_name=row["default_role_name"],
|
|
sla_days=row["sla_days"],
|
|
is_mandatory=row["is_mandatory"],
|
|
requires_review=row["requires_review"],
|
|
is_active=row["is_active"],
|
|
created_by_user_id=actor_user_id,
|
|
updated_by_user_id=actor_user_id,
|
|
)
|
|
db.add(task)
|
|
created_tasks += 1
|
|
else:
|
|
task.task_name = row["task_name"]
|
|
task.description = row["description"]
|
|
task.default_role_name = row["default_role_name"]
|
|
task.sla_days = row["sla_days"]
|
|
task.is_mandatory = row["is_mandatory"]
|
|
task.requires_review = row["requires_review"]
|
|
task.is_active = row["is_active"]
|
|
task.branch_id = branch_id
|
|
task.updated_by_user_id = actor_user_id
|
|
updated_tasks += 1
|
|
|
|
db.commit()
|
|
return {
|
|
"created_catalogue": created_catalogue,
|
|
"updated_catalogue": updated_catalogue,
|
|
"created_tasks": created_tasks,
|
|
"updated_tasks": updated_tasks,
|
|
}
|