diff --git a/app/modules/firm_admin_dashboard/__init__.py b/app/modules/firm_admin_dashboard/__init__.py
new file mode 100644
index 0000000..4bfcaef
--- /dev/null
+++ b/app/modules/firm_admin_dashboard/__init__.py
@@ -0,0 +1 @@
+"""Firm Admin setup/control dashboard package."""
diff --git a/app/modules/firm_admin_dashboard/service.py b/app/modules/firm_admin_dashboard/service.py
new file mode 100644
index 0000000..d5d8634
--- /dev/null
+++ b/app/modules/firm_admin_dashboard/service.py
@@ -0,0 +1,374 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import date, datetime
+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.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.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:
+ 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),
+ "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."}
+
+
+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 Firm Admin, Partner, Manager, Staff and role mapping.", "href": "/system-settings/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 []
diff --git a/app/modules/firm_admin_dashboard/templates/firm_admin_dashboard/dashboard.html b/app/modules/firm_admin_dashboard/templates/firm_admin_dashboard/dashboard.html
new file mode 100644
index 0000000..7b3e60c
--- /dev/null
+++ b/app/modules/firm_admin_dashboard/templates/firm_admin_dashboard/dashboard.html
@@ -0,0 +1,74 @@
+{% extends "ui/templates/base/layout.html" %}
+{% block content %}
+ Firm Setup Control Centre Configure firm details, branches, users, roles, services, task templates, financial years, branding and billing readiness.Firm Administration Dashboard
+
Latest firm-level activity, if audit log model is available.
| 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 or '-' }} |
| No audit logs found or audit model is not available. | ||||
Create branches and assign partner/users for each branch.
| Branch | Status | Partners | Users | Clients | Login / Assignment | Storage / SMTP |
|---|---|---|---|---|---|---|
{{ branch.name }} {{ branch.code }}{% if branch.is_head_office %} > Head Office{% endif %} |
+ {% if branch.is_active %}Active{% else %}Inactive{% endif %} | +{{ branch.partner_count }} | +{{ branch.user_count }} | +{{ branch.client_count }} | +Login: {{ 'Allowed' if branch.allow_login else 'Blocked' }} Assignments: {{ 'Allowed' if branch.allow_new_assignments else 'Blocked' }} |
+ Storage: {{ branch.local_storage_path or 'Not set' }} SMTP: {{ 'Configured' if branch.smtp_configured else 'Pending' }} |
+
| No branches found. Create the first branch from System Settings. | ||||||
Create, mark current, lock and manage financial years from existing FY settings.
| FY | AY | Period | Current | Lock Status |
|---|---|---|---|---|
| {{ fy.year_code }} | {{ fy.assessment_year }} | {{ fy.start_date }} to {{ fy.end_date }} | {% if fy.is_current %}Current{% else %}-{% endif %} | {% if fy.is_locked %}Locked{% else %}Open{% endif %} |
| No financial years created yet. | ||||
This tab gives shortcuts to existing firm setup pages. No duplicate settings are created here.
+Firm, branch and year configuration dashboard.
+Logo, favicon, colours, contact and display name.
+Branch/firm SMTP and email sender settings.
+Invoice series, GST, bank and payment settings.
+Branch storage node readiness.
+Firm setup and security audit trail.
+Firm Admin should complete these items before branch operations start.
+{{ report.desc }}
Enabled services and firm task template readiness.
| Service | Category | Recurrence | Task Templates | Status |
|---|---|---|---|---|
{{ service.service_name }} {{ service.service_code }} | {{ service.category }} | {{ service.recurrence_type }} | {{ service.template_count }} | {% if service.ready %}Ready{% else %}Template Pending{% endif %} |
| No firm services selected yet. | ||||
Firm Admin controls user invitations, branch mapping and role mapping.
| User | Roles | Login Status | Designation | Mobile |
|---|---|---|---|---|
{{ user.name }} {{ user.email }} |
+ {% for role in user.roles %}{{ role }}{% else %}No role{% endfor %} |
+ {% if user.is_locked %}Locked{% elif not user.is_active %}Inactive{% elif not user.allow_login %}Login Blocked{% else %}Active{% endif %}{% if user.must_change_password %} Invite/password pending {% endif %} |
+ {{ user.designation or '-' }} | +{{ user.mobile or '-' }} | +
| No users found. | ||||
These cards link to existing setup screens. They do not duplicate existing features.
+{{ wizard.desc }}