from __future__ import annotations from fastapi import APIRouter, Form, Request from fastapi.responses import RedirectResponse from app.core.db.common import CommonSessionLocal from app.core.security.csrf import get_or_create_csrf_token, validate_csrf from app.core.security.session_auth import get_current_user from app.core.templating import templates from app.modules.core.rbac.deps import get_user_permissions, get_user_roles from app.modules.core.rbac.permission_guard import require_permission from app.modules.platform_billing.services import ( ACCOUNT_TYPES, BILLING_CYCLES, CHARGE_TYPES, TAX_TYPES, create_platform_account, create_platform_invoice, create_platform_plan, create_platform_subscription, generate_audit_firm_subscription_invoices, generate_client_dashboard_subscription_invoices, generate_consultant_subscription_invoices, get_platform_invoice, list_audit_firm_accounts, list_audit_firm_subscription_rows, list_client_dashboard_accounts, list_client_dashboard_plans, list_client_dashboard_subscription_rows, list_consultant_billing_accounts, list_consultant_billing_plans, list_consultant_billing_subscription_rows, list_platform_accounts, list_platform_invoices, list_platform_plans, list_platform_subscriptions, list_reference_audit_firms, list_reference_clients, list_reference_consultants, parse_date, post_platform_invoice, record_platform_payment, sync_audit_firm_billing_accounts, sync_client_dashboard_billing_accounts, sync_consultant_billing_accounts, ) router = APIRouter(prefix="/platform-billing", tags=["platform-billing-ui"]) def _redirect_denied(): from app.core.http_responses import ui_access_denied return ui_access_denied() def _has_perm(db, user, code: str) -> bool: try: require_permission(db, user, code) return True except Exception: return False def _require_user(request: Request, db, permission_code: str): user = get_current_user(request, db=db) if not user: return None, RedirectResponse(url="/login", status_code=303) try: require_permission(db, user, permission_code) except Exception: return user, _redirect_denied() return user, None def _base_ctx(request: Request, db, user, **ctx): base = { "request": request, "current_user": user, "current_user_roles": get_user_roles(db, user.id), "current_user_permissions": get_user_permissions(db, user.id), "csrf_token": get_or_create_csrf_token(request), "account_types": ACCOUNT_TYPES, "billing_cycles": BILLING_CYCLES, "tax_types": TAX_TYPES, "charge_types": CHARGE_TYPES, } base.update(ctx) return base def _render(request: Request, template: str, db, user, **ctx): return templates.TemplateResponse(template, _base_ctx(request, db, user, **ctx)) @router.get("") def dashboard(request: Request): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_billing.view") if response: return response return _render( request, "modules/platform_billing/templates/platform_billing/dashboard.html", db, user, title="Platform Billing", plans=list_platform_plans(db)[:5], accounts=list_platform_accounts(db)[:5], audit_firm_accounts=list_audit_firm_accounts(db)[:5], client_dashboard_accounts=list_client_dashboard_accounts(db)[:5], consultant_billing_accounts=list_consultant_billing_accounts(db)[:5], subscriptions=list_platform_subscriptions(db)[:5], invoices=list_platform_invoices(db)[:5], can_manage_plans=_has_perm(db, user, "platform_plans.manage"), can_manage_subscriptions=_has_perm(db, user, "platform_subscriptions.manage"), can_generate_platform_billing_flag=_has_perm(db, user, "platform_billing.generate"), can_create_invoice=_has_perm(db, user, "platform_billing.create"), ) finally: db.close() @router.get("/plans") def plans_list(request: Request, q: str = ""): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_billing.view") if response: return response return _render(request, "modules/platform_billing/templates/platform_billing/plans/list.html", db, user, title="Platform Plans", rows=list_platform_plans(db, q=q), q=q, can_manage=_has_perm(db, user, "platform_plans.manage")) finally: db.close() @router.get("/plans/new") def plan_new(request: Request): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_plans.manage") if response: return response return _render(request, "modules/platform_billing/templates/platform_billing/plans/create.html", db, user, title="New Platform Plan") finally: db.close() @router.post("/plans/new") def plan_create(request: Request, csrf_token: str = Form(...), code: str = Form(...), name: str = Form(...), target_account_type: str = Form(...), billing_cycle: str = Form(...), base_amount: str = Form("0"), gst_rate: str = Form("18"), description: str = Form(""), feature_text: str = Form("")): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_plans.manage") if response: return response validate_csrf(request, csrf_token) create_platform_plan(db, code=code, name=name, target_account_type=target_account_type, billing_cycle=billing_cycle, base_amount=base_amount, gst_rate=gst_rate, description=description, feature_text=feature_text) return RedirectResponse(url="/platform-billing/plans", status_code=303) finally: db.close() @router.get("/accounts") def accounts_list(request: Request, q: str = "", account_type: str = ""): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_billing.view") if response: return response return _render(request, "modules/platform_billing/templates/platform_billing/accounts/list.html", db, user, title="Platform Billing Accounts", rows=list_platform_accounts(db, q=q, account_type=account_type), q=q, selected_account_type=account_type, can_create=_has_perm(db, user, "platform_billing.create")) finally: db.close() @router.get("/accounts/new") def account_new(request: Request): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_billing.create") if response: return response return _render(request, "modules/platform_billing/templates/platform_billing/accounts/create.html", db, user, title="New Platform Billing Account", audit_firms=list_reference_audit_firms(db), clients=list_reference_clients(db), consultants=list_reference_consultants(db)) finally: db.close() @router.post("/accounts/new") def account_create(request: Request, csrf_token: str = Form(...), account_type: str = Form(...), account_code: str = Form(...), display_name: str = Form(...), tenant_id: str = Form(""), client_id: str = Form(""), consultant_id: str = Form(""), email: str = Form(""), mobile: str = Form(""), gstin: str = Form(""), pan: str = Form(""), billing_address: str = Form(""), state: str = Form(""), notes: str = Form("")): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_billing.create") if response: return response validate_csrf(request, csrf_token) create_platform_account(db, account_type=account_type, account_code=account_code, display_name=display_name, tenant_id=int(tenant_id) if tenant_id else None, client_id=int(client_id) if client_id else None, consultant_id=int(consultant_id) if consultant_id else None, email=email, mobile=mobile, gstin=gstin, pan=pan, billing_address=billing_address, state=state, notes=notes, user_id=user.id) return RedirectResponse(url="/platform-billing/accounts", status_code=303) finally: db.close() @router.get("/audit-firm-subscriptions") def audit_firm_subscriptions(request: Request, q: str = "", synced: str = ""): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_billing.view") if response: return response return _render( request, "modules/platform_billing/templates/platform_billing/audit_firms/list.html", db, user, title="Audit Firm Subscription Billing", rows=list_audit_firm_subscription_rows(db, q=q), audit_firm_accounts=list_audit_firm_accounts(db, q=q), q=q, synced=synced, can_manage=_has_perm(db, user, "platform_subscriptions.manage"), can_generate=_has_perm(db, user, "platform_billing.generate"), ) finally: db.close() @router.post("/audit-firm-subscriptions/sync-accounts") def audit_firm_accounts_sync(request: Request, csrf_token: str = Form(...)): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_subscriptions.manage") if response: return response validate_csrf(request, csrf_token) result = sync_audit_firm_billing_accounts(db, user_id=user.id) return RedirectResponse(url=f"/platform-billing/audit-firm-subscriptions?synced=created-{result['created']}-updated-{result['updated']}", status_code=303) finally: db.close() @router.get("/audit-firm-subscriptions/generate") def audit_firm_generate_form(request: Request, generated: int = 0, skipped: int = 0): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_billing.generate") if response: return response return _render( request, "modules/platform_billing/templates/platform_billing/audit_firms/generate.html", db, user, title="Generate Audit Firm Platform Bills", rows=list_audit_firm_subscription_rows(db), generated=generated, skipped=skipped, errors=[], ) finally: db.close() @router.post("/audit-firm-subscriptions/generate") def audit_firm_generate_submit(request: Request, csrf_token: str = Form(...), subscription_ids: list[int] = Form(default=[]), period_from: str = Form(...), period_to: str = Form(...), invoice_date: str = Form(...), due_date: str = Form(""), tax_type: str = Form("CGST_SGST"), client_rate: str = Form("0"), employee_rate: str = Form("0"), consultant_rate: str = Form("0"), branch_rate: str = Form("0"), include_zero_usage_lines: str = Form("off")): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_billing.generate") if response: return response validate_csrf(request, csrf_token) result = generate_audit_firm_subscription_invoices( db, subscription_ids=subscription_ids, period_from=parse_date(period_from), period_to=parse_date(period_to), invoice_date=parse_date(invoice_date), due_date=parse_date(due_date), tax_type=tax_type, client_rate=client_rate, employee_rate=employee_rate, consultant_rate=consultant_rate, branch_rate=branch_rate, include_zero_usage_lines=include_zero_usage_lines == "on", user_id=user.id, ) return _render( request, "modules/platform_billing/templates/platform_billing/audit_firms/generate.html", db, user, title="Generate Audit Firm Platform Bills", rows=list_audit_firm_subscription_rows(db), generated=result["created"], skipped=result["skipped"], errors=result["errors"], invoices=result["invoices"], ) finally: db.close() @router.get("/client-dashboard-subscriptions") def client_dashboard_subscriptions(request: Request, q: str = "", synced: str = "", created: str = ""): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_billing.view") if response: return response return _render( request, "modules/platform_billing/templates/platform_billing/clients/list.html", db, user, title="Client Compliance Dashboard Billing", rows=list_client_dashboard_subscription_rows(db, q=q), client_accounts=list_client_dashboard_accounts(db, q=q), client_plans=list_client_dashboard_plans(db), q=q, synced=synced, created=created, can_manage=_has_perm(db, user, "platform_subscriptions.manage"), can_generate=_has_perm(db, user, "platform_billing.generate"), ) finally: db.close() @router.post("/client-dashboard-subscriptions/sync-accounts") def client_dashboard_accounts_sync(request: Request, csrf_token: str = Form(...)): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_subscriptions.manage") if response: return response validate_csrf(request, csrf_token) result = sync_client_dashboard_billing_accounts(db, user_id=user.id) return RedirectResponse(url=f"/platform-billing/client-dashboard-subscriptions?synced=created-{result['created']}-updated-{result['updated']}", status_code=303) finally: db.close() @router.post("/client-dashboard-subscriptions/new") def client_dashboard_subscription_create(request: Request, csrf_token: str = Form(...), account_id: int = Form(...), plan_id: int = Form(...), subscription_code: str = Form(...), start_date: str = Form(...), end_date: str = Form(""), billing_cycle: str = Form("Monthly"), amount: str = Form("0"), gst_rate: str = Form("18"), auto_generate_invoice: str = Form("on"), notes: str = Form("")): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_subscriptions.manage") if response: return response validate_csrf(request, csrf_token) create_platform_subscription( db, account_id=account_id, plan_id=plan_id, subscription_code=subscription_code, start_date=parse_date(start_date), end_date=parse_date(end_date), billing_cycle=billing_cycle, amount=amount, gst_rate=gst_rate, auto_generate_invoice=auto_generate_invoice == "on", notes=notes, user_id=user.id, ) return RedirectResponse(url="/platform-billing/client-dashboard-subscriptions?created=1", status_code=303) finally: db.close() @router.get("/client-dashboard-subscriptions/generate") def client_dashboard_generate_form(request: Request, generated: int = 0, skipped: int = 0): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_billing.generate") if response: return response return _render( request, "modules/platform_billing/templates/platform_billing/clients/generate.html", db, user, title="Generate Client Dashboard Platform Bills", rows=list_client_dashboard_subscription_rows(db), generated=generated, skipped=skipped, errors=[], ) finally: db.close() @router.post("/client-dashboard-subscriptions/generate") def client_dashboard_generate_submit(request: Request, csrf_token: str = Form(...), subscription_ids: list[int] = Form(default=[]), period_from: str = Form(...), period_to: str = Form(...), invoice_date: str = Form(...), due_date: str = Form(""), tax_type: str = Form("CGST_SGST"), pan_rate: str = Form("0"), gstin_rate: str = Form("0"), module_rate: str = Form("0"), include_zero_usage_lines: str = Form("off")): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_billing.generate") if response: return response validate_csrf(request, csrf_token) result = generate_client_dashboard_subscription_invoices( db, subscription_ids=subscription_ids, period_from=parse_date(period_from), period_to=parse_date(period_to), invoice_date=parse_date(invoice_date), due_date=parse_date(due_date), tax_type=tax_type, pan_rate=pan_rate, gstin_rate=gstin_rate, module_rate=module_rate, include_zero_usage_lines=include_zero_usage_lines == "on", user_id=user.id, ) return _render( request, "modules/platform_billing/templates/platform_billing/clients/generate.html", db, user, title="Generate Client Dashboard Platform Bills", rows=list_client_dashboard_subscription_rows(db), generated=result["created"], skipped=result["skipped"], errors=result["errors"], invoices=result["invoices"], ) finally: db.close() @router.get("/consultant-subscriptions") def consultant_subscriptions(request: Request, q: str = "", synced: str = "", created: str = ""): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_billing.view") if response: return response return _render( request, "modules/platform_billing/templates/platform_billing/consultants/list.html", db, user, title="Consultant SaaS/Tool Billing", rows=list_consultant_billing_subscription_rows(db, q=q), consultant_accounts=list_consultant_billing_accounts(db, q=q), consultant_plans=list_consultant_billing_plans(db), q=q, synced=synced, created=created, can_manage=_has_perm(db, user, "platform_subscriptions.manage"), can_generate=_has_perm(db, user, "platform_billing.generate"), ) finally: db.close() @router.post("/consultant-subscriptions/sync-accounts") def consultant_accounts_sync(request: Request, csrf_token: str = Form(...)): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_subscriptions.manage") if response: return response validate_csrf(request, csrf_token) result = sync_consultant_billing_accounts(db, user_id=user.id) return RedirectResponse(url=f"/platform-billing/consultant-subscriptions?synced=created-{result['created']}-updated-{result['updated']}", status_code=303) finally: db.close() @router.post("/consultant-subscriptions/new") def consultant_subscription_create(request: Request, csrf_token: str = Form(...), account_id: int = Form(...), plan_id: int = Form(...), subscription_code: str = Form(...), start_date: str = Form(...), end_date: str = Form(""), billing_cycle: str = Form("Monthly"), amount: str = Form("0"), gst_rate: str = Form("18"), auto_generate_invoice: str = Form("on"), notes: str = Form("")): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_subscriptions.manage") if response: return response validate_csrf(request, csrf_token) create_platform_subscription( db, account_id=account_id, plan_id=plan_id, subscription_code=subscription_code, start_date=parse_date(start_date), end_date=parse_date(end_date), billing_cycle=billing_cycle, amount=amount, gst_rate=gst_rate, auto_generate_invoice=auto_generate_invoice == "on", notes=notes, user_id=user.id, ) return RedirectResponse(url="/platform-billing/consultant-subscriptions?created=1", status_code=303) finally: db.close() @router.get("/consultant-subscriptions/generate") def consultant_generate_form(request: Request, generated: int = 0, skipped: int = 0): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_billing.generate") if response: return response return _render( request, "modules/platform_billing/templates/platform_billing/consultants/generate.html", db, user, title="Generate Consultant Platform Bills", rows=list_consultant_billing_subscription_rows(db), generated=generated, skipped=skipped, errors=[], ) finally: db.close() @router.post("/consultant-subscriptions/generate") def consultant_generate_submit(request: Request, csrf_token: str = Form(...), subscription_ids: list[int] = Form(default=[]), period_from: str = Form(...), period_to: str = Form(...), invoice_date: str = Form(...), due_date: str = Form(""), tax_type: str = Form("CGST_SGST"), managed_client_rate: str = Form("0"), user_account_rate: str = Form("0"), include_zero_usage_lines: str = Form("off")): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_billing.generate") if response: return response validate_csrf(request, csrf_token) result = generate_consultant_subscription_invoices( db, subscription_ids=subscription_ids, period_from=parse_date(period_from), period_to=parse_date(period_to), invoice_date=parse_date(invoice_date), due_date=parse_date(due_date), tax_type=tax_type, managed_client_rate=managed_client_rate, user_account_rate=user_account_rate, include_zero_usage_lines=include_zero_usage_lines == "on", user_id=user.id, ) return _render( request, "modules/platform_billing/templates/platform_billing/consultants/generate.html", db, user, title="Generate Consultant Platform Bills", rows=list_consultant_billing_subscription_rows(db), generated=result["created"], skipped=result["skipped"], errors=result["errors"], invoices=result["invoices"], ) finally: db.close() @router.get("/subscriptions") def subscriptions_list(request: Request, q: str = ""): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_billing.view") if response: return response return _render(request, "modules/platform_billing/templates/platform_billing/subscriptions/list.html", db, user, title="Platform Subscriptions", rows=list_platform_subscriptions(db, q=q), q=q, can_manage=_has_perm(db, user, "platform_subscriptions.manage")) finally: db.close() @router.get("/subscriptions/new") def subscription_new(request: Request): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_subscriptions.manage") if response: return response return _render(request, "modules/platform_billing/templates/platform_billing/subscriptions/create.html", db, user, title="New Platform Subscription", accounts=list_platform_accounts(db), plans=list_platform_plans(db)) finally: db.close() @router.post("/subscriptions/new") def subscription_create(request: Request, csrf_token: str = Form(...), account_id: int = Form(...), plan_id: int = Form(...), subscription_code: str = Form(...), start_date: str = Form(...), end_date: str = Form(""), billing_cycle: str = Form(...), amount: str = Form("0"), gst_rate: str = Form("18"), auto_generate_invoice: str = Form("off"), notes: str = Form("")): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_subscriptions.manage") if response: return response validate_csrf(request, csrf_token) create_platform_subscription(db, account_id=account_id, plan_id=plan_id, subscription_code=subscription_code, start_date=parse_date(start_date), end_date=parse_date(end_date), billing_cycle=billing_cycle, amount=amount, gst_rate=gst_rate, auto_generate_invoice=auto_generate_invoice == "on", notes=notes, user_id=user.id) return RedirectResponse(url="/platform-billing/subscriptions", status_code=303) finally: db.close() @router.get("/invoices") def invoices_list(request: Request, q: str = ""): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_billing.view") if response: return response return _render(request, "modules/platform_billing/templates/platform_billing/invoices/list.html", db, user, title="Platform Invoices", rows=list_platform_invoices(db, q=q), q=q, can_create=_has_perm(db, user, "platform_billing.create")) finally: db.close() @router.get("/invoices/new") def invoice_new(request: Request): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_billing.create") if response: return response return _render(request, "modules/platform_billing/templates/platform_billing/invoices/create.html", db, user, title="New Platform Invoice", accounts=list_platform_accounts(db), subscriptions=list_platform_subscriptions(db)) finally: db.close() @router.post("/invoices/new") def invoice_create(request: Request, csrf_token: str = Form(...), account_id: int = Form(...), subscription_id: str = Form(""), invoice_no: str = Form(...), invoice_date: str = Form(...), due_date: str = Form(""), billing_period_from: str = Form(""), billing_period_to: str = Form(""), tax_type: str = Form(...), charge_type: str = Form("SUBSCRIPTION"), description: str = Form(...), quantity: str = Form("1"), rate: str = Form("0"), discount_amount: str = Form("0"), gst_rate: str = Form("18"), notes: str = Form("")): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_billing.create") if response: return response validate_csrf(request, csrf_token) invoice = create_platform_invoice( db, account_id=account_id, subscription_id=int(subscription_id) if subscription_id else None, invoice_no=invoice_no, invoice_date=parse_date(invoice_date), due_date=parse_date(due_date), billing_period_from=parse_date(billing_period_from), billing_period_to=parse_date(billing_period_to), tax_type=tax_type, line_items=[{"charge_type": charge_type, "description": description, "quantity": quantity, "rate": rate, "discount_amount": discount_amount, "gst_rate": gst_rate}], notes=notes, user_id=user.id, ) return RedirectResponse(url=f"/platform-billing/invoices/{invoice.id}", status_code=303) finally: db.close() @router.get("/invoices/{invoice_id}") def invoice_detail(request: Request, invoice_id: int): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_billing.view") if response: return response invoice = get_platform_invoice(db, invoice_id) if not invoice: return RedirectResponse(url="/platform-billing/invoices", status_code=303) return _render(request, "modules/platform_billing/templates/platform_billing/invoices/detail.html", db, user, title=f"Platform Invoice {invoice.invoice_no}", invoice=invoice, can_post=_has_perm(db, user, "platform_billing.post"), can_record_payment=_has_perm(db, user, "platform_billing.payment.create")) finally: db.close() @router.post("/invoices/{invoice_id}/post") def invoice_post(request: Request, invoice_id: int, csrf_token: str = Form(...)): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_billing.post") if response: return response validate_csrf(request, csrf_token) invoice = get_platform_invoice(db, invoice_id) if invoice and invoice.status == "DRAFT": post_platform_invoice(db, invoice, user.id) return RedirectResponse(url=f"/platform-billing/invoices/{invoice_id}", status_code=303) finally: db.close() @router.post("/invoices/{invoice_id}/payments") def payment_create(request: Request, invoice_id: int, csrf_token: str = Form(...), amount: str = Form(...), mode: str = Form("Bank"), reference_no: str = Form(""), notes: str = Form("")): db = CommonSessionLocal() try: user, response = _require_user(request, db, "platform_billing.payment.create") if response: return response validate_csrf(request, csrf_token) invoice = get_platform_invoice(db, invoice_id) if invoice: record_platform_payment(db, invoice=invoice, amount=amount, mode=mode, reference_no=reference_no, notes=notes, user_id=user.id) return RedirectResponse(url=f"/platform-billing/invoices/{invoice_id}", status_code=303) finally: db.close()