A R R R Associates
355 lines
17 KiB
Python
355 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.modules.clients.models import Client, ClientBusinessUnit, ClientBranch
|
|
from app.modules.registrations.models import ClientRegistration, RegistrationType
|
|
from app.modules.services.models import ClientServicePlan, ServiceCatalogue
|
|
|
|
VALID_SCOPE_TYPES = {"client", "business_unit", "client_branch", "registration"}
|
|
VALID_REGISTRATION_TYPES = {"GST", "TAN", "PF", "ESI", "PT", "IEC", "FSSAI", "UDYAM", "OTHER"}
|
|
|
|
|
|
@dataclass
|
|
class ScopeTarget:
|
|
token: str
|
|
scope_type: str
|
|
scope_key: str
|
|
client_id: int
|
|
business_unit_id: int | None
|
|
client_branch_id: int | None
|
|
registration_id: int | None
|
|
client_code: str
|
|
client_name: str
|
|
pan: str
|
|
business_unit: str
|
|
client_branch: str
|
|
registration_type: str
|
|
registration_number: str
|
|
trade_name: str
|
|
state: str
|
|
entity_type: str
|
|
partner_id: int | None
|
|
existing_plan_status: str = ""
|
|
|
|
|
|
def normalize_scope_type(value: str | None) -> str:
|
|
value = (value or "client").strip().lower().replace("-", "_").replace(" ", "_")
|
|
return value if value in VALID_SCOPE_TYPES else "client"
|
|
|
|
|
|
def normalize_registration_type(value: str | None) -> str | None:
|
|
value = (value or "").strip().upper()
|
|
return value if value in VALID_REGISTRATION_TYPES else None
|
|
|
|
|
|
def scope_key(scope_type: str, target_id: int) -> str:
|
|
prefixes = {
|
|
"client": "CLIENT",
|
|
"business_unit": "BUSINESS",
|
|
"client_branch": "BRANCH",
|
|
"registration": "REGISTRATION",
|
|
}
|
|
return f"{prefixes[scope_type]}:{int(target_id)}"
|
|
|
|
|
|
def _plan_status_map(db: Session, tenant_id: int, service_catalogue_id: int) -> dict[str, str]:
|
|
rows = db.execute(
|
|
select(ClientServicePlan.scope_key, ClientServicePlan.status).where(
|
|
ClientServicePlan.tenant_id == tenant_id,
|
|
ClientServicePlan.service_catalogue_id == service_catalogue_id,
|
|
)
|
|
).all()
|
|
return {key: status for key, status in rows if key}
|
|
|
|
|
|
def list_scope_targets(
|
|
db: Session,
|
|
*,
|
|
tenant_id: int,
|
|
clients: list[Client],
|
|
catalogue: ServiceCatalogue,
|
|
) -> list[ScopeTarget]:
|
|
client_map = {int(c.id): c for c in clients}
|
|
client_ids = list(client_map)
|
|
if not client_ids:
|
|
return []
|
|
|
|
selected_scope = normalize_scope_type(catalogue.service_scope_type)
|
|
required_registration = normalize_registration_type(catalogue.required_registration_type)
|
|
plan_status = _plan_status_map(db, tenant_id, catalogue.id)
|
|
result: list[ScopeTarget] = []
|
|
|
|
if selected_scope == "client":
|
|
for client in clients:
|
|
key = scope_key("client", client.id)
|
|
result.append(ScopeTarget(
|
|
token=f"client:{client.id}", scope_type="client", scope_key=key,
|
|
client_id=client.id, business_unit_id=None, client_branch_id=None, registration_id=None,
|
|
client_code=client.client_code or "", client_name=client.client_name or "",
|
|
pan=client.pan or "", business_unit="", client_branch="", registration_type="",
|
|
registration_number="", trade_name=client.trade_name or "", state=client.state or "",
|
|
entity_type=client.client_type or "", partner_id=client.partner_id,
|
|
existing_plan_status=plan_status.get(key, ""),
|
|
))
|
|
return result
|
|
|
|
businesses = db.execute(
|
|
select(ClientBusinessUnit).where(
|
|
ClientBusinessUnit.tenant_id == tenant_id,
|
|
ClientBusinessUnit.client_id.in_(client_ids),
|
|
ClientBusinessUnit.is_active.is_(True),
|
|
).order_by(ClientBusinessUnit.client_id, ClientBusinessUnit.is_primary.desc(), ClientBusinessUnit.business_name)
|
|
).scalars().all()
|
|
|
|
if selected_scope == "business_unit":
|
|
actual_clients = set()
|
|
for business in businesses:
|
|
client = client_map[business.client_id]
|
|
actual_clients.add(client.id)
|
|
key = scope_key("business_unit", business.id)
|
|
result.append(ScopeTarget(
|
|
token=f"business:{business.id}", scope_type="business_unit", scope_key=key,
|
|
client_id=client.id, business_unit_id=business.id, client_branch_id=None, registration_id=None,
|
|
client_code=client.client_code or "", client_name=client.client_name or "", pan=client.pan or "",
|
|
business_unit=business.business_name, client_branch="", registration_type="", registration_number="",
|
|
trade_name=business.trade_name or client.trade_name or "", state=client.state or "",
|
|
entity_type=client.client_type or "", partner_id=client.partner_id,
|
|
existing_plan_status=plan_status.get(key, ""),
|
|
))
|
|
for client in clients:
|
|
if client.id not in actual_clients:
|
|
result.append(ScopeTarget(
|
|
token=f"business:auto:{client.id}", scope_type="business_unit", scope_key=f"AUTO_BUSINESS:{client.id}",
|
|
client_id=client.id, business_unit_id=None, client_branch_id=None, registration_id=None,
|
|
client_code=client.client_code or "", client_name=client.client_name or "", pan=client.pan or "",
|
|
business_unit=client.trade_name or client.client_name, client_branch="", registration_type="",
|
|
registration_number="", trade_name=client.trade_name or "", state=client.state or "",
|
|
entity_type=client.client_type or "", partner_id=client.partner_id, existing_plan_status="",
|
|
))
|
|
return result
|
|
|
|
business_map = {b.id: b for b in businesses}
|
|
branches = db.execute(
|
|
select(ClientBranch).where(
|
|
ClientBranch.tenant_id == tenant_id,
|
|
ClientBranch.client_id.in_(client_ids),
|
|
ClientBranch.is_active.is_(True),
|
|
).order_by(ClientBranch.client_id, ClientBranch.is_primary.desc(), ClientBranch.branch_name)
|
|
).scalars().all()
|
|
|
|
if selected_scope == "client_branch":
|
|
for branch in branches:
|
|
client = client_map[branch.client_id]
|
|
business = business_map.get(branch.business_unit_id)
|
|
key = scope_key("client_branch", branch.id)
|
|
result.append(ScopeTarget(
|
|
token=f"branch:{branch.id}", scope_type="client_branch", scope_key=key,
|
|
client_id=client.id, business_unit_id=branch.business_unit_id, client_branch_id=branch.id, registration_id=None,
|
|
client_code=client.client_code or "", client_name=client.client_name or "", pan=client.pan or "",
|
|
business_unit=business.business_name if business else "", client_branch=branch.branch_name,
|
|
registration_type="", registration_number="", trade_name=(business.trade_name if business else "") or client.trade_name or "",
|
|
state=branch.state or client.state or "", entity_type=client.client_type or "", partner_id=client.partner_id,
|
|
existing_plan_status=plan_status.get(key, ""),
|
|
))
|
|
return result
|
|
|
|
branch_map = {b.id: b for b in branches}
|
|
registration_rows = db.execute(
|
|
select(ClientRegistration, RegistrationType).join(
|
|
RegistrationType, RegistrationType.id == ClientRegistration.registration_type_id
|
|
).where(
|
|
ClientRegistration.tenant_id == tenant_id,
|
|
ClientRegistration.client_id.in_(client_ids),
|
|
ClientRegistration.status.in_(("active", "valid", "registered")),
|
|
RegistrationType.is_active.is_(True),
|
|
).order_by(ClientRegistration.client_id, RegistrationType.code, ClientRegistration.registration_number)
|
|
).all()
|
|
|
|
actual_legacy = set()
|
|
for registration, registration_type in registration_rows:
|
|
registration_code = (registration_type.code or "").strip().upper()
|
|
if required_registration and registration_code != required_registration:
|
|
continue
|
|
client = client_map[registration.client_id]
|
|
business = business_map.get(registration.business_unit_id)
|
|
branch = branch_map.get(registration.client_branch_id)
|
|
key = scope_key("registration", registration.id)
|
|
actual_legacy.add((client.id, registration_code))
|
|
result.append(ScopeTarget(
|
|
token=f"registration:{registration.id}", scope_type="registration", scope_key=key,
|
|
client_id=client.id, business_unit_id=registration.business_unit_id,
|
|
client_branch_id=registration.client_branch_id, registration_id=registration.id,
|
|
client_code=client.client_code or "", client_name=client.client_name or "", pan=client.pan or "",
|
|
business_unit=business.business_name if business else "", client_branch=branch.branch_name if branch else "",
|
|
registration_type=registration_code, registration_number=registration.registration_number,
|
|
trade_name=registration.trade_name or (business.trade_name if business else "") or client.trade_name or "",
|
|
state=registration.state or (branch.state if branch else "") or client.state or "",
|
|
entity_type=client.client_type or "", partner_id=client.partner_id,
|
|
existing_plan_status=plan_status.get(key, ""),
|
|
))
|
|
|
|
# Existing legacy GSTIN/TAN values remain usable immediately after migration.
|
|
for client in clients:
|
|
legacy_pairs = []
|
|
if required_registration in (None, "GST") and client.gstin:
|
|
legacy_pairs.append(("GST", client.gstin))
|
|
if required_registration in (None, "TAN") and client.tan:
|
|
legacy_pairs.append(("TAN", client.tan))
|
|
for reg_type, number in legacy_pairs:
|
|
if (client.id, reg_type) in actual_legacy:
|
|
continue
|
|
result.append(ScopeTarget(
|
|
token=f"registration:legacy:{reg_type}:{client.id}", scope_type="registration",
|
|
scope_key=f"LEGACY_{reg_type}:{client.id}", client_id=client.id,
|
|
business_unit_id=None, client_branch_id=None, registration_id=None,
|
|
client_code=client.client_code or "", client_name=client.client_name or "", pan=client.pan or "",
|
|
business_unit=client.trade_name or client.client_name, client_branch="Primary Branch",
|
|
registration_type=reg_type, registration_number=number, trade_name=client.trade_name or "",
|
|
state=client.state or "", entity_type=client.client_type or "", partner_id=client.partner_id,
|
|
existing_plan_status="",
|
|
))
|
|
return result
|
|
|
|
|
|
def _next_code(prefix: str, value: int) -> str:
|
|
return f"{prefix}{int(value):05d}"
|
|
|
|
|
|
def resolve_scope_target(
|
|
db: Session,
|
|
*,
|
|
tenant_id: int,
|
|
clients_by_id: dict[int, Client],
|
|
token: str,
|
|
actor_user_id: int,
|
|
):
|
|
parts = (token or "").split(":")
|
|
if not parts:
|
|
raise ValueError("Invalid subscription scope.")
|
|
|
|
kind = parts[0]
|
|
if kind == "client" and len(parts) == 2:
|
|
client = clients_by_id.get(int(parts[1]))
|
|
if not client:
|
|
raise ValueError("Client is outside the permitted scope.")
|
|
return client, "client", scope_key("client", client.id), None, None, None
|
|
|
|
if kind == "business":
|
|
if len(parts) == 2:
|
|
business = db.get(ClientBusinessUnit, int(parts[1]))
|
|
if not business or business.tenant_id != tenant_id or business.client_id not in clients_by_id:
|
|
raise ValueError("Business Unit is outside the permitted scope.")
|
|
elif len(parts) == 3 and parts[1] == "auto":
|
|
client = clients_by_id.get(int(parts[2]))
|
|
if not client:
|
|
raise ValueError("Client is outside the permitted scope.")
|
|
business = ClientBusinessUnit(
|
|
tenant_id=tenant_id, client_id=client.id,
|
|
business_code=_next_code("BU", client.id),
|
|
business_name=client.trade_name or client.client_name,
|
|
trade_name=client.trade_name, is_primary=True, is_active=True,
|
|
created_by_user_id=actor_user_id, updated_by_user_id=actor_user_id,
|
|
)
|
|
db.add(business); db.flush()
|
|
else:
|
|
raise ValueError("Invalid Business Unit.")
|
|
client = clients_by_id[business.client_id]
|
|
return client, "business_unit", scope_key("business_unit", business.id), business.id, None, None
|
|
|
|
if kind == "branch" and len(parts) == 2:
|
|
branch = db.get(ClientBranch, int(parts[1]))
|
|
if not branch or branch.tenant_id != tenant_id or branch.client_id not in clients_by_id:
|
|
raise ValueError("Client Branch is outside the permitted scope.")
|
|
client = clients_by_id[branch.client_id]
|
|
return client, "client_branch", scope_key("client_branch", branch.id), branch.business_unit_id, branch.id, None
|
|
|
|
if kind == "registration":
|
|
if len(parts) == 2:
|
|
reg = db.get(ClientRegistration, int(parts[1]))
|
|
if not reg or reg.tenant_id != tenant_id or reg.client_id not in clients_by_id:
|
|
raise ValueError("Registration is outside the permitted scope.")
|
|
elif len(parts) == 4 and parts[1] == "legacy":
|
|
reg_type = parts[2].upper()
|
|
client = clients_by_id.get(int(parts[3]))
|
|
if not client:
|
|
raise ValueError("Client is outside the permitted scope.")
|
|
number = client.gstin if reg_type == "GST" else client.tan if reg_type == "TAN" else None
|
|
if not number:
|
|
raise ValueError("Legacy registration is no longer available.")
|
|
business = db.execute(
|
|
select(ClientBusinessUnit).where(
|
|
ClientBusinessUnit.tenant_id == tenant_id,
|
|
ClientBusinessUnit.client_id == client.id,
|
|
ClientBusinessUnit.is_active.is_(True),
|
|
).order_by(ClientBusinessUnit.is_primary.desc(), ClientBusinessUnit.id)
|
|
).scalars().first()
|
|
if not business:
|
|
business = ClientBusinessUnit(
|
|
tenant_id=tenant_id, client_id=client.id,
|
|
business_code=_next_code("BU", client.id),
|
|
business_name=client.trade_name or client.client_name,
|
|
trade_name=client.trade_name, is_primary=True, is_active=True,
|
|
created_by_user_id=actor_user_id, updated_by_user_id=actor_user_id,
|
|
)
|
|
db.add(business); db.flush()
|
|
branch = db.execute(
|
|
select(ClientBranch).where(
|
|
ClientBranch.tenant_id == tenant_id,
|
|
ClientBranch.business_unit_id == business.id,
|
|
ClientBranch.is_active.is_(True),
|
|
).order_by(ClientBranch.is_primary.desc(), ClientBranch.id)
|
|
).scalars().first()
|
|
if not branch:
|
|
branch = ClientBranch(
|
|
tenant_id=tenant_id, client_id=client.id, business_unit_id=business.id,
|
|
branch_code=_next_code("BR", client.id), branch_name="Primary Branch",
|
|
branch_type="head_office", state=client.state, is_primary=True, is_active=True,
|
|
created_by_user_id=actor_user_id, updated_by_user_id=actor_user_id,
|
|
)
|
|
db.add(branch); db.flush()
|
|
registration_type = db.execute(
|
|
select(RegistrationType).where(RegistrationType.code == reg_type)
|
|
).scalar_one_or_none()
|
|
if not registration_type:
|
|
registration_type = RegistrationType(
|
|
code=reg_type,
|
|
name=reg_type,
|
|
category="registration",
|
|
identifier_label=f"{reg_type} Number",
|
|
supports_expiry=False,
|
|
supports_related_person=False,
|
|
is_system=True,
|
|
is_active=True,
|
|
sort_order=0,
|
|
)
|
|
db.add(registration_type)
|
|
db.flush()
|
|
reg = ClientRegistration(
|
|
tenant_id=tenant_id,
|
|
branch_id=getattr(client, "branch_id", None),
|
|
client_id=client.id,
|
|
business_unit_id=business.id,
|
|
client_branch_id=branch.id,
|
|
registration_type_id=registration_type.id,
|
|
registration_number=number.strip().upper(),
|
|
legal_name=client.client_name,
|
|
trade_name=client.trade_name,
|
|
state=client.state,
|
|
jurisdiction=client.state,
|
|
status="active",
|
|
primary_registration=True,
|
|
responsible_party="firm",
|
|
auto_create_task=True,
|
|
created_by_user_id=actor_user_id,
|
|
)
|
|
db.add(reg); db.flush()
|
|
else:
|
|
raise ValueError("Invalid registration.")
|
|
client = clients_by_id[reg.client_id]
|
|
return client, "registration", scope_key("registration", reg.id), reg.business_unit_id, reg.client_branch_id, reg.id
|
|
|
|
raise ValueError("Invalid subscription scope.")
|