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.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.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}