"""Customer account statements.
Generates printable, one-customer-per-page account statements for customers with
overdue invoices, formatted for a standard double-window envelope. Statement
generation also books a late-payment fee to the ledger (see the late-fee helpers
added alongside the generator).
"""
import json
import frappe
from frappe import _
from frappe.contacts.doctype.address.address import get_address_display, get_default_address
from frappe.utils import flt, fmt_money, getdate, nowdate
# Dedicated naming series so late-fee invoices are easy to spot and filter.
LATE_FEE_NAMING_SERIES = "LPF-.YYYY.-"
# Roles allowed to run collections/statement actions.
ALLOWED_ROLES = [
"System Manager",
"Sales User",
"Sales Manager",
"Accounts User",
"Accounts Manager",
]
@frappe.whitelist()
def get_customers_with_overdue_invoices():
"""Return one row per customer that has at least one overdue Sales Invoice.
A Sales Invoice is overdue when it is submitted, still has an outstanding
balance, and its due date is in the past.
"""
frappe.only_for(ALLOWED_ROLES)
today = nowdate()
rows = frappe.get_all(
"Sales Invoice",
filters={
"docstatus": 1,
"outstanding_amount": [">", 0],
"due_date": ["<", today],
},
fields=[
"customer",
"customer_name",
"count(name) as overdue_count",
"sum(outstanding_amount) as total_outstanding",
"min(due_date) as oldest_due_date",
],
group_by="customer, customer_name",
order_by="total_outstanding desc",
)
for row in rows:
row["max_days_overdue"] = (
(getdate(today) - getdate(row.oldest_due_date)).days
if row.oldest_due_date
else 0
)
return rows
def _aging_bucket(days_overdue):
"""Map days-overdue to a standard aging bucket label."""
if days_overdue <= 0:
return "Current"
if days_overdue <= 30:
return "1-30"
if days_overdue <= 60:
return "31-60"
if days_overdue <= 90:
return "61-90"
return "90+"
def _get_outstanding_invoices(customer):
"""Return all open (submitted, unpaid) Sales Invoices for a customer.
The statement lists the full open balance, so this includes not-yet-due
invoices; each row is annotated with days overdue, an overdue flag, and its
aging bucket.
"""
today = getdate(nowdate())
invoices = frappe.get_all(
"Sales Invoice",
filters={
"customer": customer,
"docstatus": 1,
"outstanding_amount": [">", 0],
},
fields=[
"name",
"posting_date",
"due_date",
"outstanding_amount",
"grand_total",
"company",
],
order_by="due_date asc",
)
for inv in invoices:
due = getdate(inv.due_date) if inv.due_date else None
days = (today - due).days if due else 0
inv["days_overdue"] = days if days > 0 else 0
inv["is_overdue"] = days > 0
inv["aging_bucket"] = _aging_bucket(inv["days_overdue"])
return invoices
def _address_display(doctype, name):
"""Return the formatted (HTML) default address for a party, or ''."""
address_name = get_default_address(doctype, name)
if not address_name:
return ""
return get_address_display(frappe.get_doc("Address", address_name).as_dict()) or ""
def _resolve_company(invoices):
"""Pick the company for the statement header/return address."""
if invoices:
return invoices[0].company
return frappe.defaults.get_user_default("Company") or frappe.db.get_single_value(
"Global Defaults", "default_company"
)
def get_statement_data(customer, invoices=None):
"""Assemble everything the statement template needs for one customer.
Late-fee charges are billed as Sales Invoices, so they appear in the invoice
list like any other open item (flagged `is_late_fee`); there is no separate
fee total to add.
"""
cust = frappe.get_doc("Customer", customer)
if invoices is None:
invoices = _get_outstanding_invoices(customer)
fee_names = _late_fee_invoice_names(customer)
company = _resolve_company(invoices)
company_doc = frappe.get_doc("Company", company) if company else None
aging = {"Current": 0.0, "1-30": 0.0, "31-60": 0.0, "61-90": 0.0, "90+": 0.0}
total_due = 0.0
for inv in invoices:
inv["is_late_fee"] = inv["name"] in fee_names
aging[inv["aging_bucket"]] += flt(inv["outstanding_amount"])
total_due += flt(inv["outstanding_amount"])
return {
"customer": cust.name,
"customer_name": cust.customer_name,
"customer_address": _address_display("Customer", cust.name),
"company": company,
"company_name": company_doc.company_name if company_doc else "",
"return_address": _address_display("Company", company) if company else "",
"currency": (company_doc.default_currency if company_doc else None)
or frappe.db.get_single_value("Global Defaults", "default_currency"),
"invoices": invoices,
"aging": aging,
"total_due": total_due,
"statement_date": nowdate(),
}
def _render_page(data):
path = frappe.get_app_path(
"ns_app", "templates", "statements", "customer_statement.html"
)
with open(path) as f:
template = f.read()
return frappe.render_template(template, {"s": data})
def _wrap_document(pages):
"""Wrap rendered per-customer pages in a printable HTML document."""
body = "\n".join(pages)
return f"""
Customer Statements
{body}
"""
# ── Late-payment fee (billed as a Sales Invoice on generation) ───────────────
#
# Fee schedule/amounts come from ERPNext's existing **Dunning Type** settings
# (rate_of_interest is a yearly %, plus a flat dunning_fee), editable in the
# desk. Interest is computed with ERPNext's own Dunning formula so the numbers
# match a Dunning document. The fee is billed as a submitted **Sales Invoice**
# (item -> Dunning Type income account) so it both hits the ledger and is
# collectible by the app's existing payment flow (Run Payment / AutoPay /
# multi-invoice), which settles Sales Invoices.
DUNNING_TYPE_FIELDS = [
"name",
"rate_of_interest",
"dunning_fee",
"income_account",
"cost_center",
"company",
"custom_late_fee_item",
]
def _get_fee_settings(company):
"""Resolve the Dunning Type used for late fees for a company.
Nothing is auto-created: the user must configure a Dunning Type (rate of
interest, fee, income account) for the company in ERPNext. If none exists we
stop with a clear, actionable error rather than inventing default values.
"""
def first(filters):
rows = frappe.get_all(
"Dunning Type", filters=filters, fields=DUNNING_TYPE_FIELDS, limit=1
)
return rows[0] if rows else None
dt = first({"company": company, "is_default": 1}) or first({"company": company})
if not dt:
frappe.throw(
_(
"No Dunning Type is configured for {0}. Create one under "
"Accounting > Dunning Type — set the rate of interest, dunning "
"fee, and income account — before generating statements."
).format(company)
)
return dt
def _late_fee_period():
"""Statement period key used for idempotency (one fee per calendar month)."""
return getdate(nowdate()).strftime("%Y-%m")
def _get_fee_invoices(customer, company, fee_item):
"""Return submitted late-fee Sales Invoices for a customer (by fee item)."""
if not fee_item:
return []
return frappe.db.sql(
"""
select si.name, si.posting_date
from `tabSales Invoice` si
inner join `tabSales Invoice Item` sii on sii.parent = si.name
where si.customer = %s and si.company = %s and si.docstatus = 1
and sii.item_code = %s
""",
(customer, company, fee_item),
as_dict=True,
)
def _post_late_fee_invoice(customer, company, overdue_invoices, period):
"""Bill a late fee as a submitted Sales Invoice (idempotent per month).
Returns the fee invoice name, or None if nothing was billed.
"""
if not overdue_invoices:
return None
settings = _get_fee_settings(company)
fee_item = settings.get("custom_late_fee_item")
if not fee_item:
frappe.throw(
_("Set a Late Fee Item on Dunning Type {0} before generating statements.").format(
settings.name
)
)
if not settings.income_account:
frappe.throw(
_("Set an Income Account on Dunning Type {0} before generating statements.").format(
settings.name
)
)
fee_invoices = _get_fee_invoices(customer, company, fee_item)
# Idempotency: at most one fee invoice per (customer, company, month).
month_start = getdate(period + "-01")
for fi in fee_invoices:
if getdate(fi.posting_date) >= month_start:
return fi.name
# Interest on overdue balances, excluding prior fee invoices (no fee-on-fee).
prior_fee_names = {fi.name for fi in fee_invoices}
daily_interest = flt(settings.rate_of_interest) / 100.0 / 365.0
interest = sum(
flt(inv["outstanding_amount"]) * daily_interest * inv["days_overdue"]
for inv in overdue_invoices
if inv["name"] not in prior_fee_names
)
fee = round(interest + flt(settings.dunning_fee), 2)
if fee <= 0:
return None
cost_center = settings.cost_center or frappe.get_cached_value(
"Company", company, "cost_center"
)
si = frappe.new_doc("Sales Invoice")
si.naming_series = LATE_FEE_NAMING_SERIES
si.customer = customer
si.company = company
si.posting_date = nowdate()
si.due_date = nowdate()
si.append(
"items",
{
"item_code": fee_item,
"qty": 1,
"rate": fee,
"income_account": settings.income_account,
"cost_center": cost_center,
"description": _("Late payment fee for statement period {0}").format(period),
},
)
# Late fees are not taxed. A single zero "Actual" tax line keeps the taxes
# table non-empty, which stops ERPNext from auto-applying the company or
# item tax templates; being zero it posts nothing to the ledger.
si.taxes_and_charges = ""
si.append(
"taxes",
{
"charge_type": "Actual",
"account_head": settings.income_account,
"description": _("Late fees are not taxed"),
"tax_amount": 0,
"rate": 0,
},
)
si.insert(ignore_permissions=True)
si.submit()
frappe.db.commit()
return si.name
def _late_fee_invoice_names(customer):
"""Names of the customer's submitted late-fee Sales Invoices (any company)."""
fee_items = [
d.custom_late_fee_item
for d in frappe.get_all("Dunning Type", fields=["custom_late_fee_item"])
if d.custom_late_fee_item
]
if not fee_items:
return set()
rows = frappe.db.sql(
"""
select distinct sii.parent
from `tabSales Invoice Item` sii
inner join `tabSales Invoice` si on si.name = sii.parent
where si.customer = %s and si.docstatus = 1 and sii.item_code in %s
""",
(customer, tuple(fee_items)),
as_dict=True,
)
return {r.parent for r in rows}
def _record_statement_activity(customer, data, fee_invoice_names, skip_late_fee):
"""Log statement generation on the customer's timeline (audit trail)."""
total = fmt_money(data["total_due"], currency=data["currency"])
if skip_late_fee:
fee_note = _("late fee skipped")
elif fee_invoice_names:
fee_note = _("late fee invoice {0}").format(", ".join(fee_invoice_names))
else:
fee_note = _("no late fee")
frappe.get_doc("Customer", customer).add_comment(
"Info", _("Statement generated — Total Due {0} ({1}).").format(total, fee_note)
)
@frappe.whitelist()
def generate_statements(customers, skip_late_fee=0):
"""Render printable statements (one page per customer) for the selection.
Side effect (unless `skip_late_fee`): a late-payment fee is billed as a
Sales Invoice (once per customer per month) for each customer with overdue
invoices. Each generation is recorded on the customer's timeline.
`customers` may arrive as a JSON-encoded list from the client.
"""
frappe.only_for(ALLOWED_ROLES)
if isinstance(customers, str):
try:
customers = json.loads(customers)
except (ValueError, TypeError):
customers = [customers]
if not customers:
frappe.throw(_("No customers selected"))
skip_late_fee = int(skip_late_fee or 0)
period = _late_fee_period()
pages, rendered, skipped = [], [], []
for customer in customers:
invoices = _get_outstanding_invoices(customer)
if not invoices:
skipped.append(customer)
continue
# Bill the late fee per company (on overdue invoices only).
fee_invoice_names = []
if not skip_late_fee:
overdue_by_company = {}
for inv in invoices:
if inv["is_overdue"]:
overdue_by_company.setdefault(inv["company"], []).append(inv)
for comp, invs in overdue_by_company.items():
name = _post_late_fee_invoice(customer, comp, invs, period)
if name:
fee_invoice_names.append(name)
# Re-fetch so the statement includes the freshly billed fee invoice(s).
data = get_statement_data(customer)
pages.append(_render_page(data))
rendered.append(customer)
_record_statement_activity(customer, data, fee_invoice_names, skip_late_fee)
if not pages:
frappe.throw(_("None of the selected customers have an outstanding balance."))
return {"html": _wrap_document(pages), "rendered": rendered, "skipped": skipped}