diff --git a/app/modules/system_admin_dashboard/__init__.py b/app/modules/system_admin_dashboard/__init__.py
new file mode 100644
index 0000000..7f4808c
--- /dev/null
+++ b/app/modules/system_admin_dashboard/__init__.py
@@ -0,0 +1 @@
+"""System Admin utility dashboard package."""
diff --git a/app/modules/system_admin_dashboard/service.py b/app/modules/system_admin_dashboard/service.py
new file mode 100644
index 0000000..ce07c79
--- /dev/null
+++ b/app/modules/system_admin_dashboard/service.py
@@ -0,0 +1,358 @@
+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, Tenant, YearBackupExport
+from app.modules.services.models import (
+ ServiceCatalogue,
+ ServiceCategory,
+ ServiceDefaultTaskTemplate,
+ ServiceDueDateRule,
+)
+
+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]
+
+
+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 _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,
+ "security": None,
+ "message": "Platform SMTP model not available. Apply W2 SMTP migration/package 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,
+ "security": None,
+ "message": "Platform SMTP is not configured.",
+ }
+ configured = bool(setting.smtp_host and setting.smtp_port and setting.from_email)
+ return {
+ "configured": configured,
+ "active": bool(getattr(setting, "is_active", False)),
+ "host": setting.smtp_host,
+ "port": setting.smtp_port,
+ "from_email": setting.from_email,
+ "from_name": setting.from_name,
+ "security": setting.smtp_security,
+ "message": "Platform SMTP is ready." if configured and setting.is_active else "Platform SMTP is saved but not 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 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),
+ }
+
+
+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)
+ return {
+ "total_firms": total_firms,
+ "active_firms": active_firms,
+ "inactive_firms": inactive_firms,
+ "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,
+ "storage": storage,
+ "smtp": smtp,
+ }
+
+
+def list_firms(db: Session, limit: int = 100) -> list[dict[str, Any]]:
+ 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()
+ )
+ role_ids = _firm_admin_role_ids(db)
+ primary_admins: dict[int, str] = {}
+ if role_ids:
+ rows = db.execute(
+ select(User.tenant_id, User.email, User.full_name)
+ .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 in rows:
+ if tenant_id not in primary_admins:
+ primary_admins[tenant_id] = full_name or email
+
+ 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": int(branch_counts.get(tenant.id, 0) or 0),
+ "user_count": int(user_counts.get(tenant.id, 0) or 0),
+ "primary_admin": primary_admins.get(tenant.id, "-"),
+ "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(100)).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()
+ )
+ 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))),
+ "default_task_count": scalar_count(db, select(func.count(ServiceDefaultTaskTemplate.id))),
+ "due_rule_count": scalar_count(db, select(func.count(ServiceDueDateRule.id))),
+ "categories": categories,
+ "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),
+ }
+ for service in services
+ ],
+ }
+
+
+def get_smtp_summary(db: Session) -> dict[str, Any]:
+ return _platform_smtp_status(db)
+
+
+def get_report_cards(db: Session) -> list[dict[str, str]]:
+ overview = get_overview(db)
+ return [
+ {
+ "title": "Firm Setup Completeness",
+ "description": "Review firms, admin invite status, branches and basic readiness.",
+ "metric": f"{overview['total_firms']} firms",
+ "href": "/system-admin/dashboard?tab=firms",
+ },
+ {
+ "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": "Service Catalogue Readiness",
+ "description": "Check categories, services, default task templates and due-date rules.",
+ "metric": f"{overview['service_count']} services / {overview['default_task_count']} tasks",
+ "href": "/system-admin/dashboard?tab=catalogue",
+ },
+ {
+ "title": "Platform SMTP Status",
+ "description": "Check whether system emails can be sent for firm invites and security 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() -> list[dict[str, str]]:
+ return [
+ {
+ "title": "Create Firm Wizard",
+ "description": "Create a firm, 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",
+ },
+ {
+ "title": "Service Catalogue",
+ "description": "Manage master service catalogue and default task templates.",
+ "href": "/services/catalogue",
+ "action": "Open Catalogue",
+ },
+ {
+ "title": "Bulk Service Imports",
+ "description": "Use existing bulk import screens for service master and default tasks.",
+ "href": "/services/bulk-imports",
+ "action": "Open Imports",
+ },
+ ]
+
+
+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()
diff --git a/app/modules/system_admin_dashboard/templates/system_admin_dashboard/dashboard.html b/app/modules/system_admin_dashboard/templates/system_admin_dashboard/dashboard.html
new file mode 100644
index 0000000..fac4b9f
--- /dev/null
+++ b/app/modules/system_admin_dashboard/templates/system_admin_dashboard/dashboard.html
@@ -0,0 +1,97 @@
+{% extends "base/layout.html" %}
+{% block content %}
+ Platform Utility Centre Monitor firms, service catalogue, SMTP, storage, reports, wizards and audit logs from one place.System Admin Dashboard
+
Latest 25 audit events.
| Time | User | Action | Entity | Status |
|---|---|---|---|---|
| {{ log.created_at_utc }} | {{ log.actor_email or '-' }} | {{ log.action }} | {{ log.entity_type }}{% if log.entity_name %} > {{ log.entity_name }}{% endif %} | {{ log.status }} |
| No audit logs found. | ||||
First 100 catalogue services with default task/rule counts.
| Code | Service | Category | Recurrence | Tasks | Rules | Status |
|---|---|---|---|---|---|---|
| {{ service.code }} | {{ service.name }} | {{ service.category }} | {{ service.recurrence_type }} | {{ service.task_count }} | {{ service.rule_count }} | {% if service.is_active %}Active{% else %}Inactive{% endif %} |
| No service catalogue found. | ||||||
Latest firms created on the platform.
| Code | Firm | Status | Firm Admin | Branches | Users | Action |
|---|---|---|---|---|---|---|
| {{ firm.code }} | +{{ firm.name }} {{ firm.firm_type or '-' }}{% if firm.contact_email %} > {{ firm.contact_email }}{% endif %} |
+ {% if firm.is_active %}Active{% else %}Inactive{% endif %} | +{{ firm.primary_admin }} | +{{ firm.branch_count }} | +{{ firm.user_count }} | +Open Settings | +
| No firms found. | ||||||
{{ card.description }}
+Used for Firm Creation Wizard invites and system-level emails.
+{{ smtp.message }}
+{{ storage.storage_root }}{{ storage.branding_root }}{{ card.description }}
+