A R R R Associates
359 lines
13 KiB
Python
359 lines
13 KiB
Python
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()
|