from __future__ import annotations from fastapi import APIRouter, Form, Request from fastapi.responses import RedirectResponse from sqlalchemy import select from urllib.parse import urlencode from app.core.db.common import CommonSessionLocal 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.clients.models import Client, ClientBusinessUnit, ClientBranch from app.modules.registrations.models import ClientRegistration, RegistrationType from app.modules.core.rbac.deps import get_user_permissions, get_user_roles from app.modules.core.rbac.permission_guard import require_permission from app.modules.credential_vault.models import CredentialVaultEntry from app.modules.credential_vault.service import can_manage_vault, can_open_vault, can_view_entry router = APIRouter(prefix="/clients", tags=["client-business-structure"]) CREDENTIAL_CATEGORY_LABELS = { "gst_portal": "GST Portal", "income_tax_portal": "Income Tax Portal", "traces_tds": "TRACES / TDS", "mca_portal": "MCA Portal", "eway_bill": "E-Way Bill", "einvoice": "E-Invoice", "government_portal": "Government / Registration Portal", "banking": "Banking", "email": "Email", "software": "Software", "api_key": "API / Provider", "digital_signature": "Digital Signature / Token", "other": "Other", } def _tenant_id(request, user): return int(request.session.get("active_tenant_id") or request.session.get("tenant_id") or user.tenant_id) def _load_client(db, tenant_id: int, client_id: int): return db.execute(select(Client).where(Client.id == client_id, Client.tenant_id == tenant_id)).scalar_one_or_none() def _structure_redirect(client_id: int, *, message: str | None = None, error: str | None = None, anchor: str | None = None): params = {} if message: params["message"] = message if error: params["error"] = error url = f"/clients/{client_id}/business-structure" if params: url += "?" + urlencode(params) if anchor: url += f"#{anchor}" return RedirectResponse(url, 303) def _owned_business(db, tenant_id: int, client_id: int, business_unit_id: int): return db.execute( select(ClientBusinessUnit).where( ClientBusinessUnit.id == business_unit_id, ClientBusinessUnit.tenant_id == tenant_id, ClientBusinessUnit.client_id == client_id, ) ).scalar_one_or_none() def _owned_branch(db, tenant_id: int, client_id: int, branch_id: int): return db.execute( select(ClientBranch).where( ClientBranch.id == branch_id, ClientBranch.tenant_id == tenant_id, ClientBranch.client_id == client_id, ) ).scalar_one_or_none() def _owned_registration(db, tenant_id: int, client_id: int, registration_id: int): return db.execute( select(ClientRegistration).where( ClientRegistration.id == registration_id, ClientRegistration.tenant_id == tenant_id, ClientRegistration.client_id == client_id, ) ).scalar_one_or_none() def _context(request, db, user, **extra): data = { "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), } data.update(extra) return data @router.get("/{client_id}/business-structure") def business_structure_page(request: Request, client_id: int): db = CommonSessionLocal() try: user = get_current_user(request, db=db) if not user: return RedirectResponse("/login", 303) require_permission(db, user, "clients.view") client = _load_client(db, _tenant_id(request, user), client_id) if not client: return RedirectResponse("/clients", 303) businesses = db.execute(select(ClientBusinessUnit).where( ClientBusinessUnit.client_id == client.id ).order_by(ClientBusinessUnit.is_active.desc(), ClientBusinessUnit.is_primary.desc(), ClientBusinessUnit.business_name)).scalars().all() branches = db.execute(select(ClientBranch).where( ClientBranch.client_id == client.id ).order_by(ClientBranch.is_active.desc(), ClientBranch.is_primary.desc(), ClientBranch.branch_name)).scalars().all() registration_rows = db.execute( select(ClientRegistration, RegistrationType).join( RegistrationType, RegistrationType.id == ClientRegistration.registration_type_id ).where(ClientRegistration.client_id == client.id).order_by( ClientRegistration.status, RegistrationType.code, ClientRegistration.registration_number ) ).all() registrations = [row[0] for row in registration_rows] registration_type_codes = {row[0].id: row[1].code for row in registration_rows} registration_types = db.execute( select(RegistrationType).order_by(RegistrationType.sort_order, RegistrationType.name) ).scalars().all() # Credential Vault stays the single source of truth. This page only surfaces # metadata/actions for credentials linked to the registration rows the user can # already see; encrypted values are never decrypted here. registration_credentials: dict[int, list[CredentialVaultEntry]] = {row.id: [] for row in registrations} can_open_credentials = can_open_vault(db, user) can_manage_credentials = can_manage_vault(db, user) if registrations and can_open_credentials: registration_ids = [int(row.id) for row in registrations] credential_rows = db.execute( select(CredentialVaultEntry).where( CredentialVaultEntry.tenant_id == client.tenant_id, CredentialVaultEntry.client_id == client.id, CredentialVaultEntry.registration_id.in_(registration_ids), ).order_by( CredentialVaultEntry.status, CredentialVaultEntry.category, CredentialVaultEntry.title, ) ).scalars().all() raw_active_branch_id = request.session.get("active_branch_id") active_branch_id = ( int(raw_active_branch_id) if raw_active_branch_id not in (None, "", 0, "0") else getattr(user, "branch_id", None) ) for credential in credential_rows: if can_view_entry(db, user, credential, active_branch_id): registration_credentials.setdefault(int(credential.registration_id), []).append(credential) return templates.TemplateResponse( "modules/clients/templates/clients/business_structure.html", _context(request, db, user, title="Client Business Structure", client=client, businesses=businesses, branches=branches, registrations=registrations, registration_type_codes=registration_type_codes, registration_types=registration_types, registration_credentials=registration_credentials, credential_category_labels=CREDENTIAL_CATEGORY_LABELS, can_open_credentials=can_open_credentials, can_manage_credentials=can_manage_credentials, can_edit="clients.edit" in set(get_user_permissions(db, user.id))), ) finally: db.close() @router.post("/{client_id}/business-units") def add_business_unit(request: Request, client_id: int, business_code: str = Form(...), business_name: str = Form(...), trade_name: str = Form(""), nature_of_business: str = Form(""), is_primary: str | None = Form(None), csrf_token: str = Form(...)): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user = get_current_user(request, db=db); require_permission(db, user, "clients.edit") tenant_id = _tenant_id(request, user); client = _load_client(db, tenant_id, client_id) if not client: return RedirectResponse("/clients", 303) if is_primary: db.query(ClientBusinessUnit).filter_by(client_id=client.id).update({"is_primary": False}) db.add(ClientBusinessUnit( tenant_id=tenant_id, client_id=client.id, business_code=business_code.strip().upper(), business_name=business_name.strip(), trade_name=trade_name.strip() or None, nature_of_business=nature_of_business.strip() or None, is_primary=bool(is_primary), is_active=True, created_by_user_id=user.id, updated_by_user_id=user.id, )) db.commit() return RedirectResponse(f"/clients/{client_id}/business-structure", 303) finally: db.close() @router.post("/{client_id}/branches") def add_client_branch(request: Request, client_id: int, business_unit_id: int = Form(...), branch_code: str = Form(...), branch_name: str = Form(...), branch_type: str = Form("branch"), city: str = Form(""), state: str = Form(""), pincode: str = Form(""), is_primary: str | None = Form(None), csrf_token: str = Form(...)): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user = get_current_user(request, db=db); require_permission(db, user, "clients.edit") tenant_id = _tenant_id(request, user); client = _load_client(db, tenant_id, client_id) business = db.get(ClientBusinessUnit, business_unit_id) if not client or not business or business.client_id != client.id or business.tenant_id != tenant_id: return RedirectResponse("/clients", 303) if is_primary: db.query(ClientBranch).filter_by(business_unit_id=business.id).update({"is_primary": False}) db.add(ClientBranch( tenant_id=tenant_id, client_id=client.id, business_unit_id=business.id, branch_code=branch_code.strip().upper(), branch_name=branch_name.strip(), branch_type=branch_type.strip() or "branch", city=city.strip() or None, state=state.strip() or None, pincode=pincode.strip() or None, is_primary=bool(is_primary), is_active=True, created_by_user_id=user.id, updated_by_user_id=user.id, )) db.commit() return RedirectResponse(f"/clients/{client_id}/business-structure", 303) finally: db.close() @router.post("/{client_id}/registrations") def add_registration(request: Request, client_id: int, business_unit_id: str = Form(""), client_branch_id: str = Form(""), registration_type_id: int = Form(...), registration_number: str = Form(...), legal_name: str = Form(""), trade_name: str = Form(""), state: str = Form(""), csrf_token: str = Form(...)): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user = get_current_user(request, db=db); require_permission(db, user, "clients.edit") tenant_id = _tenant_id(request, user); client = _load_client(db, tenant_id, client_id) if not client: return RedirectResponse("/clients", 303) bu_id = int(business_unit_id) if business_unit_id else None br_id = int(client_branch_id) if client_branch_id else None if bu_id: business = db.get(ClientBusinessUnit, bu_id) if not business or business.client_id != client.id: return RedirectResponse(f"/clients/{client_id}/business-structure", 303) if br_id: branch = db.get(ClientBranch, br_id) if not branch or branch.client_id != client.id: return RedirectResponse(f"/clients/{client_id}/business-structure", 303) bu_id = branch.business_unit_id registration_type = db.get(RegistrationType, registration_type_id) if not registration_type or not registration_type.is_active: return RedirectResponse(f"/clients/{client_id}/business-structure", 303) db.add(ClientRegistration( tenant_id=tenant_id, branch_id=getattr(client, "branch_id", None), client_id=client.id, business_unit_id=bu_id, client_branch_id=br_id, registration_type_id=registration_type.id, registration_number=registration_number.strip().upper(), legal_name=legal_name.strip() or client.client_name, trade_name=trade_name.strip() or None, state=state.strip() or None, jurisdiction=state.strip() or None, status="active", primary_registration=False, responsible_party="firm", auto_create_task=True, created_by_user_id=user.id, )) db.commit() return RedirectResponse(f"/clients/{client_id}/business-structure", 303) finally: db.close() @router.post("/{client_id}/business-units/{business_unit_id}/edit") def edit_business_unit( request: Request, client_id: int, business_unit_id: int, business_code: str = Form(...), business_name: str = Form(...), trade_name: str = Form(""), nature_of_business: str = Form(""), is_primary: str | None = Form(None), csrf_token: str = Form(...), ): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user = get_current_user(request, db=db) require_permission(db, user, "clients.edit") tenant_id = _tenant_id(request, user) client = _load_client(db, tenant_id, client_id) row = _owned_business(db, tenant_id, client_id, business_unit_id) if client else None if not client or not row: return RedirectResponse("/clients", 303) code = business_code.strip().upper() name = business_name.strip() if not code or not name: return _structure_redirect(client_id, error="Business code and business unit name are required.", anchor="business-units") duplicate = db.execute( select(ClientBusinessUnit.id).where( ClientBusinessUnit.tenant_id == tenant_id, ClientBusinessUnit.client_id == client_id, ClientBusinessUnit.business_code == code, ClientBusinessUnit.id != row.id, ) ).scalar_one_or_none() if duplicate: return _structure_redirect(client_id, error=f"Business code {code} is already used for this client.", anchor="business-units") if is_primary: db.query(ClientBusinessUnit).filter( ClientBusinessUnit.tenant_id == tenant_id, ClientBusinessUnit.client_id == client_id, ).update({"is_primary": False}, synchronize_session=False) row.business_code = code row.business_name = name row.trade_name = trade_name.strip() or None row.nature_of_business = nature_of_business.strip() or None row.is_primary = bool(is_primary) row.updated_by_user_id = user.id db.commit() return _structure_redirect(client_id, message="Business Unit updated successfully.", anchor="business-units") finally: db.close() @router.post("/{client_id}/branches/{client_branch_id}/edit") def edit_client_branch( request: Request, client_id: int, client_branch_id: int, business_unit_id: int = Form(...), branch_code: str = Form(...), branch_name: str = Form(...), branch_type: str = Form("branch"), city: str = Form(""), state: str = Form(""), pincode: str = Form(""), is_primary: str | None = Form(None), csrf_token: str = Form(...), ): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user = get_current_user(request, db=db) require_permission(db, user, "clients.edit") tenant_id = _tenant_id(request, user) client = _load_client(db, tenant_id, client_id) row = _owned_branch(db, tenant_id, client_id, client_branch_id) if client else None business = _owned_business(db, tenant_id, client_id, business_unit_id) if client else None if not client or not row or not business: return RedirectResponse("/clients", 303) code = branch_code.strip().upper() name = branch_name.strip() if not code or not name: return _structure_redirect(client_id, error="Branch code and branch name are required.", anchor="client-branches") duplicate = db.execute( select(ClientBranch.id).where( ClientBranch.tenant_id == tenant_id, ClientBranch.business_unit_id == business.id, ClientBranch.branch_code == code, ClientBranch.id != row.id, ) ).scalar_one_or_none() if duplicate: return _structure_redirect(client_id, error=f"Branch code {code} is already used under the selected Business Unit.", anchor="client-branches") if is_primary: db.query(ClientBranch).filter( ClientBranch.tenant_id == tenant_id, ClientBranch.business_unit_id == business.id, ).update({"is_primary": False}, synchronize_session=False) row.business_unit_id = business.id row.branch_code = code row.branch_name = name row.branch_type = branch_type.strip() or "branch" row.city = city.strip() or None row.state = state.strip() or None row.pincode = pincode.strip() or None row.is_primary = bool(is_primary) row.updated_by_user_id = user.id db.commit() return _structure_redirect(client_id, message="Client Branch updated successfully.", anchor="client-branches") finally: db.close() @router.post("/{client_id}/registrations/{registration_id}/edit") def edit_registration( request: Request, client_id: int, registration_id: int, business_unit_id: str = Form(""), client_branch_id: str = Form(""), registration_type_id: int = Form(...), registration_number: str = Form(...), legal_name: str = Form(""), trade_name: str = Form(""), state: str = Form(""), csrf_token: str = Form(...), ): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user = get_current_user(request, db=db) require_permission(db, user, "clients.edit") tenant_id = _tenant_id(request, user) client = _load_client(db, tenant_id, client_id) row = _owned_registration(db, tenant_id, client_id, registration_id) if client else None if not client or not row: return RedirectResponse("/clients", 303) try: bu_id = int(business_unit_id) if business_unit_id else None br_id = int(client_branch_id) if client_branch_id else None except (TypeError, ValueError): return _structure_redirect(client_id, error="Invalid Business Unit or Client Branch selection.", anchor="registrations") business = _owned_business(db, tenant_id, client_id, bu_id) if bu_id else None if bu_id and not business: return _structure_redirect(client_id, error="Selected Business Unit does not belong to this client.", anchor="registrations") branch = _owned_branch(db, tenant_id, client_id, br_id) if br_id else None if br_id and not branch: return _structure_redirect(client_id, error="Selected Client Branch does not belong to this client.", anchor="registrations") if branch: bu_id = branch.business_unit_id registration_type = db.get(RegistrationType, registration_type_id) if not registration_type: return _structure_redirect(client_id, error="Registration Type was not found.", anchor="registrations") if not registration_type.is_active and registration_type.id != row.registration_type_id: return _structure_redirect(client_id, error="An inactive Registration Type cannot be newly selected.", anchor="registrations") number = registration_number.strip().upper() if not number: return _structure_redirect(client_id, error="Registration number is required.", anchor="registrations") duplicate = db.execute( select(ClientRegistration.id).where( ClientRegistration.tenant_id == tenant_id, ClientRegistration.registration_type_id == registration_type.id, ClientRegistration.registration_number == number, ClientRegistration.id != row.id, ) ).scalar_one_or_none() if duplicate: return _structure_redirect( client_id, error=f"{registration_type.code} registration number {number} already exists.", anchor="registrations", ) old_state = row.state row.business_unit_id = bu_id row.client_branch_id = br_id row.registration_type_id = registration_type.id row.registration_number = number row.legal_name = legal_name.strip() or client.client_name row.trade_name = trade_name.strip() or None row.state = state.strip() or None if not row.jurisdiction or row.jurisdiction == old_state: row.jurisdiction = row.state db.commit() return _structure_redirect(client_id, message="Registration updated successfully.", anchor="registrations") finally: db.close() @router.post("/{client_id}/business-structure/{entity}/{entity_id}/toggle") def toggle_scope_record(request: Request, client_id: int, entity: str, entity_id: int, csrf_token: str = Form(...)): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user = get_current_user(request, db=db); require_permission(db, user, "clients.edit") tenant_id = _tenant_id(request, user); client = _load_client(db, tenant_id, client_id) model = {"business": ClientBusinessUnit, "branch": ClientBranch, "registration": ClientRegistration}.get(entity) row = db.get(model, entity_id) if model else None if not client or not row or row.client_id != client.id or row.tenant_id != tenant_id: return RedirectResponse("/clients", 303) if entity == "registration": row.status = "inactive" if row.status in {"active", "valid", "registered"} else "active" else: row.is_active = not bool(row.is_active) row.updated_by_user_id = user.id db.commit() return RedirectResponse(f"/clients/{client_id}/business-structure", 303) finally: db.close()