A R R R Associates
126 lines
4.0 KiB
Python
126 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Request
|
|
from fastapi.responses import RedirectResponse
|
|
|
|
from app.core.db.common import CommonSessionLocal
|
|
from app.core.http_responses import ui_access_denied
|
|
from app.core.security.csrf import get_or_create_csrf_token
|
|
from app.core.security.session_auth import get_current_user
|
|
from app.core.templating import templates
|
|
from app.modules.core.rbac.deps import get_user_permissions, get_user_roles
|
|
from app.modules.system_admin_dashboard.service import (
|
|
get_billing_readiness,
|
|
get_catalogue_summary,
|
|
get_firm_setup_health,
|
|
get_overview,
|
|
get_recent_audit_logs,
|
|
get_report_cards,
|
|
get_smtp_summary,
|
|
get_storage_status,
|
|
get_wizard_cards,
|
|
is_system_admin,
|
|
list_firms,
|
|
)
|
|
|
|
router = APIRouter(prefix="/system-admin", tags=["system-admin-dashboard-ui"])
|
|
|
|
_ALLOWED_TABS = {
|
|
"overview",
|
|
"firms",
|
|
"firm-setup-health",
|
|
"catalogue",
|
|
"smtp",
|
|
"storage",
|
|
"billing",
|
|
"reports",
|
|
"wizards",
|
|
"audit-logs",
|
|
}
|
|
|
|
|
|
def _login_redirect() -> RedirectResponse:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
|
|
|
|
def _base_context(request: Request, db, current_user, **extra):
|
|
ctx = {
|
|
"request": request,
|
|
"current_user": current_user,
|
|
"current_user_roles": get_user_roles(db, current_user.id),
|
|
"current_user_permissions": get_user_permissions(db, current_user.id),
|
|
"csrf_token": get_or_create_csrf_token(request),
|
|
}
|
|
ctx.update(extra)
|
|
return ctx
|
|
|
|
|
|
def _system_admin_or_response(request: Request, db):
|
|
current_user = get_current_user(request, db=db)
|
|
if not current_user:
|
|
return None, _login_redirect()
|
|
if not is_system_admin(db, current_user):
|
|
return None, ui_access_denied()
|
|
return current_user, None
|
|
|
|
|
|
@router.get("/dashboard")
|
|
def dashboard(request: Request, tab: str = "overview"):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
current_user, response = _system_admin_or_response(request, db)
|
|
if response:
|
|
return response
|
|
active_tab = tab if tab in _ALLOWED_TABS else "overview"
|
|
return templates.TemplateResponse(
|
|
"modules/system_admin_dashboard/templates/system_admin_dashboard/dashboard.html",
|
|
_base_context(
|
|
request,
|
|
db,
|
|
current_user,
|
|
title="System Admin Dashboard",
|
|
active_tab=active_tab,
|
|
),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/dashboard/tab/{tab_name}")
|
|
def dashboard_tab(request: Request, tab_name: str):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
current_user, response = _system_admin_or_response(request, db)
|
|
if response:
|
|
return response
|
|
if tab_name not in _ALLOWED_TABS:
|
|
tab_name = "overview"
|
|
|
|
extra = {"active_tab": tab_name}
|
|
template = f"modules/system_admin_dashboard/templates/system_admin_dashboard/partials/{tab_name.replace('-', '_')}.html"
|
|
|
|
if tab_name == "overview":
|
|
extra["overview"] = get_overview(db)
|
|
elif tab_name == "firms":
|
|
extra["firms"] = list_firms(db)
|
|
elif tab_name == "firm-setup-health":
|
|
extra["health"] = get_firm_setup_health(db)
|
|
elif tab_name == "catalogue":
|
|
extra["catalogue"] = get_catalogue_summary(db)
|
|
elif tab_name == "smtp":
|
|
extra["smtp"] = get_smtp_summary(db)
|
|
elif tab_name == "storage":
|
|
extra["storage"] = get_storage_status(db)
|
|
elif tab_name == "billing":
|
|
extra["billing"] = get_billing_readiness(db)
|
|
elif tab_name == "reports":
|
|
extra["report_groups"] = get_report_cards(db)
|
|
elif tab_name == "wizards":
|
|
extra["wizard_groups"] = get_wizard_cards()
|
|
elif tab_name == "audit-logs":
|
|
extra["audit_logs"] = get_recent_audit_logs(db)
|
|
|
|
return templates.TemplateResponse(template, _base_context(request, db, current_user, **extra))
|
|
finally:
|
|
db.close()
|