A R R R Associates
428 lines
16 KiB
Python
428 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
from math import ceil
|
|
|
|
from sqlalchemy import asc, case, desc, func, or_, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.security.passwords import hash_password
|
|
from app.modules.clients.association_models import ClientAssociation
|
|
from app.modules.clients.constants import CLIENT_SORT_FIELDS
|
|
from app.modules.clients.models import Client, ClientAuditLog
|
|
from app.modules.client_groups.models import ClientGroup
|
|
from app.modules.core.iam.models import User
|
|
from app.modules.core.rbac.models import Role, UserRole
|
|
from app.modules.core.tenancy.models import Branch, Tenant
|
|
|
|
|
|
def _safe_sort(sort_by: str, sort_order: str):
|
|
attr_name = CLIENT_SORT_FIELDS.get(sort_by, "client_name")
|
|
column = getattr(Client, attr_name)
|
|
return desc(column) if sort_order == "desc" else asc(column)
|
|
|
|
|
|
def build_clients_query(
|
|
*,
|
|
tenant_id: int,
|
|
branch_id: int | None = None,
|
|
allow_cross_branch: bool = False,
|
|
allow_all_clients: bool = False,
|
|
partner_id: int | None = None,
|
|
q: str = "",
|
|
status: str = "",
|
|
client_type: str = "",
|
|
client_group_id: int | None = None,
|
|
include_archived: bool = False,
|
|
):
|
|
assoc = ClientAssociation
|
|
stmt = (
|
|
select(
|
|
Client,
|
|
User.full_name.label("partner_name"),
|
|
Branch.name.label("branch_name"),
|
|
Tenant.name.label("tenant_name"),
|
|
ClientGroup.group_name.label("client_group_name"),
|
|
ClientGroup.group_code.label("client_group_code"),
|
|
assoc.association_type.label("association_type"),
|
|
assoc.firm_tenant_id.label("assoc_firm_tenant_id"),
|
|
assoc.consultant_id.label("assoc_consultant_id"),
|
|
assoc.partner_user_id.label("assoc_partner_user_id"),
|
|
assoc.created_source.label("assoc_created_source"),
|
|
)
|
|
.outerjoin(assoc, assoc.client_id == Client.id)
|
|
.outerjoin(ClientGroup, ClientGroup.id == Client.client_group_id)
|
|
.join(User, User.id == Client.partner_id, isouter=True)
|
|
.join(Branch, Branch.id == Client.branch_id, isouter=True)
|
|
.join(Tenant, Tenant.id == Client.tenant_id, isouter=True)
|
|
)
|
|
if not allow_all_clients:
|
|
stmt = stmt.where(Client.tenant_id == tenant_id)
|
|
if not include_archived:
|
|
stmt = stmt.where(Client.is_archived.is_(False))
|
|
if branch_id and not allow_all_clients and not allow_cross_branch:
|
|
stmt = stmt.where(Client.branch_id == branch_id)
|
|
if partner_id:
|
|
stmt = stmt.where((Client.partner_id == partner_id) | (assoc.partner_user_id == partner_id))
|
|
if status:
|
|
stmt = stmt.where(Client.status == status)
|
|
if client_type:
|
|
stmt = stmt.where(Client.client_type == client_type)
|
|
if client_group_id:
|
|
stmt = stmt.where(Client.client_group_id == client_group_id)
|
|
if q:
|
|
like = f"%{q.strip()}%"
|
|
stmt = stmt.where(or_(
|
|
Client.client_code.ilike(like), Client.client_name.ilike(like), Client.trade_name.ilike(like),
|
|
Client.pan.ilike(like), Client.gstin.ilike(like), Client.mobile.ilike(like), Client.email.ilike(like),
|
|
ClientGroup.group_name.ilike(like), ClientGroup.group_code.ilike(like),
|
|
))
|
|
return stmt
|
|
|
|
|
|
def list_clients(
|
|
db: Session, *, tenant_id: int, branch_id: int | None = None,
|
|
allow_cross_branch: bool = False, allow_all_clients: bool = False,
|
|
partner_id: int | None = None, q: str = "", status: str = "", client_type: str = "",
|
|
client_group_id: int | None = None, include_archived: bool = False,
|
|
page: int = 1, per_page: int = 25, sort_by: str = "client_name", sort_order: str = "asc",
|
|
) -> dict:
|
|
stmt = build_clients_query(
|
|
tenant_id=tenant_id, branch_id=branch_id, allow_cross_branch=allow_cross_branch,
|
|
allow_all_clients=allow_all_clients, partner_id=partner_id, q=q, status=status,
|
|
client_type=client_type, client_group_id=client_group_id, include_archived=include_archived,
|
|
)
|
|
total = db.execute(select(func.count()).select_from(stmt.subquery())).scalar_one()
|
|
per_page = min(max(int(per_page or 25), 1), 100)
|
|
pages = max(ceil(total / per_page), 1)
|
|
page = min(max(int(page or 1), 1), pages)
|
|
offset = (page - 1) * per_page
|
|
result = db.execute(
|
|
stmt.order_by(_safe_sort(sort_by, sort_order)).offset(offset).limit(per_page)
|
|
).all()
|
|
rows=[]
|
|
for client, partner_name, branch_name, tenant_name, client_group_name, client_group_code, association_type, assoc_firm_tenant_id, assoc_consultant_id, assoc_partner_user_id, assoc_created_source in result:
|
|
row={**client.__dict__}; row.pop("_sa_instance_state",None)
|
|
row.update({
|
|
"partner_name":partner_name,"branch_name":branch_name,"tenant_name":tenant_name,
|
|
"client_group_name":client_group_name,"client_group_code":client_group_code,
|
|
"association_type":association_type,"assoc_firm_tenant_id":assoc_firm_tenant_id,
|
|
"assoc_consultant_id":assoc_consultant_id,"assoc_partner_user_id":assoc_partner_user_id,
|
|
"assoc_created_source":assoc_created_source,"effective_partner_id":assoc_partner_user_id or row.get("partner_id"),
|
|
}); rows.append(row)
|
|
stats_stmt=select(func.count(Client.id),func.sum(case((Client.status=="active",1),else_=0)),func.sum(case((Client.status=="inactive",1),else_=0)),func.sum(case((Client.status=="archived",1),else_=0)))
|
|
if not allow_all_clients:
|
|
stats_stmt=stats_stmt.where(Client.tenant_id==tenant_id)
|
|
if branch_id and not allow_cross_branch: stats_stmt=stats_stmt.where(Client.branch_id==branch_id)
|
|
if partner_id: stats_stmt=stats_stmt.where(Client.partner_id==partner_id)
|
|
total_all,active,inactive,archived=db.execute(stats_stmt).one()
|
|
return {"rows":rows,"meta":{"total":total,"page":page,"per_page":per_page,"pages":pages},"stats":{"total":int(total_all or 0),"active":int(active or 0),"inactive":int(inactive or 0),"archived":int(archived or 0)}}
|
|
|
|
|
|
def get_client_detail_payload(db: Session, client_id: int):
|
|
assoc=ClientAssociation
|
|
stmt=(select(
|
|
Client, ClientGroup.group_name.label("client_group_name"), ClientGroup.group_code.label("client_group_code"),
|
|
assoc.association_type.label("association_type"), assoc.firm_tenant_id.label("assoc_firm_tenant_id"),
|
|
assoc.consultant_id.label("assoc_consultant_id"), assoc.partner_user_id.label("assoc_partner_user_id"),
|
|
assoc.created_source.label("assoc_created_source"),
|
|
).outerjoin(assoc,assoc.client_id==Client.id).outerjoin(ClientGroup,ClientGroup.id==Client.client_group_id).where(Client.id==client_id))
|
|
result=db.execute(stmt).one_or_none()
|
|
if not result: return None
|
|
client,client_group_name,client_group_code,association_type,assoc_firm_tenant_id,assoc_consultant_id,assoc_partner_user_id,assoc_created_source=result
|
|
row={**client.__dict__}; row.pop("_sa_instance_state",None)
|
|
row.update({"client_group_name":client_group_name,"client_group_code":client_group_code,"association_type":association_type,
|
|
"assoc_firm_tenant_id":assoc_firm_tenant_id,"assoc_consultant_id":assoc_consultant_id,
|
|
"assoc_partner_user_id":assoc_partner_user_id,"assoc_created_source":assoc_created_source,
|
|
"effective_partner_id":assoc_partner_user_id or row.get("partner_id")})
|
|
return row
|
|
|
|
|
|
def get_client_by_id(db: Session, client_id: int):
|
|
return db.get(Client, client_id)
|
|
|
|
|
|
def get_client_by_code(db: Session, *, tenant_id: int, client_code: str):
|
|
return db.execute(
|
|
select(Client).where(Client.tenant_id == tenant_id, Client.client_code == client_code)
|
|
).scalar_one_or_none()
|
|
|
|
|
|
def get_client_by_pan(db: Session, *, tenant_id: int, pan: str):
|
|
return db.execute(
|
|
select(Client).where(Client.tenant_id == tenant_id, Client.pan == pan)
|
|
).scalar_one_or_none()
|
|
|
|
|
|
def get_client_by_gstin(db: Session, *, tenant_id: int, gstin: str):
|
|
return db.execute(
|
|
select(Client).where(Client.tenant_id == tenant_id, Client.gstin == gstin)
|
|
).scalar_one_or_none()
|
|
|
|
|
|
def create_client(db: Session, payload: dict):
|
|
row = Client(**payload)
|
|
db.add(row)
|
|
db.commit()
|
|
db.refresh(row)
|
|
return row
|
|
|
|
|
|
def update_client(db: Session, row: Client, payload: dict):
|
|
for key, value in payload.items():
|
|
setattr(row, key, value)
|
|
db.add(row)
|
|
db.commit()
|
|
db.refresh(row)
|
|
return row
|
|
|
|
|
|
def write_audit_log(db: Session, **kwargs):
|
|
row = ClientAuditLog(**kwargs)
|
|
db.add(row)
|
|
db.commit()
|
|
db.refresh(row)
|
|
return row
|
|
|
|
|
|
def list_audit_logs(db: Session, *, client_id: int, limit: int = 50):
|
|
stmt = (
|
|
select(ClientAuditLog)
|
|
.where(ClientAuditLog.client_id == client_id)
|
|
.order_by(ClientAuditLog.created_at_utc.desc())
|
|
.limit(limit)
|
|
)
|
|
return db.execute(stmt).scalars().all()
|
|
|
|
|
|
def list_tenants(db: Session):
|
|
return db.execute(
|
|
select(Tenant).where(Tenant.is_active.is_(True)).order_by(Tenant.name.asc())
|
|
).scalars().all()
|
|
|
|
|
|
def list_branches_for_tenant(db: Session, tenant_id: int):
|
|
stmt = (
|
|
select(Branch)
|
|
.where(Branch.tenant_id == tenant_id, Branch.is_active.is_(True))
|
|
.order_by(Branch.name.asc())
|
|
)
|
|
return db.execute(stmt).scalars().all()
|
|
|
|
|
|
def list_partners_for_scope(db: Session, *, tenant_id: int, branch_id: int | None = None):
|
|
stmt = (
|
|
select(User)
|
|
.join(UserRole, UserRole.user_id == User.id)
|
|
.join(Role, Role.id == UserRole.role_id)
|
|
.where(
|
|
Role.name == "Partner",
|
|
User.tenant_id == tenant_id,
|
|
User.is_active.is_(True),
|
|
User.deleted_at.is_(None),
|
|
)
|
|
.order_by(User.full_name.asc(), User.email.asc())
|
|
)
|
|
if branch_id:
|
|
stmt = stmt.where(User.branch_id == branch_id)
|
|
return db.execute(stmt).scalars().all()
|
|
|
|
|
|
def get_branch(db: Session, branch_id: int):
|
|
return db.execute(
|
|
select(Branch).where(Branch.id == branch_id, Branch.is_active.is_(True))
|
|
).scalar_one_or_none()
|
|
|
|
|
|
def get_partner(db: Session, partner_id: int):
|
|
return db.execute(
|
|
select(User).where(User.id == partner_id, User.is_active.is_(True), User.deleted_at.is_(None))
|
|
).scalar_one_or_none()
|
|
|
|
def list_all_branches(db: Session):
|
|
stmt = (
|
|
select(
|
|
Branch,
|
|
Tenant.name.label("tenant_name"),
|
|
)
|
|
.join(Tenant, Tenant.id == Branch.tenant_id)
|
|
.where(Branch.is_active.is_(True))
|
|
.order_by(Tenant.name.asc(), Branch.name.asc())
|
|
)
|
|
|
|
rows = []
|
|
for branch, tenant_name in db.execute(stmt).all():
|
|
branch.tenant_name = tenant_name
|
|
rows.append(branch)
|
|
return rows
|
|
|
|
def list_all_partners(db: Session):
|
|
stmt = (
|
|
select(
|
|
User,
|
|
Tenant.name.label("tenant_name"),
|
|
)
|
|
.join(UserRole, UserRole.user_id == User.id)
|
|
.join(Role, Role.id == UserRole.role_id)
|
|
.join(Tenant, Tenant.id == User.tenant_id, isouter=True)
|
|
.where(
|
|
Role.name == "Partner",
|
|
User.is_active.is_(True),
|
|
User.deleted_at.is_(None),
|
|
)
|
|
.order_by(Tenant.name.asc(), User.full_name.asc(), User.email.asc())
|
|
)
|
|
|
|
rows = []
|
|
for user, tenant_name in db.execute(stmt).all():
|
|
user.tenant_name = tenant_name
|
|
rows.append(user)
|
|
return rows
|
|
|
|
def list_all_branches(db: Session):
|
|
stmt = (
|
|
select(
|
|
Branch,
|
|
Tenant.name.label("tenant_name"),
|
|
)
|
|
.join(Tenant, Tenant.id == Branch.tenant_id)
|
|
.where(Branch.is_active.is_(True))
|
|
.order_by(Tenant.name.asc(), Branch.name.asc())
|
|
)
|
|
|
|
rows = []
|
|
for branch, tenant_name in db.execute(stmt).all():
|
|
branch.tenant_name = tenant_name
|
|
rows.append(branch)
|
|
return rows
|
|
|
|
|
|
def list_all_partners(db: Session):
|
|
stmt = (
|
|
select(
|
|
User,
|
|
Tenant.name.label("tenant_name"),
|
|
)
|
|
.join(UserRole, UserRole.user_id == User.id)
|
|
.join(Role, Role.id == UserRole.role_id)
|
|
.join(Tenant, Tenant.id == User.tenant_id, isouter=True)
|
|
.where(
|
|
Role.name == "Partner",
|
|
User.is_active.is_(True),
|
|
User.deleted_at.is_(None),
|
|
)
|
|
.order_by(Tenant.name.asc(), User.full_name.asc(), User.email.asc())
|
|
)
|
|
|
|
rows = []
|
|
for user, tenant_name in db.execute(stmt).all():
|
|
user.tenant_name = tenant_name
|
|
rows.append(user)
|
|
return rows
|
|
|
|
|
|
def get_portal_client_for_user(db: Session, *, user: User):
|
|
email = (getattr(user, "email", "") or "").strip().lower()
|
|
tenant_id = getattr(user, "tenant_id", None)
|
|
if not email or not tenant_id:
|
|
return None
|
|
|
|
stmt = (
|
|
select(
|
|
Client,
|
|
User.full_name.label("partner_name"),
|
|
Branch.name.label("branch_name"),
|
|
Tenant.name.label("tenant_name"),
|
|
)
|
|
.join(User, User.id == Client.partner_id, isouter=True)
|
|
.join(Branch, Branch.id == Client.branch_id, isouter=True)
|
|
.join(Tenant, Tenant.id == Client.tenant_id, isouter=True)
|
|
.where(
|
|
Client.tenant_id == tenant_id,
|
|
Client.is_archived.is_(False),
|
|
or_(Client.email.ilike(email), Client.alternate_email.ilike(email)),
|
|
)
|
|
.order_by(
|
|
case((Client.status == "active", 0), else_=1),
|
|
Client.client_name.asc(),
|
|
Client.id.asc(),
|
|
)
|
|
)
|
|
result = db.execute(stmt).first()
|
|
if not result:
|
|
return None
|
|
|
|
client, partner_name, branch_name, tenant_name = result
|
|
row = {**client.__dict__}
|
|
row.pop("_sa_instance_state", None)
|
|
row.update(
|
|
{
|
|
"partner_name": partner_name,
|
|
"branch_name": branch_name,
|
|
"tenant_name": tenant_name,
|
|
}
|
|
)
|
|
return row
|
|
|
|
|
|
def get_user_by_email(db: Session, *, email: str, exclude_user_id: int | None = None):
|
|
email_clean = (email or "").strip().lower()
|
|
if not email_clean:
|
|
return None
|
|
stmt = select(User).where(User.email.ilike(email_clean), User.deleted_at.is_(None))
|
|
if exclude_user_id:
|
|
stmt = stmt.where(User.id != exclude_user_id)
|
|
return db.execute(stmt).scalar_one_or_none()
|
|
|
|
|
|
def get_tenant(db: Session, tenant_id: int):
|
|
return db.execute(select(Tenant).where(Tenant.id == tenant_id, Tenant.is_active.is_(True))).scalar_one_or_none()
|
|
|
|
|
|
def get_partner_for_tenant(db: Session, *, partner_user_id: int, tenant_id: int):
|
|
stmt = (
|
|
select(User)
|
|
.join(UserRole, UserRole.user_id == User.id)
|
|
.join(Role, Role.id == UserRole.role_id)
|
|
.where(
|
|
User.id == partner_user_id,
|
|
User.tenant_id == tenant_id,
|
|
User.is_active.is_(True),
|
|
User.deleted_at.is_(None),
|
|
Role.name == "Partner",
|
|
)
|
|
)
|
|
return db.execute(stmt).scalar_one_or_none()
|
|
|
|
|
|
def get_role_by_name(db: Session, role_name: str):
|
|
return db.execute(select(Role).where(Role.name == role_name, Role.is_active.is_(True))).scalar_one_or_none()
|
|
|
|
|
|
def create_portal_user(db: Session, *, email: str, full_name: str, tenant_id: int, branch_id: int, password: str):
|
|
row = User(
|
|
email=(email or '').strip().lower(),
|
|
full_name=(full_name or '').strip(),
|
|
password_hash=hash_password(password),
|
|
tenant_id=tenant_id,
|
|
branch_id=branch_id,
|
|
is_active=True,
|
|
allow_login=True,
|
|
is_locked=False,
|
|
must_change_password=False,
|
|
)
|
|
db.add(row)
|
|
db.commit()
|
|
db.refresh(row)
|
|
return row
|
|
|
|
|
|
def ensure_user_role(db: Session, *, user_id: int, role_id: int):
|
|
existing = db.execute(select(UserRole).where(UserRole.user_id == user_id, UserRole.role_id == role_id)).scalar_one_or_none()
|
|
if existing:
|
|
return existing
|
|
row = UserRole(user_id=user_id, role_id=role_id)
|
|
db.add(row)
|
|
db.commit()
|
|
db.refresh(row)
|
|
return row
|