from __future__ import annotations from dataclasses import dataclass from pathlib import Path from typing import Any from sqlalchemy import func, select from sqlalchemy.orm import Session from app.modules.core.iam.models import User from app.modules.core.iam.password_flows_models import InviteToken from app.modules.core.rbac.models import Role, UserRole from app.modules.core.tenancy.models import Branch, FinancialYear, Tenant, YearBackupExport from app.modules.services.models import ( FirmServiceSelection, ServiceCatalogue, ServiceCategory, ServiceDefaultTaskTemplate, ServiceDueDateRule, ) try: from app.modules.billing.models import BillingSettings except Exception: # pragma: no cover BillingSettings = None # type: ignore[assignment] try: from app.modules.core.audit.models import AuditLog except Exception: # pragma: no cover - keeps dashboard usable if audit module is unavailable AuditLog = None # type: ignore[assignment] try: from app.modules.email_integration.models import PlatformEmailSetting except Exception: # pragma: no cover - dashboard falls back to not configured PlatformEmailSetting = None # type: ignore[assignment] try: from app.modules.platform_billing.models import ( PlatformBillingAccount, PlatformInvoice, PlatformPlan, PlatformSubscription, ) except Exception: # pragma: no cover - platform billing can remain optional PlatformBillingAccount = None # type: ignore[assignment] PlatformInvoice = None # type: ignore[assignment] PlatformPlan = None # type: ignore[assignment] PlatformSubscription = None # type: ignore[assignment] STORAGE_ROOT = Path("/app/data/storage") BRANDING_ROOT = STORAGE_ROOT / "uploads" / "branding" def scalar_count(db: Session, statement) -> int: value = db.execute(statement).scalar() return int(value or 0) def system_admin_role_names(db: Session, user_id: int) -> set[str]: rows = db.execute( select(Role.name) .join(UserRole, UserRole.role_id == Role.id) .where(UserRole.user_id == user_id) ).all() return {name for (name,) in rows if name} def is_system_admin(db: Session, user: User | None) -> bool: if not user: return False return "System Admin" in system_admin_role_names(db, user.id) def _firm_admin_role_ids(db: Session) -> list[int]: names = ["Firm Admin", "firm_admin", "FirmAdmin"] return [role_id for (role_id,) in db.execute(select(Role.id).where(Role.name.in_(names))).all()] def _pending_firm_admin_invite_count(db: Session) -> int: role_ids = _firm_admin_role_ids(db) if not role_ids: return 0 return scalar_count( db, select(func.count(func.distinct(User.id))) .join(UserRole, UserRole.user_id == User.id) .join(InviteToken, InviteToken.user_id == User.id) .where(UserRole.role_id.in_(role_ids)) .where(InviteToken.used_at_utc.is_(None)) .where(User.must_change_password.is_(True)), ) def _pending_firm_admin_invites_by_tenant(db: Session) -> dict[int, int]: role_ids = _firm_admin_role_ids(db) if not role_ids: return {} rows = db.execute( select(User.tenant_id, func.count(func.distinct(User.id))) .join(UserRole, UserRole.user_id == User.id) .join(InviteToken, InviteToken.user_id == User.id) .where(UserRole.role_id.in_(role_ids)) .where(InviteToken.used_at_utc.is_(None)) .where(User.must_change_password.is_(True)) .group_by(User.tenant_id) ).all() return {int(tenant_id): int(count or 0) for tenant_id, count in rows if tenant_id is not None} def _platform_smtp_status(db: Session) -> dict[str, Any]: if PlatformEmailSetting is None: return { "configured": False, "active": False, "host": None, "port": None, "from_email": None, "from_name": None, "reply_to_email": None, "security": None, "send_auth_emails": False, "message": "Platform SMTP model not available. Apply the platform SMTP package/migration first.", } setting = db.execute(select(PlatformEmailSetting).order_by(PlatformEmailSetting.id.asc())).scalars().first() if not setting: return { "configured": False, "active": False, "host": None, "port": None, "from_email": None, "from_name": None, "reply_to_email": None, "security": None, "send_auth_emails": False, "message": "Platform SMTP is not configured.", } configured = bool(setting.smtp_host and setting.smtp_port and setting.from_email) active = bool(getattr(setting, "is_active", False)) send_auth_emails = bool(getattr(setting, "send_auth_emails", False)) return { "configured": configured, "active": active, "host": setting.smtp_host, "port": setting.smtp_port, "from_email": setting.from_email, "from_name": setting.from_name, "reply_to_email": getattr(setting, "reply_to_email", None), "security": setting.smtp_security, "send_auth_emails": send_auth_emails, "message": "Platform SMTP is ready for firm invites." if configured and active and send_auth_emails else "Platform SMTP needs attention before system emails are fully ready.", } def _dir_size_bytes(path: Path) -> int: if not path.exists(): return 0 total = 0 for item in path.rglob("*"): try: if item.is_file(): total += item.stat().st_size except OSError: continue return total def _file_count(path: Path) -> int: if not path.exists(): return 0 count = 0 for item in path.rglob("*"): try: if item.is_file(): count += 1 except OSError: continue return count def format_bytes(size: int) -> str: units = ["B", "KB", "MB", "GB", "TB"] value = float(size) for unit in units: if value < 1024 or unit == units[-1]: if unit == "B": return f"{int(value)} {unit}" return f"{value:.1f} {unit}" value /= 1024 return f"{size} B" def _count_distinct_tenants(rows: list[tuple[Any, Any]]) -> int: return len({int(tenant_id) for tenant_id, _ in rows if tenant_id is not None}) def _common_count_maps(db: Session) -> dict[str, dict[int, int]]: branch_counts = dict(db.execute(select(Branch.tenant_id, func.count(Branch.id)).group_by(Branch.tenant_id)).all()) user_counts = dict(db.execute(select(User.tenant_id, func.count(User.id)).group_by(User.tenant_id)).all()) fy_counts = dict(db.execute(select(FinancialYear.tenant_id, func.count(FinancialYear.id)).group_by(FinancialYear.tenant_id)).all()) current_fy_counts = dict(db.execute(select(FinancialYear.tenant_id, func.count(FinancialYear.id)).where(FinancialYear.is_current.is_(True)).group_by(FinancialYear.tenant_id)).all()) selected_service_counts = dict(db.execute(select(FirmServiceSelection.tenant_id, func.count(FirmServiceSelection.id)).where(FirmServiceSelection.is_enabled.is_(True)).group_by(FirmServiceSelection.tenant_id)).all()) billing_setting_counts: dict[int, int] = {} if BillingSettings is not None: billing_setting_counts = dict(db.execute(select(BillingSettings.tenant_id, func.count(BillingSettings.id)).group_by(BillingSettings.tenant_id)).all()) return { "branch": {int(k): int(v or 0) for k, v in branch_counts.items() if k is not None}, "user": {int(k): int(v or 0) for k, v in user_counts.items() if k is not None}, "fy": {int(k): int(v or 0) for k, v in fy_counts.items() if k is not None}, "current_fy": {int(k): int(v or 0) for k, v in current_fy_counts.items() if k is not None}, "selected_service": {int(k): int(v or 0) for k, v in selected_service_counts.items() if k is not None}, "billing_setting": {int(k): int(v or 0) for k, v in billing_setting_counts.items() if k is not None}, } def _primary_branch_names(db: Session) -> dict[int, str]: branches = db.execute(select(Branch).order_by(Branch.tenant_id.asc(), Branch.is_head_office.desc(), Branch.id.asc())).scalars().all() result: dict[int, str] = {} for branch in branches: if branch.tenant_id not in result: result[branch.tenant_id] = branch.name or branch.code return result def _primary_firm_admins(db: Session) -> dict[int, str]: role_ids = _firm_admin_role_ids(db) primary_admins: dict[int, str] = {} if not role_ids: return primary_admins rows = db.execute( select(User.tenant_id, User.email, User.full_name, User.must_change_password, User.is_active) .join(UserRole, UserRole.user_id == User.id) .where(UserRole.role_id.in_(role_ids)) .order_by(User.id.asc()) ).all() for tenant_id, email, full_name, must_change_password, is_active in rows: if tenant_id not in primary_admins: label = full_name or email if must_change_password: label = f"{label} (invite pending)" elif not is_active: label = f"{label} (inactive)" primary_admins[int(tenant_id)] = label return primary_admins def _firm_admin_counts(db: Session) -> dict[int, int]: role_ids = _firm_admin_role_ids(db) if not role_ids: return {} rows = db.execute( select(User.tenant_id, func.count(func.distinct(User.id))) .join(UserRole, UserRole.user_id == User.id) .where(UserRole.role_id.in_(role_ids)) .group_by(User.tenant_id) ).all() return {int(tenant_id): int(count or 0) for tenant_id, count in rows if tenant_id is not None} def get_storage_status(db: Session) -> dict[str, Any]: root_exists = STORAGE_ROOT.exists() branding_exists = BRANDING_ROOT.exists() root_size = _dir_size_bytes(STORAGE_ROOT) branding_files = _file_count(BRANDING_ROOT) backup_exports = scalar_count(db, select(func.count(YearBackupExport.id))) return { "storage_root": str(STORAGE_ROOT), "storage_root_exists": root_exists, "branding_root": str(BRANDING_ROOT), "branding_root_exists": branding_exists, "branding_file_count": branding_files, "backup_export_count": backup_exports, "used_bytes": root_size, "used_display": format_bytes(root_size), "status_label": "Ready" if root_exists and branding_exists else "Needs attention", } def get_firm_setup_health(db: Session, limit: int = 200) -> dict[str, Any]: tenants = db.execute(select(Tenant).order_by(Tenant.id.desc()).limit(limit)).scalars().all() maps = _common_count_maps(db) admin_counts = _firm_admin_counts(db) pending_invites = _pending_firm_admin_invites_by_tenant(db) primary_admins = _primary_firm_admins(db) primary_branches = _primary_branch_names(db) rows: list[dict[str, Any]] = [] incomplete = 0 missing_branch = 0 missing_admin = 0 missing_fy = 0 missing_services = 0 missing_branding = 0 for tenant in tenants: tid = int(tenant.id) branch_ok = maps["branch"].get(tid, 0) > 0 admin_ok = admin_counts.get(tid, 0) > 0 fy_ok = maps["fy"].get(tid, 0) > 0 current_fy_ok = maps["current_fy"].get(tid, 0) > 0 services_ok = maps["selected_service"].get(tid, 0) > 0 users_ok = maps["user"].get(tid, 0) > 0 branding_ok = bool((tenant.display_name or tenant.name) and (tenant.logo_path or tenant.primary_color or tenant.contact_email)) billing_ok = maps["billing_setting"].get(tid, 0) > 0 checks = [branch_ok, admin_ok, fy_ok, services_ok, users_ok] score = sum(1 for item in checks if item) status = "Ready" if all(checks) else "Incomplete" if not all(checks): incomplete += 1 if not branch_ok: missing_branch += 1 if not admin_ok: missing_admin += 1 if not fy_ok: missing_fy += 1 if not services_ok: missing_services += 1 if not branding_ok: missing_branding += 1 missing = [] if not branch_ok: missing.append("Branch") if not admin_ok: missing.append("Firm Admin") if not fy_ok: missing.append("FY") if not services_ok: missing.append("Services") if not users_ok: missing.append("Users") rows.append({ "id": tid, "code": tenant.code, "name": tenant.display_name or tenant.name, "is_active": tenant.is_active, "branch_ok": branch_ok, "admin_ok": admin_ok, "fy_ok": fy_ok, "current_fy_ok": current_fy_ok, "services_ok": services_ok, "users_ok": users_ok, "branding_ok": branding_ok, "billing_ok": billing_ok, "branch_count": maps["branch"].get(tid, 0), "user_count": maps["user"].get(tid, 0), "fy_count": maps["fy"].get(tid, 0), "selected_service_count": maps["selected_service"].get(tid, 0), "billing_setting_count": maps["billing_setting"].get(tid, 0), "primary_admin": primary_admins.get(tid, "-"), "primary_branch": primary_branches.get(tid, "-"), "pending_invites": pending_invites.get(tid, 0), "score": score, "score_total": len(checks), "status": status, "missing_text": ", ".join(missing) if missing else "None", }) return { "rows": rows, "total": len(rows), "ready": len(rows) - incomplete, "incomplete": incomplete, "missing_branch": missing_branch, "missing_admin": missing_admin, "missing_fy": missing_fy, "missing_services": missing_services, "missing_branding": missing_branding, } def get_overview(db: Session) -> dict[str, Any]: total_firms = scalar_count(db, select(func.count(Tenant.id))) active_firms = scalar_count(db, select(func.count(Tenant.id)).where(Tenant.is_active.is_(True))) inactive_firms = max(total_firms - active_firms, 0) service_count = scalar_count(db, select(func.count(ServiceCatalogue.id))) active_service_count = scalar_count(db, select(func.count(ServiceCatalogue.id)).where(ServiceCatalogue.is_active.is_(True))) category_count = scalar_count(db, select(func.count(ServiceCategory.id))) default_task_count = scalar_count(db, select(func.count(ServiceDefaultTaskTemplate.id))) due_rule_count = scalar_count(db, select(func.count(ServiceDueDateRule.id))) pending_invites = _pending_firm_admin_invite_count(db) storage = get_storage_status(db) smtp = _platform_smtp_status(db) health = get_firm_setup_health(db, limit=500) catalogue = get_catalogue_summary(db) billing = get_billing_readiness(db) needs_attention = [] if health["missing_branch"]: needs_attention.append({"label": "Firms without branch", "count": health["missing_branch"], "tab": "firm-setup-health"}) if health["missing_admin"]: needs_attention.append({"label": "Firms without Firm Admin", "count": health["missing_admin"], "tab": "firm-setup-health"}) if health["missing_fy"]: needs_attention.append({"label": "Firms without FY", "count": health["missing_fy"], "tab": "firm-setup-health"}) if health["missing_services"]: needs_attention.append({"label": "Firms without selected services", "count": health["missing_services"], "tab": "firm-setup-health"}) if pending_invites: needs_attention.append({"label": "Firm Admin invites pending", "count": pending_invites, "tab": "firms"}) if catalogue["services_without_tasks_count"]: needs_attention.append({"label": "Services without default tasks", "count": catalogue["services_without_tasks_count"], "tab": "catalogue"}) if not (smtp["configured"] and smtp["active"] and smtp["send_auth_emails"]): needs_attention.append({"label": "Platform SMTP not fully ready", "count": 1, "tab": "smtp"}) if not (storage["storage_root_exists"] and storage["branding_root_exists"]): needs_attention.append({"label": "Persistent storage path needs attention", "count": 1, "tab": "storage"}) return { "total_firms": total_firms, "active_firms": active_firms, "inactive_firms": inactive_firms, "setup_incomplete_firms": health["incomplete"], "pending_invites": pending_invites, "service_count": service_count, "active_service_count": active_service_count, "category_count": category_count, "default_task_count": default_task_count, "due_rule_count": due_rule_count, "services_without_tasks": catalogue["services_without_tasks_count"], "storage": storage, "smtp": smtp, "billing": billing, "needs_attention": needs_attention, } def list_firms(db: Session, limit: int = 100) -> list[dict[str, Any]]: maps = _common_count_maps(db) primary_admins = _primary_firm_admins(db) primary_branches = _primary_branch_names(db) pending_invites = _pending_firm_admin_invites_by_tenant(db) tenants = db.execute(select(Tenant).order_by(Tenant.id.desc()).limit(limit)).scalars().all() return [ { "id": tenant.id, "code": tenant.code, "name": tenant.display_name or tenant.name, "legal_name": tenant.name, "firm_type": tenant.firm_type, "is_active": tenant.is_active, "branch_count": maps["branch"].get(tenant.id, 0), "user_count": maps["user"].get(tenant.id, 0), "fy_count": maps["fy"].get(tenant.id, 0), "selected_service_count": maps["selected_service"].get(tenant.id, 0), "primary_admin": primary_admins.get(tenant.id, "-"), "primary_branch": primary_branches.get(tenant.id, "-"), "pending_invites": pending_invites.get(tenant.id, 0), "contact_email": tenant.contact_email, } for tenant in tenants ] def get_catalogue_summary(db: Session) -> dict[str, Any]: categories = db.execute(select(ServiceCategory).order_by(ServiceCategory.sort_order.asc(), ServiceCategory.name.asc())).scalars().all() services = db.execute(select(ServiceCatalogue).order_by(ServiceCatalogue.sort_order.asc(), ServiceCatalogue.service_name.asc()).limit(150)).scalars().all() task_counts = dict( db.execute( select(ServiceDefaultTaskTemplate.service_catalogue_id, func.count(ServiceDefaultTaskTemplate.id)) .group_by(ServiceDefaultTaskTemplate.service_catalogue_id) ).all() ) rule_counts = dict( db.execute( select(ServiceDueDateRule.service_catalogue_id, func.count(ServiceDueDateRule.id)) .group_by(ServiceDueDateRule.service_catalogue_id) ).all() ) services_without_tasks = [service for service in services if int(task_counts.get(service.id, 0) or 0) == 0] inactive_services = [service for service in services if not service.is_active] return { "category_count": scalar_count(db, select(func.count(ServiceCategory.id))), "service_count": scalar_count(db, select(func.count(ServiceCatalogue.id))), "active_service_count": scalar_count(db, select(func.count(ServiceCatalogue.id)).where(ServiceCatalogue.is_active.is_(True))), "inactive_service_count": scalar_count(db, select(func.count(ServiceCatalogue.id)).where(ServiceCatalogue.is_active.is_(False))), "default_task_count": scalar_count(db, select(func.count(ServiceDefaultTaskTemplate.id))), "due_rule_count": scalar_count(db, select(func.count(ServiceDueDateRule.id))), "services_without_tasks_count": scalar_count( db, select(func.count(ServiceCatalogue.id)).outerjoin( ServiceDefaultTaskTemplate, ServiceDefaultTaskTemplate.service_catalogue_id == ServiceCatalogue.id, ).group_by(ServiceCatalogue.id).having(func.count(ServiceDefaultTaskTemplate.id) == 0) ) if False else len(services_without_tasks), "categories": categories, "services_without_tasks_preview": [ {"id": service.id, "code": service.service_code, "name": service.service_name} for service in services_without_tasks[:10] ], "services": [ { "id": service.id, "code": service.service_code, "name": service.service_name, "category": service.category or (service.service_category.name if service.service_category else "-"), "recurrence_type": service.recurrence_type or "-", "engagement_type": service.engagement_type, "is_active": service.is_active, "task_count": int(task_counts.get(service.id, 0) or 0), "rule_count": int(rule_counts.get(service.id, 0) or 0), "readiness": "Ready" if int(task_counts.get(service.id, 0) or 0) > 0 else "No default tasks", } for service in services ], } def get_smtp_summary(db: Session) -> dict[str, Any]: return _platform_smtp_status(db) def get_billing_readiness(db: Session) -> dict[str, Any]: firm_count = scalar_count(db, select(func.count(Tenant.id))) firm_billing_settings = 0 if BillingSettings is not None: firm_billing_settings = scalar_count(db, select(func.count(func.distinct(BillingSettings.tenant_id)))) if PlatformPlan is None or PlatformBillingAccount is None or PlatformSubscription is None: return { "available": False, "plan_count": 0, "active_plan_count": 0, "audit_firm_account_count": 0, "active_subscription_count": 0, "expired_subscription_count": 0, "firms_without_account": firm_count, "firm_billing_settings": firm_billing_settings, "firms_without_billing_settings": max(firm_count - firm_billing_settings, 0), "message": "Platform billing models are not available in this deployment.", } plan_count = scalar_count(db, select(func.count(PlatformPlan.id))) active_plan_count = scalar_count(db, select(func.count(PlatformPlan.id)).where(PlatformPlan.is_active.is_(True))) audit_firm_account_count = scalar_count(db, select(func.count(PlatformBillingAccount.id)).where(PlatformBillingAccount.account_type == "AUDIT_FIRM")) active_subscription_count = scalar_count(db, select(func.count(PlatformSubscription.id)).where(PlatformSubscription.status == "ACTIVE")) expired_subscription_count = scalar_count(db, select(func.count(PlatformSubscription.id)).where(PlatformSubscription.status.in_(["EXPIRED", "CANCELLED", "SUSPENDED"]))) invoice_count = 0 if PlatformInvoice is not None: invoice_count = scalar_count(db, select(func.count(PlatformInvoice.id))) return { "available": True, "plan_count": plan_count, "active_plan_count": active_plan_count, "audit_firm_account_count": audit_firm_account_count, "active_subscription_count": active_subscription_count, "expired_subscription_count": expired_subscription_count, "platform_invoice_count": invoice_count, "firms_without_account": max(firm_count - audit_firm_account_count, 0), "firm_billing_settings": firm_billing_settings, "firms_without_billing_settings": max(firm_count - firm_billing_settings, 0), "message": "Platform billing module is available." if plan_count else "Platform billing exists but plans may need setup.", } def get_report_cards(db: Session) -> dict[str, list[dict[str, str]]]: overview = get_overview(db) return { "Firm Reports": [ { "title": "Firm Setup Completeness", "description": "Checklist of branch, Firm Admin, FY, selected services and users.", "metric": f"{overview['setup_incomplete_firms']} incomplete", "href": "/system-admin/dashboard?tab=firm-setup-health", }, { "title": "Firm Admin Invite Status", "description": "Track Firm Admin users whose invite/password setup is still pending.", "metric": f"{overview['pending_invites']} pending", "href": "/system-admin/dashboard?tab=firms", }, { "title": "Firm User Count Report", "description": "Review branch, user, FY and service count per firm.", "metric": f"{overview['total_firms']} firms", "href": "/system-admin/dashboard?tab=firms", }, ], "Service Reports": [ { "title": "Catalogue Readiness", "description": "Check service categories, services, templates and due-date rules.", "metric": f"{overview['service_count']} services", "href": "/system-admin/dashboard?tab=catalogue", }, { "title": "Services Without Default Tasks", "description": "Services that may not generate execution tasks correctly.", "metric": f"{overview['services_without_tasks']} services", "href": "/system-admin/dashboard?tab=catalogue", }, ], "System Reports": [ { "title": "Platform SMTP Status", "description": "Check whether system emails can be sent for firm invites and notices.", "metric": "Ready" if overview["smtp"]["configured"] and overview["smtp"]["active"] else "Needs setup", "href": "/system-admin/dashboard?tab=smtp", }, { "title": "Storage Status", "description": "Verify persistent storage, branding files and FY backup exports.", "metric": overview["storage"]["used_display"], "href": "/system-admin/dashboard?tab=storage", }, { "title": "Audit Log Review", "description": "Review recent platform and system-level actions.", "metric": "Latest 25", "href": "/system-admin/dashboard?tab=audit-logs", }, ], } def get_wizard_cards() -> dict[str, list[dict[str, str]]]: return { "Setup Wizards": [ { "title": "Create Firm Wizard", "description": "Create tenant, primary branch, Firm Admin, employee link and invite.", "href": "/wizards/system/firm/new", "action": "Open Wizard", }, { "title": "Platform SMTP Settings", "description": "Configure SMTP used for firm creation invites and system emails.", "href": "/email/platform-smtp", "action": "Configure SMTP", }, ], "Catalogue Wizards": [ { "title": "Service Catalogue", "description": "Manage master service catalogue.", "href": "/services/catalogue", "action": "Open Catalogue", }, { "title": "Default Task Templates", "description": "Manage default task templates used by firms.", "href": "/services/default-templates", "action": "Open Templates", }, { "title": "Bulk Service Imports", "description": "Use existing bulk import screens for service master and default tasks.", "href": "/services/bulk-imports", "action": "Open Imports", }, ], "Control Shortcuts": [ { "title": "Platform Billing", "description": "Open existing platform billing module.", "href": "/platform-billing", "action": "Open Billing", }, { "title": "Audit Logs", "description": "Review system audit logs.", "href": "/audit-logs", "action": "Open Logs", }, ], } def get_recent_audit_logs(db: Session, limit: int = 25) -> list[Any]: if AuditLog is None: return [] return db.execute(select(AuditLog).order_by(AuditLog.created_at_utc.desc()).limit(limit)).scalars().all()