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.consultants.models import ConsultantProfile
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",
        "requires_employee": True,
    },
    "manager": {
        "role_name": "Branch Manager",
        "label": "Manager",
        "default_designation": "Branch Manager",
        "default_department": "Operations",
        "default_employment_type": "full_time",
        "requires_employee": True,
    },
    "staff": {
        "role_name": "Staff",
        "label": "Staff",
        "default_designation": "Staff",
        "default_department": "Operations",
        "default_employment_type": "full_time",
        "requires_employee": True,
    },
    "consultant": {
        "role_name": "Consultant",
        "label": "Consultant",
        "default_designation": "Consultant",
        "default_department": "Consulting",
        "default_employment_type": "consultant",
        "requires_employee": False,
    },
}


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,
    consultant_type: str | None = None,
    firm_name: str | None = None,
    specialisation: str | None = None,
    gstin: str | None = None,
    pan: str | None = None,
    address: str | None = None,
    remarks: 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))

    consultant_profile = None
    if role_key == "consultant":
        duplicate_profile = db.execute(
            select(ConsultantProfile).where(
                ConsultantProfile.tenant_id == int(tenant_id),
                ConsultantProfile.email == email_clean,
            )
        ).scalar_one_or_none()
        if duplicate_profile:
            raise ValueError("A consultant profile with this email already exists in this firm.")

        consultant_profile = ConsultantProfile(
            tenant_id=int(tenant_id),
            branch_id=int(branch.id),
            user_id=int(user.id),
            consultant_type=(consultant_type or "external_consultant").strip() or "external_consultant",
            firm_name=(firm_name or "").strip() or None,
            contact_person=name_clean,
            email=email_clean,
            mobile=(mobile or "").strip() or None,
            specialisation=(specialisation or "").strip() or None,
            gstin=(gstin or "").strip().upper() or None,
            pan=(pan or "").strip().upper() or None,
            address=(address or "").strip() or None,
            status="active",
            onboarding_status="invited",
            is_platform_partner=False,
            is_franchise_partner=False,
            is_saas_customer=False,
            is_active=True,
            remarks=(remarks or "").strip() or None,
            created_by_user_id=actor.id,
            updated_by_user_id=actor.id,
        )
        db.add(consultant_profile)
        db.flush()
        employee_link_result = {
            "required": False,
            "created": False,
            "linked_existing": False,
            "employee_id": None,
        }
    else:
        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,
        "consultant_profile": consultant_profile,
        "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 []