from __future__ import annotations from datetime import date, datetime, timezone from decimal import Decimal, InvalidOperation, ROUND_HALF_UP from io import BytesIO from typing import Any from openpyxl import Workbook, load_workbook from sqlalchemy import and_, or_, select from sqlalchemy.orm import Session, selectinload from app.modules.billing.models import ( BillingFeeGroup, BillingFeeGroupService, BillingInvoice, BillingInvoiceLine, BillingInvoiceGenerationBatch, BillingPayment, BillingSettings, BillingOnlinePaymentTransaction, ) from app.modules.clients.models import Client from app.modules.email_integration.event_service import send_invoice_issued_email, send_payment_receipt_email from app.modules.services.models import ClientServiceSubscription, ServiceCatalogue TAX_TYPES = ["CGST_SGST", "IGST", "NO_GST"] BILLING_MODES = ["PACKAGE", "SERVICE_WISE"] FREQUENCIES = ["Monthly", "Quarterly", "Half-Yearly", "Yearly", "One-time"] INVOICE_STATUSES = ["DRAFT", "ISSUED", "PARTLY_PAID", "PAID", "OVERDUE", "CANCELLED", "WRITTEN_OFF"] PAYMENT_MODES = ["CASH", "BANK", "UPI", "CHEQUE", "ONLINE", "ADJUSTMENT"] PAYMENT_STATUSES = ["RECEIVED", "CANCELLED", "REFUNDED"] def money(value: Any) -> Decimal: try: if value in (None, ""): return Decimal("0.00") return Decimal(str(value)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) except (InvalidOperation, ValueError): return Decimal("0.00") def parse_date(value: Any) -> date | None: if value in (None, ""): return None if isinstance(value, datetime): return value.date() if isinstance(value, date): return value value = str(value).strip() if not value: return None return date.fromisoformat(value) def normalize_code(value: Any) -> str: return str(value or "").strip().upper().replace(" ", "-") def normalize_yes_no(value: Any) -> bool: return str(value or "").strip().lower() in {"yes", "y", "true", "1", "active"} def get_or_create_settings(db: Session, *, tenant_id: int, branch_id: int | None = 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 _financial_year_label(value: date | None = None) -> str: value = value or date.today() start_year = value.year if value.month >= 4 else value.year - 1 return f"{start_year}-{str(start_year + 1)[-2:]}" def billing_financial_year(*, billing_period_from: date | None = None, invoice_date: date | None = None, fallback: date | None = None) -> str: return _financial_year_label(billing_period_from or invoice_date or fallback or date.today()) def next_invoice_number(db: Session, *, tenant_id: int, branch_id: int | None = None, financial_year: str | None = None) -> str: settings = get_or_create_settings(db, tenant_id=tenant_id, branch_id=branch_id) serial = str(settings.next_invoice_no).zfill(settings.padding or 4) fy = financial_year or _financial_year_label() fmt = (getattr(settings, "invoice_number_format", None) or "{prefix}/{fy}/{number}").strip() try: number = fmt.format(prefix=settings.invoice_prefix or "INV", fy=fy, number=serial, branch_id=branch_id or "") except Exception: number = f"{settings.invoice_prefix or 'INV'}/{fy}/{serial}" settings.next_invoice_no += 1 settings.updated_at_utc = datetime.now(timezone.utc) return number def preview_invoice_number(settings: BillingSettings, *, branch_id: int | None = None, financial_year: str | None = None) -> str: serial = str((settings.next_invoice_no or 1)).zfill(settings.padding or 4) fy = financial_year or _financial_year_label() fmt = (getattr(settings, "invoice_number_format", None) or "{prefix}/{fy}/{number}").strip() try: return fmt.format(prefix=settings.invoice_prefix or "INV", fy=fy, number=serial, branch_id=branch_id or "") except Exception: return f"{settings.invoice_prefix or 'INV'}/{fy}/{serial}" _ONES = ["", "One", "Two", "Three", "Four", "Five", "Six", "Seven", "Eight", "Nine", "Ten", "Eleven", "Twelve", "Thirteen", "Fourteen", "Fifteen", "Sixteen", "Seventeen", "Eighteen", "Nineteen"] _TENS = ["", "", "Twenty", "Thirty", "Forty", "Fifty", "Sixty", "Seventy", "Eighty", "Ninety"] def _words_below_1000(n: int) -> str: parts: list[str] = [] if n >= 100: parts.append(_ONES[n // 100] + " Hundred") n %= 100 if n >= 20: parts.append(_TENS[n // 10]) n %= 10 if n > 0: parts.append(_ONES[n]) return " ".join(parts) def amount_to_indian_words(value: Any) -> str: amount = money(value) rupees = int(amount) paise = int((amount - Decimal(rupees)) * 100) if rupees == 0: words = "Zero" else: parts: list[str] = [] crore, rupees = divmod(rupees, 10000000) lakh, rupees = divmod(rupees, 100000) thousand, rupees = divmod(rupees, 1000) if crore: parts.append(_words_below_1000(crore) + " Crore") if lakh: parts.append(_words_below_1000(lakh) + " Lakh") if thousand: parts.append(_words_below_1000(thousand) + " Thousand") if rupees: parts.append(_words_below_1000(rupees)) words = " ".join(parts) result = f"Rupees {words} Only" if paise: result = f"Rupees {words} and Paise {_words_below_1000(paise)} Only" return result def _client_address_snapshot(client: Client) -> str | None: parts = [ getattr(client, "address_line_1", None), getattr(client, "address_line_2", None), getattr(client, "city", None), getattr(client, "state", None), getattr(client, "pincode", None), getattr(client, "country", None), ] return ", ".join([str(p).strip() for p in parts if str(p or "").strip()]) or None def _state_code_from_gstin(gstin: str | None) -> str | None: value = str(gstin or "").strip() if len(value) >= 2 and value[:2].isdigit(): return value[:2] return None def get_effective_billing_settings(db: Session, *, tenant_id: int, branch_id: int | None = None) -> BillingSettings: if branch_id: row = db.execute(select(BillingSettings).where(BillingSettings.tenant_id == tenant_id, BillingSettings.branch_id == branch_id)).scalar_one_or_none() if row: return row return get_or_create_settings(db, tenant_id=tenant_id, branch_id=None) def build_invoice_print_context(db: Session, invoice: BillingInvoice) -> dict[str, Any]: settings = get_effective_billing_settings(db, tenant_id=invoice.tenant_id, branch_id=invoice.branch_id) return { "invoice": invoice, "settings": settings, "invoice_title": invoice.invoice_title or settings.invoice_title or "Tax Invoice", "firm_name": settings.legal_name or "Audit Firm", "firm_address": settings.billing_address, "firm_gstin": settings.gstin, "firm_pan": settings.pan, "firm_state_code": settings.state_code, "firm_contact_email": settings.contact_email, "firm_contact_mobile": settings.contact_mobile, "firm_website": settings.website_url, "bank_name": settings.bank_name, "bank_account_name": settings.bank_account_name, "bank_account_number": settings.bank_account_number, "bank_ifsc": settings.bank_ifsc, "upi_id": settings.upi_id, "bank_details": settings.bank_details, "declaration": invoice.notes or settings.declaration, "terms": invoice.terms or settings.terms, "footer_note": settings.footer_note, "authorised_signatory_name": settings.authorised_signatory_name, } def calculate_line(*, quantity: Decimal, rate: Decimal, discount: Decimal, gst_rate: Decimal, tax_type: str) -> dict[str, Decimal]: taxable = (quantity * rate - discount).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) if taxable < 0: taxable = Decimal("0.00") cgst = sgst = igst = Decimal("0.00") if tax_type == "IGST": igst = (taxable * gst_rate / Decimal("100.00")).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) elif tax_type == "CGST_SGST": half = (taxable * gst_rate / Decimal("200.00")).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) cgst = half sgst = half total = taxable + cgst + sgst + igst return {"taxable": taxable, "cgst": cgst, "sgst": sgst, "igst": igst, "total": total} def recalc_invoice(invoice: BillingInvoice) -> None: subtotal = Decimal("0.00") discount = Decimal("0.00") taxable = Decimal("0.00") cgst = Decimal("0.00") sgst = Decimal("0.00") igst = Decimal("0.00") total = Decimal("0.00") for line in invoice.lines: subtotal += money(line.quantity) * money(line.rate) discount += money(line.discount_amount) taxable += money(line.taxable_amount) cgst += money(line.cgst_amount) sgst += money(line.sgst_amount) igst += money(line.igst_amount) total += money(line.line_total) invoice.subtotal = money(subtotal) invoice.discount_amount = money(discount) invoice.taxable_amount = money(taxable) invoice.cgst_amount = money(cgst) invoice.sgst_amount = money(sgst) invoice.igst_amount = money(igst) invoice.total_amount = money(total + money(invoice.round_off)) update_invoice_payment_totals(invoice) def update_invoice_payment_totals(invoice: BillingInvoice) -> None: paid = Decimal("0.00") tds = Decimal("0.00") charges = Decimal("0.00") for payment in getattr(invoice, "payments", []) or []: if getattr(payment, "status", "RECEIVED") != "CANCELLED": paid += money(payment.amount_received) tds += money(payment.tds_deducted) charges += money(payment.bank_charges) invoice.amount_received = money(paid) invoice.tds_deducted = money(tds) invoice.bank_charges = money(charges) invoice.balance_amount = money(money(invoice.total_amount) - paid - tds) if invoice.balance_amount < Decimal("0.00"): invoice.balance_amount = Decimal("0.00") if invoice.status not in {"DRAFT", "CANCELLED", "WRITTEN_OFF"}: if invoice.balance_amount <= Decimal("0.00") and money(invoice.total_amount) > Decimal("0.00"): invoice.status = "PAID" elif paid > Decimal("0.00") or tds > Decimal("0.00"): invoice.status = "PARTLY_PAID" elif invoice.status == "PAID": invoice.status = "ISSUED" def next_receipt_number(db: Session, *, tenant_id: int, branch_id: int | None = None, financial_year: str | None = None) -> str: fy = financial_year or _financial_year_label() prefix = "RCT" like_prefix = f"{prefix}/{fy}/%" last = db.execute( select(BillingPayment.receipt_no) .where(BillingPayment.tenant_id == tenant_id, BillingPayment.receipt_no.ilike(like_prefix)) .order_by(BillingPayment.id.desc()) ).scalar_one_or_none() next_no = 1 if last: try: next_no = int(str(last).split("/")[-1]) + 1 except Exception: next_no = 1 return f"{prefix}/{fy}/{str(next_no).zfill(4)}" def record_invoice_payment( db: Session, *, invoice: BillingInvoice, payment_date: date, amount_received: Decimal, tds_deducted: Decimal = Decimal("0.00"), bank_charges: Decimal = Decimal("0.00"), mode: str = "BANK", reference_no: str | None = None, remarks: str | None = None, created_by_user_id: int | None = None, payment_gateway: str | None = None, gateway_transaction_id: str | None = None, ) -> BillingPayment: if invoice.status in {"DRAFT", "CANCELLED"}: raise ValueError("Payment can be recorded only after invoice is issued.") mode = mode if mode in PAYMENT_MODES else "BANK" payment = BillingPayment( tenant_id=invoice.tenant_id, branch_id=invoice.branch_id, invoice_id=invoice.id, client_id=invoice.client_id, receipt_no=next_receipt_number(db, tenant_id=invoice.tenant_id, branch_id=invoice.branch_id, financial_year=getattr(invoice, "financial_year", None)), receipt_date=payment_date, payment_date=payment_date, financial_year=getattr(invoice, "financial_year", None) or billing_financial_year(invoice_date=payment_date), amount_received=money(amount_received), tds_deducted=money(tds_deducted), bank_charges=money(bank_charges), mode=mode, reference_no=(reference_no or "").strip() or None, payment_gateway=(payment_gateway or "").strip() or None, gateway_transaction_id=(gateway_transaction_id or "").strip() or None, remarks=(remarks or "").strip() or None, created_by_user_id=created_by_user_id, status="RECEIVED", ) db.add(payment) db.flush() if payment not in invoice.payments: invoice.payments.append(payment) update_invoice_payment_totals(invoice) invoice.updated_at_utc = datetime.now(timezone.utc) db.flush() try: send_payment_receipt_email(db, payment) except Exception: # Email failure should not block payment posting or receipt generation. pass return payment def list_payments(db: Session, *, tenant_id: int, branch_id: int | None = None, partner_id: int | None = None, financial_year: str | None = None, q: str = ""): stmt = select(BillingPayment).options(selectinload(BillingPayment.invoice), selectinload(BillingPayment.client)).where(BillingPayment.tenant_id == tenant_id) if branch_id: stmt = stmt.where(BillingPayment.branch_id == branch_id) if partner_id: stmt = stmt.join(Client, Client.id == BillingPayment.client_id).where(Client.partner_id == partner_id) if financial_year and financial_year.upper() != "ALL": stmt = stmt.where(BillingPayment.financial_year == financial_year) if q.strip(): term = f"%{q.strip()}%" stmt = stmt.join(BillingInvoice, BillingInvoice.id == BillingPayment.invoice_id).join(Client, Client.id == BillingPayment.client_id).where(or_(BillingPayment.receipt_no.ilike(term), BillingInvoice.invoice_no.ilike(term), Client.client_name.ilike(term), Client.client_code.ilike(term))) return db.execute(stmt.order_by(BillingPayment.payment_date.desc(), BillingPayment.id.desc())).scalars().unique().all() def get_payment(db: Session, *, payment_id: int, tenant_id: int, partner_id: int | None = None, financial_year: str | None = None) -> BillingPayment | None: stmt = select(BillingPayment).options(selectinload(BillingPayment.invoice).selectinload(BillingInvoice.lines), selectinload(BillingPayment.client)).where(BillingPayment.id == payment_id, BillingPayment.tenant_id == tenant_id) if partner_id: stmt = stmt.join(Client, Client.id == BillingPayment.client_id).where(Client.partner_id == partner_id) if financial_year and financial_year.upper() != "ALL": stmt = stmt.where(BillingPayment.financial_year == financial_year) return db.execute(stmt).scalars().unique().one_or_none() def list_clients_for_billing(db: Session, *, tenant_id: int, branch_id: int | None = None, partner_id: int | None = None, q: str = ""): stmt = select(Client).where(Client.tenant_id == tenant_id, Client.is_archived.is_(False)) if branch_id: stmt = stmt.where(Client.branch_id == branch_id) if partner_id: stmt = stmt.where(Client.partner_id == partner_id) if q.strip(): term = f"%{q.strip()}%" stmt = stmt.where(or_(Client.client_name.ilike(term), Client.client_code.ilike(term), Client.pan.ilike(term), Client.gstin.ilike(term))) return db.execute(stmt.order_by(Client.client_name.asc())).scalars().all() def list_services_for_billing(db: Session): return db.execute( select(ServiceCatalogue).where(ServiceCatalogue.is_active.is_(True)).order_by(ServiceCatalogue.service_name.asc()) ).scalars().all() def list_invoices(db: Session, *, tenant_id: int, branch_id: int | None = None, partner_id: int | None = None, financial_year: str | None = None, q: str = ""): stmt = select(BillingInvoice).options(selectinload(BillingInvoice.client), selectinload(BillingInvoice.lines), selectinload(BillingInvoice.payments)).where(BillingInvoice.tenant_id == tenant_id) if branch_id: stmt = stmt.where(BillingInvoice.branch_id == branch_id) if partner_id: stmt = stmt.join(Client, Client.id == BillingInvoice.client_id).where(Client.partner_id == partner_id) if financial_year and financial_year.upper() != "ALL": stmt = stmt.where(BillingInvoice.financial_year == financial_year) if q.strip(): term = f"%{q.strip()}%" stmt = stmt.join(Client, Client.id == BillingInvoice.client_id).where(or_(BillingInvoice.invoice_no.ilike(term), Client.client_name.ilike(term), Client.client_code.ilike(term))) return db.execute(stmt.order_by(BillingInvoice.invoice_date.desc(), BillingInvoice.id.desc())).scalars().unique().all() def build_billing_report_summary(db: Session, *, tenant_id: int, branch_id: int | None = None, partner_id: int | None = None, financial_year: str | None = None) -> dict[str, Any]: invoices = list_invoices(db, tenant_id=tenant_id, branch_id=branch_id, partner_id=partner_id, financial_year=financial_year) payments = list_payments(db, tenant_id=tenant_id, branch_id=branch_id, partner_id=partner_id, financial_year=financial_year) total_billed = sum((money(row.total_amount) for row in invoices if row.status != "CANCELLED"), Decimal("0.00")) total_received = sum((money(row.amount_received) for row in payments if row.status == "RECEIVED"), Decimal("0.00")) total_tds = sum((money(row.tds_deducted) for row in payments if row.status == "RECEIVED"), Decimal("0.00")) outstanding = sum((money(row.balance_amount) for row in invoices if row.status in {"ISSUED", "PARTLY_PAID", "OVERDUE"}), Decimal("0.00")) draft_count = sum(1 for row in invoices if row.status == "DRAFT") issued_count = sum(1 for row in invoices if row.status in {"ISSUED", "PARTLY_PAID", "OVERDUE"}) paid_count = sum(1 for row in invoices if row.status == "PAID") return { "invoice_count": len(invoices), "draft_count": draft_count, "issued_count": issued_count, "paid_count": paid_count, "payment_count": len(payments), "total_billed": money(total_billed), "total_received": money(total_received), "total_tds": money(total_tds), "total_collected_with_tds": money(total_received + total_tds), "outstanding": money(outstanding), } def get_invoice(db: Session, *, invoice_id: int, tenant_id: int, partner_id: int | None = None, financial_year: str | None = None) -> BillingInvoice | None: stmt = select(BillingInvoice).options(selectinload(BillingInvoice.client), selectinload(BillingInvoice.lines).selectinload(BillingInvoiceLine.service), selectinload(BillingInvoice.payments)).where(BillingInvoice.id == invoice_id, BillingInvoice.tenant_id == tenant_id) if partner_id: stmt = stmt.join(Client, Client.id == BillingInvoice.client_id).where(Client.partner_id == partner_id) if financial_year and financial_year.upper() != "ALL": stmt = stmt.where(BillingInvoice.financial_year == financial_year) return db.execute(stmt).scalars().unique().one_or_none() def create_invoice( db: Session, *, tenant_id: int, branch_id: int | None, client_id: int, invoice_date: date, due_date: date | None, billing_period_from: date | None, billing_period_to: date | None, tax_type: str, notes: str | None, terms: str | None, place_of_supply: str | None = None, client_state_code: str | None = None, reverse_charge: bool = False, created_by_user_id: int, raw_lines: list[dict[str, Any]], generation_batch_id: int | None = None, engagement_id: int | None = None, financial_year: str | None = None, ) -> BillingInvoice: settings = get_effective_billing_settings(db, tenant_id=tenant_id, branch_id=branch_id) client = db.get(Client, client_id) if client is None: raise ValueError("Client not found.") snapshot_state_code = (client_state_code or _state_code_from_gstin(getattr(client, "gstin", None)) or "").strip()[:2] or None invoice_financial_year = financial_year or billing_financial_year(billing_period_from=billing_period_from, invoice_date=invoice_date) invoice = BillingInvoice( tenant_id=tenant_id, branch_id=branch_id, client_id=client_id, engagement_id=engagement_id, invoice_no=next_invoice_number(db, tenant_id=tenant_id, branch_id=branch_id, financial_year=invoice_financial_year), invoice_date=invoice_date, due_date=due_date, billing_period_from=billing_period_from, billing_period_to=billing_period_to, financial_year=invoice_financial_year, invoice_title=settings.invoice_title or "Tax Invoice", place_of_supply=(place_of_supply or getattr(client, "state", None) or "").strip() or None, reverse_charge=bool(reverse_charge), client_legal_name=getattr(client, "client_name", None), client_trade_name=getattr(client, "trade_name", None), client_gstin=(getattr(client, "gstin", None) or "").strip().upper() or None, client_pan=(getattr(client, "pan", None) or "").strip().upper() or None, client_billing_address=_client_address_snapshot(client), client_state=getattr(client, "state", None), client_state_code=snapshot_state_code, client_email=getattr(client, "email", None), client_mobile=getattr(client, "mobile", None), tax_type=tax_type if tax_type in TAX_TYPES else (settings.default_tax_type if settings.default_tax_type in TAX_TYPES else "CGST_SGST"), notes=(notes or "").strip() or None, terms=(terms or settings.terms or "").strip() or None, created_by_user_id=created_by_user_id, generation_batch_id=generation_batch_id, status="DRAFT", balance_amount=Decimal("0.00"), ) db.add(invoice) db.flush() sort_order = 1 for raw in raw_lines: description = str(raw.get("description") or "").strip() if not description: continue qty = money(raw.get("quantity") or 1) if qty <= 0: qty = Decimal("1.00") rate = money(raw.get("rate")) disc = money(raw.get("discount_amount")) gst_rate = money(raw.get("gst_rate") or settings.default_gst_rate) sac_code = str(raw.get("sac_code") or settings.default_sac_code or "").strip()[:20] or None calc = calculate_line(quantity=qty, rate=rate, discount=disc, gst_rate=gst_rate, tax_type=invoice.tax_type) service_id = raw.get("service_id") or None fee_group_id = raw.get("fee_group_id") or None raw_engagement_id = raw.get("engagement_id") or None line = BillingInvoiceLine( invoice_id=invoice.id, service_id=int(service_id) if service_id else None, fee_group_id=int(fee_group_id) if fee_group_id else None, engagement_id=int(raw_engagement_id) if raw_engagement_id else None, description=description, sac_code=sac_code, billing_period_from=billing_period_from, billing_period_to=billing_period_to, quantity=qty, rate=rate, discount_amount=disc, taxable_amount=calc["taxable"], gst_rate=gst_rate, cgst_amount=calc["cgst"], sgst_amount=calc["sgst"], igst_amount=calc["igst"], line_total=calc["total"], sort_order=sort_order, ) db.add(line) invoice.lines.append(line) sort_order += 1 if sort_order == 1: raise ValueError("At least one invoice line with description is required.") recalc_invoice(invoice) invoice.amount_in_words = amount_to_indian_words(invoice.total_amount) db.flush() return invoice def issue_invoice(db: Session, invoice: BillingInvoice, *, user_id: int | None = None) -> BillingInvoice: issued_now = False if invoice.status == "DRAFT": invoice.status = "ISSUED" invoice.approved_by_user_id = user_id invoice.posted_at_utc = datetime.now(timezone.utc) update_invoice_payment_totals(invoice) invoice.updated_at_utc = datetime.now(timezone.utc) issued_now = True if issued_now: db.flush() try: send_invoice_issued_email(db, invoice) except Exception: # Email failure should be recorded in email logs and must not block invoice issue. pass return invoice def list_fee_groups(db: Session, *, tenant_id: int, branch_id: int | None = None, partner_id: int | None = None, q: str = ""): stmt = select(BillingFeeGroup).options(selectinload(BillingFeeGroup.client), selectinload(BillingFeeGroup.services).selectinload(BillingFeeGroupService.service)).where(BillingFeeGroup.tenant_id == tenant_id) if branch_id: stmt = stmt.where(BillingFeeGroup.branch_id == branch_id) if partner_id: stmt = stmt.where(BillingFeeGroup.partner_id == partner_id) if q.strip(): term = f"%{q.strip()}%" stmt = stmt.join(Client, Client.id == BillingFeeGroup.client_id).where(or_(BillingFeeGroup.group_code.ilike(term), BillingFeeGroup.group_name.ilike(term), Client.client_name.ilike(term), Client.client_code.ilike(term))) return db.execute(stmt.order_by(BillingFeeGroup.group_code.asc())).scalars().unique().all() def period_label(period_from: date, period_to: date) -> str: if period_from.year == period_to.year and period_from.month == period_to.month: return period_from.strftime("%B %Y") return f"{period_from.isoformat()} to {period_to.isoformat()}" def list_fee_groups_for_generation( db: Session, *, tenant_id: int, branch_id: int | None = None, partner_id: int | None = None, frequency: str | None = None, auto_generate_only: bool = True, q: str = "", ): stmt = ( select(BillingFeeGroup) .options( selectinload(BillingFeeGroup.client), selectinload(BillingFeeGroup.services).selectinload(BillingFeeGroupService.service), ) .where(BillingFeeGroup.tenant_id == tenant_id, BillingFeeGroup.is_active.is_(True)) ) if branch_id: stmt = stmt.where(BillingFeeGroup.branch_id == branch_id) if partner_id: stmt = stmt.where(BillingFeeGroup.partner_id == partner_id) if frequency: stmt = stmt.where(BillingFeeGroup.frequency == frequency) if auto_generate_only: stmt = stmt.where(BillingFeeGroup.auto_generate.is_(True)) if q.strip(): term = f"%{q.strip()}%" stmt = stmt.join(Client, Client.id == BillingFeeGroup.client_id).where( or_(BillingFeeGroup.group_code.ilike(term), BillingFeeGroup.group_name.ilike(term), Client.client_name.ilike(term), Client.client_code.ilike(term)) ) return db.execute(stmt.order_by(BillingFeeGroup.group_code.asc())).scalars().unique().all() def fee_group_already_billed(db: Session, *, tenant_id: int, fee_group_id: int, period_from: date, period_to: date) -> BillingInvoice | None: stmt = ( select(BillingInvoice) .join(BillingInvoiceLine, BillingInvoiceLine.invoice_id == BillingInvoice.id) .where( BillingInvoice.tenant_id == tenant_id, BillingInvoice.status != "CANCELLED", BillingInvoice.billing_period_from == period_from, BillingInvoice.billing_period_to == period_to, BillingInvoiceLine.fee_group_id == fee_group_id, ) .order_by(BillingInvoice.id.desc()) ) return db.execute(stmt).scalars().first() def _financial_year_from_period(period_from: date) -> str: return _financial_year_label(period_from) def _billing_engagement_lookup( db: Session, *, tenant_id: int, branch_id: int | None, client_ids: set[int], service_ids: set[int], financial_year: str, ) -> dict[tuple[int, int], ClientServiceSubscription]: if not client_ids or not service_ids: return {} stmt = ( select(ClientServiceSubscription) .options(selectinload(ClientServiceSubscription.client), selectinload(ClientServiceSubscription.catalogue)) .where( ClientServiceSubscription.tenant_id == tenant_id, ClientServiceSubscription.client_id.in_(client_ids), ClientServiceSubscription.service_catalogue_id.in_(service_ids), ClientServiceSubscription.financial_year == financial_year, ClientServiceSubscription.is_active.is_(True), ) ) if branch_id: stmt = stmt.where(ClientServiceSubscription.branch_id == branch_id) rows = db.execute(stmt.order_by(ClientServiceSubscription.id.desc())).scalars().unique().all() lookup: dict[tuple[int, int], ClientServiceSubscription] = {} for row in rows: lookup.setdefault((row.client_id, row.service_catalogue_id), row) return lookup def _annotate_invoice_lines_with_engagements( raw_lines: list[dict[str, Any]], *, client_id: int, lookup: dict[tuple[int, int], ClientServiceSubscription], ) -> tuple[list[dict[str, Any]], int | None, list[ClientServiceSubscription]]: engagement_ids: set[int] = set() linked: list[ClientServiceSubscription] = [] for line in raw_lines: service_id = line.get("service_id") if not service_id: continue subscription = lookup.get((client_id, int(service_id))) if not subscription: continue line["engagement_id"] = subscription.id engagement_ids.add(subscription.id) linked.append(subscription) invoice_engagement_id = next(iter(engagement_ids)) if len(engagement_ids) == 1 else None return raw_lines, invoice_engagement_id, linked def _fee_group_invoice_lines(fee_group: BillingFeeGroup, *, period_from: date, period_to: date) -> list[dict[str, Any]]: label = period_label(period_from, period_to) if fee_group.billing_mode == "SERVICE_WISE": lines: list[dict[str, Any]] = [] for item in sorted(fee_group.services, key=lambda x: x.sort_order or 100): rate = money(item.line_amount) if rate <= 0 and item.percentage is not None: rate = (money(fee_group.fee_amount) * money(item.percentage) / Decimal("100.00")).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) description = item.line_description or (item.service.service_name if item.service else fee_group.group_name) lines.append({ "description": f"{description} - {label}"[:500], "service_id": item.service_id, "fee_group_id": fee_group.id, "quantity": "1", "rate": rate, "discount_amount": "0", "gst_rate": fee_group.gst_rate, }) if lines: return lines included = [] for item in sorted(fee_group.services, key=lambda x: x.sort_order or 100): if item.service: included.append(item.line_description or item.service.service_name) desc = f"{fee_group.group_name} - {label}" if included: desc = desc + "\nIncluded services: " + ", ".join(included) return [{ "description": desc[:500], "service_id": None, "fee_group_id": fee_group.id, "quantity": "1", "rate": fee_group.fee_amount, "discount_amount": "0", "gst_rate": fee_group.gst_rate, }] def generate_draft_invoices_from_fee_groups( db: Session, *, tenant_id: int, branch_id: int | None, partner_id: int | None, generated_by_user_id: int, billing_period_from: date, billing_period_to: date, frequency: str | None, fee_group_ids: list[int], skip_duplicates: bool = True, ) -> dict[str, Any]: if billing_period_to < billing_period_from: raise ValueError("Billing Period To cannot be earlier than Billing Period From.") if not fee_group_ids: raise ValueError("Select at least one fee structure to generate invoices.") stmt = ( select(BillingFeeGroup) .options(selectinload(BillingFeeGroup.client), selectinload(BillingFeeGroup.services).selectinload(BillingFeeGroupService.service)) .where(BillingFeeGroup.tenant_id == tenant_id, BillingFeeGroup.id.in_(fee_group_ids), BillingFeeGroup.is_active.is_(True)) ) if branch_id: stmt = stmt.where(BillingFeeGroup.branch_id == branch_id) if partner_id: stmt = stmt.where(BillingFeeGroup.partner_id == partner_id) if frequency: stmt = stmt.where(BillingFeeGroup.frequency == frequency) fee_groups = db.execute(stmt.order_by(BillingFeeGroup.group_code.asc())).scalars().unique().all() batch = BillingInvoiceGenerationBatch( tenant_id=tenant_id, branch_id=branch_id, billing_period_from=billing_period_from, billing_period_to=billing_period_to, financial_year=billing_financial_year(billing_period_from=billing_period_from, invoice_date=billing_period_to), frequency=frequency or None, selected_count=len(fee_group_ids), generated_by_user_id=generated_by_user_id, status="DRAFT_CREATED", ) db.add(batch) db.flush() created: list[BillingInvoice] = [] skipped: list[str] = [] errors: list[str] = [] found_ids = {g.id for g in fee_groups} financial_year = _financial_year_from_period(billing_period_from) client_ids = {int(g.client_id) for g in fee_groups if g.client_id} service_ids = {int(item.service_id) for g in fee_groups for item in (g.services or []) if item.service_id} engagement_lookup = _billing_engagement_lookup( db, tenant_id=tenant_id, branch_id=branch_id, client_ids=client_ids, service_ids=service_ids, financial_year=financial_year, ) for missing_id in sorted(set(fee_group_ids) - found_ids): skipped.append(f"Fee structure ID {missing_id} is not available in the active Audit Firm/Branch context.") for fee_group in fee_groups: try: existing = fee_group_already_billed( db, tenant_id=tenant_id, fee_group_id=fee_group.id, period_from=billing_period_from, period_to=billing_period_to, ) if existing and skip_duplicates: skipped.append(f"{fee_group.group_code}: already billed in invoice {existing.invoice_no}.") continue raw_lines = _fee_group_invoice_lines(fee_group, period_from=billing_period_from, period_to=billing_period_to) raw_lines, invoice_engagement_id, linked_subscriptions = _annotate_invoice_lines_with_engagements( raw_lines, client_id=fee_group.client_id, lookup=engagement_lookup, ) linked_note = "" if linked_subscriptions: linked_labels = [] for sub in linked_subscriptions: service_name = sub.catalogue.service_name if getattr(sub, "catalogue", None) else f"Service {sub.service_catalogue_id}" linked_labels.append(f"{service_name} / {sub.financial_year}") linked_note = " Linked service subscriptions: " + "; ".join(linked_labels[:5]) + "." invoice = create_invoice( db, tenant_id=tenant_id, branch_id=fee_group.branch_id or branch_id, client_id=fee_group.client_id, invoice_date=date.today(), due_date=None, billing_period_from=billing_period_from, billing_period_to=billing_period_to, tax_type=fee_group.tax_type if fee_group.tax_type in TAX_TYPES else "CGST_SGST", notes=f"Draft generated from fee structure {fee_group.group_code}.{linked_note}", terms=None, created_by_user_id=generated_by_user_id, raw_lines=raw_lines, generation_batch_id=batch.id, engagement_id=invoice_engagement_id, financial_year=financial_year, ) created.append(invoice) except Exception as exc: # keep batch generation resilient per client/package errors.append(f"{fee_group.group_code}: {exc}") batch.created_invoice_count = len(created) batch.skipped_count = len(skipped) batch.error_count = len(errors) if errors and created: batch.status = "PARTIAL" elif errors and not created: batch.status = "FAILED" batch.remarks = "\n".join(skipped + errors) or None db.flush() return {"batch": batch, "created": created, "skipped": skipped, "errors": errors} def build_fee_structure_template() -> bytes: wb = Workbook() ws = wb.active ws.title = "Fee_Structure" ws.append([ "Client Code", "Billing Group Code", "Billing Group Name", "Billing Mode", "Frequency", "Fee Amount", "GST Rate", "Tax Type", "Effective From", "Effective To", "Auto Generate", "Notes" ]) ws.append(["ABC001", "ABC-GST-MONTHLY", "Monthly GST Compliance", "PACKAGE", "Monthly", 2500, 18, "CGST_SGST", "2026-04-01", "", "Yes", "GSTR-1 and GSTR-3B package"]) ws2 = wb.create_sheet("Fee_Services") ws2.append(["Billing Group Code", "Service Code", "Line Description", "Allocation Type", "Line Amount", "Percentage", "Sort Order"]) ws2.append(["ABC-GST-MONTHLY", "GSTR1", "GSTR-1 Filing", "Included", 0, "", 1]) ws2.append(["ABC-GST-MONTHLY", "GSTR3B", "GSTR-3B Filing", "Included", 0, "", 2]) bio = BytesIO() wb.save(bio) return bio.getvalue() def import_fee_structure_excel(db: Session, *, tenant_id: int, branch_id: int | None, created_by_user_id: int, file_bytes: bytes) -> dict[str, Any]: wb = load_workbook(BytesIO(file_bytes), data_only=True) if "Fee_Structure" not in wb.sheetnames or "Fee_Services" not in wb.sheetnames: raise ValueError("Excel must contain Fee_Structure and Fee_Services sheets.") clients = {c.client_code.strip().upper(): c for c in db.execute(select(Client).where(Client.tenant_id == tenant_id)).scalars().all()} services = {s.service_code.strip().upper(): s for s in db.execute(select(ServiceCatalogue)).scalars().all()} ws = wb["Fee_Structure"] header = [str(c.value or "").strip() for c in ws[1]] rows = [] errors: list[str] = [] for idx, row in enumerate(ws.iter_rows(min_row=2, values_only=True), start=2): data = dict(zip(header, row)) if not any(data.values()): continue client_code = normalize_code(data.get("Client Code")) group_code = normalize_code(data.get("Billing Group Code")) if not client_code or client_code not in clients: errors.append(f"Fee_Structure row {idx}: Client Code not found: {client_code}") continue if not group_code: errors.append(f"Fee_Structure row {idx}: Billing Group Code is required") continue mode = str(data.get("Billing Mode") or "PACKAGE").strip().upper().replace(" ", "_") if mode not in BILLING_MODES: errors.append(f"Fee_Structure row {idx}: Billing Mode must be PACKAGE or SERVICE_WISE") continue frequency = str(data.get("Frequency") or "Monthly").strip() or "Monthly" rows.append({ "client": clients[client_code], "group_code": group_code, "group_name": str(data.get("Billing Group Name") or group_code).strip(), "billing_mode": mode, "frequency": frequency, "fee_amount": money(data.get("Fee Amount")), "gst_rate": money(data.get("GST Rate") or 18), "tax_type": str(data.get("Tax Type") or "CGST_SGST").strip().upper() if str(data.get("Tax Type") or "").strip().upper() in TAX_TYPES else "CGST_SGST", "effective_from": parse_date(data.get("Effective From")), "effective_to": parse_date(data.get("Effective To")), "auto_generate": normalize_yes_no(data.get("Auto Generate")), "notes": str(data.get("Notes") or "").strip() or None, }) service_rows_by_group: dict[str, list[dict[str, Any]]] = {} ws2 = wb["Fee_Services"] header2 = [str(c.value or "").strip() for c in ws2[1]] for idx, row in enumerate(ws2.iter_rows(min_row=2, values_only=True), start=2): data = dict(zip(header2, row)) if not any(data.values()): continue group_code = normalize_code(data.get("Billing Group Code")) service_code = normalize_code(data.get("Service Code")) if not group_code: errors.append(f"Fee_Services row {idx}: Billing Group Code is required") continue if not service_code or service_code not in services: errors.append(f"Fee_Services row {idx}: Service Code not found: {service_code}") continue service_rows_by_group.setdefault(group_code, []).append({ "service": services[service_code], "line_description": str(data.get("Line Description") or services[service_code].service_name).strip(), "allocation_type": str(data.get("Allocation Type") or "Included").strip() or "Included", "line_amount": money(data.get("Line Amount")), "percentage": money(data.get("Percentage")) if data.get("Percentage") not in (None, "") else None, "sort_order": int(data.get("Sort Order") or 100), }) if errors: return {"success": False, "created": 0, "updated": 0, "errors": errors} created = updated = 0 for row in rows: existing = db.execute(select(BillingFeeGroup).where(BillingFeeGroup.tenant_id == tenant_id, BillingFeeGroup.group_code == row["group_code"])).scalar_one_or_none() if existing: fee_group = existing updated += 1 else: fee_group = BillingFeeGroup(tenant_id=tenant_id, group_code=row["group_code"], created_by_user_id=created_by_user_id) db.add(fee_group) created += 1 fee_group.branch_id = branch_id or row["client"].branch_id fee_group.client_id = row["client"].id fee_group.partner_id = row["client"].partner_id fee_group.group_name = row["group_name"] fee_group.billing_mode = row["billing_mode"] fee_group.frequency = row["frequency"] fee_group.fee_amount = row["fee_amount"] fee_group.gst_rate = row["gst_rate"] fee_group.tax_type = row["tax_type"] fee_group.effective_from = row["effective_from"] fee_group.effective_to = row["effective_to"] fee_group.auto_generate = row["auto_generate"] fee_group.notes = row["notes"] fee_group.updated_by_user_id = created_by_user_id db.flush() for old in list(fee_group.services): db.delete(old) db.flush() for service_row in service_rows_by_group.get(row["group_code"], []): db.add(BillingFeeGroupService( fee_group_id=fee_group.id, service_id=service_row["service"].id, line_description=service_row["line_description"], allocation_type=service_row["allocation_type"], line_amount=service_row["line_amount"], percentage=service_row["percentage"], sort_order=service_row["sort_order"], )) db.commit() return {"success": True, "created": created, "updated": updated, "errors": []} # --------------------------------------------------------------------------- # Phase 7R.6 - PayUMoney / PayU redirect integration helpers # --------------------------------------------------------------------------- import base64 import hashlib import hmac import json from urllib.parse import urlencode from urllib.request import Request as UrlRequest, urlopen from urllib.error import HTTPError, URLError from app.modules.billing.models import BillingOnlinePaymentTransaction PAYUMONEY_PROVIDER = "PAYUMONEY" PAYUMONEY_TEST_URL = "https://test.payu.in/_payment" PAYUMONEY_PROD_URL = "https://secure.payu.in/_payment" PAYUMONEY_MODES = ["TEST", "LIVE"] def payumoney_checkout_url(settings: BillingSettings) -> str: return PAYUMONEY_PROD_URL if str(getattr(settings, "payumoney_mode", "TEST")).upper() == "LIVE" else PAYUMONEY_TEST_URL def is_payumoney_ready(settings: BillingSettings | None) -> bool: return bool( settings and getattr(settings, "payumoney_enabled", False) and (getattr(settings, "payumoney_merchant_key", None) or "").strip() and (getattr(settings, "payumoney_merchant_salt", None) or "").strip() ) def generate_payumoney_hash(*, key: str, txnid: str, amount: str, productinfo: str, firstname: str, email: str, salt: str, udf1: str = "", udf2: str = "", udf3: str = "", udf4: str = "", udf5: str = "") -> str: hash_string = f"{key}|{txnid}|{amount}|{productinfo}|{firstname}|{email}|{udf1}|{udf2}|{udf3}|{udf4}|{udf5}||||||{salt}" return hashlib.sha512(hash_string.encode("utf-8")).hexdigest().lower() def verify_payumoney_response_hash(*, response_data: dict[str, Any], salt: str, key: str) -> bool: # PayU redirect response hash: salt|status||||||udf5|udf4|udf3|udf2|udf1|email|firstname|productinfo|amount|txnid|key received_hash = str(response_data.get("hash") or "").strip().lower() if not received_hash: return False status = str(response_data.get("status") or "") txnid = str(response_data.get("txnid") or "") amount = str(response_data.get("amount") or "") productinfo = str(response_data.get("productinfo") or "") firstname = str(response_data.get("firstname") or "") email = str(response_data.get("email") or "") udf1 = str(response_data.get("udf1") or "") udf2 = str(response_data.get("udf2") or "") udf3 = str(response_data.get("udf3") or "") udf4 = str(response_data.get("udf4") or "") udf5 = str(response_data.get("udf5") or "") reverse = f"{salt}|{status}||||||{udf5}|{udf4}|{udf3}|{udf2}|{udf1}|{email}|{firstname}|{productinfo}|{amount}|{txnid}|{key}" expected = hashlib.sha512(reverse.encode("utf-8")).hexdigest().lower() return expected == received_hash def create_payumoney_transaction(db: Session, *, invoice: BillingInvoice, settings: BillingSettings, base_url: str, client_ip: str | None = None) -> dict[str, Any]: if not is_payumoney_ready(settings): raise ValueError("PayUMoney is not enabled or merchant credentials are missing in Billing Settings.") if invoice.status not in {"ISSUED", "PARTLY_PAID", "OVERDUE"} or money(invoice.balance_amount) <= Decimal("0.00"): raise ValueError("Only issued invoices with balance can be paid online.") amount = f"{money(invoice.balance_amount):.2f}" txnid = f"AF{invoice.tenant_id}I{invoice.id}T{int(datetime.now(timezone.utc).timestamp())}" key = (settings.payumoney_merchant_key or "").strip() salt = (settings.payumoney_merchant_salt or "").strip() productinfo = (settings.payumoney_product_info or f"Invoice {invoice.invoice_no}").strip()[:250] firstname = (invoice.client_legal_name or invoice.client_trade_name or getattr(invoice.client, "client_name", None) or "Client").strip()[:120] email = (invoice.client_email or getattr(invoice.client, "email", None) or settings.contact_email or "no-reply@example.com").strip() phone = (invoice.client_mobile or getattr(invoice.client, "mobile", None) or settings.contact_mobile or "9999999999").strip() udf1, udf2, udf3, udf4, udf5 = str(invoice.id), str(invoice.client_id), str(invoice.tenant_id), str(invoice.branch_id or ""), "audit_firm_erp" hash_value = generate_payumoney_hash(key=key, txnid=txnid, amount=amount, productinfo=productinfo, firstname=firstname, email=email, salt=salt, udf1=udf1, udf2=udf2, udf3=udf3, udf4=udf4, udf5=udf5) transaction = BillingOnlinePaymentTransaction( tenant_id=invoice.tenant_id, branch_id=invoice.branch_id, invoice_id=invoice.id, client_id=invoice.client_id, provider=PAYUMONEY_PROVIDER, mode=(settings.payumoney_mode or "TEST").upper(), txnid=txnid, amount=money(amount), productinfo=productinfo, firstname=firstname, email=email, phone=phone, status="INITIATED", gateway_status="created", ) db.add(transaction) db.flush() surl = f"{base_url.rstrip('/')}/client/billing/payumoney/success" furl = f"{base_url.rstrip('/')}/client/billing/payumoney/failure" payload = { "key": key, "txnid": txnid, "amount": amount, "productinfo": productinfo, "firstname": firstname, "email": email, "phone": phone, "surl": surl, "furl": furl, "hash": hash_value, "udf1": udf1, "udf2": udf2, "udf3": udf3, "udf4": udf4, "udf5": udf5, } if getattr(settings, "payumoney_merchant_id", None): payload["merchant_id"] = settings.payumoney_merchant_id return {"transaction": transaction, "payload": payload, "checkout_url": payumoney_checkout_url(settings)} def get_online_transaction_by_txnid(db: Session, *, txnid: str) -> BillingOnlinePaymentTransaction | None: return db.execute( select(BillingOnlinePaymentTransaction) .options(selectinload(BillingOnlinePaymentTransaction.invoice).selectinload(BillingInvoice.payments)) .where(BillingOnlinePaymentTransaction.txnid == txnid) ).scalars().unique().one_or_none() def process_payumoney_response(db: Session, *, response_data: dict[str, Any]) -> BillingOnlinePaymentTransaction | None: txnid = str(response_data.get("txnid") or "").strip() if not txnid: return None transaction = get_online_transaction_by_txnid(db, txnid=txnid) if not transaction: return None invoice = transaction.invoice settings = get_effective_billing_settings(db, tenant_id=invoice.tenant_id, branch_id=invoice.branch_id) hash_ok = verify_payumoney_response_hash(response_data=response_data, salt=(settings.payumoney_merchant_salt or ""), key=(settings.payumoney_merchant_key or "")) gateway_status = str(response_data.get("status") or "").lower() transaction.gateway_status = gateway_status or None transaction.response_hash = str(response_data.get("hash") or "") or None transaction.payu_payment_id = str(response_data.get("payuMoneyId") or response_data.get("payu_money_id") or "") or None transaction.mihpayid = str(response_data.get("mihpayid") or "") or None transaction.bank_ref_num = str(response_data.get("bank_ref_num") or response_data.get("bank_ref_no") or "") or None transaction.raw_response = json.dumps({k: str(v) for k, v in response_data.items()}, ensure_ascii=False) transaction.updated_at_utc = datetime.now(timezone.utc) if not hash_ok: transaction.status = "HASH_FAILED" return transaction if gateway_status == "success": transaction.status = "SUCCESS" transaction.completed_at_utc = datetime.now(timezone.utc) if not transaction.receipt_payment_id: payment = record_invoice_payment( db, invoice=invoice, payment_date=date.today(), amount_received=money(response_data.get("amount") or transaction.amount), tds_deducted=Decimal("0.00"), bank_charges=Decimal("0.00"), mode="ONLINE", reference_no=transaction.bank_ref_num or transaction.mihpayid or transaction.txnid, remarks=f"Online payment received through PayUMoney. Txn ID: {transaction.txnid}", created_by_user_id=None, payment_gateway=PAYUMONEY_PROVIDER, gateway_transaction_id=transaction.mihpayid or transaction.payu_payment_id or transaction.txnid, ) transaction.receipt_payment_id = payment.id else: transaction.status = "FAILED" return transaction # --------------------------------------------------------------------------- # Phase 7R.6A - Cashfree Payment Gateway Integration helpers # --------------------------------------------------------------------------- CASHFREE_PROVIDER = "CASHFREE" CASHFREE_API_VERSION_DEFAULT = "2023-08-01" CASHFREE_TEST_BASE_URL = "https://sandbox.cashfree.com/pg" CASHFREE_PROD_BASE_URL = "https://api.cashfree.com/pg" CASHFREE_MODES = ["TEST", "LIVE"] def cashfree_base_url(settings: BillingSettings) -> str: return CASHFREE_PROD_BASE_URL if str(getattr(settings, "cashfree_mode", "TEST")).upper() == "LIVE" else CASHFREE_TEST_BASE_URL def is_cashfree_ready(settings: BillingSettings | None) -> bool: return bool( settings and getattr(settings, "cashfree_enabled", False) and (getattr(settings, "cashfree_client_id", None) or "").strip() and (getattr(settings, "cashfree_client_secret", None) or "").strip() ) def _cashfree_headers(settings: BillingSettings) -> dict[str, str]: return { "Content-Type": "application/json", "Accept": "application/json", "x-api-version": (getattr(settings, "cashfree_api_version", None) or CASHFREE_API_VERSION_DEFAULT).strip() or CASHFREE_API_VERSION_DEFAULT, "x-client-id": (settings.cashfree_client_id or "").strip(), "x-client-secret": (settings.cashfree_client_secret or "").strip(), } def _cashfree_api_request(settings: BillingSettings, *, method: str, path: str, payload: dict[str, Any] | None = None) -> dict[str, Any]: url = cashfree_base_url(settings).rstrip("/") + path body = json.dumps(payload or {}).encode("utf-8") if payload is not None else None req = UrlRequest(url, data=body, headers=_cashfree_headers(settings), method=method.upper()) try: with urlopen(req, timeout=30) as response: raw = response.read().decode("utf-8") return json.loads(raw or "{}") except HTTPError as exc: raw = exc.read().decode("utf-8", errors="replace") raise ValueError(f"Cashfree API error {exc.code}: {raw}") from exc except URLError as exc: raise ValueError(f"Cashfree API connection error: {exc.reason}") from exc def create_cashfree_transaction(db: Session, *, invoice: BillingInvoice, settings: BillingSettings, base_url: str, client_ip: str | None = None) -> dict[str, Any]: if not is_cashfree_ready(settings): raise ValueError("Cashfree is not enabled or client credentials are missing in Billing Settings.") if invoice.status not in {"ISSUED", "PARTLY_PAID", "OVERDUE"} or money(invoice.balance_amount) <= Decimal("0.00"): raise ValueError("Only issued invoices with balance can be paid online.") amount = money(invoice.balance_amount) order_id = f"AF{invoice.tenant_id}I{invoice.id}C{int(datetime.now(timezone.utc).timestamp())}" customer_name = (invoice.client_legal_name or invoice.client_trade_name or getattr(invoice.client, "client_name", None) or "Client").strip()[:120] customer_email = (invoice.client_email or getattr(invoice.client, "email", None) or settings.contact_email or "no-reply@example.com").strip() customer_phone = (invoice.client_mobile or getattr(invoice.client, "mobile", None) or settings.contact_mobile or "9999999999").strip() note = (settings.cashfree_order_note or f"Invoice {invoice.invoice_no}").strip()[:250] return_url = f"{base_url.rstrip('/')}/client/billing/cashfree/return?order_id={{order_id}}" notify_url = f"{base_url.rstrip('/')}/client/billing/cashfree/webhook" payload = { "order_id": order_id, "order_amount": float(amount), "order_currency": "INR", "customer_details": { "customer_id": str(invoice.client_id), "customer_name": customer_name, "customer_email": customer_email, "customer_phone": customer_phone, }, "order_meta": { "return_url": return_url, "notify_url": notify_url, }, "order_note": note, "order_tags": { "tenant_id": str(invoice.tenant_id), "branch_id": str(invoice.branch_id or ""), "invoice_id": str(invoice.id), "invoice_no": str(invoice.invoice_no), "source": "audit_firm_erp", }, } response = _cashfree_api_request(settings, method="POST", path="/orders", payload=payload) payment_session_id = str(response.get("payment_session_id") or "").strip() if not payment_session_id: raise ValueError(f"Cashfree order created without payment_session_id: {response}") transaction = BillingOnlinePaymentTransaction( tenant_id=invoice.tenant_id, branch_id=invoice.branch_id, invoice_id=invoice.id, client_id=invoice.client_id, provider=CASHFREE_PROVIDER, mode=(settings.cashfree_mode or "TEST").upper(), txnid=order_id, amount=amount, productinfo=note, firstname=customer_name, email=customer_email, phone=customer_phone, status="INITIATED", gateway_status=str(response.get("order_status") or "ACTIVE"), cashfree_order_id=order_id, cashfree_cf_order_id=str(response.get("cf_order_id") or "") or None, cashfree_payment_session_id=payment_session_id, raw_response=json.dumps(response, ensure_ascii=False, default=str), ) db.add(transaction) db.flush() return {"transaction": transaction, "payment_session_id": payment_session_id, "order_response": response} def get_cashfree_transaction_by_order_id(db: Session, *, order_id: str) -> BillingOnlinePaymentTransaction | None: return db.execute( select(BillingOnlinePaymentTransaction) .options(selectinload(BillingOnlinePaymentTransaction.invoice).selectinload(BillingInvoice.payments)) .where( BillingOnlinePaymentTransaction.provider == CASHFREE_PROVIDER, BillingOnlinePaymentTransaction.txnid == order_id, ) ).scalars().unique().one_or_none() def fetch_cashfree_order_status(settings: BillingSettings, *, order_id: str) -> dict[str, Any]: return _cashfree_api_request(settings, method="GET", path=f"/orders/{order_id}") def _mark_cashfree_success(db: Session, transaction: BillingOnlinePaymentTransaction, *, amount: Any, reference_no: str | None, raw_payload: dict[str, Any] | None = None) -> BillingOnlinePaymentTransaction: invoice = transaction.invoice transaction.status = "SUCCESS" transaction.gateway_status = "PAID" transaction.completed_at_utc = datetime.now(timezone.utc) transaction.updated_at_utc = datetime.now(timezone.utc) if raw_payload is not None: transaction.raw_response = json.dumps(raw_payload, ensure_ascii=False, default=str) if not transaction.receipt_payment_id: payment = record_invoice_payment( db, invoice=invoice, payment_date=date.today(), amount_received=money(amount or transaction.amount), tds_deducted=Decimal("0.00"), bank_charges=Decimal("0.00"), mode="ONLINE", reference_no=reference_no or transaction.cashfree_payment_id or transaction.txnid, remarks=f"Online payment received through Cashfree. Order ID: {transaction.txnid}", created_by_user_id=None, payment_gateway=CASHFREE_PROVIDER, gateway_transaction_id=reference_no or transaction.cashfree_payment_id or transaction.txnid, ) transaction.receipt_payment_id = payment.id return transaction def process_cashfree_return(db: Session, *, order_id: str) -> BillingOnlinePaymentTransaction | None: transaction = get_cashfree_transaction_by_order_id(db, order_id=order_id) if not transaction: return None invoice = transaction.invoice settings = get_effective_billing_settings(db, tenant_id=invoice.tenant_id, branch_id=invoice.branch_id) order = fetch_cashfree_order_status(settings, order_id=order_id) transaction.gateway_status = str(order.get("order_status") or "") or transaction.gateway_status transaction.cashfree_cf_order_id = str(order.get("cf_order_id") or transaction.cashfree_cf_order_id or "") or None transaction.raw_response = json.dumps(order, ensure_ascii=False, default=str) transaction.updated_at_utc = datetime.now(timezone.utc) if str(order.get("order_status") or "").upper() == "PAID": return _mark_cashfree_success(db, transaction, amount=order.get("order_amount") or transaction.amount, reference_no=str(order.get("cf_order_id") or order_id), raw_payload=order) if str(order.get("order_status") or "").upper() in {"EXPIRED", "TERMINATED", "CANCELLED"}: transaction.status = "FAILED" return transaction def verify_cashfree_webhook_signature(*, raw_body: bytes, timestamp: str, signature: str, secret: str) -> bool: if not raw_body or not timestamp or not signature or not secret: return False signed_payload = timestamp.encode("utf-8") + raw_body digest = hmac.new(secret.encode("utf-8"), signed_payload, hashlib.sha256).digest() expected = base64.b64encode(digest).decode("utf-8") return hmac.compare_digest(expected, signature) def process_cashfree_webhook(db: Session, *, raw_body: bytes, headers: dict[str, str]) -> BillingOnlinePaymentTransaction | None: payload = json.loads(raw_body.decode("utf-8") or "{}") data = payload.get("data") or payload order = data.get("order") or data payment = data.get("payment") or {} order_id = str(order.get("order_id") or data.get("order_id") or "").strip() if not order_id: return None transaction = get_cashfree_transaction_by_order_id(db, order_id=order_id) if not transaction: return None invoice = transaction.invoice settings = get_effective_billing_settings(db, tenant_id=invoice.tenant_id, branch_id=invoice.branch_id) timestamp = headers.get("x-webhook-timestamp") or headers.get("X-Webhook-Timestamp") or "" signature = headers.get("x-webhook-signature") or headers.get("X-Webhook-Signature") or "" if not verify_cashfree_webhook_signature(raw_body=raw_body, timestamp=timestamp, signature=signature, secret=(settings.cashfree_client_secret or "")): transaction.status = "HASH_FAILED" transaction.gateway_status = "WEBHOOK_SIGNATURE_FAILED" transaction.raw_response = raw_body.decode("utf-8", errors="replace") transaction.updated_at_utc = datetime.now(timezone.utc) return transaction event_id = str(payload.get("event_id") or payload.get("cf_event_id") or "") or None if event_id and transaction.webhook_event_id == event_id and transaction.status == "SUCCESS": return transaction transaction.webhook_event_id = event_id transaction.cashfree_payment_id = str(payment.get("cf_payment_id") or payment.get("payment_id") or "") or transaction.cashfree_payment_id transaction.gateway_status = str(payment.get("payment_status") or order.get("order_status") or payload.get("type") or "") or None transaction.raw_response = raw_body.decode("utf-8", errors="replace") transaction.updated_at_utc = datetime.now(timezone.utc) status_text = (transaction.gateway_status or "").upper() if "SUCCESS" in status_text or status_text == "PAID" or str(order.get("order_status") or "").upper() == "PAID": amount = payment.get("payment_amount") or order.get("order_amount") or transaction.amount reference = transaction.cashfree_payment_id or str(payment.get("bank_reference") or order.get("cf_order_id") or order_id) return _mark_cashfree_success(db, transaction, amount=amount, reference_no=reference, raw_payload=payload) if "FAILED" in status_text or "CANCELLED" in status_text or "EXPIRED" in status_text: transaction.status = "FAILED" return transaction