Add unified hierarchical client management interface
This commit is contained in:
A R R R Associates
@@ -0,0 +1,333 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from math import ceil
|
||||
|
||||
from fastapi import APIRouter, Request
|
||||
from fastapi.responses import RedirectResponse
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.core.db.common import CommonSessionLocal
|
||||
from app.core.security.session_auth import get_current_user
|
||||
from app.core.templating import templates
|
||||
from app.modules.clients.access import (
|
||||
build_scope,
|
||||
can_view_client_row,
|
||||
enforce_partner_scope,
|
||||
is_partner_role,
|
||||
)
|
||||
from app.modules.clients.models import Client, ClientBusinessUnit, ClientBranch
|
||||
from app.modules.clients.service import list_clients_payload
|
||||
from app.modules.core.rbac.deps import get_user_permissions, get_user_roles
|
||||
from app.modules.registrations.models import ClientRegistration, RegistrationType
|
||||
|
||||
router = APIRouter(prefix="/clients", tags=["unified-client-management"])
|
||||
|
||||
|
||||
def _has_perm_factory(db, user):
|
||||
permissions = set(get_user_permissions(db, user.id))
|
||||
return lambda code: code in permissions
|
||||
|
||||
|
||||
def _role_names(db, user) -> set[str]:
|
||||
return {str(role).strip().lower() for role in get_user_roles(db, user.id)}
|
||||
|
||||
|
||||
def _apply_role_scope(scope, user, role_names: set[str]):
|
||||
if "system admin" in role_names:
|
||||
scope.allow_all_clients = True
|
||||
scope.allow_cross_tenant = True
|
||||
scope.allow_cross_branch = True
|
||||
scope.own_only = False
|
||||
scope.locked_partner_id = None
|
||||
return enforce_partner_scope(scope, user=user, role_names=role_names)
|
||||
|
||||
|
||||
def _viewer_partner_id(user, role_names: set[str]):
|
||||
return int(user.id) if is_partner_role(role_names) else None
|
||||
|
||||
|
||||
def _norm(value) -> str:
|
||||
return str(value or "").strip().lower()
|
||||
|
||||
|
||||
def _matches(needle: str, *values) -> bool:
|
||||
if not needle:
|
||||
return True
|
||||
return any(needle in _norm(value) for value in values)
|
||||
|
||||
|
||||
@router.get("/management")
|
||||
def unified_client_management(
|
||||
request: Request,
|
||||
q: str = "",
|
||||
status: str = "",
|
||||
client_type: str = "",
|
||||
registration_type: str = "",
|
||||
registration_status: str = "",
|
||||
page: int = 1,
|
||||
per_page: int = 25,
|
||||
):
|
||||
db = CommonSessionLocal()
|
||||
try:
|
||||
user = get_current_user(request, db=db)
|
||||
if not user:
|
||||
return RedirectResponse("/login", 303)
|
||||
|
||||
permissions = set(get_user_permissions(db, user.id))
|
||||
if "clients.view" not in permissions:
|
||||
return RedirectResponse("/dashboard", 303)
|
||||
|
||||
roles = _role_names(db, user)
|
||||
# This management console is intentionally exposed to the two requested
|
||||
# operational roles. System Admin retains access for support/admin use.
|
||||
if not ({"partner", "firm admin", "system admin"} & roles):
|
||||
return RedirectResponse("/clients", 303)
|
||||
|
||||
has = _has_perm_factory(db, user)
|
||||
scope = _apply_role_scope(build_scope(request, user, has), user, roles)
|
||||
|
||||
# Reuse the production client-list scope engine rather than inventing a
|
||||
# second visibility rule. Fetch a broad scoped set, then enrich it.
|
||||
base = list_clients_payload(
|
||||
db,
|
||||
tenant_id=scope.tenant_id,
|
||||
branch_id=scope.branch_id,
|
||||
allow_cross_branch=scope.allow_cross_branch,
|
||||
allow_all_clients=scope.allow_all_clients,
|
||||
partner_id=None,
|
||||
viewer_partner_id=_viewer_partner_id(user, roles),
|
||||
q="",
|
||||
status="",
|
||||
client_type="",
|
||||
client_group_id=None,
|
||||
include_archived=True,
|
||||
page=1,
|
||||
per_page=10000,
|
||||
sort_by="client_name",
|
||||
sort_order="asc",
|
||||
)
|
||||
visible_rows = [
|
||||
row for row in base.get("rows", [])
|
||||
if can_view_client_row(scope, row, user_id=user.id)
|
||||
]
|
||||
visible_ids = [int(row["id"]) for row in visible_rows if row.get("id") is not None]
|
||||
row_meta = {int(row["id"]): row for row in visible_rows if row.get("id") is not None}
|
||||
|
||||
if not visible_ids:
|
||||
clients = []
|
||||
else:
|
||||
clients = db.execute(
|
||||
select(Client)
|
||||
.where(Client.id.in_(visible_ids))
|
||||
.order_by(Client.client_name.asc(), Client.id.asc())
|
||||
).scalars().all()
|
||||
|
||||
businesses = []
|
||||
branches = []
|
||||
registration_pairs = []
|
||||
if visible_ids:
|
||||
businesses = db.execute(
|
||||
select(ClientBusinessUnit)
|
||||
.where(ClientBusinessUnit.client_id.in_(visible_ids))
|
||||
.order_by(
|
||||
ClientBusinessUnit.client_id,
|
||||
ClientBusinessUnit.is_active.desc(),
|
||||
ClientBusinessUnit.is_primary.desc(),
|
||||
ClientBusinessUnit.business_name.asc(),
|
||||
)
|
||||
).scalars().all()
|
||||
branches = db.execute(
|
||||
select(ClientBranch)
|
||||
.where(ClientBranch.client_id.in_(visible_ids))
|
||||
.order_by(
|
||||
ClientBranch.client_id,
|
||||
ClientBranch.is_active.desc(),
|
||||
ClientBranch.is_primary.desc(),
|
||||
ClientBranch.branch_name.asc(),
|
||||
)
|
||||
).scalars().all()
|
||||
registration_pairs = db.execute(
|
||||
select(ClientRegistration, RegistrationType)
|
||||
.join(RegistrationType, RegistrationType.id == ClientRegistration.registration_type_id)
|
||||
.where(ClientRegistration.client_id.in_(visible_ids))
|
||||
.order_by(
|
||||
ClientRegistration.client_id,
|
||||
RegistrationType.sort_order.asc(),
|
||||
RegistrationType.code.asc(),
|
||||
ClientRegistration.registration_number.asc(),
|
||||
)
|
||||
).all()
|
||||
|
||||
businesses_by_client: dict[int, list] = {}
|
||||
businesses_by_id: dict[int, ClientBusinessUnit] = {}
|
||||
for row in businesses:
|
||||
businesses_by_client.setdefault(row.client_id, []).append(row)
|
||||
businesses_by_id[row.id] = row
|
||||
|
||||
branches_by_client: dict[int, list] = {}
|
||||
branches_by_business: dict[int, list] = {}
|
||||
branches_by_id: dict[int, ClientBranch] = {}
|
||||
for row in branches:
|
||||
branches_by_client.setdefault(row.client_id, []).append(row)
|
||||
branches_by_business.setdefault(row.business_unit_id, []).append(row)
|
||||
branches_by_id[row.id] = row
|
||||
|
||||
regs_by_client: dict[int, list[tuple[ClientRegistration, RegistrationType]]] = {}
|
||||
regs_by_business: dict[int, list[tuple[ClientRegistration, RegistrationType]]] = {}
|
||||
regs_by_branch: dict[int, list[tuple[ClientRegistration, RegistrationType]]] = {}
|
||||
for reg, reg_type in registration_pairs:
|
||||
pair = (reg, reg_type)
|
||||
regs_by_client.setdefault(reg.client_id, []).append(pair)
|
||||
if reg.business_unit_id:
|
||||
regs_by_business.setdefault(reg.business_unit_id, []).append(pair)
|
||||
if reg.client_branch_id:
|
||||
regs_by_branch.setdefault(reg.client_branch_id, []).append(pair)
|
||||
|
||||
needle = _norm(q)
|
||||
reg_type_filter = _norm(registration_type)
|
||||
reg_status_filter = _norm(registration_status)
|
||||
client_status_filter = _norm(status)
|
||||
client_type_filter = _norm(client_type)
|
||||
|
||||
filtered_clients = []
|
||||
for client in clients:
|
||||
if client_status_filter and _norm(client.status) != client_status_filter:
|
||||
continue
|
||||
if client_type_filter and _norm(client.client_type) != client_type_filter:
|
||||
continue
|
||||
|
||||
client_businesses = businesses_by_client.get(client.id, [])
|
||||
client_branches = branches_by_client.get(client.id, [])
|
||||
client_regs = regs_by_client.get(client.id, [])
|
||||
|
||||
if reg_type_filter and not any(_norm(rt.code) == reg_type_filter for _, rt in client_regs):
|
||||
continue
|
||||
if reg_status_filter and not any(_norm(reg.status) == reg_status_filter for reg, _ in client_regs):
|
||||
continue
|
||||
|
||||
if needle:
|
||||
hit = _matches(
|
||||
needle,
|
||||
client.client_name, client.trade_name, client.client_code,
|
||||
client.pan, client.gstin, client.tan, client.cin_llpin,
|
||||
client.mobile, client.email, client.city, client.state,
|
||||
)
|
||||
if not hit:
|
||||
hit = any(_matches(
|
||||
needle, b.business_name, b.trade_name, b.business_code, b.nature_of_business
|
||||
) for b in client_businesses)
|
||||
if not hit:
|
||||
hit = any(_matches(
|
||||
needle, br.branch_name, br.branch_code, br.city, br.state, br.pincode
|
||||
) for br in client_branches)
|
||||
if not hit:
|
||||
hit = any(_matches(
|
||||
needle, reg.registration_number, reg.legal_name, reg.trade_name,
|
||||
reg.state, reg.jurisdiction, rt.code, rt.name
|
||||
) for reg, rt in client_regs)
|
||||
if not hit:
|
||||
continue
|
||||
filtered_clients.append(client)
|
||||
|
||||
per_page = per_page if per_page in {10, 25, 50, 100} else 25
|
||||
total = len(filtered_clients)
|
||||
pages = max(1, ceil(total / per_page))
|
||||
page = max(1, min(int(page or 1), pages))
|
||||
start = (page - 1) * per_page
|
||||
page_clients = filtered_clients[start:start + per_page]
|
||||
|
||||
hierarchy = []
|
||||
for client in page_clients:
|
||||
business_nodes = []
|
||||
assigned_reg_ids: set[int] = set()
|
||||
for business in businesses_by_client.get(client.id, []):
|
||||
branch_nodes = []
|
||||
for branch in branches_by_business.get(business.id, []):
|
||||
branch_regs = regs_by_branch.get(branch.id, [])
|
||||
assigned_reg_ids.update(reg.id for reg, _ in branch_regs)
|
||||
branch_nodes.append({
|
||||
"row": branch,
|
||||
"registrations": branch_regs,
|
||||
})
|
||||
direct_business_regs = [
|
||||
pair for pair in regs_by_business.get(business.id, [])
|
||||
if not pair[0].client_branch_id
|
||||
]
|
||||
assigned_reg_ids.update(reg.id for reg, _ in direct_business_regs)
|
||||
business_nodes.append({
|
||||
"row": business,
|
||||
"branches": branch_nodes,
|
||||
"registrations": direct_business_regs,
|
||||
})
|
||||
|
||||
unassigned_regs = [
|
||||
pair for pair in regs_by_client.get(client.id, [])
|
||||
if pair[0].id not in assigned_reg_ids
|
||||
]
|
||||
hierarchy.append({
|
||||
"client": client,
|
||||
"meta": row_meta.get(client.id, {}),
|
||||
"businesses": business_nodes,
|
||||
"unassigned_registrations": unassigned_regs,
|
||||
"business_count": len(businesses_by_client.get(client.id, [])),
|
||||
"branch_count": len(branches_by_client.get(client.id, [])),
|
||||
"registration_count": len(regs_by_client.get(client.id, [])),
|
||||
})
|
||||
|
||||
all_visible_client_ids = {c.id for c in filtered_clients}
|
||||
filtered_businesses = [b for b in businesses if b.client_id in all_visible_client_ids]
|
||||
filtered_branches = [b for b in branches if b.client_id in all_visible_client_ids]
|
||||
filtered_regs = [(r, t) for r, t in registration_pairs if r.client_id in all_visible_client_ids]
|
||||
|
||||
counters = {
|
||||
"clients": len(filtered_clients),
|
||||
"businesses": len(filtered_businesses),
|
||||
"branches": len(filtered_branches),
|
||||
"registrations": len(filtered_regs),
|
||||
"gstin": sum(1 for _, rt in filtered_regs if _norm(rt.code) == "gstin"),
|
||||
"tan": sum(1 for _, rt in filtered_regs if _norm(rt.code) == "tan"),
|
||||
"active_registrations": sum(
|
||||
1 for reg, _ in filtered_regs
|
||||
if _norm(reg.status) in {"active", "valid", "registered"}
|
||||
),
|
||||
}
|
||||
|
||||
registration_types = db.execute(
|
||||
select(RegistrationType)
|
||||
.where(RegistrationType.is_active.is_(True))
|
||||
.order_by(RegistrationType.sort_order.asc(), RegistrationType.name.asc())
|
||||
).scalars().all()
|
||||
|
||||
client_types = sorted({str(c.client_type) for c in clients if c.client_type})
|
||||
registration_statuses = sorted({
|
||||
str(reg.status) for reg, _ in registration_pairs if reg.status
|
||||
})
|
||||
|
||||
return templates.TemplateResponse(
|
||||
"modules/clients/templates/clients/management.html",
|
||||
{
|
||||
"request": request,
|
||||
"current_user": user,
|
||||
"current_user_roles": get_user_roles(db, user.id),
|
||||
"current_user_permissions": list(permissions),
|
||||
"title": "Client Management",
|
||||
"hierarchy": hierarchy,
|
||||
"counters": counters,
|
||||
"q": q,
|
||||
"status": status,
|
||||
"client_type": client_type,
|
||||
"registration_type": registration_type,
|
||||
"registration_status": registration_status,
|
||||
"registration_types": registration_types,
|
||||
"registration_statuses": registration_statuses,
|
||||
"client_types": client_types,
|
||||
"page": page,
|
||||
"pages": pages,
|
||||
"per_page": per_page,
|
||||
"total": total,
|
||||
"can_create": "clients.create" in permissions,
|
||||
"can_edit": "clients.edit" in permissions,
|
||||
},
|
||||
)
|
||||
finally:
|
||||
db.close()
|
||||
Reference in New Issue
Block a user