refactor: convert payment schedule and billing validation to service objects

Introduce PaymentScheduleService and BillingValidationService classes so
call sites read PaymentScheduleService(doc).set_payment_schedule() instead
of the opaque self.set_payment_schedule() shim. Removes 15 shim methods
from AccountsController and updates all 11 call sites across the codebase.
This commit is contained in:
Nabin Hait
2026-05-28 17:13:23 +05:30
parent 8aaa3a72ef
commit 6c1ac51d7a
11 changed files with 443 additions and 483 deletions

View File

@@ -290,7 +290,9 @@ class PurchaseInvoice(BuyingController):
self.validate_expense_account()
self.set_against_expense_account()
self.validate_write_off_account()
self.validate_multiple_billing("Purchase Receipt", "pr_detail", "amount")
from erpnext.accounts.services.billing_validation import BillingValidationService
BillingValidationService(self).validate_multiple_billing("Purchase Receipt", "pr_detail", "amount")
self.set_status()
self.validate_purchase_receipt_if_update_stock()
validate_inter_company_party(

View File

@@ -362,7 +362,9 @@ class SalesInvoice(SellingController):
if not self.is_return:
self.validate_time_sheets_are_submitted()
self.validate_multiple_billing("Delivery Note", "dn_detail", "amount")
from erpnext.accounts.services.billing_validation import BillingValidationService
BillingValidationService(self).validate_multiple_billing("Delivery Note", "dn_detail", "amount")
if self.is_return and self.return_against:
for row in self.timesheets:

View File

@@ -9,139 +9,143 @@ from frappe.query_builder.functions import Sum
from frappe.utils import cint, flt, fmt_money
def validate_multiple_billing(doc, ref_dt: str, item_ref_dn: str, based_on: str) -> None:
from erpnext.controllers.status_updater import get_allowance_for
class BillingValidationService:
def __init__(self, doc):
self.doc = doc
ref_wise_billed_amount = get_reference_wise_billed_amt(doc, ref_dt, item_ref_dn, based_on)
if not ref_wise_billed_amount:
return
def validate_multiple_billing(self, ref_dt: str, item_ref_dn: str, based_on: str) -> None:
from erpnext.controllers.status_updater import get_allowance_for
total_overbilled_amt = 0.0
overbilled_items = []
precision = doc.precision(based_on, "items")
precision_allowance = 1 / (10**precision)
ref_wise_billed_amount = self.get_reference_wise_billed_amt(ref_dt, item_ref_dn, based_on)
if not ref_wise_billed_amount:
return
role_allowed_to_overbill = frappe.get_single_value("Accounts Settings", "role_allowed_to_over_bill")
is_overbilling_allowed = role_allowed_to_overbill in frappe.get_roles()
total_overbilled_amt = 0.0
overbilled_items = []
precision = self.doc.precision(based_on, "items")
precision_allowance = 1 / (10**precision)
for row in ref_wise_billed_amount.values():
total_billed_amt = row.billed_amt
allowance = get_allowance_for(row.item_code, {}, None, None, "amount")[0]
max_allowed_amt = flt(row.ref_amt * (100 + allowance) / 100)
role_allowed_to_overbill = frappe.get_single_value("Accounts Settings", "role_allowed_to_over_bill")
is_overbilling_allowed = role_allowed_to_overbill in frappe.get_roles()
if total_billed_amt < 0 and max_allowed_amt < 0:
total_billed_amt, max_allowed_amt = abs(total_billed_amt), abs(max_allowed_amt)
for row in ref_wise_billed_amount.values():
total_billed_amt = row.billed_amt
allowance = get_allowance_for(row.item_code, {}, None, None, "amount")[0]
max_allowed_amt = flt(row.ref_amt * (100 + allowance) / 100)
overbill_amt = total_billed_amt - max_allowed_amt
row["max_allowed_amt"] = max_allowed_amt
total_overbilled_amt += overbill_amt
if total_billed_amt < 0 and max_allowed_amt < 0:
total_billed_amt, max_allowed_amt = abs(total_billed_amt), abs(max_allowed_amt)
if overbill_amt > precision_allowance and not is_overbilling_allowed:
if doc.doctype != "Purchase Invoice" or not cint(
frappe.db.get_single_value(
"Buying Settings", "bill_for_rejected_quantity_in_purchase_invoice"
)
):
overbilled_items.append(row)
overbill_amt = total_billed_amt - max_allowed_amt
row["max_allowed_amt"] = max_allowed_amt
total_overbilled_amt += overbill_amt
if overbilled_items:
throw_overbill_exception(doc, overbilled_items, precision)
if overbill_amt > precision_allowance and not is_overbilling_allowed:
if self.doc.doctype != "Purchase Invoice" or not cint(
frappe.db.get_single_value(
"Buying Settings", "bill_for_rejected_quantity_in_purchase_invoice"
)
):
overbilled_items.append(row)
if is_overbilling_allowed and total_overbilled_amt > 0.1:
frappe.msgprint(
_("Overbilling of {} ignored because you have {} role.").format(
total_overbilled_amt, role_allowed_to_overbill
),
indicator="orange",
alert=True,
)
if overbilled_items:
self.throw_overbill_exception(overbilled_items, precision)
def get_reference_wise_billed_amt(doc, ref_dt: str, item_ref_dn: str, based_on: str) -> dict | None:
"""Return sum of billed amounts per reference row, including previously submitted invoices."""
reference_names = [d.get(item_ref_dn) for d in doc.items if d.get(item_ref_dn)]
if not reference_names:
return
precision = doc.precision(based_on, "items")
reference_details = get_billing_reference_details(doc, reference_names, ref_dt + " Item", based_on)
already_billed = get_already_billed_amount(doc, reference_names, item_ref_dn, based_on)
ref_wise_billed_amount = {}
for item in doc.items:
key = item.get(item_ref_dn)
if not key:
continue
ref_amt = flt(reference_details.get(key), precision)
current_amount = flt(item.get(based_on), precision)
if not ref_amt:
if current_amount:
frappe.msgprint(
_("System will not check over billing since amount for Item {0} in {1} is zero").format(
item.item_code, ref_dt
),
title=_("Warning"),
indicator="orange",
)
continue
ref_wise_billed_amount.setdefault(
key,
frappe._dict(item_code=item.item_code, billed_amt=0.0, ref_amt=ref_amt, rows=[]),
)
ref_wise_billed_amount[key]["rows"].append(item.idx)
ref_wise_billed_amount[key]["ref_amt"] = ref_amt
ref_wise_billed_amount[key]["billed_amt"] += current_amount
if key in already_billed:
ref_wise_billed_amount[key]["billed_amt"] += flt(already_billed.pop(key, 0), precision)
return ref_wise_billed_amount
def get_billing_reference_details(
doc, reference_names: list, reference_doctype: str, based_on: str
) -> frappe._dict:
return frappe._dict(
frappe.get_all(
reference_doctype,
filters={"name": ("in", reference_names)},
fields=["name", based_on],
as_list=1,
)
)
def get_already_billed_amount(doc, reference_names: list, item_ref_dn: str, based_on: str) -> frappe._dict:
item_doctype = frappe.qb.DocType(doc.items[0].doctype)
based_on_field = frappe.qb.Field(based_on)
join_field = frappe.qb.Field(item_ref_dn)
return frappe._dict(
(
frappe.qb.from_(item_doctype)
.select(join_field, Sum(based_on_field))
.where(join_field.isin(reference_names))
.where((item_doctype.docstatus == 1) & (item_doctype.parent != doc.name))
.groupby(join_field)
).run()
)
def throw_overbill_exception(doc, overbilled_items: list, precision: int) -> None:
message = (
_("<p>Cannot overbill for the following Items:</p>")
+ "<ul>"
+ "".join(
_("<li>Item {0} in row(s) {1} billed more than {2}</li>").format(
frappe.bold(item.item_code),
", ".join(str(x) for x in item.rows),
frappe.bold(fmt_money(item.max_allowed_amt, precision=precision, currency=doc.currency)),
if is_overbilling_allowed and total_overbilled_amt > 0.1:
frappe.msgprint(
_("Overbilling of {} ignored because you have {} role.").format(
total_overbilled_amt, role_allowed_to_overbill
),
indicator="orange",
alert=True,
)
def get_reference_wise_billed_amt(self, ref_dt: str, item_ref_dn: str, based_on: str) -> dict | None:
"""Return sum of billed amounts per reference row, including previously submitted invoices."""
reference_names = [d.get(item_ref_dn) for d in self.doc.items if d.get(item_ref_dn)]
if not reference_names:
return
precision = self.doc.precision(based_on, "items")
reference_details = self.get_billing_reference_details(reference_names, ref_dt + " Item", based_on)
already_billed = self.get_already_billed_amount(reference_names, item_ref_dn, based_on)
ref_wise_billed_amount = {}
for item in self.doc.items:
key = item.get(item_ref_dn)
if not key:
continue
ref_amt = flt(reference_details.get(key), precision)
current_amount = flt(item.get(based_on), precision)
if not ref_amt:
if current_amount:
frappe.msgprint(
_(
"System will not check over billing since amount for Item {0} in {1} is zero"
).format(item.item_code, ref_dt),
title=_("Warning"),
indicator="orange",
)
continue
ref_wise_billed_amount.setdefault(
key,
frappe._dict(item_code=item.item_code, billed_amt=0.0, ref_amt=ref_amt, rows=[]),
)
ref_wise_billed_amount[key]["rows"].append(item.idx)
ref_wise_billed_amount[key]["ref_amt"] = ref_amt
ref_wise_billed_amount[key]["billed_amt"] += current_amount
if key in already_billed:
ref_wise_billed_amount[key]["billed_amt"] += flt(already_billed.pop(key, 0), precision)
return ref_wise_billed_amount
def get_billing_reference_details(
self, reference_names: list, reference_doctype: str, based_on: str
) -> frappe._dict:
return frappe._dict(
frappe.get_all(
reference_doctype,
filters={"name": ("in", reference_names)},
fields=["name", based_on],
as_list=1,
)
for item in overbilled_items
)
+ "</ul>"
)
message += _("<p>To allow over-billing, please set allowance in Accounts Settings.</p>")
frappe.throw(_(message))
def get_already_billed_amount(
self, reference_names: list, item_ref_dn: str, based_on: str
) -> frappe._dict:
item_doctype = frappe.qb.DocType(self.doc.items[0].doctype)
based_on_field = frappe.qb.Field(based_on)
join_field = frappe.qb.Field(item_ref_dn)
return frappe._dict(
(
frappe.qb.from_(item_doctype)
.select(join_field, Sum(based_on_field))
.where(join_field.isin(reference_names))
.where((item_doctype.docstatus == 1) & (item_doctype.parent != self.doc.name))
.groupby(join_field)
).run()
)
def throw_overbill_exception(self, overbilled_items: list, precision: int) -> None:
message = (
_("<p>Cannot overbill for the following Items:</p>")
+ "<ul>"
+ "".join(
_("<li>Item {0} in row(s) {1} billed more than {2}</li>").format(
frappe.bold(item.item_code),
", ".join(str(x) for x in item.rows),
frappe.bold(
fmt_money(item.max_allowed_amt, precision=precision, currency=self.doc.currency)
),
)
for item in overbilled_items
)
+ "</ul>"
)
message += _("<p>To allow over-billing, please set allowance in Accounts Settings.</p>")
frappe.throw(_(message))

View File

@@ -10,256 +10,37 @@ from frappe.utils import DateTimeLikeObject, add_days, add_months, cint, flt, ge
from erpnext.accounts.party import get_party_account_currency
def set_payment_schedule(doc) -> None:
if (doc.doctype == "Sales Invoice" and doc.is_pos) or doc.get("is_opening") == "Yes":
doc.payment_terms_template = ""
return
class PaymentScheduleService:
def __init__(self, doc):
self.doc = doc
party_account_currency = doc.get("party_account_currency")
if not party_account_currency:
party_type, party = doc.get_party()
if party_type and party:
party_account_currency = get_party_account_currency(party_type, party, doc.company)
def set_payment_schedule(self) -> None:
doc = self.doc
if (doc.doctype == "Sales Invoice" and doc.is_pos) or doc.get("is_opening") == "Yes":
doc.payment_terms_template = ""
return
posting_date = doc.get("bill_date") or doc.get("posting_date") or doc.get("transaction_date")
due_date = doc.get("due_date") or posting_date
party_account_currency = doc.get("party_account_currency")
if not party_account_currency:
party_type, party = doc.get_party()
if party_type and party:
party_account_currency = get_party_account_currency(party_type, party, doc.company)
base_grand_total = flt(doc.get("base_rounded_total") or doc.base_grand_total)
grand_total = flt(doc.get("rounded_total") or doc.grand_total)
automatically_fetch_payment_terms = 0
if doc.doctype in ("Sales Invoice", "Purchase Invoice", "Sales Order"):
po_or_so, doctype, fieldname = get_order_details(doc)
automatically_fetch_payment_terms = cint(
frappe.get_single_value("Accounts Settings", "automatically_fetch_payment_terms")
)
if doc.doctype != "Sales Order":
base_grand_total = base_grand_total - flt(doc.base_write_off_amount)
grand_total = grand_total - flt(doc.write_off_amount)
if doc.get("total_advance"):
if party_account_currency == doc.company_currency:
base_grand_total -= doc.get("total_advance")
grand_total = flt(base_grand_total / doc.get("conversion_rate"), doc.precision("grand_total"))
else:
grand_total -= doc.get("total_advance")
base_grand_total = flt(
grand_total * doc.get("conversion_rate"), doc.precision("base_grand_total")
)
if not doc.get("payment_schedule"):
if (
doc.doctype in ["Sales Invoice", "Purchase Invoice", "Sales Order"]
and automatically_fetch_payment_terms
and linked_order_has_payment_terms(doc, po_or_so, fieldname, doctype)
):
fetch_payment_terms_from_order(
doc, po_or_so, doctype, grand_total, base_grand_total, automatically_fetch_payment_terms
)
if doc.get("payment_terms_template"):
doc.ignore_default_payment_terms_template = 1
elif doc.get("payment_terms_template"):
data = get_payment_terms(doc.payment_terms_template, posting_date, grand_total, base_grand_total)
for item in data:
doc.append("payment_schedule", item)
elif doc.doctype not in ["Purchase Receipt"]:
doc.append(
"payment_schedule",
dict(
due_date=due_date,
invoice_portion=100,
payment_amount=grand_total,
base_payment_amount=base_grand_total,
),
)
allocate_payment_based_on_payment_terms = frappe.db.get_value(
"Payment Terms Template",
doc.payment_terms_template,
"allocate_payment_based_on_payment_terms",
)
if not (
automatically_fetch_payment_terms
and allocate_payment_based_on_payment_terms
and linked_order_has_payment_terms(doc, po_or_so, fieldname, doctype)
):
for d in doc.get("payment_schedule"):
if d.invoice_portion:
d.payment_amount = flt(
grand_total * flt(d.invoice_portion) / 100, d.precision("payment_amount")
)
d.base_payment_amount = flt(
base_grand_total * flt(d.invoice_portion) / 100, d.precision("base_payment_amount")
)
d.outstanding = d.payment_amount
d.base_outstanding = d.base_payment_amount
elif not d.invoice_portion:
d.base_payment_amount = flt(
d.payment_amount * doc.get("conversion_rate"), d.precision("base_payment_amount")
)
d.base_outstanding = d.base_payment_amount
else:
fetch_payment_terms_from_order(
doc, po_or_so, doctype, grand_total, base_grand_total, automatically_fetch_payment_terms
)
doc.ignore_default_payment_terms_template = 1
def get_order_details(doc) -> tuple:
if not doc.get("items"):
return None, None, None
if doc.doctype == "Sales Invoice":
prev_doc = doc.get("items")[0].get("sales_order")
prev_doctype = "Sales Order"
prev_doctype_name = "sales_order"
elif doc.doctype == "Purchase Invoice":
prev_doc = doc.get("items")[0].get("purchase_order")
prev_doctype = "Purchase Order"
prev_doctype_name = "purchase_order"
else:
prev_doc = doc.get("items")[0].get("prevdoc_docname")
prev_doctype = "Quotation"
prev_doctype_name = "prevdoc_docname"
return prev_doc, prev_doctype, prev_doctype_name
def linked_order_has_payment_terms(doc, po_or_so, fieldname, doctype) -> bool:
if po_or_so and all_items_have_same_po_or_so(doc, po_or_so, fieldname):
if linked_order_has_payment_terms_template(po_or_so, doctype):
return True
elif linked_order_has_payment_schedule(po_or_so):
return True
return False
def all_items_have_same_po_or_so(doc, po_or_so, fieldname) -> bool:
for item in doc.get("items"):
if item.get(fieldname) != po_or_so:
return False
return True
def linked_order_has_payment_terms_template(po_or_so, doctype) -> str | None:
return frappe.get_value(doctype, po_or_so, "payment_terms_template")
def linked_order_has_payment_schedule(po_or_so) -> list:
return frappe.get_all("Payment Schedule", filters={"parent": po_or_so})
def fetch_payment_terms_from_order(
doc, po_or_so, po_or_so_doctype, grand_total, base_grand_total, automatically_fetch_payment_terms
) -> None:
"""Fetch Payment Terms from Purchase/Sales Order when creating a new invoice."""
po_or_so = frappe.get_cached_doc(po_or_so_doctype, po_or_so)
doc.payment_schedule = []
doc.payment_terms_template = po_or_so.payment_terms_template
posting_date = doc.get("bill_date") or doc.get("posting_date") or doc.get("transaction_date")
for schedule in po_or_so.payment_schedule:
payment_schedule = {
"payment_term": schedule.payment_term,
"due_date": schedule.due_date,
"invoice_portion": schedule.invoice_portion,
"mode_of_payment": schedule.mode_of_payment,
"description": schedule.description,
"paid_amount": schedule.paid_amount,
}
if automatically_fetch_payment_terms:
if schedule.due_date_based_on:
payment_schedule["due_date"] = get_due_date(schedule, posting_date)
payment_schedule["due_date_based_on"] = schedule.due_date_based_on
payment_schedule["credit_days"] = cint(schedule.credit_days)
payment_schedule["credit_months"] = cint(schedule.credit_months)
if schedule.discount_validity_based_on:
payment_schedule["discount_date"] = get_discount_date(schedule, posting_date)
payment_schedule["discount_validity_based_on"] = schedule.discount_validity_based_on
payment_schedule["discount_validity"] = cint(schedule.discount_validity)
payment_schedule["payment_amount"] = flt(
grand_total * flt(payment_schedule["invoice_portion"]) / 100,
schedule.precision("payment_amount"),
)
payment_schedule["base_payment_amount"] = flt(
base_grand_total * flt(payment_schedule["invoice_portion"]) / 100,
schedule.precision("base_payment_amount"),
)
payment_schedule["outstanding"] = payment_schedule["payment_amount"]
else:
payment_schedule["base_payment_amount"] = flt(
schedule.base_payment_amount * doc.get("conversion_rate"),
schedule.precision("base_payment_amount"),
)
if schedule.discount_type == "Percentage":
payment_schedule["discount_type"] = schedule.discount_type
payment_schedule["discount"] = schedule.discount
if not schedule.invoice_portion:
payment_schedule["payment_amount"] = schedule.payment_amount
doc.append("payment_schedule", payment_schedule)
def set_due_date(doc) -> None:
due_dates = [d.due_date for d in doc.get("payment_schedule") if d.due_date]
if due_dates:
doc.due_date = max(due_dates)
def validate_payment_schedule_dates(doc) -> None:
dates = []
li = []
if doc.doctype == "Sales Invoice" and doc.is_pos:
return
for d in doc.get("payment_schedule"):
d.validate_from_to_dates("discount_date", "due_date")
if doc.doctype in ["Sales Order", "Quotation"] and getdate(d.due_date) < getdate(
doc.transaction_date
):
frappe.throw(
_("Row {0}: Due Date in the Payment Terms table cannot be before Posting Date").format(d.idx)
)
elif d.due_date in dates:
li.append(_("{0} in row {1}").format(d.due_date, d.idx))
dates.append(d.due_date)
if li:
frappe.throw(
_("Rows with duplicate due dates in other rows were found: {0}").format("<br>" + "<br>".join(li)),
title=_("Payment Schedule"),
)
def validate_payment_schedule_amount(doc) -> None:
if (doc.doctype == "Sales Invoice" and doc.is_pos) or doc.get("is_opening") == "Yes":
return
party_account_currency = doc.get("party_account_currency")
if not party_account_currency:
party_type, party = doc.get_party()
if party_type and party:
party_account_currency = get_party_account_currency(party_type, party, doc.company)
if doc.get("payment_schedule"):
total = 0
base_total = 0
for d in doc.get("payment_schedule"):
total += flt(d.payment_amount, d.precision("payment_amount"))
base_total += flt(d.base_payment_amount, d.precision("base_payment_amount"))
posting_date = doc.get("bill_date") or doc.get("posting_date") or doc.get("transaction_date")
due_date = doc.get("due_date") or posting_date
base_grand_total = flt(doc.get("base_rounded_total") or doc.base_grand_total)
grand_total = flt(doc.get("rounded_total") or doc.grand_total)
automatically_fetch_payment_terms = 0
if doc.doctype in ("Sales Invoice", "Purchase Invoice"):
base_grand_total = base_grand_total - flt(doc.base_write_off_amount)
grand_total = grand_total - flt(doc.write_off_amount)
if doc.doctype in ("Sales Invoice", "Purchase Invoice", "Sales Order"):
po_or_so, doctype, fieldname = self.get_order_details()
automatically_fetch_payment_terms = cint(
frappe.get_single_value("Accounts Settings", "automatically_fetch_payment_terms")
)
if doc.doctype != "Sales Order":
base_grand_total = base_grand_total - flt(doc.base_write_off_amount)
grand_total = grand_total - flt(doc.write_off_amount)
if doc.get("total_advance"):
if party_account_currency == doc.company_currency:
@@ -271,16 +52,252 @@ def validate_payment_schedule_amount(doc) -> None:
grand_total * doc.get("conversion_rate"), doc.precision("base_grand_total")
)
if (
abs(flt(total, doc.precision("grand_total")) - flt(grand_total, doc.precision("grand_total")))
> 0.1
or abs(
flt(base_total, doc.precision("base_grand_total"))
- flt(base_grand_total, doc.precision("base_grand_total"))
)
> 0.1
if not doc.get("payment_schedule"):
if (
doc.doctype in ["Sales Invoice", "Purchase Invoice", "Sales Order"]
and automatically_fetch_payment_terms
and self.linked_order_has_payment_terms(po_or_so, fieldname, doctype)
):
self.fetch_payment_terms_from_order(
po_or_so, doctype, grand_total, base_grand_total, automatically_fetch_payment_terms
)
if doc.get("payment_terms_template"):
doc.ignore_default_payment_terms_template = 1
elif doc.get("payment_terms_template"):
data = get_payment_terms(
doc.payment_terms_template, posting_date, grand_total, base_grand_total
)
for item in data:
doc.append("payment_schedule", item)
elif doc.doctype not in ["Purchase Receipt"]:
doc.append(
"payment_schedule",
dict(
due_date=due_date,
invoice_portion=100,
payment_amount=grand_total,
base_payment_amount=base_grand_total,
),
)
allocate_payment_based_on_payment_terms = frappe.db.get_value(
"Payment Terms Template",
doc.payment_terms_template,
"allocate_payment_based_on_payment_terms",
)
if not (
automatically_fetch_payment_terms
and allocate_payment_based_on_payment_terms
and self.linked_order_has_payment_terms(po_or_so, fieldname, doctype)
):
frappe.throw(_("Total Payment Amount in Payment Schedule must be equal to Grand / Rounded Total"))
for d in doc.get("payment_schedule"):
if d.invoice_portion:
d.payment_amount = flt(
grand_total * flt(d.invoice_portion) / 100, d.precision("payment_amount")
)
d.base_payment_amount = flt(
base_grand_total * flt(d.invoice_portion) / 100, d.precision("base_payment_amount")
)
d.outstanding = d.payment_amount
d.base_outstanding = d.base_payment_amount
elif not d.invoice_portion:
d.base_payment_amount = flt(
d.payment_amount * doc.get("conversion_rate"), d.precision("base_payment_amount")
)
d.base_outstanding = d.base_payment_amount
else:
self.fetch_payment_terms_from_order(
po_or_so, doctype, grand_total, base_grand_total, automatically_fetch_payment_terms
)
doc.ignore_default_payment_terms_template = 1
def get_order_details(self) -> tuple:
doc = self.doc
if not doc.get("items"):
return None, None, None
if doc.doctype == "Sales Invoice":
prev_doc = doc.get("items")[0].get("sales_order")
prev_doctype = "Sales Order"
prev_doctype_name = "sales_order"
elif doc.doctype == "Purchase Invoice":
prev_doc = doc.get("items")[0].get("purchase_order")
prev_doctype = "Purchase Order"
prev_doctype_name = "purchase_order"
else:
prev_doc = doc.get("items")[0].get("prevdoc_docname")
prev_doctype = "Quotation"
prev_doctype_name = "prevdoc_docname"
return prev_doc, prev_doctype, prev_doctype_name
def linked_order_has_payment_terms(self, po_or_so, fieldname, doctype) -> bool:
if po_or_so and self.all_items_have_same_po_or_so(po_or_so, fieldname):
if linked_order_has_payment_terms_template(po_or_so, doctype):
return True
elif linked_order_has_payment_schedule(po_or_so):
return True
return False
def all_items_have_same_po_or_so(self, po_or_so, fieldname) -> bool:
for item in self.doc.get("items"):
if item.get(fieldname) != po_or_so:
return False
return True
def fetch_payment_terms_from_order(
self,
po_or_so,
po_or_so_doctype,
grand_total,
base_grand_total,
automatically_fetch_payment_terms,
) -> None:
"""Fetch Payment Terms from Purchase/Sales Order when creating a new invoice."""
doc = self.doc
po_or_so = frappe.get_cached_doc(po_or_so_doctype, po_or_so)
doc.payment_schedule = []
doc.payment_terms_template = po_or_so.payment_terms_template
posting_date = doc.get("bill_date") or doc.get("posting_date") or doc.get("transaction_date")
for schedule in po_or_so.payment_schedule:
payment_schedule = {
"payment_term": schedule.payment_term,
"due_date": schedule.due_date,
"invoice_portion": schedule.invoice_portion,
"mode_of_payment": schedule.mode_of_payment,
"description": schedule.description,
"paid_amount": schedule.paid_amount,
}
if automatically_fetch_payment_terms:
if schedule.due_date_based_on:
payment_schedule["due_date"] = get_due_date(schedule, posting_date)
payment_schedule["due_date_based_on"] = schedule.due_date_based_on
payment_schedule["credit_days"] = cint(schedule.credit_days)
payment_schedule["credit_months"] = cint(schedule.credit_months)
if schedule.discount_validity_based_on:
payment_schedule["discount_date"] = get_discount_date(schedule, posting_date)
payment_schedule["discount_validity_based_on"] = schedule.discount_validity_based_on
payment_schedule["discount_validity"] = cint(schedule.discount_validity)
payment_schedule["payment_amount"] = flt(
grand_total * flt(payment_schedule["invoice_portion"]) / 100,
schedule.precision("payment_amount"),
)
payment_schedule["base_payment_amount"] = flt(
base_grand_total * flt(payment_schedule["invoice_portion"]) / 100,
schedule.precision("base_payment_amount"),
)
payment_schedule["outstanding"] = payment_schedule["payment_amount"]
else:
payment_schedule["base_payment_amount"] = flt(
schedule.base_payment_amount * doc.get("conversion_rate"),
schedule.precision("base_payment_amount"),
)
if schedule.discount_type == "Percentage":
payment_schedule["discount_type"] = schedule.discount_type
payment_schedule["discount"] = schedule.discount
if not schedule.invoice_portion:
payment_schedule["payment_amount"] = schedule.payment_amount
doc.append("payment_schedule", payment_schedule)
def set_due_date(self) -> None:
due_dates = [d.due_date for d in self.doc.get("payment_schedule") if d.due_date]
if due_dates:
self.doc.due_date = max(due_dates)
def validate_payment_schedule_dates(self) -> None:
dates = []
li = []
doc = self.doc
if doc.doctype == "Sales Invoice" and doc.is_pos:
return
for d in doc.get("payment_schedule"):
d.validate_from_to_dates("discount_date", "due_date")
if doc.doctype in ["Sales Order", "Quotation"] and getdate(d.due_date) < getdate(
doc.transaction_date
):
frappe.throw(
_("Row {0}: Due Date in the Payment Terms table cannot be before Posting Date").format(
d.idx
)
)
elif d.due_date in dates:
li.append(_("{0} in row {1}").format(d.due_date, d.idx))
dates.append(d.due_date)
if li:
frappe.throw(
_("Rows with duplicate due dates in other rows were found: {0}").format(
"<br>" + "<br>".join(li)
),
title=_("Payment Schedule"),
)
def validate_payment_schedule_amount(self) -> None:
doc = self.doc
if (doc.doctype == "Sales Invoice" and doc.is_pos) or doc.get("is_opening") == "Yes":
return
party_account_currency = doc.get("party_account_currency")
if not party_account_currency:
party_type, party = doc.get_party()
if party_type and party:
party_account_currency = get_party_account_currency(party_type, party, doc.company)
if doc.get("payment_schedule"):
total = 0
base_total = 0
for d in doc.get("payment_schedule"):
total += flt(d.payment_amount, d.precision("payment_amount"))
base_total += flt(d.base_payment_amount, d.precision("base_payment_amount"))
base_grand_total = flt(doc.get("base_rounded_total") or doc.base_grand_total)
grand_total = flt(doc.get("rounded_total") or doc.grand_total)
if doc.doctype in ("Sales Invoice", "Purchase Invoice"):
base_grand_total = base_grand_total - flt(doc.base_write_off_amount)
grand_total = grand_total - flt(doc.write_off_amount)
if doc.get("total_advance"):
if party_account_currency == doc.company_currency:
base_grand_total -= doc.get("total_advance")
grand_total = flt(
base_grand_total / doc.get("conversion_rate"), doc.precision("grand_total")
)
else:
grand_total -= doc.get("total_advance")
base_grand_total = flt(
grand_total * doc.get("conversion_rate"), doc.precision("base_grand_total")
)
if (
abs(flt(total, doc.precision("grand_total")) - flt(grand_total, doc.precision("grand_total")))
> 0.1
or abs(
flt(base_total, doc.precision("base_grand_total"))
- flt(base_grand_total, doc.precision("base_grand_total"))
)
> 0.1
):
frappe.throw(
_("Total Payment Amount in Payment Schedule must be equal to Grand / Rounded Total")
)
def linked_order_has_payment_terms_template(po_or_so, doctype) -> str | None:
return frappe.get_value(doctype, po_or_so, "payment_terms_template")
def linked_order_has_payment_schedule(po_or_so) -> list:
return frappe.get_all("Payment Schedule", filters={"parent": po_or_so})
def get_payment_terms(

View File

@@ -842,7 +842,9 @@ def get_mapped_purchase_invoice(source_name, target_doc=None, ignore_permissions
if target.get("allocate_advances_automatically"):
target.set_advances()
target.set_payment_schedule()
from erpnext.accounts.services.payment_schedule import PaymentScheduleService
PaymentScheduleService(target).set_payment_schedule()
target.credit_to = get_party_account("Supplier", source.supplier, source.company)
def get_billed_qty(po_item_name):

View File

@@ -125,7 +125,9 @@ class AccountsController(TransactionBase):
"Sales Invoice",
)
if self.doctype in relevant_docs:
self.set_payment_schedule()
from erpnext.accounts.services.payment_schedule import PaymentScheduleService
PaymentScheduleService(self).set_payment_schedule()
def on_update(self):
from erpnext.controllers.taxes_and_totals import process_item_wise_tax_details
@@ -647,18 +649,24 @@ class AccountsController(TransactionBase):
if self.is_return:
return
self.validate_payment_schedule_dates()
self.set_due_date()
self.set_payment_schedule()
from erpnext.accounts.services.payment_schedule import PaymentScheduleService
ps = PaymentScheduleService(self)
ps.validate_payment_schedule_dates()
ps.set_due_date()
ps.set_payment_schedule()
if not self.get("ignore_default_payment_terms_template"):
self.validate_payment_schedule_amount()
ps.validate_payment_schedule_amount()
self.validate_due_date()
self.validate_advance_entries()
def validate_non_invoice_documents_schedule(self):
self.set_payment_schedule()
self.validate_payment_schedule_dates()
self.validate_payment_schedule_amount()
from erpnext.accounts.services.payment_schedule import PaymentScheduleService
ps = PaymentScheduleService(self)
ps.set_payment_schedule()
ps.validate_payment_schedule_dates()
ps.validate_payment_schedule_amount()
def validate_all_documents_schedule(self):
if self.doctype in ("Sales Invoice", "Purchase Invoice"):
@@ -1466,35 +1474,6 @@ class AccountsController(TransactionBase):
frappe.msgprint(_("Purchase Orders {0} are un-linked").format("\n".join(linked_po)))
def validate_multiple_billing(self, ref_dt: str, item_ref_dn: str, based_on: str) -> None:
from erpnext.accounts.services.billing_validation import validate_multiple_billing
validate_multiple_billing(self, ref_dt, item_ref_dn, based_on)
def get_billing_reference_details(
self, reference_names: list, reference_doctype: str, based_on: str
) -> frappe._dict:
from erpnext.accounts.services.billing_validation import get_billing_reference_details
return get_billing_reference_details(self, reference_names, reference_doctype, based_on)
def get_reference_wise_billed_amt(self, ref_dt: str, item_ref_dn: str, based_on: str) -> dict | None:
from erpnext.accounts.services.billing_validation import get_reference_wise_billed_amt
return get_reference_wise_billed_amt(self, ref_dt, item_ref_dn, based_on)
def get_already_billed_amount(
self, reference_names: list, item_ref_dn: str, based_on: str
) -> frappe._dict:
from erpnext.accounts.services.billing_validation import get_already_billed_amount
return get_already_billed_amount(self, reference_names, item_ref_dn, based_on)
def throw_overbill_exception(self, overbilled_items: list, precision: int) -> None:
from erpnext.accounts.services.billing_validation import throw_overbill_exception
throw_overbill_exception(self, overbilled_items, precision)
def get_company_default(self, fieldname, ignore_validation=False):
from erpnext.accounts.utils import get_company_default
@@ -1681,65 +1660,6 @@ class AccountsController(TransactionBase):
for item in duplicate_list:
self.remove(item)
def set_payment_schedule(self) -> None:
from erpnext.accounts.services.payment_schedule import set_payment_schedule
set_payment_schedule(self)
def get_order_details(self) -> tuple:
from erpnext.accounts.services.payment_schedule import get_order_details
return get_order_details(self)
def linked_order_has_payment_terms(self, po_or_so, fieldname, doctype) -> bool:
from erpnext.accounts.services.payment_schedule import linked_order_has_payment_terms
return linked_order_has_payment_terms(self, po_or_so, fieldname, doctype)
def all_items_have_same_po_or_so(self, po_or_so, fieldname) -> bool:
from erpnext.accounts.services.payment_schedule import all_items_have_same_po_or_so
return all_items_have_same_po_or_so(self, po_or_so, fieldname)
def linked_order_has_payment_terms_template(self, po_or_so, doctype) -> str | None:
from erpnext.accounts.services.payment_schedule import linked_order_has_payment_terms_template
return linked_order_has_payment_terms_template(po_or_so, doctype)
def linked_order_has_payment_schedule(self, po_or_so) -> list:
from erpnext.accounts.services.payment_schedule import linked_order_has_payment_schedule
return linked_order_has_payment_schedule(po_or_so)
def fetch_payment_terms_from_order(
self,
po_or_so,
po_or_so_doctype,
grand_total,
base_grand_total,
automatically_fetch_payment_terms,
) -> None:
from erpnext.accounts.services.payment_schedule import fetch_payment_terms_from_order
fetch_payment_terms_from_order(
self, po_or_so, po_or_so_doctype, grand_total, base_grand_total, automatically_fetch_payment_terms
)
def set_due_date(self) -> None:
from erpnext.accounts.services.payment_schedule import set_due_date
set_due_date(self)
def validate_payment_schedule_dates(self) -> None:
from erpnext.accounts.services.payment_schedule import validate_payment_schedule_dates
validate_payment_schedule_dates(self)
def validate_payment_schedule_amount(self) -> None:
from erpnext.accounts.services.payment_schedule import validate_payment_schedule_amount
validate_payment_schedule_amount(self)
def is_rounded_total_disabled(self):
if self.meta.get_field("disable_rounded_total"):
return self.disable_rounded_total
@@ -2538,7 +2458,9 @@ def update_child_qty_rate(
)
if parent_doctype != "Supplier Quotation":
parent.set_payment_schedule()
from erpnext.accounts.services.payment_schedule import PaymentScheduleService
PaymentScheduleService(parent).set_payment_schedule()
if parent_doctype == "Purchase Order":
parent.validate_minimum_order_qty()
parent.validate_budget()

View File

@@ -140,7 +140,9 @@ def update_totals(vat_tax, base_vat_tax, doc):
doc.in_words = money_in_words(doc.grand_total, doc.currency)
doc.base_in_words = money_in_words(doc.base_grand_total, erpnext.get_company_currency(doc.company))
doc.set_payment_schedule()
from erpnext.accounts.services.payment_schedule import PaymentScheduleService
PaymentScheduleService(doc).set_payment_schedule()
def make_regional_gl_entries(gl_entries, doc):

View File

@@ -480,7 +480,9 @@ def _make_sales_order(source_name, target_doc=None, ignore_permissions=False, ar
)
if automatically_fetch_payment_terms:
doclist.set_payment_schedule()
from erpnext.accounts.services.payment_schedule import PaymentScheduleService
PaymentScheduleService(doclist).set_payment_schedule()
return doclist

View File

@@ -1606,7 +1606,9 @@ def make_sales_invoice(
frappe.get_single_value("Accounts Settings", "automatically_fetch_payment_terms")
)
if automatically_fetch_payment_terms:
doclist.set_payment_schedule()
from erpnext.accounts.services.payment_schedule import PaymentScheduleService
PaymentScheduleService(doclist).set_payment_schedule()
return doclist

View File

@@ -1000,9 +1000,12 @@ def make_sales_invoice(
)
if not doc.is_return:
so, doctype, fieldname = doc.get_order_details()
from erpnext.accounts.services.payment_schedule import PaymentScheduleService
ps = PaymentScheduleService(doc)
so, doctype, fieldname = ps.get_order_details()
if (
doc.linked_order_has_payment_terms(so, fieldname, doctype)
ps.linked_order_has_payment_terms(so, fieldname, doctype)
and not automatically_fetch_payment_terms
):
payment_terms_template = frappe.db.get_value(doctype, so, "payment_terms_template")
@@ -1016,7 +1019,7 @@ def make_sales_invoice(
)
elif automatically_fetch_payment_terms:
doc.set_payment_schedule()
ps.set_payment_schedule()
return doc

View File

@@ -1112,7 +1112,9 @@ def make_purchase_invoice(
merge_taxes(source, doc)
doc.run_method("calculate_taxes_and_totals")
doc.set_payment_schedule()
from erpnext.accounts.services.payment_schedule import PaymentScheduleService
PaymentScheduleService(doc).set_payment_schedule()
def update_item(source_doc, target_doc, source_parent):
target_doc.qty, returned_qty = get_pending_qty(source_doc)