A R R R Associates
256 lines
13 KiB
Python
256 lines
13 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.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 _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).where(RegistrationType.is_active.is_(True)).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-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()
|