from __future__ import annotations from datetime import date, datetime, timezone from decimal import Decimal, ROUND_HALF_UP from typing import Iterable from sqlalchemy import and_, func, or_, select from sqlalchemy.orm import joinedload from app.modules.clients.models import Client from app.modules.consultants.models import ConsultantManagedClient, ConsultantProfile from app.modules.employees.models import Employee from app.modules.core.tenancy.models import Branch, Tenant from app.modules.platform_billing.models import ( PlatformBillingAccount, PlatformInvoice, PlatformInvoiceLine, PlatformPayment, PlatformPlan, PlatformPlanFeature, PlatformSubscription, ) ACCOUNT_TYPES = ["AUDIT_FIRM", "CLIENT", "CONSULTANT", "MARKETPLACE_CUSTOMER"] BILLING_CYCLES = ["Monthly", "Quarterly", "Yearly", "One-time"] TAX_TYPES = ["CGST_SGST", "IGST", "NO_GST"] CHARGE_TYPES = [ "SUBSCRIPTION", "CLIENT_USAGE", "CONSULTANT_USAGE", "EMPLOYEE_USAGE", "BRANCH_USAGE", "MODULE_CHARGE", "COMPLIANCE_DASHBOARD", "CONSULTANT_PORTAL", "MANAGED_CLIENT_USAGE", "USER_ACCOUNT_USAGE", "SERVICE_REQUEST_USAGE", "GSTIN_USAGE", "PAN_USAGE", "COMPLIANCE_MODULE_USAGE", "LEAD_FEE", "LEAD_COMMISSION", "AI_CREDITS", "STORAGE", "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 parse_date(value: str | None) -> date | None: if not value: return None return datetime.strptime(value, "%Y-%m-%d").date() def _tax_split(taxable: Decimal, gst_rate: Decimal, tax_type: str) -> tuple[Decimal, Decimal, Decimal]: taxable = money(taxable) gst_rate = money(gst_rate) if tax_type == "NO_GST" or gst_rate <= 0: return money(0), money(0), money(0) tax = money(taxable * gst_rate / Decimal("100")) if tax_type == "IGST": return money(0), money(0), tax half = money(tax / Decimal("2")) return half, money(tax - half), money(0) def list_platform_plans(db, q: str = "") -> list[PlatformPlan]: stmt = select(PlatformPlan).order_by(PlatformPlan.target_account_type.asc(), PlatformPlan.name.asc()) if q: like = f"%{q}%" stmt = stmt.where(or_(PlatformPlan.code.ilike(like), PlatformPlan.name.ilike(like))) return list(db.execute(stmt).scalars().all()) def create_platform_plan(db, *, code: str, name: str, target_account_type: str, billing_cycle: str, base_amount, gst_rate, description: str | None, feature_text: str | None) -> PlatformPlan: plan = PlatformPlan( code=code.strip(), name=name.strip(), target_account_type=target_account_type, billing_cycle=billing_cycle, base_amount=money(base_amount), gst_rate=money(gst_rate), description=description or None, is_active=True, ) db.add(plan) db.flush() for idx, raw in enumerate((feature_text or "").splitlines(), start=1): raw = raw.strip() if not raw: continue code_part = raw.upper().replace(" ", "_")[:80] db.add(PlatformPlanFeature(plan_id=plan.id, feature_code=code_part, feature_name=raw, sort_order=idx)) db.commit() db.refresh(plan) return plan def list_platform_accounts(db, q: str = "", account_type: str = "") -> list[PlatformBillingAccount]: stmt = select(PlatformBillingAccount).order_by(PlatformBillingAccount.account_type.asc(), PlatformBillingAccount.display_name.asc()) if account_type: stmt = stmt.where(PlatformBillingAccount.account_type == account_type) if q: like = f"%{q}%" stmt = stmt.where(or_(PlatformBillingAccount.account_code.ilike(like), PlatformBillingAccount.display_name.ilike(like), PlatformBillingAccount.email.ilike(like))) return list(db.execute(stmt).scalars().all()) def list_reference_audit_firms(db) -> list[Tenant]: return list(db.execute(select(Tenant).order_by(Tenant.name.asc())).scalars().all()) def list_reference_clients(db) -> list[Client]: return list(db.execute(select(Client).order_by(Client.client_name.asc())).scalars().all()) def list_reference_consultants(db) -> list[ConsultantProfile]: return list(db.execute(select(ConsultantProfile).order_by(ConsultantProfile.contact_person.asc())).scalars().all()) def create_platform_account(db, *, account_type: str, account_code: str, display_name: str, tenant_id: int | None, client_id: int | None, consultant_id: int | None, email: str | None, mobile: str | None, gstin: str | None, pan: str | None, billing_address: str | None, state: str | None, notes: str | None, user_id: int | None) -> PlatformBillingAccount: account = PlatformBillingAccount( account_type=account_type, account_code=account_code.strip(), display_name=display_name.strip(), tenant_id=tenant_id, client_id=client_id, consultant_id=consultant_id, email=email or None, mobile=mobile or None, gstin=gstin or None, pan=pan or None, billing_address=billing_address or None, state=state or None, notes=notes or None, created_by_user_id=user_id, updated_by_user_id=user_id, status="ACTIVE", ) db.add(account) db.commit() db.refresh(account) return account def list_platform_subscriptions(db, q: str = "") -> list[PlatformSubscription]: stmt = select(PlatformSubscription).options(joinedload(PlatformSubscription.account), joinedload(PlatformSubscription.plan)).order_by(PlatformSubscription.id.desc()) if q: like = f"%{q}%" stmt = stmt.join(PlatformBillingAccount).where(or_(PlatformSubscription.subscription_code.ilike(like), PlatformBillingAccount.display_name.ilike(like))) return list(db.execute(stmt).scalars().all()) def create_platform_subscription(db, *, account_id: int, plan_id: int, subscription_code: str, start_date: date, end_date: date | None, billing_cycle: str, amount, gst_rate, auto_generate_invoice: bool, notes: str | None, user_id: int | None) -> PlatformSubscription: sub = PlatformSubscription( account_id=account_id, plan_id=plan_id, subscription_code=subscription_code.strip(), start_date=start_date, end_date=end_date, billing_cycle=billing_cycle, amount=money(amount), gst_rate=money(gst_rate), auto_generate_invoice=auto_generate_invoice, notes=notes or None, status="ACTIVE", created_by_user_id=user_id, ) db.add(sub) db.commit() db.refresh(sub) return sub # ----------------------------------------------------------------------------- # PB2 - Audit Firm Subscription Billing helpers # ----------------------------------------------------------------------------- def list_audit_firm_accounts(db, q: str = "") -> list[PlatformBillingAccount]: stmt = ( select(PlatformBillingAccount) .where(PlatformBillingAccount.account_type == "AUDIT_FIRM") .order_by(PlatformBillingAccount.display_name.asc()) ) if q: like = f"%{q}%" stmt = stmt.where(or_(PlatformBillingAccount.account_code.ilike(like), PlatformBillingAccount.display_name.ilike(like), PlatformBillingAccount.email.ilike(like))) return list(db.execute(stmt).scalars().all()) def sync_audit_firm_billing_accounts(db, *, user_id: int | None = None) -> dict: """Create/update platform billing accounts for every active Audit Firm. This only touches platform_billing_accounts with account_type=AUDIT_FIRM. It does not alter tenants, firm billing invoices, or any firm-owned records. """ tenants = list(db.execute(select(Tenant).order_by(Tenant.name.asc())).scalars().all()) created = 0 updated = 0 for tenant in tenants: existing = db.execute( select(PlatformBillingAccount).where( or_( and_(PlatformBillingAccount.account_type == "AUDIT_FIRM", PlatformBillingAccount.tenant_id == tenant.id), and_(PlatformBillingAccount.account_type == "AUDIT_FIRM", PlatformBillingAccount.account_code == tenant.code), ) ) ).scalars().first() if existing: changed = False if existing.tenant_id != tenant.id: existing.tenant_id = tenant.id changed = True if existing.account_code != tenant.code: existing.account_code = tenant.code changed = True if existing.display_name != tenant.name: existing.display_name = tenant.name changed = True if existing.status != "ACTIVE" and getattr(tenant, "is_active", True): existing.status = "ACTIVE" changed = True if changed: existing.updated_by_user_id = user_id updated += 1 continue db.add( PlatformBillingAccount( account_type="AUDIT_FIRM", account_code=tenant.code, display_name=tenant.name, tenant_id=tenant.id, status="ACTIVE" if getattr(tenant, "is_active", True) else "INACTIVE", created_by_user_id=user_id, updated_by_user_id=user_id, ) ) created += 1 db.commit() return {"created": created, "updated": updated, "total": len(tenants)} def list_audit_firm_subscriptions(db, q: str = "") -> list[PlatformSubscription]: stmt = ( select(PlatformSubscription) .join(PlatformBillingAccount, PlatformSubscription.account_id == PlatformBillingAccount.id) .options(joinedload(PlatformSubscription.account), joinedload(PlatformSubscription.plan)) .where(PlatformBillingAccount.account_type == "AUDIT_FIRM") .order_by(PlatformBillingAccount.display_name.asc(), PlatformSubscription.id.desc()) ) if q: like = f"%{q}%" stmt = stmt.where(or_(PlatformSubscription.subscription_code.ilike(like), PlatformBillingAccount.display_name.ilike(like))) return list(db.execute(stmt).scalars().all()) def _count_scalar(db, stmt) -> int: value = db.execute(stmt).scalar() return int(value or 0) def get_audit_firm_usage_counts(db, tenant_id: int | None) -> dict[str, int]: if not tenant_id: return {"clients": 0, "branches": 0, "employees": 0, "consultants": 0} return { "clients": _count_scalar(db, select(func.count()).select_from(Client).where(Client.tenant_id == tenant_id, Client.is_archived.is_(False), Client.is_active.is_(True))), "branches": _count_scalar(db, select(func.count()).select_from(Branch).where(Branch.tenant_id == tenant_id, Branch.is_active.is_(True))), "employees": _count_scalar(db, select(func.count()).select_from(Employee).where(Employee.tenant_id == tenant_id, Employee.is_active.is_(True))), "consultants": _count_scalar(db, select(func.count()).select_from(ConsultantProfile).where(ConsultantProfile.tenant_id == tenant_id, ConsultantProfile.is_active.is_(True))), } def list_audit_firm_subscription_rows(db, q: str = "") -> list[dict]: rows = [] for sub in list_audit_firm_subscriptions(db, q=q): rows.append({"subscription": sub, "usage": get_audit_firm_usage_counts(db, sub.account.tenant_id if sub.account else None)}) return rows def _platform_invoice_exists_for_subscription_period(db, *, subscription_id: int, period_from: date, period_to: date) -> PlatformInvoice | None: return db.execute( select(PlatformInvoice).where( PlatformInvoice.subscription_id == subscription_id, PlatformInvoice.billing_period_from == period_from, PlatformInvoice.billing_period_to == period_to, PlatformInvoice.status != "CANCELLED", ) ).scalars().first() def _next_audit_firm_platform_invoice_no(db, *, subscription: PlatformSubscription, invoice_date: date) -> str: base = f"PB/AF/{invoice_date.strftime('%Y%m')}/{subscription.id:05d}" candidate = base suffix = 1 while db.execute(select(PlatformInvoice.id).where(PlatformInvoice.invoice_no == candidate)).scalar() is not None: suffix += 1 candidate = f"{base}-{suffix}" return candidate def generate_audit_firm_subscription_invoices( db, *, subscription_ids: list[int], period_from: date, period_to: date, invoice_date: date, due_date: date | None, tax_type: str, client_rate, employee_rate, consultant_rate, branch_rate, include_zero_usage_lines: bool = False, user_id: int | None = None, ) -> dict: """Generate draft platform invoices for Audit Firm subscriptions. Duplicate prevention is based on subscription + billing period. Generated invoices are kept as DRAFT for review/posting by System Admin. """ clean_ids = [int(x) for x in subscription_ids if str(x).strip()] if not clean_ids: return {"created": 0, "skipped": 0, "errors": ["No subscriptions selected."], "invoices": []} stmt = ( select(PlatformSubscription) .join(PlatformBillingAccount, PlatformSubscription.account_id == PlatformBillingAccount.id) .options(joinedload(PlatformSubscription.account), joinedload(PlatformSubscription.plan)) .where(PlatformSubscription.id.in_(clean_ids), PlatformBillingAccount.account_type == "AUDIT_FIRM") ) subscriptions = list(db.execute(stmt).scalars().all()) created = 0 skipped = 0 errors: list[str] = [] invoices: list[PlatformInvoice] = [] rate_map = { "CLIENT_USAGE": money(client_rate), "EMPLOYEE_USAGE": money(employee_rate), "CONSULTANT_USAGE": money(consultant_rate), "BRANCH_USAGE": money(branch_rate), } for sub in subscriptions: account = sub.account if not account or account.account_type != "AUDIT_FIRM" or not account.tenant_id: skipped += 1 errors.append(f"Skipped subscription {sub.subscription_code}: not linked to an Audit Firm account.") continue if sub.status != "ACTIVE" or not sub.auto_generate_invoice: skipped += 1 errors.append(f"Skipped {account.display_name}: subscription is not active/auto-generate enabled.") continue existing = _platform_invoice_exists_for_subscription_period(db, subscription_id=sub.id, period_from=period_from, period_to=period_to) if existing: skipped += 1 errors.append(f"Skipped {account.display_name}: invoice already exists for this period ({existing.invoice_no}).") continue usage = get_audit_firm_usage_counts(db, account.tenant_id) line_items: list[dict] = [] if money(sub.amount) > 0: line_items.append({ "charge_type": "SUBSCRIPTION", "description": f"{sub.plan.name if sub.plan else 'Audit Firm Subscription'} - {period_from.strftime('%d-%m-%Y')} to {period_to.strftime('%d-%m-%Y')}", "quantity": "1", "rate": sub.amount, "discount_amount": "0", "gst_rate": sub.gst_rate, "reference_type": "PLATFORM_SUBSCRIPTION", "reference_id": sub.id, }) usage_specs = [ ("CLIENT_USAGE", "Client usage", usage["clients"]), ("EMPLOYEE_USAGE", "Employee usage", usage["employees"]), ("CONSULTANT_USAGE", "Consultant usage", usage["consultants"]), ("BRANCH_USAGE", "Branch usage", usage["branches"]), ] for charge_type, label, count in usage_specs: rate = rate_map[charge_type] if rate > 0 and (count > 0 or include_zero_usage_lines): line_items.append({ "charge_type": charge_type, "description": f"{label} for {account.display_name} ({period_from.strftime('%b %Y')})", "quantity": str(count), "rate": rate, "discount_amount": "0", "gst_rate": sub.gst_rate, "reference_type": "AUDIT_FIRM", "reference_id": account.tenant_id, }) if not line_items: skipped += 1 errors.append(f"Skipped {account.display_name}: no billable line items.") continue invoice = create_platform_invoice( db, account_id=account.id, subscription_id=sub.id, invoice_no=_next_audit_firm_platform_invoice_no(db, subscription=sub, invoice_date=invoice_date), invoice_date=invoice_date, due_date=due_date, billing_period_from=period_from, billing_period_to=period_to, tax_type=tax_type, line_items=line_items, notes="Generated from Audit Firm subscription billing (PB2).", user_id=user_id, ) created += 1 invoices.append(invoice) return {"created": created, "skipped": skipped, "errors": errors, "invoices": invoices} # ----------------------------------------------------------------------------- # PB3 - Client Compliance Dashboard Billing helpers # ----------------------------------------------------------------------------- CLIENT_COMPLIANCE_FLAGS = [ ("gst_applicable", "GST"), ("income_tax_applicable", "Income Tax"), ("tds_applicable", "TDS"), ("roc_applicable", "ROC"), ("audit_applicable", "Audit"), ("pf_applicable", "PF"), ("esi_applicable", "ESI"), ("professional_tax_applicable", "Professional Tax"), ("payroll_applicable", "Payroll"), ("msme_applicable", "MSME"), ("import_export_applicable", "Import/Export"), ] def _client_account_code(client: Client) -> str: return f"CL-{client.id:06d}" def list_client_dashboard_accounts(db, q: str = "") -> list[PlatformBillingAccount]: stmt = ( select(PlatformBillingAccount) .where(PlatformBillingAccount.account_type == "CLIENT") .order_by(PlatformBillingAccount.display_name.asc()) ) if q: like = f"%{q}%" stmt = stmt.where(or_(PlatformBillingAccount.account_code.ilike(like), PlatformBillingAccount.display_name.ilike(like), PlatformBillingAccount.email.ilike(like))) return list(db.execute(stmt).scalars().all()) def sync_client_dashboard_billing_accounts(db, *, user_id: int | None = None) -> dict: """Create/update platform billing accounts for active clients. This is for Platform/System Admin billing of client compliance dashboard access. It only touches platform_billing_accounts with account_type=CLIENT. It does not alter client master records or firm-level billing invoices. """ clients = list( db.execute( select(Client) .where(Client.is_archived.is_(False), Client.is_active.is_(True)) .order_by(Client.client_name.asc()) ).scalars().all() ) created = 0 updated = 0 for client in clients: code = _client_account_code(client) display_name = client.client_name address_parts = [client.address_line_1, client.address_line_2, client.city, client.state, client.pincode] billing_address = ", ".join([part for part in address_parts if part]) or None existing = db.execute( select(PlatformBillingAccount).where( or_( and_(PlatformBillingAccount.account_type == "CLIENT", PlatformBillingAccount.client_id == client.id), and_(PlatformBillingAccount.account_type == "CLIENT", PlatformBillingAccount.account_code == code), ) ) ).scalars().first() if existing: changed = False updates = { "account_code": code, "display_name": display_name, "tenant_id": client.tenant_id, "client_id": client.id, "email": client.email, "mobile": client.mobile, "gstin": client.gstin, "pan": client.pan, "billing_address": billing_address, "state": client.state, "status": "ACTIVE", } for field, value in updates.items(): if getattr(existing, field) != value: setattr(existing, field, value) changed = True if changed: existing.updated_by_user_id = user_id updated += 1 continue db.add( PlatformBillingAccount( account_type="CLIENT", account_code=code, display_name=display_name, tenant_id=client.tenant_id, client_id=client.id, email=client.email, mobile=client.mobile, gstin=client.gstin, pan=client.pan, billing_address=billing_address, state=client.state, status="ACTIVE", created_by_user_id=user_id, updated_by_user_id=user_id, ) ) created += 1 db.commit() return {"created": created, "updated": updated, "total": len(clients)} def list_client_dashboard_plans(db) -> list[PlatformPlan]: return list( db.execute( select(PlatformPlan) .where(PlatformPlan.target_account_type == "CLIENT", PlatformPlan.is_active.is_(True)) .order_by(PlatformPlan.name.asc()) ).scalars().all() ) def list_client_dashboard_subscriptions(db, q: str = "") -> list[PlatformSubscription]: stmt = ( select(PlatformSubscription) .join(PlatformBillingAccount, PlatformSubscription.account_id == PlatformBillingAccount.id) .options(joinedload(PlatformSubscription.account), joinedload(PlatformSubscription.plan)) .where(PlatformBillingAccount.account_type == "CLIENT") .order_by(PlatformBillingAccount.display_name.asc(), PlatformSubscription.id.desc()) ) if q: like = f"%{q}%" stmt = stmt.where(or_(PlatformSubscription.subscription_code.ilike(like), PlatformBillingAccount.display_name.ilike(like), PlatformBillingAccount.account_code.ilike(like))) return list(db.execute(stmt).scalars().all()) def get_client_dashboard_usage(db, client_id: int | None) -> dict: if not client_id: return {"pan_units": 0, "gstin_units": 0, "module_count": 0, "modules": []} client = db.get(Client, client_id) if not client: return {"pan_units": 0, "gstin_units": 0, "module_count": 0, "modules": []} modules = [label for attr, label in CLIENT_COMPLIANCE_FLAGS if bool(getattr(client, attr, False))] return { "pan_units": 1 if client.pan else 0, "gstin_units": 1 if client.gstin else 0, "module_count": len(modules), "modules": modules, } def list_client_dashboard_subscription_rows(db, q: str = "") -> list[dict]: rows = [] for sub in list_client_dashboard_subscriptions(db, q=q): rows.append({"subscription": sub, "usage": get_client_dashboard_usage(db, sub.account.client_id if sub.account else None)}) return rows def _next_client_dashboard_invoice_no(db, *, subscription: PlatformSubscription, invoice_date: date) -> str: base = f"PB/CL/{invoice_date.strftime('%Y%m')}/{subscription.id:05d}" candidate = base suffix = 1 while db.execute(select(PlatformInvoice.id).where(PlatformInvoice.invoice_no == candidate)).scalar() is not None: suffix += 1 candidate = f"{base}-{suffix}" return candidate def generate_client_dashboard_subscription_invoices( db, *, subscription_ids: list[int], period_from: date, period_to: date, invoice_date: date, due_date: date | None, tax_type: str, pan_rate, gstin_rate, module_rate, include_zero_usage_lines: bool = False, user_id: int | None = None, ) -> dict: """Generate draft platform invoices for client compliance dashboard subscriptions.""" clean_ids = [int(x) for x in subscription_ids if str(x).strip()] if not clean_ids: return {"created": 0, "skipped": 0, "errors": ["No client dashboard subscriptions selected."], "invoices": []} stmt = ( select(PlatformSubscription) .join(PlatformBillingAccount, PlatformSubscription.account_id == PlatformBillingAccount.id) .options(joinedload(PlatformSubscription.account), joinedload(PlatformSubscription.plan)) .where(PlatformSubscription.id.in_(clean_ids), PlatformBillingAccount.account_type == "CLIENT") ) subscriptions = list(db.execute(stmt).scalars().all()) created = 0 skipped = 0 errors: list[str] = [] invoices: list[PlatformInvoice] = [] pan_rate = money(pan_rate) gstin_rate = money(gstin_rate) module_rate = money(module_rate) for sub in subscriptions: account = sub.account if not account or account.account_type != "CLIENT" or not account.client_id: skipped += 1 errors.append(f"Skipped subscription {sub.subscription_code}: not linked to a client account.") continue if sub.status != "ACTIVE" or not sub.auto_generate_invoice: skipped += 1 errors.append(f"Skipped {account.display_name}: subscription is not active/auto-generate enabled.") continue existing = _platform_invoice_exists_for_subscription_period(db, subscription_id=sub.id, period_from=period_from, period_to=period_to) if existing: skipped += 1 errors.append(f"Skipped {account.display_name}: invoice already exists for this period ({existing.invoice_no}).") continue usage = get_client_dashboard_usage(db, account.client_id) line_items: list[dict] = [] if money(sub.amount) > 0: line_items.append({ "charge_type": "COMPLIANCE_DASHBOARD", "description": f"{sub.plan.name if sub.plan else 'Client Compliance Dashboard'} - {period_from.strftime('%d-%m-%Y')} to {period_to.strftime('%d-%m-%Y')}", "quantity": "1", "rate": sub.amount, "discount_amount": "0", "gst_rate": sub.gst_rate, "reference_type": "CLIENT_DASHBOARD_SUBSCRIPTION", "reference_id": sub.id, }) usage_specs = [ ("PAN_USAGE", "PAN dashboard access", usage["pan_units"], pan_rate), ("GSTIN_USAGE", "GSTIN dashboard access", usage["gstin_units"], gstin_rate), ("COMPLIANCE_MODULE_USAGE", "Compliance modules", usage["module_count"], module_rate), ] for charge_type, label, quantity, rate in usage_specs: if rate > 0 and (quantity > 0 or include_zero_usage_lines): extra = "" if charge_type == "COMPLIANCE_MODULE_USAGE" and usage["modules"]: extra = f" ({', '.join(usage['modules'])})" line_items.append({ "charge_type": charge_type, "description": f"{label}{extra} for {account.display_name} ({period_from.strftime('%b %Y')})", "quantity": str(quantity), "rate": rate, "discount_amount": "0", "gst_rate": sub.gst_rate, "reference_type": "CLIENT", "reference_id": account.client_id, }) if not line_items: skipped += 1 errors.append(f"Skipped {account.display_name}: no billable line items.") continue invoice = create_platform_invoice( db, account_id=account.id, subscription_id=sub.id, invoice_no=_next_client_dashboard_invoice_no(db, subscription=sub, invoice_date=invoice_date), invoice_date=invoice_date, due_date=due_date, billing_period_from=period_from, billing_period_to=period_to, tax_type=tax_type, line_items=line_items, notes="Generated from Client Compliance Dashboard subscription billing (PB3).", user_id=user_id, ) created += 1 invoices.append(invoice) return {"created": created, "skipped": skipped, "errors": errors, "invoices": invoices} # ----------------------------------------------------------------------------- # PB4 - Consultant SaaS / Tool Access Billing helpers # ----------------------------------------------------------------------------- def _consultant_account_code(consultant: ConsultantProfile) -> str: return f"CON-{consultant.id:06d}" def _consultant_display_name(consultant: ConsultantProfile) -> str: if consultant.firm_name and consultant.contact_person: return f"{consultant.firm_name} - {consultant.contact_person}" return consultant.firm_name or consultant.contact_person or f"Consultant {consultant.id}" def list_consultant_billing_accounts(db, q: str = "") -> list[PlatformBillingAccount]: stmt = ( select(PlatformBillingAccount) .where(PlatformBillingAccount.account_type == "CONSULTANT") .order_by(PlatformBillingAccount.display_name.asc()) ) if q: like = f"%{q}%" stmt = stmt.where(or_(PlatformBillingAccount.account_code.ilike(like), PlatformBillingAccount.display_name.ilike(like), PlatformBillingAccount.email.ilike(like))) return list(db.execute(stmt).scalars().all()) def sync_consultant_billing_accounts(db, *, user_id: int | None = None) -> dict: """Create/update platform billing accounts for active consultants. This is for Platform/System Admin billing of consultant SaaS/tool access. It only touches platform_billing_accounts with account_type=CONSULTANT. It does not alter consultant profiles, consultant workspace data, firm billing, or firm-owned client invoices. """ consultants = list( db.execute( select(ConsultantProfile) .where(ConsultantProfile.is_active.is_(True)) .order_by(ConsultantProfile.contact_person.asc()) ).scalars().all() ) created = 0 updated = 0 for consultant in consultants: code = _consultant_account_code(consultant) display_name = _consultant_display_name(consultant) existing = db.execute( select(PlatformBillingAccount).where( or_( and_(PlatformBillingAccount.account_type == "CONSULTANT", PlatformBillingAccount.consultant_id == consultant.id), and_(PlatformBillingAccount.account_type == "CONSULTANT", PlatformBillingAccount.account_code == code), ) ) ).scalars().first() updates = { "account_code": code, "display_name": display_name, "tenant_id": consultant.tenant_id, "consultant_id": consultant.id, "email": consultant.email, "mobile": consultant.mobile, "gstin": consultant.gstin, "pan": consultant.pan, "billing_address": consultant.address, "status": "ACTIVE", } if existing: changed = False for field, value in updates.items(): if getattr(existing, field) != value: setattr(existing, field, value) changed = True if changed: existing.updated_by_user_id = user_id updated += 1 continue db.add( PlatformBillingAccount( account_type="CONSULTANT", created_by_user_id=user_id, updated_by_user_id=user_id, **updates, ) ) created += 1 db.commit() return {"created": created, "updated": updated, "total": len(consultants)} def list_consultant_billing_plans(db) -> list[PlatformPlan]: return list( db.execute( select(PlatformPlan) .where(PlatformPlan.target_account_type == "CONSULTANT", PlatformPlan.is_active.is_(True)) .order_by(PlatformPlan.name.asc()) ).scalars().all() ) def list_consultant_billing_subscriptions(db, q: str = "") -> list[PlatformSubscription]: stmt = ( select(PlatformSubscription) .join(PlatformBillingAccount, PlatformSubscription.account_id == PlatformBillingAccount.id) .options(joinedload(PlatformSubscription.account), joinedload(PlatformSubscription.plan)) .where(PlatformBillingAccount.account_type == "CONSULTANT") .order_by(PlatformBillingAccount.display_name.asc(), PlatformSubscription.id.desc()) ) if q: like = f"%{q}%" stmt = stmt.where(or_(PlatformSubscription.subscription_code.ilike(like), PlatformBillingAccount.display_name.ilike(like), PlatformBillingAccount.account_code.ilike(like))) return list(db.execute(stmt).scalars().all()) def get_consultant_tool_usage(db, consultant_id: int | None) -> dict: if not consultant_id: return {"managed_client_count": 0, "user_account_count": 1, "workspace_type": "-", "workspace_plan": "-"} consultant = db.get(ConsultantProfile, consultant_id) managed_client_count = db.execute( select(func.count(ConsultantManagedClient.id)).where( ConsultantManagedClient.consultant_id == consultant_id, ConsultantManagedClient.is_active.is_(True), ) ).scalar() or 0 workspace = getattr(consultant, "workspace", None) if consultant else None workspace_type = "-" if consultant: workspace_type = getattr(workspace, "workspace_type", None) or getattr(consultant, "consultant_type", "-") or "-" return { "managed_client_count": int(managed_client_count), "user_account_count": 1 if consultant and consultant.user_id else 0, "workspace_type": workspace_type, "workspace_plan": getattr(workspace, "plan_code", None) or "-", } def list_consultant_billing_subscription_rows(db, q: str = "") -> list[dict]: rows = [] for sub in list_consultant_billing_subscriptions(db, q=q): rows.append({"subscription": sub, "usage": get_consultant_tool_usage(db, sub.account.consultant_id if sub.account else None)}) return rows def _next_consultant_invoice_no(db, *, subscription: PlatformSubscription, invoice_date: date) -> str: base = f"PB/CON/{invoice_date.strftime('%Y%m')}/{subscription.id:05d}" candidate = base suffix = 1 while db.execute(select(PlatformInvoice.id).where(PlatformInvoice.invoice_no == candidate)).scalar() is not None: suffix += 1 candidate = f"{base}-{suffix}" return candidate def generate_consultant_subscription_invoices( db, *, subscription_ids: list[int], period_from: date, period_to: date, invoice_date: date, due_date: date | None, tax_type: str, managed_client_rate, user_account_rate, include_zero_usage_lines: bool = False, user_id: int | None = None, ) -> dict: """Generate draft platform invoices for consultant SaaS/tool subscriptions.""" clean_ids = [int(x) for x in subscription_ids if str(x).strip()] if not clean_ids: return {"created": 0, "skipped": 0, "errors": ["No consultant subscriptions selected."], "invoices": []} stmt = ( select(PlatformSubscription) .join(PlatformBillingAccount, PlatformSubscription.account_id == PlatformBillingAccount.id) .options(joinedload(PlatformSubscription.account), joinedload(PlatformSubscription.plan)) .where(PlatformSubscription.id.in_(clean_ids), PlatformBillingAccount.account_type == "CONSULTANT") ) subscriptions = list(db.execute(stmt).scalars().all()) created = 0 skipped = 0 errors: list[str] = [] invoices: list[PlatformInvoice] = [] managed_client_rate = money(managed_client_rate) user_account_rate = money(user_account_rate) for sub in subscriptions: account = sub.account if not account or account.account_type != "CONSULTANT" or not account.consultant_id: skipped += 1 errors.append(f"Skipped subscription {sub.subscription_code}: not linked to a consultant account.") continue if sub.status != "ACTIVE" or not sub.auto_generate_invoice: skipped += 1 errors.append(f"Skipped {account.display_name}: subscription is not active/auto-generate enabled.") continue existing = _platform_invoice_exists_for_subscription_period(db, subscription_id=sub.id, period_from=period_from, period_to=period_to) if existing: skipped += 1 errors.append(f"Skipped {account.display_name}: invoice already exists for this period ({existing.invoice_no}).") continue usage = get_consultant_tool_usage(db, account.consultant_id) line_items: list[dict] = [] if money(sub.amount) > 0: line_items.append({ "charge_type": "CONSULTANT_PORTAL", "description": f"{sub.plan.name if sub.plan else 'Consultant Tool Access'} - {period_from.strftime('%d-%m-%Y')} to {period_to.strftime('%d-%m-%Y')}", "quantity": "1", "rate": sub.amount, "discount_amount": "0", "gst_rate": sub.gst_rate, "reference_type": "CONSULTANT_SUBSCRIPTION", "reference_id": sub.id, }) usage_specs = [ ("MANAGED_CLIENT_USAGE", "Managed consultant clients", usage["managed_client_count"], managed_client_rate), ("USER_ACCOUNT_USAGE", "Consultant portal user accounts", usage["user_account_count"], user_account_rate), ] for charge_type, label, quantity, rate in usage_specs: if rate > 0 and (quantity > 0 or include_zero_usage_lines): line_items.append({ "charge_type": charge_type, "description": f"{label} for {account.display_name} ({period_from.strftime('%b %Y')})", "quantity": str(quantity), "rate": rate, "discount_amount": "0", "gst_rate": sub.gst_rate, "reference_type": "CONSULTANT", "reference_id": account.consultant_id, }) if not line_items: skipped += 1 errors.append(f"Skipped {account.display_name}: no billable line items.") continue invoice = create_platform_invoice( db, account_id=account.id, subscription_id=sub.id, invoice_no=_next_consultant_invoice_no(db, subscription=sub, invoice_date=invoice_date), invoice_date=invoice_date, due_date=due_date, billing_period_from=period_from, billing_period_to=period_to, tax_type=tax_type, line_items=line_items, notes="Generated from Consultant SaaS/tool subscription billing (PB4).", user_id=user_id, ) created += 1 invoices.append(invoice) return {"created": created, "skipped": skipped, "errors": errors, "invoices": invoices} def list_platform_invoices(db, q: str = "") -> list[PlatformInvoice]: stmt = select(PlatformInvoice).options(joinedload(PlatformInvoice.account)).order_by(PlatformInvoice.invoice_date.desc(), PlatformInvoice.id.desc()) if q: like = f"%{q}%" stmt = stmt.join(PlatformBillingAccount).where(or_(PlatformInvoice.invoice_no.ilike(like), PlatformBillingAccount.display_name.ilike(like))) return list(db.execute(stmt).scalars().all()) def get_platform_invoice(db, invoice_id: int) -> PlatformInvoice | None: return db.execute( select(PlatformInvoice).options(joinedload(PlatformInvoice.account), joinedload(PlatformInvoice.lines)).where(PlatformInvoice.id == invoice_id) ).unique().scalar_one_or_none() def create_platform_invoice(db, *, account_id: int, subscription_id: int | None, invoice_no: str, invoice_date: date, due_date: date | None, billing_period_from: date | None, billing_period_to: date | None, tax_type: str, line_items: Iterable[dict], notes: str | None, user_id: int | None) -> PlatformInvoice: invoice = PlatformInvoice( account_id=account_id, subscription_id=subscription_id, invoice_no=invoice_no.strip(), invoice_date=invoice_date, due_date=due_date, billing_period_from=billing_period_from, billing_period_to=billing_period_to, tax_type=tax_type, notes=notes or None, created_by_user_id=user_id, status="DRAFT", ) db.add(invoice) db.flush() subtotal = money(0) taxable_total = money(0) cgst_total = money(0) sgst_total = money(0) igst_total = money(0) for idx, item in enumerate(line_items, start=1): description = (item.get("description") or "").strip() if not description: continue quantity = money(item.get("quantity", 1)) rate = money(item.get("rate", 0)) discount = money(item.get("discount_amount", 0)) gst_rate = money(item.get("gst_rate", 18)) taxable = money((quantity * rate) - discount) if taxable < 0: taxable = money(0) cgst, sgst, igst = _tax_split(taxable, gst_rate, tax_type) line_total = money(taxable + cgst + sgst + igst) subtotal += money(quantity * rate) taxable_total += taxable cgst_total += cgst sgst_total += sgst igst_total += igst db.add(PlatformInvoiceLine( invoice_id=invoice.id, charge_type=item.get("charge_type") or "SUBSCRIPTION", description=description, reference_type=item.get("reference_type") or None, reference_id=item.get("reference_id") or None, quantity=quantity, rate=rate, discount_amount=discount, taxable_amount=taxable, gst_rate=gst_rate, cgst_amount=cgst, sgst_amount=sgst, igst_amount=igst, line_total=line_total, sort_order=idx, )) invoice.subtotal = money(subtotal) invoice.discount_amount = money(subtotal - taxable_total) invoice.taxable_amount = money(taxable_total) invoice.cgst_amount = money(cgst_total) invoice.sgst_amount = money(sgst_total) invoice.igst_amount = money(igst_total) invoice.total_amount = money(taxable_total + cgst_total + sgst_total + igst_total) db.commit() db.refresh(invoice) return invoice def post_platform_invoice(db, invoice: PlatformInvoice, user_id: int | None) -> PlatformInvoice: invoice.status = "POSTED" invoice.posted_by_user_id = user_id invoice.posted_at_utc = datetime.now(timezone.utc) db.commit() db.refresh(invoice) return invoice def record_platform_payment(db, *, invoice: PlatformInvoice, amount, mode: str, reference_no: str | None, notes: str | None, user_id: int | None) -> PlatformPayment: payment = PlatformPayment( invoice_id=invoice.id, account_id=invoice.account_id, amount=money(amount), mode=mode, reference_no=reference_no or None, notes=notes or None, created_by_user_id=user_id, ) db.add(payment) db.commit() db.refresh(payment) return payment