arrr-erp/app/modules/firm_admin_dashboard/service.py

from __future__ import annotations

from datetime import date, datetime, timezone
from io import BytesIO
from typing import Any

from openpyxl import Workbook, load_workbook
from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from app.core.security.passwords import hash_password
from app.core.settings import get_settings
from app.modules.core.iam.invite_service import issue_invite_token
from app.modules.email_integration.services import send_user_invite_email
from app.modules.core.iam.models import User
from app.modules.core.rbac.models import Role, UserRole
from app.modules.core.tenancy.models import Branch, FinancialYear, Tenant
from app.modules.clients.models import Client
from app.modules.employees.models import Employee
from app.modules.services.models import (
    FirmServiceSelection,
    FirmServiceTaskTemplate,
    ServiceCatalogue,
)

try:
    from app.modules.billing.models import BillingSettings
except Exception:  # pragma: no cover - optional module guard
    BillingSettings = None

try:
    from app.modules.email_integration.models import PlatformEmailSettings
except Exception:  # pragma: no cover - optional W2 guard
    PlatformEmailSettings = None

try:
    from app.modules.core.audit.models import AuditLog
except Exception:  # pragma: no cover - older schema guard
    AuditLog = None


FIRM_ADMIN_ROLES = {"Firm Admin", "System Admin"}


def _count(db: Session, stmt) -> int:
    value = db.execute(stmt).scalar()
    return int(value or 0)


def _active_tenant_id(request, current_user, roles: set[str]) -> int | None:
    tenant_id = request.session.get("active_tenant_id") or getattr(current_user, "tenant_id", None)
    if "System Admin" not in roles:
        tenant_id = getattr(current_user, "tenant_id", None)
    return int(tenant_id) if tenant_id else None


def _active_branch_id(request, current_user, roles: set[str]) -> int | None:
    branch_id = request.session.get("active_branch_id") or getattr(current_user, "branch_id", None)
    if "System Admin" in roles or "Firm Admin" in roles:
        if branch_id in (None, "", 0, "0"):
            return None
        return int(branch_id)
    return int(getattr(current_user, "branch_id", 0) or 0) or None


def _tenant(db: Session, tenant_id: int | None) -> Tenant | None:
    return db.get(Tenant, int(tenant_id)) if tenant_id else None


def _role_ids(db: Session, role_names: set[str]) -> list[int]:
    return list(db.execute(select(Role.id).where(Role.name.in_(role_names))).scalars().all())


def get_user_role_names(db: Session, user_id: int) -> list[str]:
    return list(
        db.execute(
            select(Role.name)
            .join(UserRole, UserRole.role_id == Role.id)
            .where(UserRole.user_id == int(user_id), Role.is_active.is_(True))
            .order_by(Role.name.asc())
        ).scalars().all()
    )


def can_access_firm_admin_dashboard(db: Session, current_user) -> bool:
    roles = set(get_user_role_names(db, current_user.id))
    return bool(roles.intersection(FIRM_ADMIN_ROLES))


def _branch_rows(db: Session, tenant_id: int | None) -> list[dict[str, Any]]:
    if not tenant_id:
        return []
    rows = db.execute(
        select(Branch)
        .where(Branch.tenant_id == tenant_id)
        .order_by(Branch.is_head_office.desc(), Branch.name.asc())
    ).scalars().all()
    out: list[dict[str, Any]] = []
    for branch in rows:
        partner_count = _count(
            db,
            select(func.count(func.distinct(User.id)))
            .join(UserRole, UserRole.user_id == User.id)
            .join(Role, Role.id == UserRole.role_id)
            .where(
                User.tenant_id == tenant_id,
                User.branch_id == branch.id,
                User.is_active.is_(True),
                Role.name == "Partner",
            ),
        )
        user_count = _count(
            db,
            select(func.count(User.id)).where(
                User.tenant_id == tenant_id,
                User.branch_id == branch.id,
                User.is_active.is_(True),
            ),
        )
        client_count = _count(
            db,
            select(func.count(Client.id)).where(
                Client.tenant_id == tenant_id,
                Client.branch_id == branch.id,
                Client.is_active.is_(True),
            ),
        )
        out.append(
            {
                "id": branch.id,
                "code": branch.code,
                "name": branch.name,
                "is_active": branch.is_active,
                "is_head_office": branch.is_head_office,
                "allow_login": branch.allow_login,
                "allow_new_assignments": branch.allow_new_assignments,
                "timezone": branch.timezone,
                "partner_count": partner_count,
                "user_count": user_count,
                "client_count": client_count,
                "local_storage_path": branch.local_storage_path,
                "smtp_configured": bool(branch.smtp_host and branch.smtp_port and branch.smtp_username),
            }
        )
    return out


def _user_rows(db: Session, tenant_id: int | None) -> list[dict[str, Any]]:
    if not tenant_id:
        return []
    users = db.execute(
        select(User)
        .where(User.tenant_id == tenant_id)
        .order_by(User.is_active.desc(), User.full_name.asc(), User.email.asc())
        .limit(100)
    ).scalars().all()
    rows: list[dict[str, Any]] = []
    for user in users:
        employee = db.execute(
            select(Employee).where(Employee.tenant_id == tenant_id, Employee.user_id == user.id)
        ).scalar_one_or_none()
        rows.append(
            {
                "id": user.id,
                "name": user.full_name or user.email,
                "email": user.email,
                "designation": user.designation,
                "mobile": user.mobile,
                "branch_id": user.branch_id,
                "roles": get_user_role_names(db, user.id),
                "employee_id": employee.id if employee else None,
                "employee_linked": bool(employee),
                "employee_code": employee.employee_code if employee else None,
                "is_active": user.is_active,
                "allow_login": user.allow_login,
                "is_locked": user.is_locked,
                "must_change_password": user.must_change_password,
            }
        )
    return rows


def _service_rows(db: Session, tenant_id: int | None) -> list[dict[str, Any]]:
    if not tenant_id:
        return []
    selections = db.execute(
        select(FirmServiceSelection, ServiceCatalogue)
        .join(ServiceCatalogue, ServiceCatalogue.id == FirmServiceSelection.service_catalogue_id)
        .where(FirmServiceSelection.tenant_id == tenant_id)
        .order_by(ServiceCatalogue.category.asc(), ServiceCatalogue.service_name.asc())
    ).all()
    out: list[dict[str, Any]] = []
    for selection, catalogue in selections:
        template_count = _count(
            db,
            select(func.count(FirmServiceTaskTemplate.id)).where(
                FirmServiceTaskTemplate.tenant_id == tenant_id,
                FirmServiceTaskTemplate.service_catalogue_id == catalogue.id,
                FirmServiceTaskTemplate.is_active.is_(True),
            ),
        )
        out.append(
            {
                "id": selection.id,
                "service_code": catalogue.service_code,
                "service_name": catalogue.service_name,
                "category": catalogue.category or getattr(getattr(catalogue, "service_category", None), "name", None) or "-",
                "recurrence_type": catalogue.recurrence_type or "-",
                "is_enabled": selection.is_enabled,
                "default_branch_id": selection.default_branch_id,
                "template_count": template_count,
                "ready": bool(selection.is_enabled and template_count > 0),
            }
        )
    return out


def _financial_year_rows(db: Session, tenant_id: int | None) -> list[dict[str, Any]]:
    if not tenant_id:
        return []
    years = db.execute(
        select(FinancialYear)
        .where(FinancialYear.tenant_id == tenant_id)
        .order_by(FinancialYear.start_date.desc(), FinancialYear.year_code.desc())
    ).scalars().all()
    return [
        {
            "id": fy.id,
            "year_code": fy.year_code,
            "assessment_year": fy.assessment_year,
            "start_date": fy.start_date,
            "end_date": fy.end_date,
            "is_current": fy.is_current,
            "is_locked": fy.is_locked,
            "locked_at_utc": fy.locked_at_utc,
        }
        for fy in years
    ]


def _billing_settings_count(db: Session, tenant_id: int | None) -> int:
    if not tenant_id or BillingSettings is None:
        return 0
    return _count(db, select(func.count(BillingSettings.id)).where(BillingSettings.tenant_id == tenant_id))


def _smtp_status(db: Session, tenant_id: int | None) -> dict[str, Any]:
    if PlatformEmailSettings is not None:
        try:
            row = db.execute(select(PlatformEmailSettings).order_by(PlatformEmailSettings.id.desc())).scalars().first()
            return {
                "available": True,
                "configured": bool(row and row.smtp_host and row.smtp_port and row.from_email),
                "label": "Configured" if row and row.smtp_host and row.smtp_port and row.from_email else "Pending",
                "message": getattr(row, "from_email", None) or "Platform SMTP not configured",
            }
        except Exception:
            pass
    return {"available": False, "configured": False, "label": "Check settings", "message": "Open email settings to verify SMTP."}



ONBOARDING_ROLE_CONFIG: dict[str, dict[str, str]] = {
    "partner": {
        "role_name": "Partner",
        "label": "Partner",
        "default_designation": "Partner",
        "default_department": "Management",
        "default_employment_type": "full_time",
    },
    "manager": {
        "role_name": "Branch Manager",
        "label": "Manager",
        "default_designation": "Branch Manager",
        "default_department": "Operations",
        "default_employment_type": "full_time",
    },
    "staff": {
        "role_name": "Staff",
        "label": "Staff",
        "default_designation": "Staff",
        "default_department": "Operations",
        "default_employment_type": "full_time",
    },
}


def _public_invite_url(request, invite_token: str) -> str:
    """Build an invite URL for the domain currently used by the Firm Admin.

    Tenant custom domains must remain tenant-specific.  Prefer the request host
    (including trusted proxy headers) and retain ERP_PUBLIC_BASE_URL only as a
    defensive fallback for non-HTTP callers.
    """
    forwarded_proto = (request.headers.get("x-forwarded-proto") or "").split(",", 1)[0].strip().lower()
    forwarded_host = (request.headers.get("x-forwarded-host") or "").split(",", 1)[0].strip()
    request_host = (request.headers.get("host") or "").strip()

    scheme = forwarded_proto if forwarded_proto in {"http", "https"} else str(request.url.scheme or "https").lower()
    host = forwarded_host or request_host or str(request.url.netloc or "").strip()

    # Reject header-control characters before using a host in a generated URL.
    if host and not any(ch in host for ch in "\r\n/\\"):
        base = f"{scheme}://{host}".rstrip("/")
    else:
        base = str(request.base_url).strip().rstrip("/")

    if not base:
        base = (get_settings().ERP_PUBLIC_BASE_URL or "").strip().rstrip("/")
    if not base:
        raise ValueError("Unable to determine the public ERP URL for the invite link.")

    return f"{base}/invite/accept?token={invite_token}"


def _parse_optional_date(value: str | None) -> date | None:
    value = (value or "").strip()
    if not value:
        return None
    return date.fromisoformat(value)


def _next_employee_code(db: Session, tenant_id: int) -> str:
    prefix = "EMP"
    rows = db.execute(
        select(Employee.employee_code)
        .where(Employee.tenant_id == int(tenant_id), Employee.employee_code.ilike(f"{prefix}%"))
        .order_by(Employee.employee_code.desc())
    ).all()
    max_no = 0
    for (code,) in rows:
        suffix = "".join(ch for ch in str(code or "") if ch.isdigit())
        if suffix:
            max_no = max(max_no, int(suffix))
    return f"{prefix}{max_no + 1:05d}"


def _ensure_employee_for_onboarded_user(
    db: Session,
    *,
    actor: User,
    user_obj: User,
    employee_code: str | None,
    mobile: str | None,
    department: str | None,
    designation: str | None,
    date_of_joining: str | None,
    employment_type: str | None,
) -> dict[str, Any]:
    if not user_obj.tenant_id or not user_obj.branch_id:
        raise ValueError("Tenant and Branch are mandatory for internal firm users.")

    existing_linked = db.execute(
        select(Employee).where(Employee.tenant_id == int(user_obj.tenant_id), Employee.user_id == int(user_obj.id))
    ).scalar_one_or_none()
    if existing_linked:
        return {"required": True, "created": False, "linked_existing": True, "employee_id": existing_linked.id}

    email = (user_obj.email or "").strip().lower()
    existing_unlinked = None
    if email:
        existing_unlinked = db.execute(
            select(Employee).where(
                Employee.tenant_id == int(user_obj.tenant_id),
                Employee.user_id.is_(None),
                Employee.email == email,
            )
        ).scalar_one_or_none()

    if existing_unlinked:
        existing_unlinked.user_id = user_obj.id
        existing_unlinked.branch_id = user_obj.branch_id
        existing_unlinked.full_name = user_obj.full_name or existing_unlinked.full_name
        existing_unlinked.mobile = (mobile or "").strip() or existing_unlinked.mobile
        existing_unlinked.department = (department or "").strip() or existing_unlinked.department
        existing_unlinked.designation = (designation or "").strip() or existing_unlinked.designation
        existing_unlinked.updated_by_user_id = actor.id
        existing_unlinked.updated_at_utc = datetime.now(timezone.utc)
        return {"required": True, "created": False, "linked_existing": True, "employee_id": existing_unlinked.id}

    code = (employee_code or "").strip() or _next_employee_code(db, int(user_obj.tenant_id))
    duplicate = db.execute(
        select(Employee).where(Employee.tenant_id == int(user_obj.tenant_id), Employee.employee_code == code)
    ).scalar_one_or_none()
    if duplicate:
        raise ValueError(f"Employee code '{code}' already exists in this firm.")

    emp_type = (employment_type or "full_time").strip() or "full_time"
    if emp_type not in {"full_time", "part_time", "article_assistant", "intern", "consultant", "contract"}:
        emp_type = "full_time"

    employee = Employee(
        tenant_id=int(user_obj.tenant_id),
        branch_id=int(user_obj.branch_id),
        user_id=int(user_obj.id),
        employee_code=code,
        full_name=(user_obj.full_name or email or code).strip(),
        email=email or None,
        mobile=(mobile or "").strip() or None,
        date_of_joining=_parse_optional_date(date_of_joining),
        employment_type=emp_type,
        status="active",
        is_active=True,
        department=(department or "").strip() or None,
        designation=(designation or "").strip() or None,
        created_by_user_id=actor.id,
        updated_by_user_id=actor.id,
    )
    db.add(employee)
    db.flush()
    return {"required": True, "created": True, "linked_existing": False, "employee_id": employee.id}


def get_onboarding_role_config(role_key: str) -> dict[str, str] | None:
    return ONBOARDING_ROLE_CONFIG.get((role_key or "").strip().lower())


def build_user_onboarding_payload(db: Session, request, current_user, role_key: str) -> dict[str, Any]:
    roles = set(get_user_role_names(db, current_user.id))
    tenant_id = _active_tenant_id(request, current_user, roles)
    tenant = _tenant(db, tenant_id)
    config = get_onboarding_role_config(role_key)
    branches = _branch_rows(db, tenant_id)
    return {
        "tenant": tenant,
        "tenant_id": tenant_id,
        "role_key": (role_key or "").strip().lower(),
        "role_config": config,
        "branches": branches,
        "employee_code_suggestion": _next_employee_code(db, int(tenant_id)) if tenant_id else "",
    }


def create_firm_internal_user(
    db: Session,
    *,
    request,
    actor: User,
    role_key: str,
    email: str,
    full_name: str,
    branch_id: int,
    employee_code: str | None = None,
    mobile: str | None = None,
    department: str | None = None,
    designation: str | None = None,
    date_of_joining: str | None = None,
    employment_type: str | None = None,
) -> dict[str, Any]:
    actor_roles = set(get_user_role_names(db, actor.id))
    if not actor_roles.intersection(FIRM_ADMIN_ROLES):
        raise PermissionError("Only Firm Admin or System Admin can add firm users.")

    config = get_onboarding_role_config(role_key)
    if not config:
        raise ValueError("Invalid onboarding role.")

    tenant_id = _active_tenant_id(request, actor, actor_roles)
    if not tenant_id:
        raise ValueError("Active firm context is required before adding users.")

    branch = db.get(Branch, int(branch_id)) if branch_id else None
    if not branch or int(branch.tenant_id) != int(tenant_id):
        raise ValueError("Please select a valid branch for the active firm.")
    if not branch.is_active:
        raise ValueError("Selected branch is inactive.")

    role = db.execute(
        select(Role).where(Role.name == config["role_name"], Role.is_active.is_(True))
    ).scalar_one_or_none()
    if not role:
        raise ValueError(f"Role '{config['role_name']}' is not available. Please seed startup roles first.")

    email_clean = (email or "").strip().lower()
    name_clean = (full_name or "").strip()
    if not email_clean or "@" not in email_clean:
        raise ValueError("A valid email is required.")
    if not name_clean:
        raise ValueError("Full name is required.")
    if db.execute(select(User).where(User.email == email_clean)).scalar_one_or_none():
        raise ValueError("Email already exists.")

    temp_password = hash_password(__import__("secrets").token_urlsafe(18))
    user = User(
        email=email_clean,
        full_name=name_clean,
        password_hash=temp_password,
        tenant_id=int(tenant_id),
        branch_id=int(branch.id),
        is_active=True,
        allow_login=True,
        is_locked=False,
        deleted_at=None,
        must_change_password=True,
        password_changed_at_utc=None,
        designation=(designation or config.get("default_designation") or "").strip() or None,
        mobile=(mobile or "").strip() or None,
    )
    db.add(user)
    db.flush()
    db.add(UserRole(user_id=user.id, role_id=role.id))

    employee_link_result = _ensure_employee_for_onboarded_user(
        db,
        actor=actor,
        user_obj=user,
        employee_code=employee_code,
        mobile=mobile,
        department=(department or config.get("default_department") or ""),
        designation=(designation or config.get("default_designation") or ""),
        date_of_joining=date_of_joining,
        employment_type=(employment_type or config.get("default_employment_type") or "full_time"),
    )

    try:
        db.commit()
    except IntegrityError as exc:
        db.rollback()
        raise ValueError("Could not create user due to duplicate or invalid data.") from exc

    db.refresh(user)
    invite_token = issue_invite_token(db, user)
    invite_url = _public_invite_url(request, invite_token)
    email_status = "not_attempted"
    email_error = None
    try:
        email_log = send_user_invite_email(db, user=user, invite_token=invite_token)
        db.commit()
        email_status = getattr(email_log, "status", None) or "attempted"
        email_error = getattr(email_log, "error_message", None)
    except Exception as exc:  # invite link fallback remains available
        db.rollback()
        email_status = "failed"
        email_error = str(exc)

    return {
        "user": user,
        "role_name": role.name,
        "role_label": config["label"],
        "branch": branch,
        "employee_link_result": employee_link_result,
        "invite_url": invite_url,
        "email_status": email_status,
        "email_error": email_error,
    }


FIRM_USER_IMPORT_HEADERS = [
    "role",
    "full_name",
    "email",
    "branch_id",
    "branch_code",
    "branch_name",
    "employee_code",
    "mobile",
    "department",
    "designation",
    "date_of_joining",
    "employment_type",
]

FIRM_USER_IMPORT_SAMPLE_ROWS = [
    ["partner", "Sample Partner", "partner@example.com", "", "HO", "", "PTR001", "9999999999", "Management", "Partner", "2026-04-01", "full_time"],
    ["manager", "Sample Manager", "manager@example.com", "", "HO", "", "MGR001", "9999999998", "Operations", "Branch Manager", "2026-04-01", "full_time"],
    ["staff", "Sample Staff", "staff@example.com", "", "HO", "", "EMP001", "9999999997", "Audit", "Associate", "2026-04-01", "full_time"],
]


def build_firm_user_import_template(db: Session, request, current_user) -> bytes:
    roles = set(get_user_role_names(db, current_user.id))
    tenant_id = _active_tenant_id(request, current_user, roles)
    tenant = _tenant(db, tenant_id)
    branches = _branch_rows(db, tenant_id)

    wb = Workbook()
    ws = wb.active
    ws.title = "Firm Users"
    ws.append(FIRM_USER_IMPORT_HEADERS)
    for row in FIRM_USER_IMPORT_SAMPLE_ROWS:
        ws.append(row)

    notes = wb.create_sheet("Instructions")
    notes.append(["Firm", getattr(tenant, "display_name", None) or getattr(tenant, "name", None) or "Active firm"])
    notes.append(["Allowed role values", "partner, manager, staff"])
    notes.append(["Manager mapping", "manager imports as existing Branch Manager role"])
    notes.append(["Branch selection", "Use branch_id OR branch_code OR exact branch_name from Branches sheet"])
    notes.append(["Employment type values", "full_time, part_time, article_assistant, intern, consultant, contract"])
    notes.append(["Date format", "YYYY-MM-DD preferred"])
    notes.append(["Invite", "Each valid row creates user + employee link + invite token. Invite link is shown in import result."])

    branch_sheet = wb.create_sheet("Branches")
    branch_sheet.append(["branch_id", "branch_code", "branch_name", "is_active", "is_head_office"])
    for branch in branches:
        branch_sheet.append([branch.get("id"), branch.get("code"), branch.get("name"), branch.get("is_active"), branch.get("is_head_office")])

    for sheet in wb.worksheets:
        for cell in sheet[1]:
            cell.font = cell.font.copy(bold=True)
        for column_cells in sheet.columns:
            letter = column_cells[0].column_letter
            width = max(14, min(34, max(len(str(c.value or "")) for c in column_cells) + 3))
            sheet.column_dimensions[letter].width = width

    bio = BytesIO()
    wb.save(bio)
    return bio.getvalue()


def _cell_text(value: Any) -> str:
    if value is None:
        return ""
    if isinstance(value, datetime):
        return value.date().isoformat()
    if isinstance(value, date):
        return value.isoformat()
    return str(value).strip()


def _load_firm_user_import_rows(content: bytes) -> list[dict[str, Any]]:
    try:
        wb = load_workbook(BytesIO(content), data_only=True)
    except Exception as exc:
        raise ValueError(f"Unable to read Excel file: {exc}") from exc
    ws = wb["Firm Users"] if "Firm Users" in wb.sheetnames else wb.active
    headers = [str(cell.value or "").strip().lower() for cell in ws[1]]
    if not any(headers):
        raise ValueError("Excel file has no header row.")
    rows: list[dict[str, Any]] = []
    for row_no, row in enumerate(ws.iter_rows(min_row=2, values_only=True), start=2):
        if not any(value not in (None, "") for value in row):
            continue
        item = {headers[idx]: _cell_text(row[idx] if idx < len(row) else None) for idx in range(len(headers)) if headers[idx]}
        item["_row_no"] = row_no
        rows.append(item)
    if not rows:
        raise ValueError("Excel file has no data rows.")
    return rows


def _resolve_import_branch_id(db: Session, tenant_id: int, row: dict[str, Any]) -> int:
    raw_id = _cell_text(row.get("branch_id"))
    raw_code = _cell_text(row.get("branch_code"))
    raw_name = _cell_text(row.get("branch_name"))

    branch = None
    if raw_id:
        try:
            branch = db.get(Branch, int(float(raw_id)))
        except Exception as exc:
            raise ValueError(f"Invalid branch_id '{raw_id}'.") from exc
    elif raw_code:
        branch = db.execute(
            select(Branch).where(Branch.tenant_id == int(tenant_id), func.lower(Branch.code) == raw_code.lower())
        ).scalar_one_or_none()
    elif raw_name:
        branch = db.execute(
            select(Branch).where(Branch.tenant_id == int(tenant_id), func.lower(Branch.name) == raw_name.lower())
        ).scalar_one_or_none()
    else:
        active = db.execute(
            select(Branch)
            .where(Branch.tenant_id == int(tenant_id), Branch.is_active.is_(True))
            .order_by(Branch.is_head_office.desc(), Branch.id.asc())
        ).scalars().all()
        if len(active) == 1:
            branch = active[0]
        else:
            raise ValueError("Branch is required when the firm has multiple active branches.")

    if not branch or int(branch.tenant_id) != int(tenant_id):
        raise ValueError("Branch does not belong to the active firm.")
    if not branch.is_active:
        raise ValueError("Selected branch is inactive.")
    return int(branch.id)


def import_firm_internal_users(db: Session, *, request, actor: User, content: bytes) -> dict[str, Any]:
    actor_roles = set(get_user_role_names(db, actor.id))
    if not actor_roles.intersection(FIRM_ADMIN_ROLES):
        raise PermissionError("Only Firm Admin or System Admin can import firm users.")
    tenant_id = _active_tenant_id(request, actor, actor_roles)
    if not tenant_id:
        raise ValueError("Active firm context is required before importing users.")

    rows = _load_firm_user_import_rows(content)
    results: list[dict[str, Any]] = []
    created = failed = 0
    invite_links: list[dict[str, Any]] = []

    for row in rows:
        row_no = int(row.get("_row_no") or 0)
        role_key = _cell_text(row.get("role") or "staff").lower()
        email = _cell_text(row.get("email")).lower()
        full_name = _cell_text(row.get("full_name"))
        try:
            if not get_onboarding_role_config(role_key):
                raise ValueError("Role must be one of partner, manager or staff.")
            branch_id = _resolve_import_branch_id(db, int(tenant_id), row)
            result = create_firm_internal_user(
                db,
                request=request,
                actor=actor,
                role_key=role_key,
                email=email,
                full_name=full_name,
                branch_id=branch_id,
                employee_code=_cell_text(row.get("employee_code")),
                mobile=_cell_text(row.get("mobile")),
                department=_cell_text(row.get("department")),
                designation=_cell_text(row.get("designation")),
                date_of_joining=_cell_text(row.get("date_of_joining")),
                employment_type=_cell_text(row.get("employment_type")) or None,
            )
            created += 1
            invite_url = result.get("invite_url")
            invite_links.append({
                "row_no": row_no,
                "name": full_name,
                "email": email,
                "role": result.get("role_name"),
                "invite_url": invite_url,
                "email_status": result.get("email_status"),
            })
            results.append({"row_no": row_no, "status": "created", "email": email, "name": full_name, "role": result.get("role_name"), "message": "User, role, employee link and invite created."})
        except Exception as exc:
            db.rollback()
            failed += 1
            results.append({"row_no": row_no, "status": "failed", "email": email, "name": full_name, "role": role_key, "message": str(exc)})

    return {
        "summary": {"total": len(rows), "created": created, "failed": failed},
        "rows": results,
        "invite_links": invite_links,
    }

def build_dashboard_payload(db: Session, request, current_user) -> dict[str, Any]:
    roles = set(get_user_role_names(db, current_user.id))
    tenant_id = _active_tenant_id(request, current_user, roles)
    branch_id = _active_branch_id(request, current_user, roles)
    tenant = _tenant(db, tenant_id)
    branches = _branch_rows(db, tenant_id)
    users = _user_rows(db, tenant_id)
    services = _service_rows(db, tenant_id)
    financial_years = _financial_year_rows(db, tenant_id)
    billing_settings_count = _billing_settings_count(db, tenant_id)

    firm_admin_roles = _role_ids(db, {"Firm Admin"})
    firm_admin_count = 0
    if tenant_id and firm_admin_roles:
        firm_admin_count = _count(
            db,
            select(func.count(func.distinct(User.id)))
            .join(UserRole, UserRole.user_id == User.id)
            .where(User.tenant_id == tenant_id, UserRole.role_id.in_(firm_admin_roles), User.is_active.is_(True)),
        )

    partner_count = _count(
        db,
        select(func.count(func.distinct(User.id)))
        .join(UserRole, UserRole.user_id == User.id)
        .join(Role, Role.id == UserRole.role_id)
        .where(User.tenant_id == tenant_id, Role.name == "Partner", User.is_active.is_(True)),
    ) if tenant_id else 0

    client_count = _count(db, select(func.count(Client.id)).where(Client.tenant_id == tenant_id, Client.is_active.is_(True))) if tenant_id else 0
    user_count = _count(db, select(func.count(User.id)).where(User.tenant_id == tenant_id, User.is_active.is_(True))) if tenant_id else 0
    service_count = len([s for s in services if s["is_enabled"]])
    services_without_templates = len([s for s in services if s["is_enabled"] and s["template_count"] <= 0])
    current_fy = next((fy for fy in financial_years if fy["is_current"]), None)

    branding_ready = bool(tenant and (tenant.display_name or tenant.logo_path or tenant.primary_color or tenant.contact_email))
    setup_checks = [
        {"key": "branches", "label": "Branches created", "ok": len(branches) > 0, "detail": f"{len(branches)} branch(es)"},
        {"key": "firm_admin", "label": "Firm Admin user", "ok": firm_admin_count > 0, "detail": f"{firm_admin_count} active"},
        {"key": "partners", "label": "Partners assigned", "ok": partner_count > 0, "detail": f"{partner_count} active"},
        {"key": "services", "label": "Services selected", "ok": service_count > 0, "detail": f"{service_count} enabled"},
        {"key": "templates", "label": "Task templates ready", "ok": services_without_templates == 0 and service_count > 0, "detail": f"{services_without_templates} service(s) pending"},
        {"key": "fy", "label": "Current financial year", "ok": bool(current_fy), "detail": current_fy["year_code"] if current_fy else "Not set"},
        {"key": "branding", "label": "Branding/contact", "ok": branding_ready, "detail": "Started" if branding_ready else "Pending"},
        {"key": "billing", "label": "Billing settings", "ok": billing_settings_count > 0, "detail": f"{billing_settings_count} setup row(s)"},
    ]
    setup_score = sum(1 for item in setup_checks if item["ok"])

    overview = {
        "tenant": tenant,
        "tenant_id": tenant_id,
        "branch_id": branch_id,
        "branch_count": len(branches),
        "active_branch_count": len([b for b in branches if b["is_active"]]),
        "user_count": user_count,
        "firm_admin_count": firm_admin_count,
        "partner_count": partner_count,
        "client_count": client_count,
        "enabled_service_count": service_count,
        "services_without_templates": services_without_templates,
        "fy_count": len(financial_years),
        "current_fy": current_fy,
        "billing_settings_count": billing_settings_count,
        "branding_ready": branding_ready,
        "setup_score": setup_score,
        "setup_total": len(setup_checks),
        "setup_percent": round((setup_score / len(setup_checks)) * 100) if setup_checks else 0,
        "smtp": _smtp_status(db, tenant_id),
        "today": date.today(),
    }

    return {
        "roles": sorted(roles),
        "tenant": tenant,
        "overview": overview,
        "setup_checks": setup_checks,
        "branches": branches,
        "users": users,
        "services": services,
        "financial_years": financial_years,
        "reports": _report_cards(),
        "wizards": _wizard_cards(),
        "audit_logs": _audit_rows(db, tenant_id),
    }


def _report_cards() -> list[dict[str, str]]:
    return [
        {"group": "Setup Reports", "title": "Firm Setup Completeness", "desc": "Branches, users, roles, services, FY, branding and billing readiness.", "href": "/firm-admin/dashboard?tab=overview"},
        {"group": "Branch Reports", "title": "Branch Readiness", "desc": "Active branches, partner assignment, users and local storage readiness.", "href": "/firm-admin/dashboard?tab=branches"},
        {"group": "User Reports", "title": "Users & Roles", "desc": "Firm Admin, Partner, Manager, Staff and login readiness.", "href": "/firm-admin/dashboard?tab=users"},
        {"group": "Service Reports", "title": "Service Setup Readiness", "desc": "Enabled services and missing task templates.", "href": "/firm-admin/dashboard?tab=services"},
        {"group": "FY Reports", "title": "Financial Year Status", "desc": "Current, locked and historical financial years.", "href": "/firm-admin/dashboard?tab=financial-years"},
    ]


def _wizard_cards() -> list[dict[str, str]]:
    return [
        {"title": "Add / Manage Branches", "desc": "Create branch, office timing, login control and local storage path.", "href": "/system-settings/branches"},
        {"title": "Invite Users", "desc": "Create Partner, Manager and Staff directly from Firm Admin.", "href": "/firm-admin/dashboard?tab=users"},
        {"title": "Firm Profile & Branding", "desc": "Display name, logo, colours and contact details.", "href": "/system-settings/branding"},
        {"title": "Select Firm Services", "desc": "Enable services from system catalogue for this firm.", "href": "/services"},
        {"title": "Firm Task Templates", "desc": "Customize task list and document requirements for enabled services.", "href": "/services/templates"},
        {"title": "Financial Years", "desc": "Create, switch, lock and manage financial years.", "href": "/system-settings/financial-years"},
        {"title": "Billing Settings", "desc": "Invoice prefix, GST settings, payment gateway and bank details.", "href": "/billing/settings"},
    ]


def _audit_rows(db: Session, tenant_id: int | None) -> list[Any]:
    if AuditLog is None or not tenant_id:
        return []
    try:
        return list(
            db.execute(
                select(AuditLog)
                .where(getattr(AuditLog, "target_tenant_id", tenant_id) == tenant_id)
                .order_by(AuditLog.created_at_utc.desc())
                .limit(25)
            ).scalars().all()
        )
    except Exception:
        try:
            return list(db.execute(select(AuditLog).order_by(AuditLog.created_at_utc.desc()).limit(25)).scalars().all())
        except Exception:
            return []