Add firm admin bulk user import with template
This commit is contained in:
A R R R Associates
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user