from __future__ import annotations from datetime import date from decimal import Decimal from fastapi import APIRouter, File, Form, Request, UploadFile from fastapi.responses import RedirectResponse, StreamingResponse from sqlalchemy import select 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.billing.models import BillingFeeGroup, BillingSettings from app.modules.billing.services import ( BILLING_MODES, FREQUENCIES, PAYMENT_MODES, TAX_TYPES, build_fee_structure_template, build_invoice_print_context, build_billing_report_summary, billing_financial_year, create_invoice, fee_group_already_billed, generate_draft_invoices_from_fee_groups, get_invoice, import_fee_structure_excel, issue_invoice, list_clients_for_billing, list_fee_groups, list_fee_groups_for_generation, list_invoices, list_payments, list_services_for_billing, parse_date, preview_invoice_number, record_invoice_payment, get_payment, ) 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.core.tenancy.models import Branch, Tenant from app.modules.core.tenancy.year_control import redirect_if_financial_year_locked, is_row_financial_year_locked router = APIRouter(prefix="/billing", tags=["billing-ui"]) def _base_ctx(request: Request, user, db, **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), "tax_types": TAX_TYPES, "billing_modes": BILLING_MODES, "frequencies": FREQUENCIES, "payment_modes": PAYMENT_MODES, } base.update(ctx) return base def _render(request: Request, template: str, db, user, **ctx): return templates.TemplateResponse(template, _base_ctx(request, user, db, **ctx)) def _redirect_denied(): return RedirectResponse(url="/system-settings", status_code=303) def _has_perm(db, user, code: str) -> bool: try: require_permission(db, user, code) return True except Exception: return False def _role_names(db, user) -> set[str]: return {str(r or "").strip() for r in get_user_roles(db, user.id)} def _can_manage_billing_settings(db, user) -> bool: roles = _role_names(db, user) return bool({"System Admin", "Firm Admin", "Partner"}.intersection(roles)) or _has_perm(db, user, "billing.edit") def _get_or_create_billing_settings(db, *, tenant_id: int, branch_id: int | None) -> BillingSettings: row = db.execute( select(BillingSettings).where(BillingSettings.tenant_id == tenant_id, BillingSettings.branch_id == branch_id) ).scalar_one_or_none() if row: return row row = BillingSettings(tenant_id=tenant_id, branch_id=branch_id) db.add(row) db.flush() return row def _decimal_form(value: str | None, default: str = "0.00") -> Decimal: try: return Decimal(str(value or default)).quantize(Decimal("0.01")) except Exception: return Decimal(default).quantize(Decimal("0.01")) def _int_form(value: str | int | None, default: int, minimum: int | None = None, maximum: int | None = None) -> int: try: parsed = int(value) except Exception: parsed = default if minimum is not None: parsed = max(minimum, parsed) if maximum is not None: parsed = min(maximum, parsed) return parsed def _billing_context_names(db, *, tenant_id: int, branch_id: int | None) -> tuple[str, str | None]: tenant = db.get(Tenant, tenant_id) branch = db.get(Branch, branch_id) if branch_id else None return (getattr(tenant, "name", None) or f"Audit Firm {tenant_id}", getattr(branch, "name", None) if branch else None) def _active_tenant_id(request: Request, user) -> int: return int(request.session.get("active_tenant_id") or request.session.get("selected_tenant_id") or request.session.get("tenant_id") or user.tenant_id) def _active_branch_id(request: Request, user, db) -> int | None: value = request.session.get("active_branch_id") if value in (None, "", 0, "0"): if _has_perm(db, user, "billing.cross_branch"): return None return int(getattr(user, "branch_id", 0) or 0) or None return int(value) def _active_financial_year(request: Request) -> str | None: value = request.session.get("active_financial_year") or getattr(request.state, "year_code", None) value = (value or "").strip() if not value or value.upper() == "ALL": return None return value def _period_start_for_fy(financial_year: str | None) -> date: try: start_year = int(str(financial_year or "").split("-")[0]) return date(start_year, 4, 1) except Exception: today = date.today() return date(today.year if today.month >= 4 else today.year - 1, 4, 1) def _period_end_for_fy(financial_year: str | None) -> date: start = _period_start_for_fy(financial_year) return date(start.year + 1, 3, 31) def _locked_partner_id(db, user) -> int | None: return int(user.id) if _has_perm(db, user, "billing.view_own") else None def _require_billing_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 @router.get("") def invoice_list(request: Request, q: str = ""): db = CommonSessionLocal() try: user, response = _require_billing_user(request, db, "billing.view") if response: return response tenant_id = _active_tenant_id(request, user) branch_id = _active_branch_id(request, user, db) partner_id = _locked_partner_id(db, user) financial_year = _active_financial_year(request) rows = list_invoices(db, tenant_id=tenant_id, branch_id=branch_id, partner_id=partner_id, financial_year=financial_year, q=q) report_summary = build_billing_report_summary(db, tenant_id=tenant_id, branch_id=branch_id, partner_id=partner_id, financial_year=financial_year) return _render( request, "modules/billing/templates/billing/list.html", db, user, title="Billing - Invoices", q=q, active_financial_year=financial_year, rows=rows, report_summary=report_summary, can_create=_has_perm(db, user, "billing.create"), can_import_fee_structure=_has_perm(db, user, "billing_fee_structure.import"), can_generate=_has_perm(db, user, "billing_invoice.generate"), can_view_fee_structure=_has_perm(db, user, "billing_fee_structure.view"), can_record_payment=_has_perm(db, user, "billing_payment.record") or _has_perm(db, user, "billing.create"), ) finally: db.close() @router.get("/payments") def payment_list(request: Request, q: str = ""): db = CommonSessionLocal() try: user, response = _require_billing_user(request, db, "billing.view") if response: return response financial_year = _active_financial_year(request) rows = list_payments( db, tenant_id=_active_tenant_id(request, user), branch_id=_active_branch_id(request, user, db), partner_id=_locked_partner_id(db, user), financial_year=financial_year, q=q, ) return _render(request, "modules/billing/templates/billing/payments/list.html", db, user, title="Payments & Receipts", rows=rows, q=q, active_financial_year=financial_year) finally: db.close() @router.get("/payments/{payment_id}/receipt") def payment_receipt_print(request: Request, payment_id: int): db = CommonSessionLocal() try: user, response = _require_billing_user(request, db, "billing.view") if response: return response payment = get_payment(db, payment_id=payment_id, tenant_id=_active_tenant_id(request, user), partner_id=_locked_partner_id(db, user), financial_year=_active_financial_year(request)) if not payment: return _redirect_denied() invoice_ctx = build_invoice_print_context(db, payment.invoice) return _render(request, "modules/billing/templates/billing/payments/receipt_print.html", db, user, title=f"Receipt {payment.receipt_no}", payment=payment, invoice=payment.invoice, invoice_ctx=invoice_ctx) finally: db.close() @router.get("/settings") def billing_settings_page(request: Request, branch_scope: str = "active"): db = CommonSessionLocal() try: user, response = _require_billing_user(request, db, "billing.view") if response: return response tenant_id = _active_tenant_id(request, user) active_branch_id = _active_branch_id(request, user, db) branch_id = None if branch_scope == "firm" and _has_perm(db, user, "billing.cross_branch") else active_branch_id settings = _get_or_create_billing_settings(db, tenant_id=tenant_id, branch_id=branch_id) tenant_name, branch_name = _billing_context_names(db, tenant_id=tenant_id, branch_id=branch_id) return _render( request, "modules/billing/templates/billing/settings.html", db, user, title="Billing Settings", settings=settings, preview_invoice_no=preview_invoice_number(settings, branch_id=branch_id, financial_year=_active_financial_year(request)), tenant_name=tenant_name, branch_name=branch_name, branch_scope="firm" if branch_id is None else "active", can_edit_settings=_can_manage_billing_settings(db, user), ) finally: db.close() @router.post("/settings") def billing_settings_submit( request: Request, branch_scope: str = Form("active"), legal_name: str | None = Form(None), gstin: str | None = Form(None), pan: str | None = Form(None), state_code: str | None = Form(None), billing_address: str | None = Form(None), contact_email: str | None = Form(None), contact_mobile: str | None = Form(None), website_url: str | None = Form(None), invoice_title: str | None = Form(None), invoice_prefix: str = Form("INV"), invoice_number_format: str | None = Form("{prefix}/{fy}/{number}"), next_invoice_no: int = Form(1), padding: int = Form(4), default_due_days: int = Form(15), default_gst_rate: str = Form("18.00"), default_tax_type: str = Form("CGST_SGST"), default_sac_code: str | None = Form(None), bank_name: str | None = Form(None), bank_account_name: str | None = Form(None), bank_account_number: str | None = Form(None), bank_ifsc: str | None = Form(None), upi_id: str | None = Form(None), bank_details: str | None = Form(None), terms: str | None = Form(None), footer_note: str | None = Form(None), declaration: str | None = Form(None), authorised_signatory_name: str | None = Form(None), payumoney_enabled: str | None = Form(None), payumoney_mode: str = Form("TEST"), payumoney_merchant_key: str | None = Form(None), payumoney_merchant_salt: str | None = Form(None), payumoney_merchant_id: str | None = Form(None), payumoney_product_info: str | None = Form(None), cashfree_enabled: str | None = Form(None), cashfree_mode: str = Form("TEST"), cashfree_client_id: str | None = Form(None), cashfree_client_secret: str | None = Form(None), cashfree_api_version: str | None = Form("2023-08-01"), cashfree_order_note: str | None = Form(None), csrf_token: str = Form(...), ): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user, response = _require_billing_user(request, db, "billing.view") if response: return response if not _can_manage_billing_settings(db, user): return _redirect_denied() tenant_id = _active_tenant_id(request, user) active_branch_id = _active_branch_id(request, user, db) branch_id = None if branch_scope == "firm" and _has_perm(db, user, "billing.cross_branch") else active_branch_id settings = _get_or_create_billing_settings(db, tenant_id=tenant_id, branch_id=branch_id) settings.legal_name = (legal_name or "").strip() or None settings.gstin = (gstin or "").strip().upper() or None settings.pan = (pan or "").strip().upper() or None settings.state_code = (state_code or "").strip()[:2] or None settings.billing_address = (billing_address or "").strip() or None settings.contact_email = (contact_email or "").strip() or None settings.contact_mobile = (contact_mobile or "").strip() or None settings.website_url = (website_url or "").strip() or None settings.invoice_title = (invoice_title or "").strip() or None settings.invoice_prefix = (invoice_prefix or "INV").strip().upper()[:40] or "INV" settings.invoice_number_format = (invoice_number_format or "{prefix}/{fy}/{number}").strip()[:120] or "{prefix}/{fy}/{number}" settings.next_invoice_no = _int_form(next_invoice_no, 1, minimum=1) settings.padding = _int_form(padding, 4, minimum=1, maximum=10) settings.default_due_days = _int_form(default_due_days, 15, minimum=0, maximum=365) settings.default_gst_rate = _decimal_form(default_gst_rate, "18.00") settings.default_tax_type = default_tax_type if default_tax_type in TAX_TYPES else "CGST_SGST" settings.default_sac_code = (default_sac_code or "").strip()[:20] or None settings.bank_name = (bank_name or "").strip() or None settings.bank_account_name = (bank_account_name or "").strip() or None settings.bank_account_number = (bank_account_number or "").strip() or None settings.bank_ifsc = (bank_ifsc or "").strip().upper() or None settings.upi_id = (upi_id or "").strip() or None settings.bank_details = (bank_details or "").strip() or None settings.terms = (terms or "").strip() or None settings.footer_note = (footer_note or "").strip() or None settings.declaration = (declaration or "").strip() or None settings.authorised_signatory_name = (authorised_signatory_name or "").strip() or None settings.payumoney_enabled = bool(payumoney_enabled) settings.payumoney_mode = (payumoney_mode or "TEST").strip().upper() if (payumoney_mode or "TEST").strip().upper() in {"TEST", "LIVE"} else "TEST" settings.payumoney_merchant_key = (payumoney_merchant_key or "").strip() or None settings.payumoney_merchant_salt = (payumoney_merchant_salt or "").strip() or None settings.payumoney_merchant_id = (payumoney_merchant_id or "").strip() or None settings.payumoney_product_info = (payumoney_product_info or "").strip() or None db.commit() suffix = "?branch_scope=firm" if branch_id is None else "" return RedirectResponse(url=f"/billing/settings{suffix}", status_code=303) except Exception: db.rollback() raise finally: db.close() @router.get("/new") def invoice_create_page(request: Request): db = CommonSessionLocal() try: user, response = _require_billing_user(request, db, "billing.create") if response: return response tenant_id = _active_tenant_id(request, user) branch_id = _active_branch_id(request, user, db) partner_id = _locked_partner_id(db, user) clients = list_clients_for_billing(db, tenant_id=tenant_id, branch_id=branch_id, partner_id=partner_id) services = list_services_for_billing(db) settings = _get_or_create_billing_settings(db, tenant_id=tenant_id, branch_id=branch_id) return _render( request, "modules/billing/templates/billing/create.html", db, user, title="Create Invoice", active_financial_year=financial_year, default_billing_period_from=_period_start_for_fy(_active_financial_year(request)).isoformat(), default_billing_period_to=_period_end_for_fy(_active_financial_year(request)).isoformat(), clients=clients, services=services, settings=settings, today=date.today().isoformat(), ) finally: db.close() @router.post("/new") def invoice_create_submit( request: Request, client_id: int = Form(...), invoice_date: str = Form(...), due_date: str | None = Form(None), billing_period_from: str | None = Form(None), billing_period_to: str | None = Form(None), tax_type: str = Form("CGST_SGST"), place_of_supply: str | None = Form(None), client_state_code: str | None = Form(None), reverse_charge: str | None = Form(None), notes: str | None = Form(None), terms: str | None = Form(None), line_description: list[str] = Form(default=[]), line_service_id: list[str] = Form(default=[]), line_quantity: list[str] = Form(default=[]), line_rate: list[str] = Form(default=[]), line_discount: list[str] = Form(default=[]), line_gst_rate: list[str] = Form(default=[]), line_sac_code: list[str] = Form(default=[]), csrf_token: str = Form(...), ): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user, response = _require_billing_user(request, db, "billing.create") if response: return response tenant_id = _active_tenant_id(request, user) branch_id = _active_branch_id(request, user, db) partner_id = _locked_partner_id(db, user) financial_year = _active_financial_year(request) locked_response = redirect_if_financial_year_locked(db, tenant_id=tenant_id, year_code=financial_year, redirect_url=f"/billing?financial_year={financial_year or ''}") if locked_response: return locked_response allowed_clients = {c.id for c in list_clients_for_billing(db, tenant_id=tenant_id, branch_id=branch_id, partner_id=partner_id)} if client_id not in allowed_clients: return _redirect_denied() raw_lines = [] max_len = max(len(line_description), len(line_service_id), len(line_quantity), len(line_rate), len(line_discount), len(line_gst_rate), len(line_sac_code), 0) for idx in range(max_len): raw_lines.append({ "description": line_description[idx] if idx < len(line_description) else "", "service_id": line_service_id[idx] if idx < len(line_service_id) else "", "quantity": line_quantity[idx] if idx < len(line_quantity) else "1", "rate": line_rate[idx] if idx < len(line_rate) else "0", "discount_amount": line_discount[idx] if idx < len(line_discount) else "0", "gst_rate": line_gst_rate[idx] if idx < len(line_gst_rate) else "18", "sac_code": line_sac_code[idx] if idx < len(line_sac_code) else "", }) invoice = create_invoice( db, tenant_id=tenant_id, branch_id=branch_id, client_id=client_id, invoice_date=parse_date(invoice_date) or date.today(), 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, notes=notes, terms=terms, place_of_supply=place_of_supply, client_state_code=client_state_code, reverse_charge=(reverse_charge == "yes"), created_by_user_id=user.id, raw_lines=raw_lines, financial_year=_active_financial_year(request), ) db.commit() return RedirectResponse(url=f"/billing/{invoice.id}", status_code=303) except ValueError: db.rollback() return RedirectResponse(url="/billing/new", status_code=303) finally: db.close() @router.get("/fee-structures/list") def fee_structure_list(request: Request, q: str = ""): db = CommonSessionLocal() try: user, response = _require_billing_user(request, db, "billing_fee_structure.view") if response: return response rows = list_fee_groups( db, tenant_id=_active_tenant_id(request, user), branch_id=_active_branch_id(request, user, db), partner_id=_locked_partner_id(db, user), q=q, ) return _render( request, "modules/billing/templates/billing/fee_structures/list.html", db, user, title="Fee Structure", q=q, active_financial_year=financial_year, rows=rows, can_import=_has_perm(db, user, "billing_fee_structure.import"), ) finally: db.close() @router.get("/fee-structures/import") def fee_structure_import_page(request: Request): db = CommonSessionLocal() try: user, response = _require_billing_user(request, db, "billing_fee_structure.import") if response: return response return _render(request, "modules/billing/templates/billing/fee_structures/import.html", db, user, title="Import Fee Structure", result=None) finally: db.close() @router.get("/fee-structures/template") def fee_structure_template_download(request: Request): db = CommonSessionLocal() try: user, response = _require_billing_user(request, db, "billing_fee_structure.import") if response: return response data = build_fee_structure_template() return StreamingResponse( iter([data]), media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": "attachment; filename=billing_fee_structure_template.xlsx"}, ) finally: db.close() @router.post("/fee-structures/import") async def fee_structure_import_submit(request: Request, import_file: UploadFile = File(...), csrf_token: str = Form(...)): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user, response = _require_billing_user(request, db, "billing_fee_structure.import") if response: return response filename = (import_file.filename or "").lower() if not filename.endswith((".xlsx", ".xlsm")): result = {"success": False, "created": 0, "updated": 0, "errors": ["Please upload an .xlsx file."]} else: content = await import_file.read() if len(content) > 5 * 1024 * 1024: result = {"success": False, "created": 0, "updated": 0, "errors": ["File size must be 5 MB or less."]} else: result = import_fee_structure_excel( db, tenant_id=_active_tenant_id(request, user), branch_id=_active_branch_id(request, user, db), created_by_user_id=user.id, file_bytes=content, ) return _render(request, "modules/billing/templates/billing/fee_structures/import.html", db, user, title="Import Fee Structure", result=result) finally: db.close() @router.get("/generate") def generate_invoices_page( request: Request, frequency: str = "Monthly", billing_period_from: str | None = None, billing_period_to: str | None = None, auto_generate_only: str = "yes", q: str = "", ): db = CommonSessionLocal() try: user, response = _require_billing_user(request, db, "billing_invoice.generate") if response: return response tenant_id = _active_tenant_id(request, user) branch_id = _active_branch_id(request, user, db) partner_id = _locked_partner_id(db, user) financial_year = _active_financial_year(request) locked_response = redirect_if_financial_year_locked(db, tenant_id=tenant_id, year_code=financial_year, redirect_url=f"/billing?financial_year={financial_year or ''}") if locked_response: return locked_response period_from = parse_date(billing_period_from) or _period_start_for_fy(financial_year) period_to = parse_date(billing_period_to) or _period_end_for_fy(financial_year) rows = list_fee_groups_for_generation( db, tenant_id=tenant_id, branch_id=branch_id, partner_id=partner_id, frequency=frequency or None, auto_generate_only=(auto_generate_only != "no"), q=q, ) duplicate_map = { row.id: fee_group_already_billed(db, tenant_id=tenant_id, fee_group_id=row.id, period_from=period_from, period_to=period_to) for row in rows } return _render( request, "modules/billing/templates/billing/generate.html", db, user, title="Generate Draft Invoices", rows=rows, duplicate_map=duplicate_map, frequencies=FREQUENCIES, frequency=frequency, billing_period_from=period_from.isoformat(), billing_period_to=period_to.isoformat(), auto_generate_only=auto_generate_only, q=q, active_financial_year=financial_year, result=None, ) finally: db.close() @router.post("/generate") def generate_invoices_submit( request: Request, frequency: str = Form("Monthly"), billing_period_from: str = Form(...), billing_period_to: str = Form(...), auto_generate_only: str = Form("yes"), q: str = Form(""), fee_group_ids: list[int] = Form(default=[]), skip_duplicates: str = Form("yes"), csrf_token: str = Form(...), ): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user, response = _require_billing_user(request, db, "billing_invoice.generate") if response: return response tenant_id = _active_tenant_id(request, user) branch_id = _active_branch_id(request, user, db) partner_id = _locked_partner_id(db, user) financial_year = _active_financial_year(request) locked_response = redirect_if_financial_year_locked(db, tenant_id=tenant_id, year_code=financial_year, redirect_url=f"/billing/generate?year_locked=1") if locked_response: return locked_response period_from = parse_date(billing_period_from) period_to = parse_date(billing_period_to) if financial_year and period_from and billing_financial_year(billing_period_from=period_from) != financial_year: result = {"created": [], "skipped": [], "errors": [f"Billing period must fall within active FY {financial_year}."], "batch": None} elif period_from is None or period_to is None: result = {"created": [], "skipped": [], "errors": ["Billing period From and To are required."], "batch": None} else: result = generate_draft_invoices_from_fee_groups( db, tenant_id=tenant_id, branch_id=branch_id, partner_id=partner_id, generated_by_user_id=user.id, billing_period_from=period_from, billing_period_to=period_to, frequency=frequency or None, fee_group_ids=fee_group_ids, skip_duplicates=(skip_duplicates != "no"), ) db.commit() rows = list_fee_groups_for_generation( db, tenant_id=tenant_id, branch_id=branch_id, partner_id=partner_id, frequency=frequency or None, auto_generate_only=(auto_generate_only != "no"), q=q, ) duplicate_map = { row.id: fee_group_already_billed(db, tenant_id=tenant_id, fee_group_id=row.id, period_from=period_from or date.today(), period_to=period_to or date.today()) for row in rows } return _render( request, "modules/billing/templates/billing/generate.html", db, user, title="Generate Draft Invoices", rows=rows, duplicate_map=duplicate_map, frequencies=FREQUENCIES, frequency=frequency, billing_period_from=(period_from or date.today()).isoformat(), billing_period_to=(period_to or date.today()).isoformat(), auto_generate_only=auto_generate_only, q=q, active_financial_year=financial_year, result=result, ) except ValueError as exc: db.rollback() rows = [] result = {"created": [], "skipped": [], "errors": [str(exc)], "batch": None} return _render( request, "modules/billing/templates/billing/generate.html", db, user, title="Generate Draft Invoices", rows=rows, duplicate_map={}, frequencies=FREQUENCIES, frequency=frequency, billing_period_from=billing_period_from, billing_period_to=billing_period_to, auto_generate_only=auto_generate_only, q=q, active_financial_year=_active_financial_year(request), result=result, ) except Exception as exc: db.rollback() result = {"created": [], "skipped": [], "errors": [f"Generation failed: {exc}"], "batch": None} return _render( request, "modules/billing/templates/billing/generate.html", db, user, title="Generate Draft Invoices", rows=[], duplicate_map={}, frequencies=FREQUENCIES, frequency=frequency, billing_period_from=billing_period_from, billing_period_to=billing_period_to, auto_generate_only=auto_generate_only, q=q, active_financial_year=_active_financial_year(request), result=result, ) finally: db.close() @router.get("/{invoice_id}/payments/new") def invoice_payment_page(request: Request, invoice_id: int): db = CommonSessionLocal() try: user, response = _require_billing_user(request, db, "billing.view") if response: return response invoice = get_invoice(db, invoice_id=invoice_id, tenant_id=_active_tenant_id(request, user), partner_id=_locked_partner_id(db, user), financial_year=_active_financial_year(request)) if not invoice: return _redirect_denied() if invoice.status in {"DRAFT", "CANCELLED"}: return RedirectResponse(url=f"/billing/{invoice.id}", status_code=303) can_record = _has_perm(db, user, "billing_payment.record") or _has_perm(db, user, "billing.create") if not can_record: return _redirect_denied() return _render(request, "modules/billing/templates/billing/payments/new.html", db, user, title=f"Record Payment - {invoice.invoice_no}", invoice=invoice, today=date.today().isoformat()) finally: db.close() @router.post("/{invoice_id}/payments/new") def invoice_payment_submit( request: Request, invoice_id: int, payment_date: str = Form(...), amount_received: str = Form("0.00"), tds_deducted: str = Form("0.00"), bank_charges: str = Form("0.00"), mode: str = Form("BANK"), reference_no: str | None = Form(None), remarks: str | None = Form(None), csrf_token: str = Form(...), ): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user, response = _require_billing_user(request, db, "billing.view") if response: return response can_record = _has_perm(db, user, "billing_payment.record") or _has_perm(db, user, "billing.create") if not can_record: return _redirect_denied() invoice = get_invoice(db, invoice_id=invoice_id, tenant_id=_active_tenant_id(request, user), partner_id=_locked_partner_id(db, user), financial_year=_active_financial_year(request)) if not invoice: return _redirect_denied() if is_row_financial_year_locked(db, invoice): return RedirectResponse(url=f"/billing/{invoice.id}?year_locked=1", status_code=303) payment = record_invoice_payment( db, invoice=invoice, payment_date=parse_date(payment_date) or date.today(), amount_received=_decimal_form(amount_received, "0.00"), tds_deducted=_decimal_form(tds_deducted, "0.00"), bank_charges=_decimal_form(bank_charges, "0.00"), mode=mode, reference_no=reference_no, remarks=remarks, created_by_user_id=user.id, ) db.commit() return RedirectResponse(url=f"/billing/payments/{payment.id}/receipt", status_code=303) except ValueError: db.rollback() return RedirectResponse(url=f"/billing/{invoice_id}", status_code=303) except Exception: db.rollback() raise finally: db.close() @router.get("/{invoice_id}/print") def invoice_print(request: Request, invoice_id: int): db = CommonSessionLocal() try: user, response = _require_billing_user(request, db, "billing.view") if response: return response invoice = get_invoice(db, invoice_id=invoice_id, tenant_id=_active_tenant_id(request, user), partner_id=_locked_partner_id(db, user), financial_year=_active_financial_year(request)) if not invoice: return _redirect_denied() invoice_ctx = build_invoice_print_context(db, invoice) return _render(request, "modules/billing/templates/billing/invoice_print.html", db, user, title=f"Print Invoice {invoice.invoice_no}", invoice=invoice, invoice_ctx=invoice_ctx) finally: db.close() @router.post("/{invoice_id}/issue") def invoice_issue_submit(request: Request, invoice_id: int, csrf_token: str = Form(...)): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user, response = _require_billing_user(request, db, "billing.create") if response: return response invoice = get_invoice(db, invoice_id=invoice_id, tenant_id=_active_tenant_id(request, user), partner_id=_locked_partner_id(db, user), financial_year=_active_financial_year(request)) if not invoice: return _redirect_denied() if is_row_financial_year_locked(db, invoice): return RedirectResponse(url=f"/billing/{invoice.id}?year_locked=1", status_code=303) issue_invoice(db, invoice, user_id=user.id) db.commit() return RedirectResponse(url=f"/billing/{invoice.id}", status_code=303) except Exception: db.rollback() raise finally: db.close() @router.get("/{invoice_id}") def invoice_detail(request: Request, invoice_id: int): db = CommonSessionLocal() try: user, response = _require_billing_user(request, db, "billing.view") if response: return response invoice = get_invoice(db, invoice_id=invoice_id, tenant_id=_active_tenant_id(request, user), partner_id=_locked_partner_id(db, user), financial_year=_active_financial_year(request)) if not invoice: return _redirect_denied() invoice_ctx = build_invoice_print_context(db, invoice) return _render(request, "modules/billing/templates/billing/detail.html", db, user, title=f"Invoice {invoice.invoice_no}", invoice=invoice, invoice_ctx=invoice_ctx, can_record_payment=_has_perm(db, user, "billing_payment.record") or _has_perm(db, user, "billing.create")) finally: db.close()