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,
    attach_engagement_to_plan,
    normalize_financial_year,
    normalize_period_label,
    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
from app.modules.services.execution import generate_tasks_for_subscription_if_ready

TRUE_VALUES = {"1", "true", "yes", "y", "on"}
FALSE_VALUES = {"0", "false", "no", "n", "off"}

CLIENT_ASSIGNMENT_COLUMNS = [
    "client_code",
    "service_code",
    "financial_year",
    "period_label",
    "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
            period_label = normalize_period_label(
                _clean(_cell(row, headers, "period_label")),
                financial_year=financial_year,
                recurrence_type=getattr(catalogue, "recurrence_type", None),
            )

            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,
                    ClientServiceSubscription.period_label == period_label,
                )
            ).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,
                    period_label=period_label,
                    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.period_label = period_label
            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
            attach_engagement_to_plan(
                db, engagement=sub, client=client, catalogue=catalogue,
                firm_selection=firm_selection, actor_user_id=current_user.id,
            )
            apply_due_date_rule_to_subscription(db, sub, force=True)
            ensure_engagement_quality_workflow(db, subscription=sub, actor_user_id=current_user.id, create_declarations=False)
            enforce_quality_gate_on_subscription(sub)
            generate_tasks_for_subscription_if_ready(
                db,
                subscription=sub,
                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}


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}
