A R R R Associates
1125 lines
46 KiB
Python
1125 lines
46 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import date, datetime
|
|
from io import BytesIO
|
|
from typing import Any
|
|
|
|
from openpyxl import Workbook, load_workbook
|
|
from openpyxl.styles import Font
|
|
from sqlalchemy import or_, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.modules.services.models import ClientServiceSubscription
|
|
from app.modules.clients.models import Client
|
|
from app.modules.core.iam.models import User
|
|
from app.modules.core.rbac.models import Role, UserRole
|
|
from app.modules.services.models import (
|
|
FirmServiceSelection,
|
|
FirmServiceTaskTemplate,
|
|
ServiceCatalogue,
|
|
ServiceCategory,
|
|
ServiceDefaultTaskTemplate,
|
|
ServiceDueDateExtension,
|
|
ServiceDueDateRule,
|
|
)
|
|
from app.modules.services.services import normalize_code, normalize_engagement_type
|
|
from app.modules.services.client_services import assessment_year_from_financial_year, normalize_financial_year, review_partner_required_for_engagement, ensure_engagement_quality_workflow, enforce_quality_gate_on_subscription
|
|
from app.modules.services.due_dates import apply_due_date_rule_to_subscription, create_due_date_extension
|
|
|
|
TRUE_VALUES = {"1", "true", "yes", "y", "on"}
|
|
FALSE_VALUES = {"0", "false", "no", "n", "off"}
|
|
|
|
CLIENT_ASSIGNMENT_COLUMNS = [
|
|
"client_code",
|
|
"service_code",
|
|
"financial_year",
|
|
"engagement_type",
|
|
"assigned_partner_email",
|
|
"assigned_manager_email",
|
|
"assigned_staff_email",
|
|
"start_date",
|
|
"end_date",
|
|
"expiry_date",
|
|
"status",
|
|
"is_active",
|
|
"remarks",
|
|
]
|
|
|
|
SERVICE_MASTER_COLUMNS = [
|
|
"category_code",
|
|
"category_name",
|
|
"service_code",
|
|
"service_name",
|
|
"recurrence_type",
|
|
"engagement_type",
|
|
"sort_order",
|
|
"applicable_individual",
|
|
"applicable_proprietorship",
|
|
"applicable_partnership",
|
|
"applicable_llp",
|
|
"applicable_company",
|
|
"applicable_trust",
|
|
"applicable_society",
|
|
"is_active",
|
|
"is_client_requestable",
|
|
"is_consultant_requestable",
|
|
"description",
|
|
]
|
|
|
|
DEFAULT_TASK_COLUMNS = [
|
|
"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",
|
|
]
|
|
|
|
FIRM_TASK_COLUMNS = [
|
|
"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",
|
|
]
|
|
|
|
DUE_DATE_RULE_COLUMNS = [
|
|
"service_code",
|
|
"rule_name",
|
|
"period_type",
|
|
"due_year_basis",
|
|
"due_day",
|
|
"due_month",
|
|
"due_month_offset",
|
|
"days_offset_after_event",
|
|
"renewal_days_before_expiry",
|
|
"sort_order",
|
|
"is_active",
|
|
"remarks",
|
|
]
|
|
|
|
DUE_DATE_EXTENSION_COLUMNS = [
|
|
"service_code",
|
|
"rule_name",
|
|
"financial_year",
|
|
"assessment_year",
|
|
"period_label",
|
|
"extended_due_date",
|
|
"notification_reference",
|
|
"notification_date",
|
|
"remarks",
|
|
]
|
|
|
|
|
|
def _clean(value: Any) -> str:
|
|
if value is None:
|
|
return ""
|
|
return str(value).strip()
|
|
|
|
|
|
def _upper(value: Any) -> str:
|
|
return _clean(value).upper()
|
|
|
|
|
|
def _bool(value: Any, default: bool = False) -> bool:
|
|
if value is None or value == "":
|
|
return default
|
|
if isinstance(value, bool):
|
|
return value
|
|
txt = str(value).strip().lower()
|
|
if txt in TRUE_VALUES:
|
|
return True
|
|
if txt in FALSE_VALUES:
|
|
return False
|
|
return default
|
|
|
|
|
|
def _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 _date(value: Any) -> date | None:
|
|
if value in (None, ""):
|
|
return None
|
|
if isinstance(value, datetime):
|
|
return value.date()
|
|
if isinstance(value, date):
|
|
return value
|
|
txt = str(value).strip()
|
|
if not txt:
|
|
return None
|
|
for fmt in ("%Y-%m-%d", "%d-%m-%Y", "%d/%m/%Y"):
|
|
try:
|
|
return datetime.strptime(txt, fmt).date()
|
|
except Exception:
|
|
pass
|
|
return date.fromisoformat(txt)
|
|
|
|
|
|
def _headers(ws) -> dict[str, int]:
|
|
row = next(ws.iter_rows(min_row=1, max_row=1, values_only=True), [])
|
|
return {str(v).strip().lower(): idx for idx, v in enumerate(row) if v is not None and str(v).strip()}
|
|
|
|
|
|
def _cell(row: tuple[Any, ...], 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 _style_template(wb: Workbook) -> bytes:
|
|
for ws in wb.worksheets:
|
|
ws.freeze_panes = "A2"
|
|
for cell in ws[1]:
|
|
cell.font = Font(bold=True)
|
|
for col in ws.columns:
|
|
col_letter = col[0].column_letter
|
|
max_len = max(len(str(c.value or "")) for c in col)
|
|
ws.column_dimensions[col_letter].width = min(max(max_len + 2, 14), 42)
|
|
out = BytesIO()
|
|
wb.save(out)
|
|
return out.getvalue()
|
|
|
|
|
|
def build_template(template_type: str) -> bytes:
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
|
|
if template_type == "client_service_assignments":
|
|
ws.title = "client_service_assignments"
|
|
ws.append(CLIENT_ASSIGNMENT_COLUMNS)
|
|
ws.append([
|
|
"CLT-001",
|
|
"GST-MONTHLY",
|
|
"2025-26",
|
|
"",
|
|
"partner@example.com",
|
|
"manager@example.com",
|
|
"staff@example.com",
|
|
"2026-04-01",
|
|
"",
|
|
"",
|
|
"active",
|
|
"TRUE",
|
|
"Monthly GST compliance assignment",
|
|
])
|
|
notes = wb.create_sheet("instructions")
|
|
notes.append(["Column", "Instruction"])
|
|
notes.append(["client_code", "Client must belong to the active firm."])
|
|
notes.append(["service_code", "Service must be enabled for the active firm."])
|
|
notes.append(["financial_year", "Use format like 2025-26. If blank, current FY is used."])
|
|
notes.append(["engagement_type", "Optional. Leave blank to copy from service catalogue. Allowed: assurance, non_assurance. Existing unlocked engagements can be updated."])
|
|
notes.append(["assigned_*_email", "Optional, but if provided must be an active user in the same firm."])
|
|
notes.append(["expiry_date", "Optional. Required for renewal-before-expiry services such as DSC renewal. Due date = expiry_date - renewal_days_before_expiry configured in due_date_rules."])
|
|
notes.append(["Partner users", "If logged in as Partner/own-only role, assigned_partner_email must be your email and client must be your client."])
|
|
|
|
elif template_type == "service_master":
|
|
ws.title = "service_master"
|
|
ws.append(SERVICE_MASTER_COLUMNS)
|
|
ws.append([
|
|
"GST",
|
|
"GST",
|
|
"GST-MONTHLY",
|
|
"GST Monthly Return Filing",
|
|
"monthly",
|
|
"non_assurance",
|
|
10,
|
|
"TRUE",
|
|
"TRUE",
|
|
"TRUE",
|
|
"TRUE",
|
|
"TRUE",
|
|
"FALSE",
|
|
"FALSE",
|
|
"TRUE",
|
|
"TRUE",
|
|
"FALSE",
|
|
"Monthly GST compliance service",
|
|
])
|
|
ws.append([
|
|
"ROC",
|
|
"ROC / MCA Compliance",
|
|
"ROC-AOC4",
|
|
"AOC-4 Filing",
|
|
"annual",
|
|
"non_assurance",
|
|
20,
|
|
"FALSE",
|
|
"FALSE",
|
|
"FALSE",
|
|
"TRUE",
|
|
"TRUE",
|
|
"FALSE",
|
|
"FALSE",
|
|
"TRUE",
|
|
"TRUE",
|
|
"FALSE",
|
|
"Annual ROC filing service. Category will be auto-created if missing.",
|
|
])
|
|
ws.append([
|
|
"DIGITAL",
|
|
"Digital Signature / Certificates",
|
|
"DSC-RENEWAL",
|
|
"DSC Renewal",
|
|
"renewal",
|
|
"non_assurance",
|
|
30,
|
|
"TRUE",
|
|
"TRUE",
|
|
"TRUE",
|
|
"TRUE",
|
|
"TRUE",
|
|
"TRUE",
|
|
"FALSE",
|
|
"TRUE",
|
|
"TRUE",
|
|
"TRUE",
|
|
"Before-expiry renewal service. Engagement expiry_date drives due date.",
|
|
])
|
|
ws_rules = wb.create_sheet("due_date_rules")
|
|
ws_rules.append(DUE_DATE_RULE_COLUMNS)
|
|
ws_rules.append([
|
|
"GST-MONTHLY",
|
|
"Monthly 20th of next month",
|
|
"monthly",
|
|
"calendar_year",
|
|
20,
|
|
"",
|
|
1,
|
|
"",
|
|
"",
|
|
10,
|
|
"TRUE",
|
|
"GSTR-3B style monthly due date rule",
|
|
])
|
|
ws_rules.append([
|
|
"ROC-AOC4",
|
|
"30 days from AGM",
|
|
"event_based",
|
|
"calendar_year",
|
|
"",
|
|
"",
|
|
0,
|
|
30,
|
|
"",
|
|
20,
|
|
"TRUE",
|
|
"Event-based rule foundation; actual event date handling can be added later.",
|
|
])
|
|
ws_rules.append([
|
|
"DSC-RENEWAL",
|
|
"Renew 30 days before expiry",
|
|
"renewal_based",
|
|
"calendar_year",
|
|
"",
|
|
"",
|
|
0,
|
|
"",
|
|
30,
|
|
30,
|
|
"TRUE",
|
|
"For DSC/FSSAI/licence renewal: engagement expiry_date minus 30 days.",
|
|
])
|
|
|
|
ws_ext = wb.create_sheet("due_date_extensions")
|
|
ws_ext.append(DUE_DATE_EXTENSION_COLUMNS)
|
|
ws_ext.append([
|
|
"GST-MONTHLY",
|
|
"Monthly 20th of next month",
|
|
"2025-26",
|
|
"2026-27",
|
|
"Apr",
|
|
"2025-05-25",
|
|
"GST notification / advisory reference",
|
|
"2025-05-18",
|
|
"Example extension row. Delete if not required.",
|
|
])
|
|
|
|
notes = wb.create_sheet("instructions")
|
|
notes.append(["Column", "Instruction"])
|
|
notes.append(["category_code", "Optional but recommended. If category_code/category_name does not exist, System Admin import auto-creates the category."])
|
|
notes.append(["category_name", "Optional if category_code is given. If blank and category is new, category_code is used as name."])
|
|
notes.append(["service_code", "Required. Unique service catalogue code. Used by due_date_rules and due_date_extensions sheets."])
|
|
notes.append(["service_name", "Required. Display name of service."])
|
|
notes.append(["recurrence_type", "Optional values like monthly, quarterly, annual, one_time."])
|
|
notes.append(["engagement_type", "Allowed: assurance, non_assurance. Old values audit/non_audit are also normalized."])
|
|
notes.append(["due_date_rules", "Optional sheet. One service can have multiple rows/rules. Existing rules are matched by service_code + rule_name."])
|
|
notes.append(["due_date_extensions", "Optional sheet. Imports extension history for the active tenant/firm and updates matching unlocked engagements."])
|
|
notes.append(["Multiple extensions", "Upload a later extended_due_date as another row. The system stores sequence/history and keeps original due date intact."])
|
|
notes.append(["Duplicate protection", "If same service/rule/FY/AY/period/extended_due_date/notification_reference already exists, upload updates/skips instead of creating another duplicate."])
|
|
notes.append(["TRUE/FALSE columns", "Use TRUE/FALSE, Yes/No, 1/0."])
|
|
|
|
elif template_type == "due_date_extensions":
|
|
ws.title = "due_date_extensions"
|
|
ws.append(DUE_DATE_EXTENSION_COLUMNS)
|
|
ws.append([
|
|
"GST-MONTHLY",
|
|
"Monthly 20th of next month",
|
|
"2025-26",
|
|
"2026-27",
|
|
"Apr",
|
|
"2025-05-25",
|
|
"GST notification / advisory reference",
|
|
"2025-05-18",
|
|
"Example extension row. Keep one row per extension notification.",
|
|
])
|
|
notes = wb.create_sheet("instructions")
|
|
notes.append(["Column", "Instruction"])
|
|
notes.append(["service_code", "Required. Existing service catalogue code."])
|
|
notes.append(["rule_name", "Optional but recommended. If blank, the active due date rule of the service is used."])
|
|
notes.append(["financial_year", "Required. Format 2025-26."])
|
|
notes.append(["assessment_year", "Optional. If blank, derived from financial_year."])
|
|
notes.append(["period_label", "Optional. Use Apr/May/Q1/Q2 etc. for monthly/quarterly rules."])
|
|
notes.append(["extended_due_date", "Required. Use YYYY-MM-DD."])
|
|
notes.append(["notification_reference", "Recommended. Circular/order/advisory reference."])
|
|
notes.append(["notification_date", "Optional. Use YYYY-MM-DD."])
|
|
notes.append(["Multiple extensions", "Every new extended_due_date is stored as a fresh sequence. Existing duplicate rows are not duplicated."])
|
|
|
|
elif template_type == "system_default_tasks":
|
|
ws.title = "system_default_tasks"
|
|
ws.append(DEFAULT_TASK_COLUMNS)
|
|
ws.append(["GST-MONTHLY", 1, "Collect data", "Staff", "TRUE", "FALSE", "TRUE", "Collect sales/purchase data"])
|
|
ws.append(["GST-MONTHLY", 2, "Review and file", "Manager", "TRUE", "TRUE", "TRUE", "Review and file return"])
|
|
|
|
elif template_type == "firm_task_templates":
|
|
ws.title = "firm_task_templates"
|
|
ws.append(FIRM_TASK_COLUMNS)
|
|
ws.append(["GST-MONTHLY", 1, "Collect data", "Staff", "TRUE", "FALSE", "TRUE", "Firm-specific data collection task"])
|
|
ws.append(["GST-MONTHLY", 2, "Partner review", "Partner", "TRUE", "TRUE", "TRUE", "Firm-specific review task"])
|
|
|
|
else:
|
|
raise ValueError("Unknown template type")
|
|
|
|
return _style_template(wb)
|
|
|
|
|
|
def _load_sheet(file_bytes: bytes, expected_sheet: str):
|
|
wb = load_workbook(BytesIO(file_bytes), data_only=True)
|
|
if expected_sheet in wb.sheetnames:
|
|
ws = wb[expected_sheet]
|
|
else:
|
|
ws = wb[wb.sheetnames[0]]
|
|
return ws, _headers(ws)
|
|
|
|
|
|
def _validate_headers(headers: dict[str, int], required: list[str]) -> list[str]:
|
|
return [h for h in required if h not in headers]
|
|
|
|
|
|
def get_user_by_email(db: Session, *, tenant_id: int, email: str) -> User | None:
|
|
if not email:
|
|
return None
|
|
return db.execute(
|
|
select(User).where(
|
|
User.tenant_id == tenant_id,
|
|
User.email == email.strip().lower(),
|
|
User.is_active.is_(True),
|
|
)
|
|
).scalar_one_or_none()
|
|
|
|
|
|
def user_has_role(db: Session, *, user_id: int, role_names: set[str]) -> bool:
|
|
return db.execute(
|
|
select(User.id)
|
|
.join(UserRole, UserRole.user_id == User.id)
|
|
.join(Role, Role.id == UserRole.role_id)
|
|
.where(User.id == user_id, Role.name.in_(role_names))
|
|
).first() is not None
|
|
|
|
|
|
def find_client(db: Session, *, tenant_id: int, client_code: str) -> Client | None:
|
|
return db.execute(
|
|
select(Client).where(
|
|
Client.tenant_id == tenant_id,
|
|
Client.client_code == client_code,
|
|
Client.is_active.is_(True),
|
|
)
|
|
).scalar_one_or_none()
|
|
|
|
|
|
def get_or_create_service_category(
|
|
db: Session,
|
|
*,
|
|
category_code: str | None = None,
|
|
category_name: str | None = None,
|
|
) -> ServiceCategory | None:
|
|
"""Return existing service category or auto-create it during System Admin service import.
|
|
|
|
Matching is forgiving: category_code first, then category_name. If neither
|
|
exists, a new active category is created. This allows one-step service
|
|
master upload without a separate category import.
|
|
"""
|
|
clean_name = _clean(category_name)
|
|
clean_code = normalize_code(category_code) if category_code else ""
|
|
|
|
if not clean_code and clean_name:
|
|
clean_code = normalize_code(clean_name)
|
|
if not clean_name and clean_code:
|
|
clean_name = clean_code
|
|
if not clean_code and not clean_name:
|
|
return None
|
|
|
|
category = None
|
|
if clean_code:
|
|
category = db.execute(
|
|
select(ServiceCategory).where(ServiceCategory.code == clean_code)
|
|
).scalar_one_or_none()
|
|
|
|
if not category and clean_name:
|
|
category = db.execute(
|
|
select(ServiceCategory).where(ServiceCategory.name == clean_name)
|
|
).scalar_one_or_none()
|
|
|
|
if category:
|
|
if clean_name and category.name != clean_name:
|
|
category.name = clean_name
|
|
if hasattr(category, "is_active"):
|
|
category.is_active = True
|
|
return category
|
|
|
|
category = ServiceCategory(
|
|
code=clean_code,
|
|
name=clean_name,
|
|
sort_order=100,
|
|
is_active=True,
|
|
)
|
|
db.add(category)
|
|
db.flush()
|
|
return category
|
|
|
|
def find_service(db: Session, *, service_code: str) -> ServiceCatalogue | None:
|
|
return db.execute(
|
|
select(ServiceCatalogue).where(ServiceCatalogue.service_code == service_code)
|
|
).scalar_one_or_none()
|
|
|
|
|
|
def find_enabled_service(db: Session, *, tenant_id: int, service_code: str):
|
|
return db.execute(
|
|
select(FirmServiceSelection, ServiceCatalogue)
|
|
.join(ServiceCatalogue, ServiceCatalogue.id == FirmServiceSelection.service_catalogue_id)
|
|
.where(
|
|
FirmServiceSelection.tenant_id == tenant_id,
|
|
FirmServiceSelection.is_enabled.is_(True),
|
|
ServiceCatalogue.service_code == service_code,
|
|
)
|
|
).first()
|
|
|
|
|
|
|
|
def find_due_date_rule(
|
|
db: Session,
|
|
*,
|
|
service_catalogue_id: int,
|
|
rule_name: str | None,
|
|
) -> ServiceDueDateRule | None:
|
|
query = select(ServiceDueDateRule).where(ServiceDueDateRule.service_catalogue_id == service_catalogue_id)
|
|
if rule_name:
|
|
query = query.where(ServiceDueDateRule.rule_name == rule_name.strip())
|
|
else:
|
|
query = query.where(ServiceDueDateRule.is_active.is_(True)).order_by(ServiceDueDateRule.sort_order.asc(), ServiceDueDateRule.id.asc())
|
|
return db.execute(query).scalars().first()
|
|
return db.execute(query).scalar_one_or_none()
|
|
|
|
|
|
def _workbook_sheet(file_bytes: bytes, sheet_name: str):
|
|
wb = load_workbook(BytesIO(file_bytes), data_only=True)
|
|
if sheet_name not in wb.sheetnames:
|
|
return None, {}
|
|
ws = wb[sheet_name]
|
|
return ws, _headers(ws)
|
|
|
|
|
|
def _extension_duplicate(
|
|
db: Session,
|
|
*,
|
|
tenant_id: int,
|
|
catalogue_id: int,
|
|
rule_id: int | None,
|
|
financial_year: str,
|
|
assessment_year: str | None,
|
|
period_label: str | None,
|
|
extended_due_date: date,
|
|
notification_reference: str | None,
|
|
) -> ServiceDueDateExtension | None:
|
|
query = select(ServiceDueDateExtension).where(
|
|
ServiceDueDateExtension.tenant_id == tenant_id,
|
|
ServiceDueDateExtension.service_catalogue_id == catalogue_id,
|
|
ServiceDueDateExtension.financial_year == financial_year,
|
|
ServiceDueDateExtension.extended_due_date == extended_due_date,
|
|
)
|
|
if rule_id:
|
|
query = query.where(ServiceDueDateExtension.due_date_rule_id == rule_id)
|
|
else:
|
|
query = query.where(ServiceDueDateExtension.due_date_rule_id.is_(None))
|
|
if assessment_year:
|
|
query = query.where(ServiceDueDateExtension.assessment_year == assessment_year)
|
|
else:
|
|
query = query.where((ServiceDueDateExtension.assessment_year.is_(None)) | (ServiceDueDateExtension.assessment_year == ""))
|
|
if period_label:
|
|
query = query.where(ServiceDueDateExtension.period_label == period_label)
|
|
else:
|
|
query = query.where((ServiceDueDateExtension.period_label.is_(None)) | (ServiceDueDateExtension.period_label == ""))
|
|
if notification_reference:
|
|
query = query.where(ServiceDueDateExtension.notification_reference == notification_reference)
|
|
return db.execute(query.order_by(ServiceDueDateExtension.id.desc())).scalars().first()
|
|
|
|
|
|
def _import_due_date_rules_from_sheet(
|
|
db: Session,
|
|
*,
|
|
current_user,
|
|
ws,
|
|
headers: dict[str, int],
|
|
update_existing: bool,
|
|
) -> tuple[int, int, int, list[dict]]:
|
|
missing = _validate_headers(headers, ["service_code", "rule_name"])
|
|
if missing:
|
|
return 0, 0, 0, [{"row": 1, "message": f"Due date rules sheet missing columns: {', '.join(missing)}"}]
|
|
|
|
created = updated = skipped = 0
|
|
errors: list[dict] = []
|
|
for row_no, row in enumerate(ws.iter_rows(min_row=2, values_only=True), start=2):
|
|
if not any(v not in (None, "") for v in row):
|
|
continue
|
|
try:
|
|
service_code = normalize_code(_cell(row, headers, "service_code"))
|
|
rule_name = _clean(_cell(row, headers, "rule_name"))
|
|
if not service_code or not rule_name:
|
|
raise ValueError("service_code and rule_name are required.")
|
|
catalogue = find_service(db, service_code=service_code)
|
|
if not catalogue:
|
|
raise ValueError("Service code not found in service catalogue.")
|
|
|
|
rule = find_due_date_rule(db, service_catalogue_id=catalogue.id, rule_name=rule_name)
|
|
if rule and not update_existing:
|
|
skipped += 1
|
|
continue
|
|
if rule:
|
|
updated += 1
|
|
else:
|
|
rule = ServiceDueDateRule(service_catalogue_id=catalogue.id, rule_name=rule_name, created_by_user_id=current_user.id)
|
|
db.add(rule)
|
|
created += 1
|
|
|
|
rule.rule_name = rule_name
|
|
rule.period_type = (_clean(_cell(row, headers, "period_type")) or "yearly").lower()
|
|
rule.due_year_basis = _clean(_cell(row, headers, "due_year_basis")) or "assessment_year_start"
|
|
rule.due_day = _int(_cell(row, headers, "due_day"), None)
|
|
rule.due_month = _int(_cell(row, headers, "due_month"), None)
|
|
rule.due_month_offset = _int(_cell(row, headers, "due_month_offset"), 0) or 0
|
|
rule.days_offset_after_event = _int(_cell(row, headers, "days_offset_after_event"), None)
|
|
rule.renewal_days_before_expiry = _int(_cell(row, headers, "renewal_days_before_expiry"), None)
|
|
rule.sort_order = _int(_cell(row, headers, "sort_order"), 100) or 100
|
|
rule.is_active = _bool(_cell(row, headers, "is_active"), True)
|
|
rule.remarks = _clean(_cell(row, headers, "remarks")) or None
|
|
rule.updated_by_user_id = current_user.id
|
|
except Exception as exc:
|
|
errors.append({"row": row_no, "message": str(exc)})
|
|
return created, updated, skipped, errors
|
|
|
|
|
|
def _import_due_date_extensions_from_sheet(
|
|
db: Session,
|
|
*,
|
|
current_user,
|
|
tenant_id: int,
|
|
ws,
|
|
headers: dict[str, int],
|
|
update_existing: bool,
|
|
) -> tuple[int, int, int, list[dict]]:
|
|
missing = _validate_headers(headers, ["service_code", "financial_year", "extended_due_date"])
|
|
if missing:
|
|
return 0, 0, 0, [{"row": 1, "message": f"Due date extensions sheet missing columns: {', '.join(missing)}"}]
|
|
|
|
created = updated = skipped = 0
|
|
errors: list[dict] = []
|
|
for row_no, row in enumerate(ws.iter_rows(min_row=2, values_only=True), start=2):
|
|
if not any(v not in (None, "") for v in row):
|
|
continue
|
|
try:
|
|
service_code = normalize_code(_cell(row, headers, "service_code"))
|
|
financial_year = normalize_financial_year(_cell(row, headers, "financial_year"))
|
|
assessment_year = _clean(_cell(row, headers, "assessment_year")) or assessment_year_from_financial_year(financial_year)
|
|
period_label = _clean(_cell(row, headers, "period_label")) or None
|
|
extended_due_date = _date(_cell(row, headers, "extended_due_date"))
|
|
notification_reference = _clean(_cell(row, headers, "notification_reference")) or None
|
|
notification_date = _date(_cell(row, headers, "notification_date"))
|
|
remarks = _clean(_cell(row, headers, "remarks")) or None
|
|
if not service_code:
|
|
raise ValueError("service_code is required.")
|
|
if not financial_year:
|
|
raise ValueError("financial_year is required.")
|
|
if not extended_due_date:
|
|
raise ValueError("extended_due_date is required and must be a valid date.")
|
|
catalogue = find_service(db, service_code=service_code)
|
|
if not catalogue:
|
|
raise ValueError("Service code not found in service catalogue.")
|
|
rule_name = _clean(_cell(row, headers, "rule_name")) or None
|
|
rule = find_due_date_rule(db, service_catalogue_id=catalogue.id, rule_name=rule_name)
|
|
if rule_name and not rule:
|
|
raise ValueError("rule_name not found for this service.")
|
|
|
|
duplicate = _extension_duplicate(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
catalogue_id=catalogue.id,
|
|
rule_id=rule.id if rule else None,
|
|
financial_year=financial_year,
|
|
assessment_year=assessment_year,
|
|
period_label=period_label,
|
|
extended_due_date=extended_due_date,
|
|
notification_reference=notification_reference,
|
|
)
|
|
if duplicate:
|
|
if not update_existing:
|
|
skipped += 1
|
|
continue
|
|
duplicate.notification_date = notification_date
|
|
duplicate.remarks = remarks
|
|
duplicate.updated_by_user_id = current_user.id
|
|
updated += 1
|
|
continue
|
|
|
|
create_due_date_extension(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
catalogue_id=catalogue.id,
|
|
due_date_rule_id=rule.id if rule else None,
|
|
financial_year=financial_year,
|
|
assessment_year=assessment_year,
|
|
period_label=period_label,
|
|
extended_due_date=extended_due_date,
|
|
notification_reference=notification_reference,
|
|
notification_date=notification_date,
|
|
remarks=remarks,
|
|
user_id=current_user.id,
|
|
)
|
|
created += 1
|
|
except Exception as exc:
|
|
errors.append({"row": row_no, "message": str(exc)})
|
|
return created, updated, skipped, errors
|
|
|
|
def import_client_service_assignments(
|
|
db: Session,
|
|
*,
|
|
current_user,
|
|
tenant_id: int,
|
|
locked_partner_id: int | None,
|
|
file_bytes: bytes,
|
|
update_existing: bool = True,
|
|
) -> dict:
|
|
ws, headers = _load_sheet(file_bytes, "client_service_assignments")
|
|
missing = _validate_headers(headers, ["client_code", "service_code"])
|
|
if missing:
|
|
return {"created": 0, "updated": 0, "skipped": 0, "errors": [{"row": 1, "message": f"Missing columns: {', '.join(missing)}"}]}
|
|
|
|
created = updated = skipped = 0
|
|
errors: list[dict] = []
|
|
|
|
for row_no, row in enumerate(ws.iter_rows(min_row=2, values_only=True), start=2):
|
|
if not any(v not in (None, "") for v in row):
|
|
continue
|
|
try:
|
|
client_code = _clean(_cell(row, headers, "client_code"))
|
|
service_code = normalize_code(_cell(row, headers, "service_code"))
|
|
financial_year = normalize_financial_year(_cell(row, headers, "financial_year"))
|
|
client = find_client(db, tenant_id=tenant_id, client_code=client_code)
|
|
enabled = find_enabled_service(db, tenant_id=tenant_id, service_code=service_code)
|
|
if not client:
|
|
raise ValueError("Client not found in active firm.")
|
|
if locked_partner_id and int(client.partner_id or 0) != int(locked_partner_id):
|
|
raise ValueError("Partner can assign services only to own clients.")
|
|
if not enabled:
|
|
raise ValueError("Service is not enabled for this firm.")
|
|
firm_selection, catalogue = enabled
|
|
|
|
partner_email = _clean(_cell(row, headers, "assigned_partner_email"))
|
|
manager_email = _clean(_cell(row, headers, "assigned_manager_email"))
|
|
staff_email = _clean(_cell(row, headers, "assigned_staff_email"))
|
|
partner = get_user_by_email(db, tenant_id=tenant_id, email=partner_email) if partner_email else None
|
|
manager = get_user_by_email(db, tenant_id=tenant_id, email=manager_email) if manager_email else None
|
|
staff = get_user_by_email(db, tenant_id=tenant_id, email=staff_email) if staff_email else None
|
|
|
|
if locked_partner_id:
|
|
if partner_email and (not partner or int(partner.id) != int(locked_partner_id)):
|
|
raise ValueError("Partner upload must assign the partner field to the logged-in partner.")
|
|
partner = db.get(User, locked_partner_id)
|
|
if partner_email and not partner:
|
|
raise ValueError("Assigned partner email not found in active firm.")
|
|
if manager_email and not manager:
|
|
raise ValueError("Assigned manager email not found in active firm.")
|
|
if staff_email and not staff:
|
|
raise ValueError("Assigned staff email not found in active firm.")
|
|
|
|
status = _clean(_cell(row, headers, "status")) or "active"
|
|
is_active = _bool(_cell(row, headers, "is_active"), status == "active")
|
|
existing = db.execute(
|
|
select(ClientServiceSubscription).where(
|
|
ClientServiceSubscription.tenant_id == tenant_id,
|
|
ClientServiceSubscription.client_id == client.id,
|
|
ClientServiceSubscription.service_catalogue_id == catalogue.id,
|
|
ClientServiceSubscription.financial_year == financial_year,
|
|
)
|
|
).scalar_one_or_none()
|
|
|
|
if existing and not update_existing:
|
|
skipped += 1
|
|
continue
|
|
if existing:
|
|
sub = existing
|
|
updated += 1
|
|
else:
|
|
sub = ClientServiceSubscription(
|
|
tenant_id=tenant_id,
|
|
client_id=client.id,
|
|
service_catalogue_id=catalogue.id,
|
|
financial_year=financial_year,
|
|
assessment_year=assessment_year_from_financial_year(financial_year),
|
|
created_by_user_id=current_user.id,
|
|
)
|
|
db.add(sub)
|
|
created += 1
|
|
|
|
if getattr(sub, "is_locked", False):
|
|
raise ValueError("Existing engagement for this client/service/year is locked and cannot be updated.")
|
|
sub.financial_year = financial_year
|
|
sub.assessment_year = assessment_year_from_financial_year(financial_year)
|
|
imported_engagement_type = _clean(_cell(row, headers, "engagement_type"))
|
|
sub.engagement_type = normalize_engagement_type(imported_engagement_type) if imported_engagement_type else (getattr(catalogue, "engagement_type", "non_assurance") or "non_assurance")
|
|
sub.branch_id = client.branch_id
|
|
sub.firm_service_selection_id = firm_selection.id
|
|
sub.assigned_partner_user_id = partner.id if partner else (client.partner_id or None)
|
|
sub.assigned_manager_user_id = manager.id if manager else None
|
|
sub.assigned_staff_user_id = staff.id if staff else None
|
|
sub.review_partner_user_id = client.default_review_partner_user_id if review_partner_required_for_engagement(db, tenant_id=tenant_id, engagement_type=sub.engagement_type) else None
|
|
sub.start_date = _date(_cell(row, headers, "start_date"))
|
|
sub.end_date = _date(_cell(row, headers, "end_date"))
|
|
sub.expiry_date = _date(_cell(row, headers, "expiry_date"))
|
|
sub.status = status
|
|
sub.is_active = is_active
|
|
sub.remarks = _clean(_cell(row, headers, "remarks")) or None
|
|
sub.updated_by_user_id = current_user.id
|
|
apply_due_date_rule_to_subscription(db, sub)
|
|
ensure_engagement_quality_workflow(db, subscription=sub, actor_user_id=current_user.id, create_declarations=False)
|
|
enforce_quality_gate_on_subscription(sub)
|
|
except Exception as exc:
|
|
errors.append({"row": row_no, "message": str(exc)})
|
|
|
|
if not errors:
|
|
db.commit()
|
|
else:
|
|
db.rollback()
|
|
return {"created": created if not errors else 0, "updated": updated if not errors else 0, "skipped": skipped, "errors": errors}
|
|
|
|
|
|
def import_service_master(
|
|
db: Session,
|
|
*,
|
|
current_user,
|
|
file_bytes: bytes,
|
|
update_existing: bool = True,
|
|
tenant_id: int | None = None,
|
|
) -> dict:
|
|
wb = load_workbook(BytesIO(file_bytes), data_only=True)
|
|
if "service_master" in wb.sheetnames:
|
|
ws = wb["service_master"]
|
|
else:
|
|
ws = wb[wb.sheetnames[0]]
|
|
headers = _headers(ws)
|
|
missing = _validate_headers(headers, ["service_code", "service_name"])
|
|
if missing:
|
|
return {"created": 0, "updated": 0, "skipped": 0, "errors": [{"row": 1, "message": f"Missing columns: {', '.join(missing)}"}]}
|
|
|
|
created = updated = skipped = 0
|
|
errors: list[dict] = []
|
|
for row_no, row in enumerate(ws.iter_rows(min_row=2, values_only=True), start=2):
|
|
if not any(v not in (None, "") for v in row):
|
|
continue
|
|
try:
|
|
service_code = normalize_code(_cell(row, headers, "service_code"))
|
|
service_name = _clean(_cell(row, headers, "service_name"))
|
|
if not service_code or not service_name:
|
|
raise ValueError("service_code and service_name are required.")
|
|
|
|
category_id = None
|
|
category_name = _clean(_cell(row, headers, "category_name"))
|
|
category_code = _clean(_cell(row, headers, "category_code"))
|
|
category = get_or_create_service_category(
|
|
db,
|
|
category_code=category_code,
|
|
category_name=category_name,
|
|
)
|
|
if category:
|
|
category_id = category.id
|
|
category_name = category.name
|
|
|
|
catalogue = db.execute(select(ServiceCatalogue).where(ServiceCatalogue.service_code == service_code)).scalar_one_or_none()
|
|
if catalogue and not update_existing:
|
|
skipped += 1
|
|
continue
|
|
if catalogue:
|
|
updated += 1
|
|
else:
|
|
catalogue = ServiceCatalogue(service_code=service_code, service_name=service_name, created_by_user_id=current_user.id)
|
|
db.add(catalogue)
|
|
created += 1
|
|
|
|
catalogue.service_name = service_name
|
|
if hasattr(catalogue, "category_id"):
|
|
catalogue.category_id = category_id
|
|
catalogue.category = category_name or None
|
|
if hasattr(catalogue, "recurrence_type"):
|
|
catalogue.recurrence_type = _clean(_cell(row, headers, "recurrence_type")) or None
|
|
if hasattr(catalogue, "engagement_type"):
|
|
catalogue.engagement_type = normalize_engagement_type(_cell(row, headers, "engagement_type"))
|
|
if hasattr(catalogue, "sort_order"):
|
|
catalogue.sort_order = _int(_cell(row, headers, "sort_order"), 100) or 100
|
|
catalogue.description = _clean(_cell(row, headers, "description")) or None
|
|
for flag in [
|
|
"applicable_individual", "applicable_proprietorship", "applicable_partnership", "applicable_llp",
|
|
"applicable_company", "applicable_trust", "applicable_society",
|
|
"is_active", "is_client_requestable", "is_consultant_requestable",
|
|
]:
|
|
if hasattr(catalogue, flag) and flag in headers:
|
|
setattr(catalogue, flag, _bool(_cell(row, headers, flag), getattr(catalogue, flag, False)))
|
|
catalogue.updated_by_user_id = current_user.id
|
|
except Exception as exc:
|
|
errors.append({"row": row_no, "message": str(exc)})
|
|
|
|
rule_created = rule_updated = rule_skipped = 0
|
|
ext_created = ext_updated = ext_skipped = 0
|
|
if not errors and "due_date_rules" in wb.sheetnames:
|
|
r_ws = wb["due_date_rules"]
|
|
c, u, sk, rule_errors = _import_due_date_rules_from_sheet(
|
|
db,
|
|
current_user=current_user,
|
|
ws=r_ws,
|
|
headers=_headers(r_ws),
|
|
update_existing=update_existing,
|
|
)
|
|
rule_created, rule_updated, rule_skipped = c, u, sk
|
|
errors.extend(rule_errors)
|
|
|
|
if not errors and "due_date_extensions" in wb.sheetnames:
|
|
if not tenant_id:
|
|
errors.append({"row": 1, "message": "due_date_extensions sheet requires an active tenant/firm context."})
|
|
else:
|
|
e_ws = wb["due_date_extensions"]
|
|
c, u, sk, ext_errors = _import_due_date_extensions_from_sheet(
|
|
db,
|
|
current_user=current_user,
|
|
tenant_id=tenant_id,
|
|
ws=e_ws,
|
|
headers=_headers(e_ws),
|
|
update_existing=update_existing,
|
|
)
|
|
ext_created, ext_updated, ext_skipped = c, u, sk
|
|
errors.extend(ext_errors)
|
|
|
|
if not errors:
|
|
db.commit()
|
|
else:
|
|
db.rollback()
|
|
return {
|
|
"created": (created + rule_created + ext_created) if not errors else 0,
|
|
"updated": (updated + rule_updated + ext_updated) if not errors else 0,
|
|
"skipped": skipped + rule_skipped + ext_skipped,
|
|
"errors": errors,
|
|
"breakdown": {
|
|
"services_created": created,
|
|
"services_updated": updated,
|
|
"due_rules_created": rule_created,
|
|
"due_rules_updated": rule_updated,
|
|
"due_extensions_created": ext_created,
|
|
"due_extensions_updated": ext_updated,
|
|
},
|
|
}
|
|
|
|
|
|
def import_due_date_extensions(
|
|
db: Session,
|
|
*,
|
|
current_user,
|
|
tenant_id: int,
|
|
file_bytes: bytes,
|
|
update_existing: bool = True,
|
|
) -> dict:
|
|
ws, headers = _load_sheet(file_bytes, "due_date_extensions")
|
|
created, updated, skipped, errors = _import_due_date_extensions_from_sheet(
|
|
db,
|
|
current_user=current_user,
|
|
tenant_id=tenant_id,
|
|
ws=ws,
|
|
headers=headers,
|
|
update_existing=update_existing,
|
|
)
|
|
if not errors:
|
|
db.commit()
|
|
else:
|
|
db.rollback()
|
|
return {"created": created if not errors else 0, "updated": updated if not errors else 0, "skipped": skipped, "errors": errors}
|
|
|
|
|
|
def import_system_default_tasks(db: Session, *, current_user, file_bytes: bytes, update_existing: bool = True) -> dict:
|
|
ws, headers = _load_sheet(file_bytes, "system_default_tasks")
|
|
missing = _validate_headers(headers, ["service_code", "sequence_no", "task_name"])
|
|
if missing:
|
|
return {"created": 0, "updated": 0, "skipped": 0, "errors": [{"row": 1, "message": f"Missing columns: {', '.join(missing)}"}]}
|
|
created = updated = skipped = 0
|
|
errors: list[dict] = []
|
|
for row_no, row in enumerate(ws.iter_rows(min_row=2, values_only=True), start=2):
|
|
if not any(v not in (None, "") for v in row):
|
|
continue
|
|
try:
|
|
service_code = normalize_code(_cell(row, headers, "service_code"))
|
|
sequence_no = _int(_cell(row, headers, "sequence_no"), None)
|
|
task_name = _clean(_cell(row, headers, "task_name"))
|
|
catalogue = find_service(db, service_code=service_code)
|
|
if not catalogue:
|
|
raise ValueError("Service code not found in service catalogue.")
|
|
if not sequence_no or not task_name:
|
|
raise ValueError("sequence_no and task_name are required.")
|
|
task = db.execute(
|
|
select(ServiceDefaultTaskTemplate).where(
|
|
ServiceDefaultTaskTemplate.service_catalogue_id == catalogue.id,
|
|
ServiceDefaultTaskTemplate.sequence_no == sequence_no,
|
|
)
|
|
).scalar_one_or_none()
|
|
if task and not update_existing:
|
|
skipped += 1
|
|
continue
|
|
if task:
|
|
updated += 1
|
|
else:
|
|
task = ServiceDefaultTaskTemplate(service_catalogue_id=catalogue.id, sequence_no=sequence_no, task_name=task_name)
|
|
db.add(task)
|
|
created += 1
|
|
task.task_name = task_name
|
|
task.description = _clean(_cell(row, headers, "description")) or None
|
|
task.default_role_name = _clean(_cell(row, headers, "default_role_name")) or None
|
|
task.is_mandatory = _bool(_cell(row, headers, "is_mandatory"), True)
|
|
task.requires_review = _bool(_cell(row, headers, "requires_review"), False)
|
|
task.task_category = _clean(_cell(row, headers, "task_category")) or None
|
|
task.response_required = _bool(_cell(row, headers, "response_required"), False)
|
|
task.response_type = (_clean(_cell(row, headers, "response_type")) or "NONE").upper()
|
|
if task.response_type not in {"NONE", "YES_NO_NA", "YES_NO", "TEXT", "NUMBER", "DATE"}:
|
|
raise ValueError("response_type must be NONE, YES_NO_NA, YES_NO, TEXT, NUMBER or DATE.")
|
|
task.evidence_required = _bool(_cell(row, headers, "evidence_required"), False)
|
|
task.remarks_required_if_no = _bool(_cell(row, headers, "remarks_required_if_no"), False)
|
|
task.is_aqmm_task = _bool(_cell(row, headers, "is_aqmm_task"), False)
|
|
task.aqmm_mandatory = _bool(_cell(row, headers, "aqmm_mandatory"), False)
|
|
task.aqmm_evidence_required = _bool(_cell(row, headers, "aqmm_evidence_required"), False)
|
|
task.aqmm_manager_review_required = _bool(_cell(row, headers, "aqmm_manager_review_required"), False)
|
|
task.aqmm_partner_review_required = _bool(_cell(row, headers, "aqmm_partner_review_required"), False)
|
|
task.aqmm_review_partner_required = _bool(_cell(row, headers, "aqmm_review_partner_required"), False)
|
|
task.aqmm_blocks_final_release = _bool(_cell(row, headers, "aqmm_blocks_final_release"), False)
|
|
task.aqmm_reference = _clean(_cell(row, headers, "aqmm_reference")) or None
|
|
task.is_active = _bool(_cell(row, headers, "is_active"), True)
|
|
except Exception as exc:
|
|
errors.append({"row": row_no, "message": str(exc)})
|
|
if not errors:
|
|
db.commit()
|
|
else:
|
|
db.rollback()
|
|
return {"created": created if not errors else 0, "updated": updated if not errors else 0, "skipped": skipped, "errors": errors}
|
|
|
|
|
|
def import_firm_task_templates(db: Session, *, current_user, tenant_id: int, file_bytes: bytes, update_existing: bool = True) -> dict:
|
|
ws, headers = _load_sheet(file_bytes, "firm_task_templates")
|
|
missing = _validate_headers(headers, ["service_code", "sequence_no", "task_name"])
|
|
if missing:
|
|
return {"created": 0, "updated": 0, "skipped": 0, "errors": [{"row": 1, "message": f"Missing columns: {', '.join(missing)}"}]}
|
|
created = updated = skipped = 0
|
|
errors: list[dict] = []
|
|
for row_no, row in enumerate(ws.iter_rows(min_row=2, values_only=True), start=2):
|
|
if not any(v not in (None, "") for v in row):
|
|
continue
|
|
try:
|
|
service_code = normalize_code(_cell(row, headers, "service_code"))
|
|
sequence_no = _int(_cell(row, headers, "sequence_no"), None)
|
|
task_name = _clean(_cell(row, headers, "task_name"))
|
|
enabled = find_enabled_service(db, tenant_id=tenant_id, service_code=service_code)
|
|
if not enabled:
|
|
raise ValueError("Service is not enabled for this firm.")
|
|
firm_selection, catalogue = enabled
|
|
if not sequence_no or not task_name:
|
|
raise ValueError("sequence_no and task_name are required.")
|
|
task = db.execute(
|
|
select(FirmServiceTaskTemplate).where(
|
|
FirmServiceTaskTemplate.tenant_id == tenant_id,
|
|
FirmServiceTaskTemplate.service_catalogue_id == catalogue.id,
|
|
FirmServiceTaskTemplate.sequence_no == sequence_no,
|
|
)
|
|
).scalar_one_or_none()
|
|
if task and not update_existing:
|
|
skipped += 1
|
|
continue
|
|
if task:
|
|
updated += 1
|
|
else:
|
|
task = FirmServiceTaskTemplate(
|
|
tenant_id=tenant_id,
|
|
service_catalogue_id=catalogue.id,
|
|
sequence_no=sequence_no,
|
|
task_name=task_name,
|
|
created_by_user_id=current_user.id,
|
|
)
|
|
db.add(task)
|
|
created += 1
|
|
task.task_name = task_name
|
|
task.description = _clean(_cell(row, headers, "description")) or None
|
|
task.default_role_name = _clean(_cell(row, headers, "default_role_name")) or None
|
|
task.is_mandatory = _bool(_cell(row, headers, "is_mandatory"), True)
|
|
task.requires_review = _bool(_cell(row, headers, "requires_review"), False)
|
|
task.task_category = _clean(_cell(row, headers, "task_category")) or None
|
|
task.response_required = _bool(_cell(row, headers, "response_required"), False)
|
|
task.response_type = (_clean(_cell(row, headers, "response_type")) or "NONE").upper()
|
|
if task.response_type not in {"NONE", "YES_NO_NA", "YES_NO", "TEXT", "NUMBER", "DATE"}:
|
|
raise ValueError("response_type must be NONE, YES_NO_NA, YES_NO, TEXT, NUMBER or DATE.")
|
|
task.evidence_required = _bool(_cell(row, headers, "evidence_required"), False)
|
|
task.remarks_required_if_no = _bool(_cell(row, headers, "remarks_required_if_no"), False)
|
|
task.is_aqmm_task = _bool(_cell(row, headers, "is_aqmm_task"), False)
|
|
task.aqmm_mandatory = _bool(_cell(row, headers, "aqmm_mandatory"), False)
|
|
task.aqmm_evidence_required = _bool(_cell(row, headers, "aqmm_evidence_required"), False)
|
|
task.aqmm_manager_review_required = _bool(_cell(row, headers, "aqmm_manager_review_required"), False)
|
|
task.aqmm_partner_review_required = _bool(_cell(row, headers, "aqmm_partner_review_required"), False)
|
|
task.aqmm_review_partner_required = _bool(_cell(row, headers, "aqmm_review_partner_required"), False)
|
|
task.aqmm_blocks_final_release = _bool(_cell(row, headers, "aqmm_blocks_final_release"), False)
|
|
task.aqmm_reference = _clean(_cell(row, headers, "aqmm_reference")) or None
|
|
task.is_active = _bool(_cell(row, headers, "is_active"), True)
|
|
task.updated_by_user_id = current_user.id
|
|
except Exception as exc:
|
|
errors.append({"row": row_no, "message": str(exc)})
|
|
if not errors:
|
|
db.commit()
|
|
else:
|
|
db.rollback()
|
|
return {"created": created if not errors else 0, "updated": updated if not errors else 0, "skipped": skipped, "errors": errors}
|