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