Rakseeibra/arrr-erp
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
5a135a65a80941006ebb84759c4fc30dc2519662
arrr-erp/app/modules/billing/services.py
T
A R R R Associates 5c75eb6bd9 Prepare ERP source for Gitea deployment
2026-06-20 15:01:44 +05:30

1373 lines
63 KiB
Python
Raw Blame History

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
Reference in New Issue View Git Blame Copy Permalink