A R R R Associates
293 lines
10 KiB
Python
293 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Form, Request
|
|
from fastapi.responses import RedirectResponse
|
|
from sqlalchemy import select
|
|
|
|
from app.core.db.common import CommonSessionLocal
|
|
from app.core.http_responses import forbidden_response, ui_access_denied
|
|
from app.core.security.csrf import get_or_create_csrf_token, validate_csrf
|
|
from app.core.security.session_auth import get_current_user
|
|
from app.core.templating import templates
|
|
from app.modules.core.audit.service import model_snapshot, write_audit_log
|
|
from app.modules.core.rbac.deps import get_user_permissions, get_user_roles
|
|
from app.modules.core.tenancy.models import Branch, FinancialYear, Tenant
|
|
from app.modules.core.iam.models import User
|
|
from app.modules.wizards.service import (
|
|
FirmWizardError,
|
|
build_firm_wizard_payload,
|
|
create_firm_from_wizard,
|
|
default_ay_from_fy,
|
|
default_dates_from_fy,
|
|
validate_firm_wizard_payload,
|
|
)
|
|
|
|
router = APIRouter(prefix="/wizards", tags=["wizards-ui"])
|
|
|
|
|
|
def _ctx(request: Request, db, user, **extra):
|
|
ctx = {
|
|
"request": request,
|
|
"current_user": user,
|
|
"current_user_roles": get_user_roles(db, user.id),
|
|
"current_user_permissions": get_user_permissions(db, user.id),
|
|
"csrf_token": get_or_create_csrf_token(request),
|
|
}
|
|
ctx.update(extra)
|
|
return ctx
|
|
|
|
|
|
def _login_redirect():
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
|
|
|
|
def _csrf_rejected(request: Request):
|
|
return forbidden_response(request, "CSRF validation failed")
|
|
|
|
|
|
def _is_system_admin(db, user) -> bool:
|
|
return "System Admin" in get_user_roles(db, user.id)
|
|
|
|
|
|
def _system_admin_or_denied(request: Request, db):
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return None, _login_redirect()
|
|
if not _is_system_admin(db, user):
|
|
return None, ui_access_denied()
|
|
return user, None
|
|
|
|
|
|
def _default_form_values() -> dict:
|
|
start, end = default_dates_from_fy("2026-27")
|
|
return {
|
|
"tenant_code": "",
|
|
"tenant_name": "",
|
|
"firm_type": "partnership",
|
|
"default_timezone": "Asia/Kolkata",
|
|
"default_session_duration_minutes": 480,
|
|
"default_otp_required_roles_csv": "Partner,System Admin",
|
|
"default_storage_mode": "local_only",
|
|
"branch_code": "HO",
|
|
"branch_name": "Head Office",
|
|
"branch_timezone": "Asia/Kolkata",
|
|
"branch_address_line1": "",
|
|
"branch_address_line2": "",
|
|
"branch_city": "",
|
|
"branch_state": "",
|
|
"branch_pin_code": "",
|
|
"branch_gstin": "",
|
|
"branch_pan": "",
|
|
"admin_full_name": "",
|
|
"admin_email": "",
|
|
"admin_mobile": "",
|
|
"admin_designation": "Firm Admin",
|
|
"create_financial_year": True,
|
|
"fy_year_code": "2026-27",
|
|
"fy_assessment_year": default_ay_from_fy("2026-27"),
|
|
"fy_start_date": start.isoformat() if start else "",
|
|
"fy_end_date": end.isoformat() if end else "",
|
|
"fy_is_current": True,
|
|
}
|
|
|
|
|
|
@router.get("/system/firm/new")
|
|
def system_firm_new(request: Request):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, denied = _system_admin_or_denied(request, db)
|
|
if denied:
|
|
return denied
|
|
return templates.TemplateResponse(
|
|
"modules/wizards/templates/wizards/system_firm_new.html",
|
|
_ctx(
|
|
request,
|
|
db,
|
|
user,
|
|
title="Create Firm Wizard",
|
|
form=_default_form_values(),
|
|
errors=[],
|
|
),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/system/firm/preview")
|
|
def system_firm_preview(
|
|
request: Request,
|
|
tenant_code: str = Form(""),
|
|
tenant_name: str = Form(""),
|
|
firm_type: str = Form("partnership"),
|
|
default_timezone: str = Form("Asia/Kolkata"),
|
|
default_session_duration_minutes: int = Form(480),
|
|
default_otp_required_roles_csv: str = Form("Partner,System Admin"),
|
|
default_storage_mode: str = Form("local_only"),
|
|
branch_code: str = Form("HO"),
|
|
branch_name: str = Form("Head Office"),
|
|
branch_timezone: str = Form("Asia/Kolkata"),
|
|
branch_address_line1: str = Form(""),
|
|
branch_address_line2: str = Form(""),
|
|
branch_city: str = Form(""),
|
|
branch_state: str = Form(""),
|
|
branch_pin_code: str = Form(""),
|
|
branch_gstin: str = Form(""),
|
|
branch_pan: str = Form(""),
|
|
admin_full_name: str = Form(""),
|
|
admin_email: str = Form(""),
|
|
admin_mobile: str = Form(""),
|
|
admin_designation: str = Form("Firm Admin"),
|
|
create_financial_year: str | None = Form(None),
|
|
fy_year_code: str = Form(""),
|
|
fy_assessment_year: str = Form(""),
|
|
fy_start_date: str = Form(""),
|
|
fy_end_date: str = Form(""),
|
|
fy_is_current: str | None = Form(None),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
try:
|
|
validate_csrf(request, csrf_token)
|
|
except PermissionError:
|
|
return _csrf_rejected(request)
|
|
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, denied = _system_admin_or_denied(request, db)
|
|
if denied:
|
|
return denied
|
|
|
|
form = build_firm_wizard_payload(locals())
|
|
errors = validate_firm_wizard_payload(db, form)
|
|
if errors:
|
|
return templates.TemplateResponse(
|
|
"modules/wizards/templates/wizards/system_firm_new.html",
|
|
_ctx(request, db, user, title="Create Firm Wizard", form=form, errors=errors),
|
|
status_code=400,
|
|
)
|
|
|
|
return templates.TemplateResponse(
|
|
"modules/wizards/templates/wizards/system_firm_preview.html",
|
|
_ctx(request, db, user, title="Preview Firm Creation", form=form, errors=[]),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/system/firm/confirm")
|
|
def system_firm_confirm(
|
|
request: Request,
|
|
tenant_code: str = Form(""),
|
|
tenant_name: str = Form(""),
|
|
firm_type: str = Form("partnership"),
|
|
default_timezone: str = Form("Asia/Kolkata"),
|
|
default_session_duration_minutes: int = Form(480),
|
|
default_otp_required_roles_csv: str = Form("Partner,System Admin"),
|
|
default_storage_mode: str = Form("local_only"),
|
|
branch_code: str = Form("HO"),
|
|
branch_name: str = Form("Head Office"),
|
|
branch_timezone: str = Form("Asia/Kolkata"),
|
|
branch_address_line1: str = Form(""),
|
|
branch_address_line2: str = Form(""),
|
|
branch_city: str = Form(""),
|
|
branch_state: str = Form(""),
|
|
branch_pin_code: str = Form(""),
|
|
branch_gstin: str = Form(""),
|
|
branch_pan: str = Form(""),
|
|
admin_full_name: str = Form(""),
|
|
admin_email: str = Form(""),
|
|
admin_mobile: str = Form(""),
|
|
admin_designation: str = Form("Firm Admin"),
|
|
create_financial_year: str | None = Form(None),
|
|
fy_year_code: str = Form(""),
|
|
fy_assessment_year: str = Form(""),
|
|
fy_start_date: str = Form(""),
|
|
fy_end_date: str = Form(""),
|
|
fy_is_current: str | None = Form(None),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
try:
|
|
validate_csrf(request, csrf_token)
|
|
except PermissionError:
|
|
return _csrf_rejected(request)
|
|
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, denied = _system_admin_or_denied(request, db)
|
|
if denied:
|
|
return denied
|
|
|
|
form = build_firm_wizard_payload(locals())
|
|
errors = validate_firm_wizard_payload(db, form)
|
|
if errors:
|
|
return templates.TemplateResponse(
|
|
"modules/wizards/templates/wizards/system_firm_new.html",
|
|
_ctx(request, db, user, title="Create Firm Wizard", form=form, errors=errors),
|
|
status_code=400,
|
|
)
|
|
|
|
try:
|
|
result = create_firm_from_wizard(db, form)
|
|
tenant_id = result.tenant.id
|
|
branch_id = result.branch.id
|
|
admin_id = result.firm_admin.id
|
|
fy_id = result.financial_year.id if result.financial_year else None
|
|
invite_email_status = result.invite_email_status
|
|
invite_email_error = result.invite_email_error
|
|
db.commit()
|
|
except FirmWizardError as exc:
|
|
db.rollback()
|
|
return templates.TemplateResponse(
|
|
"modules/wizards/templates/wizards/system_firm_new.html",
|
|
_ctx(request, db, user, title="Create Firm Wizard", form=form, errors=[str(exc)]),
|
|
status_code=400,
|
|
)
|
|
except Exception as exc:
|
|
db.rollback()
|
|
return templates.TemplateResponse(
|
|
"modules/wizards/templates/wizards/system_firm_new.html",
|
|
_ctx(request, db, user, title="Create Firm Wizard", form=form, errors=[f"Firm was not created: {exc}"]),
|
|
status_code=400,
|
|
)
|
|
|
|
tenant = db.execute(select(Tenant).where(Tenant.id == tenant_id)).scalar_one()
|
|
branch = db.execute(select(Branch).where(Branch.id == branch_id)).scalar_one()
|
|
firm_admin = db.execute(select(User).where(User.id == admin_id)).scalar_one()
|
|
financial_year = db.execute(select(FinancialYear).where(FinancialYear.id == fy_id)).scalar_one_or_none() if fy_id else None
|
|
|
|
write_audit_log(
|
|
db,
|
|
action="wizard.system.firm.create",
|
|
entity_type="tenant",
|
|
actor=user,
|
|
request=request,
|
|
entity_id=tenant.id,
|
|
entity_name=tenant.name,
|
|
target_tenant_id=tenant.id,
|
|
target_branch_id=branch.id,
|
|
details={
|
|
"tenant": model_snapshot(tenant, ["code", "name", "firm_type", "is_active", "default_timezone", "default_storage_mode"]),
|
|
"branch": model_snapshot(branch, ["code", "name", "is_head_office", "allow_login", "allow_new_assignments"]),
|
|
"firm_admin": model_snapshot(firm_admin, ["email", "full_name", "tenant_id", "branch_id", "is_active", "allow_login", "must_change_password"]),
|
|
"financial_year": model_snapshot(financial_year, ["year_code", "assessment_year", "start_date", "end_date", "is_current"]) if financial_year else None,
|
|
},
|
|
)
|
|
|
|
return templates.TemplateResponse(
|
|
"modules/wizards/templates/wizards/system_firm_complete.html",
|
|
_ctx(
|
|
request,
|
|
db,
|
|
user,
|
|
title="Firm Created",
|
|
tenant=tenant,
|
|
branch=branch,
|
|
firm_admin=firm_admin,
|
|
financial_year=financial_year,
|
|
invite_url=result.invite_url,
|
|
invite_email_status=invite_email_status,
|
|
invite_email_error=invite_email_error,
|
|
),
|
|
)
|
|
finally:
|
|
db.close()
|