A R R R Associates
139 lines
6.2 KiB
Python
139 lines
6.2 KiB
Python
from __future__ import annotations
|
|
|
|
from decimal import Decimal
|
|
from typing import Any
|
|
from urllib.parse import quote
|
|
|
|
from sqlalchemy import or_, select
|
|
from sqlalchemy.orm import Session, selectinload
|
|
|
|
from app.modules.billing.models import BillingInvoice, BillingInvoiceLine, BillingPayment
|
|
from app.modules.billing.services import build_invoice_print_context, is_cashfree_ready, is_payumoney_ready, money
|
|
|
|
CLIENT_VISIBLE_INVOICE_STATUSES = {"ISSUED", "PARTLY_PAID", "PAID", "OVERDUE"}
|
|
|
|
|
|
def _client_ids(client_row: Any) -> tuple[int, int]:
|
|
"""Return (tenant_id, client_id) from dict/row/model style client payload."""
|
|
if isinstance(client_row, dict):
|
|
return int(client_row.get("tenant_id") or 0), int(client_row.get("id") or 0)
|
|
return int(getattr(client_row, "tenant_id", 0) or 0), int(getattr(client_row, "id", 0) or 0)
|
|
|
|
|
|
def list_client_portal_invoices(db: Session, client_row: Any, *, q: str = "", include_paid: bool = True, financial_year: str | None = None) -> list[BillingInvoice]:
|
|
tenant_id, client_id = _client_ids(client_row)
|
|
stmt = (
|
|
select(BillingInvoice)
|
|
.options(
|
|
selectinload(BillingInvoice.lines).selectinload(BillingInvoiceLine.service),
|
|
selectinload(BillingInvoice.payments),
|
|
)
|
|
.where(
|
|
BillingInvoice.tenant_id == tenant_id,
|
|
BillingInvoice.client_id == client_id,
|
|
BillingInvoice.status.in_(CLIENT_VISIBLE_INVOICE_STATUSES),
|
|
)
|
|
)
|
|
if not include_paid:
|
|
stmt = stmt.where(BillingInvoice.status != "PAID")
|
|
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.where(or_(BillingInvoice.invoice_no.ilike(term), BillingInvoice.invoice_title.ilike(term)))
|
|
return db.execute(stmt.order_by(BillingInvoice.invoice_date.desc(), BillingInvoice.id.desc())).scalars().unique().all()
|
|
|
|
|
|
def get_client_portal_invoice(db: Session, client_row: Any, invoice_id: int, *, financial_year: str | None = None) -> BillingInvoice | None:
|
|
tenant_id, client_id = _client_ids(client_row)
|
|
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,
|
|
BillingInvoice.client_id == client_id,
|
|
BillingInvoice.status.in_(CLIENT_VISIBLE_INVOICE_STATUSES),
|
|
)
|
|
)
|
|
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 get_client_portal_payment(db: Session, client_row: Any, payment_id: int, *, financial_year: str | None = None) -> BillingPayment | None:
|
|
tenant_id, client_id = _client_ids(client_row)
|
|
stmt = (
|
|
select(BillingPayment)
|
|
.options(
|
|
selectinload(BillingPayment.invoice).selectinload(BillingInvoice.lines),
|
|
selectinload(BillingPayment.client),
|
|
)
|
|
.where(
|
|
BillingPayment.id == payment_id,
|
|
BillingPayment.tenant_id == tenant_id,
|
|
BillingPayment.client_id == client_id,
|
|
BillingPayment.status == "RECEIVED",
|
|
)
|
|
)
|
|
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 build_client_billing_summary(db: Session, client_row: Any, *, financial_year: str | None = None) -> dict[str, Any]:
|
|
invoices = list_client_portal_invoices(db, client_row, include_paid=True, financial_year=financial_year)
|
|
open_invoices = [row for row in invoices if row.status in {"ISSUED", "PARTLY_PAID", "OVERDUE"} and money(row.balance_amount) > Decimal("0.00")]
|
|
paid_invoices = [row for row in invoices if row.status == "PAID"]
|
|
outstanding = sum((money(row.balance_amount) for row in open_invoices), Decimal("0.00"))
|
|
latest_invoice = invoices[0] if invoices else None
|
|
latest_due_invoice = open_invoices[0] if open_invoices else None
|
|
return {
|
|
"billing_invoices": invoices,
|
|
"billing_open_invoices": open_invoices,
|
|
"billing_paid_invoices": paid_invoices,
|
|
"billing_outstanding_amount": money(outstanding),
|
|
"billing_latest_invoice": latest_invoice,
|
|
"billing_latest_due_invoice": latest_due_invoice,
|
|
"billing_open_count": len(open_invoices),
|
|
"billing_paid_count": len(paid_invoices),
|
|
"billing_total_count": len(invoices),
|
|
}
|
|
|
|
|
|
def build_client_payment_context(db: Session, invoice: BillingInvoice) -> dict[str, Any]:
|
|
invoice_ctx = build_invoice_print_context(db, invoice)
|
|
settings = invoice_ctx.get("settings")
|
|
amount_due = money(invoice.balance_amount)
|
|
firm_name = invoice_ctx.get("firm_name") or "Audit Firm"
|
|
upi_id = getattr(settings, "upi_id", None) if settings else None
|
|
upi_link = None
|
|
if upi_id and amount_due > Decimal("0.00"):
|
|
upi_link = (
|
|
"upi://pay?"
|
|
f"pa={quote(str(upi_id))}"
|
|
f"&pn={quote(str(firm_name))}"
|
|
f"&am={quote(str(amount_due))}"
|
|
"&cu=INR"
|
|
f"&tn={quote('Invoice ' + str(invoice.invoice_no))}"
|
|
)
|
|
return {
|
|
"invoice_ctx": invoice_ctx,
|
|
"amount_due": amount_due,
|
|
"upi_link": upi_link,
|
|
"upi_id": upi_id,
|
|
"bank_name": invoice_ctx.get("bank_name"),
|
|
"bank_account_name": invoice_ctx.get("bank_account_name"),
|
|
"bank_account_number": invoice_ctx.get("bank_account_number"),
|
|
"bank_ifsc": invoice_ctx.get("bank_ifsc"),
|
|
"payment_instructions": getattr(settings, "bank_details", None) if settings else None,
|
|
"payumoney_enabled": is_payumoney_ready(settings),
|
|
"payumoney_mode": getattr(settings, "payumoney_mode", "TEST") if settings else "TEST",
|
|
"cashfree_enabled": is_cashfree_ready(settings),
|
|
"cashfree_mode": getattr(settings, "cashfree_mode", "TEST") if settings else "TEST",
|
|
}
|