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(invite_token: str) -> str: base = (get_settings().ERP_PUBLIC_BASE_URL or "").strip().rstrip("/") or "http://localhost:8000" 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(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 []