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