A R R R Associates
334 lines
13 KiB
Python
334 lines
13 KiB
Python
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()
|