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()