"""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, getdate, nowdate
# 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):
"""Assemble everything the statement template needs for one customer."""
cust = frappe.get_doc("Customer", customer)
invoices = _get_outstanding_invoices(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_outstanding = 0.0
for inv in invoices:
aging[inv["aging_bucket"]] += flt(inv["outstanding_amount"])
total_outstanding += flt(inv["outstanding_amount"])
# Late fee is booked and populated by generate_statements (later commit);
# get_statement_data on its own reports a zero fee.
late_fee = 0.0
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_outstanding": total_outstanding,
"late_fee": late_fee,
"total_due": total_outstanding + late_fee,
"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}
"""
@frappe.whitelist()
def generate_statements(customers):
"""Render printable statements (one page per customer) for the selection.
`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"))
pages, rendered, skipped = [], [], []
for customer in customers:
data = get_statement_data(customer)
if not data["invoices"]:
skipped.append(customer)
continue
pages.append(_render_page(data))
rendered.append(customer)
if not pages:
frappe.throw(_("None of the selected customers have an outstanding balance."))
return {"html": _wrap_document(pages), "rendered": rendered, "skipped": skipped}