A R R R Associates
229 lines
9.1 KiB
Python
229 lines
9.1 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from decimal import Decimal, ROUND_HALF_UP
|
|
|
|
from sqlalchemy import func, or_, select
|
|
|
|
from app.modules.clients.models import Client
|
|
from app.modules.consultants.models import ConsultantProfile
|
|
from app.modules.core.iam.models import User
|
|
from app.modules.core.tenancy.models import Branch, Tenant
|
|
from app.modules.marketplace.models import MarketplaceLead, MarketplaceLeadAssignment
|
|
|
|
LEAD_STATUSES = ["NEW", "CONTACTED", "QUALIFIED", "ASSIGNED", "ACCEPTED", "REJECTED", "CONVERTED", "LOST"]
|
|
LEAD_PRIORITIES = ["LOW", "NORMAL", "HIGH", "URGENT"]
|
|
LEAD_SOURCES = ["manual", "public_website", "consultant", "referral", "campaign", "other"]
|
|
SERVICE_CATEGORIES = ["GST", "Income Tax", "ROC", "Audit", "Accounting", "Payroll", "Registration", "Advisory", "Other"]
|
|
|
|
|
|
def money(value) -> Decimal:
|
|
if value in (None, ""):
|
|
return Decimal("0.00")
|
|
return Decimal(str(value)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
|
|
|
|
|
|
def _role_set(role_names: list[str] | None) -> set[str]:
|
|
return set(role_names or [])
|
|
|
|
|
|
def is_system_admin(role_names: list[str] | None) -> bool:
|
|
return "System Admin" in _role_set(role_names)
|
|
|
|
|
|
def is_firm_admin(role_names: list[str] | None) -> bool:
|
|
return "Firm Admin" in _role_set(role_names)
|
|
|
|
|
|
def is_partner(role_names: list[str] | None) -> bool:
|
|
return "Partner" in _role_set(role_names)
|
|
|
|
|
|
def generate_lead_no(db) -> str:
|
|
year = datetime.now().year
|
|
count = db.execute(select(func.count(MarketplaceLead.id))).scalar_one() or 0
|
|
return f"ML-{year}-{count + 1:05d}"
|
|
|
|
|
|
def list_reference_audit_firms(db) -> list[Tenant]:
|
|
return list(db.execute(select(Tenant).order_by(Tenant.name.asc())).scalars().all())
|
|
|
|
|
|
def list_reference_branches(db, tenant_id: int | None = None) -> list[Branch]:
|
|
stmt = select(Branch).order_by(Branch.name.asc())
|
|
if tenant_id:
|
|
stmt = stmt.where(Branch.tenant_id == tenant_id)
|
|
return list(db.execute(stmt).scalars().all())
|
|
|
|
|
|
def list_reference_partners(db, tenant_id: int | None = None) -> list[User]:
|
|
stmt = select(User).order_by(User.full_name.asc(), User.email.asc())
|
|
if tenant_id:
|
|
stmt = stmt.where(User.tenant_id == tenant_id)
|
|
return list(db.execute(stmt).scalars().all())
|
|
|
|
|
|
def list_marketplace_leads(db, *, q: str = "", status: str = "", user=None, role_names: list[str] | None = None) -> list[MarketplaceLead]:
|
|
stmt = select(MarketplaceLead).order_by(MarketplaceLead.created_at_utc.desc(), MarketplaceLead.id.desc())
|
|
roles = _role_set(role_names)
|
|
if not is_system_admin(role_names):
|
|
tenant_id = getattr(user, "tenant_id", None)
|
|
if tenant_id:
|
|
stmt = stmt.where(MarketplaceLead.assigned_tenant_id == tenant_id)
|
|
if is_partner(role_names):
|
|
stmt = stmt.where(MarketplaceLead.assigned_partner_user_id == getattr(user, "id", None))
|
|
if status:
|
|
stmt = stmt.where(MarketplaceLead.status == status)
|
|
if q:
|
|
like = f"%{q}%"
|
|
stmt = stmt.where(or_(MarketplaceLead.lead_no.ilike(like), MarketplaceLead.lead_name.ilike(like), MarketplaceLead.business_name.ilike(like), MarketplaceLead.mobile.ilike(like), MarketplaceLead.email.ilike(like), MarketplaceLead.service_requested.ilike(like)))
|
|
return list(db.execute(stmt).scalars().all())
|
|
|
|
|
|
def get_marketplace_lead(db, lead_id: int) -> MarketplaceLead | None:
|
|
return db.get(MarketplaceLead, int(lead_id))
|
|
|
|
|
|
def create_marketplace_lead(db, *, lead_name: str, business_name: str | None, email: str | None, mobile: str | None, city: str | None, state: str | None, service_category: str | None, service_requested: str, message: str | None, source: str = "manual", priority: str = "NORMAL", estimated_value=0, created_by_user_id: int | None = None) -> MarketplaceLead:
|
|
lead = MarketplaceLead(
|
|
lead_no=generate_lead_no(db),
|
|
source=source or "manual",
|
|
service_category=service_category or None,
|
|
service_requested=(service_requested or "General enquiry").strip(),
|
|
lead_name=lead_name.strip(),
|
|
business_name=(business_name or None),
|
|
email=(email or None),
|
|
mobile=(mobile or None),
|
|
city=(city or None),
|
|
state=(state or None),
|
|
message=(message or None),
|
|
priority=priority if priority in LEAD_PRIORITIES else "NORMAL",
|
|
estimated_value=money(estimated_value),
|
|
created_by_user_id=created_by_user_id,
|
|
)
|
|
db.add(lead)
|
|
db.commit()
|
|
db.refresh(lead)
|
|
return lead
|
|
|
|
|
|
def assign_marketplace_lead(db, *, lead: MarketplaceLead, tenant_id: int, branch_id: int | None, partner_user_id: int | None, notes: str | None, assigned_by_user_id: int | None) -> MarketplaceLead:
|
|
now = datetime.now(timezone.utc)
|
|
lead.assigned_tenant_id = tenant_id
|
|
lead.assigned_branch_id = branch_id or None
|
|
lead.assigned_partner_user_id = partner_user_id or None
|
|
lead.assigned_by_user_id = assigned_by_user_id
|
|
lead.assigned_at_utc = now
|
|
lead.status = "ASSIGNED"
|
|
lead.updated_by_user_id = assigned_by_user_id
|
|
db.add(MarketplaceLeadAssignment(lead_id=lead.id, tenant_id=tenant_id, branch_id=branch_id or None, partner_user_id=partner_user_id or None, notes=notes or None, assigned_by_user_id=assigned_by_user_id, assigned_at_utc=now))
|
|
db.commit()
|
|
db.refresh(lead)
|
|
return lead
|
|
|
|
|
|
def update_lead_status(db, *, lead: MarketplaceLead, status: str, user_id: int | None) -> MarketplaceLead:
|
|
if status not in LEAD_STATUSES:
|
|
raise ValueError("Invalid lead status")
|
|
lead.status = status
|
|
lead.updated_by_user_id = user_id
|
|
db.commit()
|
|
db.refresh(lead)
|
|
return lead
|
|
|
|
|
|
def convert_lead_to_client(db, *, lead: MarketplaceLead, tenant_id: int, branch_id: int, partner_user_id: int | None, client_code: str | None, user_id: int | None) -> Client:
|
|
if lead.converted_client_id:
|
|
existing = db.get(Client, lead.converted_client_id)
|
|
if existing:
|
|
return existing
|
|
code = (client_code or f"LEAD-{lead.id:05d}").strip().upper()
|
|
name = (lead.business_name or lead.lead_name).strip()
|
|
client = Client(
|
|
tenant_id=tenant_id,
|
|
branch_id=branch_id,
|
|
partner_id=partner_user_id or None,
|
|
engagement_mode="internal_managed",
|
|
client_code=code,
|
|
client_name=name,
|
|
trade_name=lead.business_name or None,
|
|
client_type="Other",
|
|
contact_person_name=lead.lead_name,
|
|
mobile=lead.mobile,
|
|
email=lead.email,
|
|
city=lead.city,
|
|
state=lead.state,
|
|
status="active",
|
|
is_active=True,
|
|
is_archived=False,
|
|
notes=f"Converted from marketplace lead {lead.lead_no}.\n\n{lead.message or ''}".strip(),
|
|
created_by_user_id=user_id,
|
|
updated_by_user_id=user_id,
|
|
)
|
|
db.add(client)
|
|
db.flush()
|
|
lead.converted_client_id = client.id
|
|
lead.converted_by_user_id = user_id
|
|
lead.converted_at_utc = datetime.now(timezone.utc)
|
|
lead.status = "CONVERTED"
|
|
lead.updated_by_user_id = user_id
|
|
db.commit()
|
|
db.refresh(client)
|
|
return client
|
|
|
|
|
|
def get_marketplace_overview_counts(db) -> dict[str, int]:
|
|
"""Return lightweight public marketplace counts.
|
|
|
|
This is intentionally read-only and tolerant of partially configured data so the
|
|
public marketplace page never breaks the ERP login/runtime flow.
|
|
"""
|
|
try:
|
|
firm_count = db.execute(select(func.count(Tenant.id)).where(Tenant.is_active.is_(True))).scalar_one() or 0
|
|
except Exception:
|
|
firm_count = 0
|
|
try:
|
|
consultant_count = db.execute(select(func.count(ConsultantProfile.id)).where(ConsultantProfile.is_active.is_(True))).scalar_one() or 0
|
|
except Exception:
|
|
consultant_count = 0
|
|
try:
|
|
lead_count = db.execute(select(func.count(MarketplaceLead.id))).scalar_one() or 0
|
|
except Exception:
|
|
lead_count = 0
|
|
return {"audit_firms": int(firm_count), "consultants": int(consultant_count), "service_requests": int(lead_count)}
|
|
|
|
|
|
def list_public_marketplace_audit_firms(db, limit: int = 12) -> list[Tenant]:
|
|
"""List active audit firms for public marketplace display.
|
|
|
|
More advanced publication controls can be added later. For Phase 7T.4 this uses
|
|
only active tenants and does not expose private firm records beyond basic name/type.
|
|
"""
|
|
try:
|
|
stmt = select(Tenant).where(Tenant.is_active.is_(True)).order_by(Tenant.name.asc()).limit(int(limit or 12))
|
|
return list(db.execute(stmt).scalars().all())
|
|
except Exception:
|
|
return []
|
|
|
|
|
|
def list_public_marketplace_consultants(db, limit: int = 12) -> list[ConsultantProfile]:
|
|
"""List active consultants for public marketplace display."""
|
|
try:
|
|
stmt = (
|
|
select(ConsultantProfile)
|
|
.where(ConsultantProfile.is_active.is_(True))
|
|
.order_by(ConsultantProfile.contact_person.asc(), ConsultantProfile.firm_name.asc())
|
|
.limit(int(limit or 12))
|
|
)
|
|
return list(db.execute(stmt).scalars().all())
|
|
except Exception:
|
|
return []
|
|
|
|
|
|
def is_marketplace_domain_request(request) -> bool:
|
|
"""True when Phase 7T.2 resolved the host as a marketplace domain."""
|
|
try:
|
|
return getattr(request.state, "domain_type", None) == "marketplace"
|
|
except Exception:
|
|
return False
|