diff --git a/app/modules/firm_admin_dashboard/service.py b/app/modules/firm_admin_dashboard/service.py index fdbfe21..b15f3a1 100644 --- a/app/modules/firm_admin_dashboard/service.py +++ b/app/modules/firm_admin_dashboard/service.py @@ -1,8 +1,10 @@ from __future__ import annotations from datetime import date, datetime, timezone +from io import BytesIO from typing import Any +from openpyxl import Workbook, load_workbook from sqlalchemy import func, select from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session @@ -520,6 +522,196 @@ def create_firm_internal_user( "email_error": email_error, } + +FIRM_USER_IMPORT_HEADERS = [ + "role", + "full_name", + "email", + "branch_id", + "branch_code", + "branch_name", + "employee_code", + "mobile", + "department", + "designation", + "date_of_joining", + "employment_type", +] + +FIRM_USER_IMPORT_SAMPLE_ROWS = [ + ["partner", "Sample Partner", "partner@example.com", "", "HO", "", "PTR001", "9999999999", "Management", "Partner", "2026-04-01", "full_time"], + ["manager", "Sample Manager", "manager@example.com", "", "HO", "", "MGR001", "9999999998", "Operations", "Branch Manager", "2026-04-01", "full_time"], + ["staff", "Sample Staff", "staff@example.com", "", "HO", "", "EMP001", "9999999997", "Audit", "Associate", "2026-04-01", "full_time"], +] + + +def build_firm_user_import_template(db: Session, request, current_user) -> bytes: + roles = set(get_user_role_names(db, current_user.id)) + tenant_id = _active_tenant_id(request, current_user, roles) + tenant = _tenant(db, tenant_id) + branches = _branch_rows(db, tenant_id) + + wb = Workbook() + ws = wb.active + ws.title = "Firm Users" + ws.append(FIRM_USER_IMPORT_HEADERS) + for row in FIRM_USER_IMPORT_SAMPLE_ROWS: + ws.append(row) + + notes = wb.create_sheet("Instructions") + notes.append(["Firm", getattr(tenant, "display_name", None) or getattr(tenant, "name", None) or "Active firm"]) + notes.append(["Allowed role values", "partner, manager, staff"]) + notes.append(["Manager mapping", "manager imports as existing Branch Manager role"]) + notes.append(["Branch selection", "Use branch_id OR branch_code OR exact branch_name from Branches sheet"]) + notes.append(["Employment type values", "full_time, part_time, article_assistant, intern, consultant, contract"]) + notes.append(["Date format", "YYYY-MM-DD preferred"]) + notes.append(["Invite", "Each valid row creates user + employee link + invite token. Invite link is shown in import result."]) + + branch_sheet = wb.create_sheet("Branches") + branch_sheet.append(["branch_id", "branch_code", "branch_name", "is_active", "is_head_office"]) + for branch in branches: + branch_sheet.append([branch.get("id"), branch.get("code"), branch.get("name"), branch.get("is_active"), branch.get("is_head_office")]) + + for sheet in wb.worksheets: + for cell in sheet[1]: + cell.font = cell.font.copy(bold=True) + for column_cells in sheet.columns: + letter = column_cells[0].column_letter + width = max(14, min(34, max(len(str(c.value or "")) for c in column_cells) + 3)) + sheet.column_dimensions[letter].width = width + + bio = BytesIO() + wb.save(bio) + return bio.getvalue() + + +def _cell_text(value: Any) -> str: + if value is None: + return "" + if isinstance(value, datetime): + return value.date().isoformat() + if isinstance(value, date): + return value.isoformat() + return str(value).strip() + + +def _load_firm_user_import_rows(content: bytes) -> list[dict[str, Any]]: + try: + wb = load_workbook(BytesIO(content), data_only=True) + except Exception as exc: + raise ValueError(f"Unable to read Excel file: {exc}") from exc + ws = wb["Firm Users"] if "Firm Users" in wb.sheetnames else wb.active + headers = [str(cell.value or "").strip().lower() for cell in ws[1]] + if not any(headers): + raise ValueError("Excel file has no header row.") + rows: list[dict[str, Any]] = [] + for row_no, row in enumerate(ws.iter_rows(min_row=2, values_only=True), start=2): + if not any(value not in (None, "") for value in row): + continue + item = {headers[idx]: _cell_text(row[idx] if idx < len(row) else None) for idx in range(len(headers)) if headers[idx]} + item["_row_no"] = row_no + rows.append(item) + if not rows: + raise ValueError("Excel file has no data rows.") + return rows + + +def _resolve_import_branch_id(db: Session, tenant_id: int, row: dict[str, Any]) -> int: + raw_id = _cell_text(row.get("branch_id")) + raw_code = _cell_text(row.get("branch_code")) + raw_name = _cell_text(row.get("branch_name")) + + branch = None + if raw_id: + try: + branch = db.get(Branch, int(float(raw_id))) + except Exception as exc: + raise ValueError(f"Invalid branch_id '{raw_id}'.") from exc + elif raw_code: + branch = db.execute( + select(Branch).where(Branch.tenant_id == int(tenant_id), func.lower(Branch.code) == raw_code.lower()) + ).scalar_one_or_none() + elif raw_name: + branch = db.execute( + select(Branch).where(Branch.tenant_id == int(tenant_id), func.lower(Branch.name) == raw_name.lower()) + ).scalar_one_or_none() + else: + active = db.execute( + select(Branch) + .where(Branch.tenant_id == int(tenant_id), Branch.is_active.is_(True)) + .order_by(Branch.is_head_office.desc(), Branch.id.asc()) + ).scalars().all() + if len(active) == 1: + branch = active[0] + else: + raise ValueError("Branch is required when the firm has multiple active branches.") + + if not branch or int(branch.tenant_id) != int(tenant_id): + raise ValueError("Branch does not belong to the active firm.") + if not branch.is_active: + raise ValueError("Selected branch is inactive.") + return int(branch.id) + + +def import_firm_internal_users(db: Session, *, request, actor: User, content: bytes) -> dict[str, Any]: + actor_roles = set(get_user_role_names(db, actor.id)) + if not actor_roles.intersection(FIRM_ADMIN_ROLES): + raise PermissionError("Only Firm Admin or System Admin can import firm users.") + tenant_id = _active_tenant_id(request, actor, actor_roles) + if not tenant_id: + raise ValueError("Active firm context is required before importing users.") + + rows = _load_firm_user_import_rows(content) + results: list[dict[str, Any]] = [] + created = failed = 0 + invite_links: list[dict[str, Any]] = [] + + for row in rows: + row_no = int(row.get("_row_no") or 0) + role_key = _cell_text(row.get("role") or "staff").lower() + email = _cell_text(row.get("email")).lower() + full_name = _cell_text(row.get("full_name")) + try: + if not get_onboarding_role_config(role_key): + raise ValueError("Role must be one of partner, manager or staff.") + branch_id = _resolve_import_branch_id(db, int(tenant_id), row) + result = create_firm_internal_user( + db, + request=request, + actor=actor, + role_key=role_key, + email=email, + full_name=full_name, + branch_id=branch_id, + employee_code=_cell_text(row.get("employee_code")), + mobile=_cell_text(row.get("mobile")), + department=_cell_text(row.get("department")), + designation=_cell_text(row.get("designation")), + date_of_joining=_cell_text(row.get("date_of_joining")), + employment_type=_cell_text(row.get("employment_type")) or None, + ) + created += 1 + invite_url = result.get("invite_url") + invite_links.append({ + "row_no": row_no, + "name": full_name, + "email": email, + "role": result.get("role_name"), + "invite_url": invite_url, + "email_status": result.get("email_status"), + }) + results.append({"row_no": row_no, "status": "created", "email": email, "name": full_name, "role": result.get("role_name"), "message": "User, role, employee link and invite created."}) + except Exception as exc: + db.rollback() + failed += 1 + results.append({"row_no": row_no, "status": "failed", "email": email, "name": full_name, "role": role_key, "message": str(exc)}) + + return { + "summary": {"total": len(rows), "created": created, "failed": failed}, + "rows": results, + "invite_links": invite_links, + } + 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) diff --git a/app/modules/firm_admin_dashboard/templates/firm_admin_dashboard/partials/users.html b/app/modules/firm_admin_dashboard/templates/firm_admin_dashboard/partials/users.html index 45ccd7c..b835f83 100644 --- a/app/modules/firm_admin_dashboard/templates/firm_admin_dashboard/partials/users.html +++ b/app/modules/firm_admin_dashboard/templates/firm_admin_dashboard/partials/users.html @@ -8,6 +8,8 @@ + Add Partner + Add Manager + Add Staff + Import Users + Template Manage Users diff --git a/app/modules/firm_admin_dashboard/templates/firm_admin_dashboard/user_import.html b/app/modules/firm_admin_dashboard/templates/firm_admin_dashboard/user_import.html new file mode 100644 index 0000000..129b3ff --- /dev/null +++ b/app/modules/firm_admin_dashboard/templates/firm_admin_dashboard/user_import.html @@ -0,0 +1,43 @@ +{% extends "ui/templates/base/layout.html" %} +{% block content %} +
+
+
+
+

Firm Admin Bulk Onboarding

+

Import Partner / Manager / Staff

+

Upload the Excel template to create user accounts, map roles and branches, create/link employee masters and generate invite links in bulk.

+
+ Back to Users +
+
+ + {% if flash %} +
{{ flash }}
+ {% endif %} + +
+
+

1. Download template

+

The template includes role values, sample rows and your active firm's branch list.

+ Download Excel Template +
+ +
+ +

2. Upload completed file

+

Allowed roles: partner, manager, staff. Manager will be assigned your existing Branch Manager role.

+ + +
+
+ +
+
Important
+

Do not keep the sample rows unless you want them imported. Email and employee code must be unique. If email sending fails, the fallback invite links will still be shown after import.

+
+
+{% endblock %} diff --git a/app/modules/firm_admin_dashboard/templates/firm_admin_dashboard/user_import_done.html b/app/modules/firm_admin_dashboard/templates/firm_admin_dashboard/user_import_done.html new file mode 100644 index 0000000..9837c9e --- /dev/null +++ b/app/modules/firm_admin_dashboard/templates/firm_admin_dashboard/user_import_done.html @@ -0,0 +1,54 @@ +{% extends "ui/templates/base/layout.html" %} +{% block content %} +
+
+
+
+

Firm Admin Bulk Onboarding

+

Import Result

+

Created {{ import_result.summary.created }} of {{ import_result.summary.total }} rows. Failed {{ import_result.summary.failed }} rows.

+
+
+ Import More + Back to Users +
+
+
+ + {% if import_result.invite_links %} +
+

Invite link fallback

+

Copy these links if automatic email is not configured or failed for any user.

+
+ {% for item in import_result.invite_links %} +
+
Row {{ item.row_no }} — {{ item.name }} / {{ item.email }} / {{ item.role }}
+
Email status: {{ item.email_status or 'not_attempted' }}
+ +
+ {% endfor %} +
+
+ {% endif %} + +
+ + + + + + {% for row in import_result.rows %} + + + + + + + + + {% endfor %} + +
RowStatusNameEmailRoleMessage
{{ row.row_no }}{{ row.status }}{{ row.name or '-' }}{{ row.email or '-' }}{{ row.role or '-' }}{{ row.message }}
+
+
+{% endblock %} diff --git a/app/modules/firm_admin_dashboard/ui.py b/app/modules/firm_admin_dashboard/ui.py index b5c8495..fac31f6 100644 --- a/app/modules/firm_admin_dashboard/ui.py +++ b/app/modules/firm_admin_dashboard/ui.py @@ -1,7 +1,9 @@ from __future__ import annotations -from fastapi import APIRouter, Form, Request -from fastapi.responses import RedirectResponse +from io import BytesIO + +from fastapi import APIRouter, File, Form, Request, UploadFile +from fastapi.responses import RedirectResponse, StreamingResponse from app.core.db.common import CommonSessionLocal from app.core.http_responses import ui_access_denied @@ -13,8 +15,10 @@ from app.modules.firm_admin_dashboard.service import ( build_dashboard_payload, build_user_onboarding_payload, can_access_firm_admin_dashboard, + build_firm_user_import_template, create_firm_internal_user, get_onboarding_role_config, + import_firm_internal_users, ) router = APIRouter(prefix="/firm-admin", tags=["firm-admin-dashboard-ui"]) @@ -192,3 +196,73 @@ def user_onboarding_submit( ) finally: db.close() + +@router.get("/users/import") +def user_import_form(request: Request): + db = CommonSessionLocal() + try: + current_user = get_current_user(request, db=db) + if not current_user: + return _redirect_login() + if not can_access_firm_admin_dashboard(db, current_user): + return ui_access_denied() + return templates.TemplateResponse( + "modules/firm_admin_dashboard/templates/firm_admin_dashboard/user_import.html", + _ctx(request, db, current_user, active_tab="users", title="Import Firm Users"), + ) + finally: + db.close() + + +@router.get("/users/import/template") +def user_import_template(request: Request): + db = CommonSessionLocal() + try: + current_user = get_current_user(request, db=db) + if not current_user: + return _redirect_login() + if not can_access_firm_admin_dashboard(db, current_user): + return ui_access_denied() + content = build_firm_user_import_template(db, request, current_user) + return StreamingResponse( + BytesIO(content), + media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + headers={"Content-Disposition": 'attachment; filename="firm_admin_user_import_template.xlsx"'}, + ) + finally: + db.close() + + +@router.post("/users/import") +async def user_import_submit(request: Request, import_file: UploadFile = File(...), csrf_token: str = Form(...)): + validate_csrf(request, csrf_token) + db = CommonSessionLocal() + try: + current_user = get_current_user(request, db=db) + if not current_user: + return _redirect_login() + if not can_access_firm_admin_dashboard(db, current_user): + return ui_access_denied() + if not import_file.filename or not import_file.filename.lower().endswith((".xlsx", ".xlsm")): + return templates.TemplateResponse( + "modules/firm_admin_dashboard/templates/firm_admin_dashboard/user_import.html", + _ctx(request, db, current_user, active_tab="users", title="Import Firm Users", flash="Please upload an .xlsx Excel file."), + status_code=400, + ) + try: + content = await import_file.read() + result = import_firm_internal_users(db, request=request, actor=current_user, content=content) + except Exception as exc: + db.rollback() + return templates.TemplateResponse( + "modules/firm_admin_dashboard/templates/firm_admin_dashboard/user_import.html", + _ctx(request, db, current_user, active_tab="users", title="Import Firm Users", flash=str(exc)), + status_code=400, + ) + return templates.TemplateResponse( + "modules/firm_admin_dashboard/templates/firm_admin_dashboard/user_import_done.html", + _ctx(request, db, current_user, active_tab="users", title="Import Result", import_result=result), + ) + finally: + db.close() +