refactor: extract billing, payment schedule, and exchange gain/loss into services

Move billing validation, payment schedule, and exchange gain/loss logic from
AccountsController into dedicated service modules under accounts/services/.
AccountsController retains thin shim methods that delegate to the services.
This commit is contained in:
Nabin Hait
2026-05-27 23:42:50 +05:30
parent 983d80f7c5
commit bb803a8f82
4 changed files with 837 additions and 718 deletions

View File

@@ -0,0 +1,147 @@
# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
# License: GNU General Public License v3. See license.txt
"""Billing amount validation helpers (overbilling checks)."""
import frappe
from frappe import _
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
ref_wise_billed_amount = get_reference_wise_billed_amt(doc, ref_dt, item_ref_dn, based_on)
if not ref_wise_billed_amount:
return
total_overbilled_amt = 0.0
overbilled_items = []
precision = doc.precision(based_on, "items")
precision_allowance = 1 / (10**precision)
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()
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)
if total_billed_amt < 0 and max_allowed_amt < 0:
total_billed_amt, max_allowed_amt = abs(total_billed_amt), abs(max_allowed_amt)
overbill_amt = total_billed_amt - max_allowed_amt
row["max_allowed_amt"] = max_allowed_amt
total_overbilled_amt += overbill_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)
if overbilled_items:
throw_overbill_exception(doc, overbilled_items, precision)
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(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)),
)
for item in overbilled_items
)
+ "</ul>"
)
message += _("<p>To allow over-billing, please set allowance in Accounts Settings.</p>")
frappe.throw(_(message))

View File

@@ -0,0 +1,237 @@
# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
# License: GNU General Public License v3. See license.txt
"""Exchange gain/loss journal helpers."""
import frappe
from frappe import _, qb
from frappe.utils import flt, get_link_to_form
from erpnext.accounts.doctype.accounting_dimension.accounting_dimension import get_dimensions
from erpnext.accounts.general_ledger import get_round_off_account_and_cost_center
from erpnext.accounts.utils import create_gain_loss_journal, get_currency_precision
def make_precision_loss_gl_entry(doc, gl_entries: list) -> None:
round_off_account, round_off_cost_center, _ = get_round_off_account_and_cost_center(
doc.company, "Purchase Invoice", doc.name, doc.use_company_roundoff_cost_center
)
precision_loss = doc.get("base_net_total") - flt(
doc.get("net_total") * doc.conversion_rate, doc.precision("net_total")
)
credit_or_debit = "credit" if doc.doctype == "Purchase Invoice" else "debit"
against = doc.supplier if doc.doctype == "Purchase Invoice" else doc.customer
if precision_loss:
gl_entries.append(
doc.get_gl_dict(
{
"account": round_off_account,
"against": against,
credit_or_debit: precision_loss,
"cost_center": round_off_cost_center
if doc.use_company_roundoff_cost_center
else doc.cost_center or round_off_cost_center,
"remarks": _("Net total calculation precision loss"),
}
)
)
def gain_loss_journal_already_booked(
gain_loss_account: str,
exc_gain_loss: float,
ref2_dt: str,
ref2_dn: str,
ref2_detail_no: str,
) -> bool:
"""Check if a gain/loss journal has already been booked for the given parameters."""
if res := frappe.db.get_all(
"Journal Entry Account",
filters={
"docstatus": 1,
"account": gain_loss_account,
"reference_type": ref2_dt,
"reference_name": ref2_dn,
"reference_detail_no": ref2_detail_no,
},
pluck="parent",
):
res = list({x for x in res})
if exc_vouchers := frappe.db.get_all(
"Journal Entry",
filters={"name": ["in", res], "voucher_type": "Exchange Gain Or Loss"},
fields=["voucher_type", "total_debit", "total_credit"],
):
booked_voucher = exc_vouchers[0]
if (
booked_voucher.total_debit == exc_gain_loss
and booked_voucher.total_credit == exc_gain_loss
and booked_voucher.voucher_type == "Exchange Gain Or Loss"
):
return True
return False
def make_exchange_gain_loss_journal(
doc, args: dict | None = None, dimensions_dict: dict | None = None
) -> None:
"""Make Exchange Gain/Loss journal for Invoices and Payments."""
# Cancelling existing exchange gain/loss journals is handled during the `on_cancel` event.
# see accounts/utils.py:cancel_exchange_gain_loss_journal()
if doc.docstatus != 1:
return
if dimensions_dict is None:
dimensions_dict = frappe._dict()
active_dimensions = get_dimensions()[0]
for dim in active_dimensions:
dimensions_dict[dim.fieldname] = doc.get(dim.fieldname)
if doc.get("doctype") == "Journal Entry":
if args:
precision = get_currency_precision()
for arg in args:
if (
flt(arg.get("difference_amount", 0), precision) != 0
or flt(arg.get("exchange_gain_loss", 0), precision) != 0
) and arg.get("difference_account"):
party_account = arg.get("account")
gain_loss_account = arg.get("difference_account")
difference_amount = arg.get("difference_amount") or arg.get("exchange_gain_loss")
if difference_amount > 0:
dr_or_cr = "debit" if arg.get("party_type") == "Customer" else "credit"
else:
dr_or_cr = "credit" if arg.get("party_type") == "Customer" else "debit"
reverse_dr_or_cr = "debit" if dr_or_cr == "credit" else "credit"
if not gain_loss_journal_already_booked(
gain_loss_account,
difference_amount,
doc.doctype,
doc.name,
arg.get("referenced_row"),
):
posting_date = arg.get("difference_posting_date") or frappe.db.get_value(
arg.voucher_type, arg.voucher_no, "posting_date"
)
je = create_gain_loss_journal(
doc.company,
posting_date,
arg.get("party_type"),
arg.get("party"),
party_account,
gain_loss_account,
difference_amount,
dr_or_cr,
reverse_dr_or_cr,
arg.get("against_voucher_type"),
arg.get("against_voucher"),
arg.get("idx"),
doc.doctype,
doc.name,
arg.get("referenced_row"),
arg.get("cost_center"),
dimensions_dict,
arg.get("project"),
)
frappe.msgprint(
_("Exchange Gain/Loss amount has been booked through {0}").format(
get_link_to_form("Journal Entry", je)
)
)
if doc.get("doctype") == "Payment Entry":
gain_loss_to_book = [x for x in doc.references if x.exchange_gain_loss != 0]
booked = []
if gain_loss_to_book:
je = qb.DocType("Journal Entry")
jea = qb.DocType("Journal Entry Account")
parents = (
qb.from_(jea)
.select(jea.parent)
.where(
(jea.reference_type == "Payment Entry")
& (jea.reference_name == doc.name)
& (jea.docstatus == 1)
)
.run()
)
if parents:
booked = (
qb.from_(je)
.inner_join(jea)
.on(je.name == jea.parent)
.select(jea.reference_type, jea.reference_name, jea.reference_detail_no)
.where(
(je.docstatus == 1)
& (je.name.isin(parents))
& (je.voucher_type == "Exchange Gain or Loss")
)
.run()
)
for d in gain_loss_to_book:
if d.exchange_gain_loss and ((d.reference_doctype, d.reference_name, str(d.idx)) not in booked):
if doc.book_advance_payments_in_separate_party_account:
party_account = d.account
else:
if doc.payment_type == "Receive":
party_account = doc.paid_from
elif doc.payment_type == "Pay":
party_account = doc.paid_to
dr_or_cr = "debit" if d.exchange_gain_loss > 0 else "credit"
if is_payable_account(d.reference_doctype, party_account):
dr_or_cr = "debit" if dr_or_cr == "credit" else "credit"
reverse_dr_or_cr = "debit" if dr_or_cr == "credit" else "credit"
gain_loss_account = frappe.get_cached_value(
"Company", doc.company, "exchange_gain_loss_account"
)
je = create_gain_loss_journal(
doc.company,
args.get("difference_posting_date") if args else doc.posting_date,
doc.party_type,
doc.party,
party_account,
gain_loss_account,
d.exchange_gain_loss,
dr_or_cr,
reverse_dr_or_cr,
d.reference_doctype,
d.reference_name,
d.idx,
doc.doctype,
doc.name,
d.idx,
doc.cost_center,
dimensions_dict,
doc.project,
)
frappe.msgprint(
_("Exchange Gain/Loss amount has been booked through {0}").format(
get_link_to_form("Journal Entry", je)
)
)
def is_payable_account(reference_doctype: str, account: str) -> bool:
if reference_doctype == "Purchase Invoice" or (
reference_doctype == "Journal Entry"
and frappe.get_cached_value("Account", account, "account_type") == "Payable"
):
return True
return False
def set_transaction_currency_and_rate_in_gl_map(doc, gl_entries: list) -> None:
for entry in gl_entries:
entry["transaction_currency"] = doc.currency
entry["transaction_exchange_rate"] = doc.get("conversion_rate") or 1

View File

@@ -0,0 +1,373 @@
# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
# License: GNU General Public License v3. See license.txt
"""Payment schedule and payment terms helpers."""
import frappe
from frappe import _
from frappe.utils import DateTimeLikeObject, add_days, add_months, cint, flt, get_last_day, getdate
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
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)
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", "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"))
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 get_payment_terms(
terms_template: str,
posting_date: DateTimeLikeObject | None = None,
grand_total: float | None = None,
base_grand_total: float | None = None,
bill_date: DateTimeLikeObject | None = None,
) -> list:
if not terms_template:
return
terms_doc = frappe.get_doc("Payment Terms Template", terms_template)
schedule = []
for d in terms_doc.get("terms"):
d = frappe._dict(d.as_dict())
term_details = get_payment_term_details(d, posting_date, grand_total, base_grand_total, bill_date)
schedule.append(term_details)
return schedule
@frappe.whitelist()
def get_payment_term_details(
term: str | frappe._dict,
posting_date: DateTimeLikeObject | None = None,
grand_total: float | None = None,
base_grand_total: float | None = None,
bill_date: DateTimeLikeObject | None = None,
) -> frappe._dict:
term_details = frappe._dict()
if isinstance(term, str):
term = frappe.get_doc("Payment Term", term)
else:
term_details.payment_term = term.payment_term
for field in [
"description",
"invoice_portion",
"discount_type",
"discount",
"mode_of_payment",
"due_date_based_on",
"credit_days",
"credit_months",
"discount_validity_based_on",
"discount_validity",
]:
term_details[field] = term.get(field)
term_details.payment_amount = flt(term.invoice_portion) * flt(grand_total) / 100
term_details.base_payment_amount = flt(term.invoice_portion) * flt(base_grand_total) / 100
term_details.outstanding = term_details.payment_amount
term_details.base_outstanding = term_details.base_payment_amount
if bill_date:
term_details.due_date = get_due_date(term, bill_date)
term_details.discount_date = get_discount_date(term, bill_date)
elif posting_date:
term_details.due_date = get_due_date(term, posting_date)
term_details.discount_date = get_discount_date(term, posting_date)
if posting_date and getdate(term_details.due_date) < getdate(posting_date):
term_details.due_date = posting_date
return term_details
def get_due_date(term, posting_date=None, bill_date=None):
due_date = None
date = bill_date or posting_date
if term.due_date_based_on == "Day(s) after invoice date":
due_date = add_days(date, cint(term.credit_days))
elif term.due_date_based_on == "Day(s) after the end of the invoice month":
due_date = add_days(get_last_day(date), cint(term.credit_days))
elif term.due_date_based_on == "Month(s) after the end of the invoice month":
due_date = get_last_day(add_months(date, cint(term.credit_months)))
return due_date
def get_discount_date(term, posting_date=None, bill_date=None):
discount_validity = None
date = bill_date or posting_date
if term.discount_validity_based_on == "Day(s) after invoice date":
discount_validity = add_days(date, cint(term.discount_validity))
elif term.discount_validity_based_on == "Day(s) after the end of the invoice month":
discount_validity = add_days(get_last_day(date), cint(term.discount_validity))
elif term.discount_validity_based_on == "Month(s) after the end of the invoice month":
discount_validity = get_last_day(add_months(date, cint(term.discount_validity)))
return discount_validity

View File

@@ -12,15 +12,9 @@ from frappe.model.workflow import get_workflow_name, is_transition_condition_sat
from frappe.query_builder import DocType
from frappe.query_builder.functions import Sum
from frappe.utils import (
DateTimeLikeObject,
add_days,
add_months,
cint,
comma_and,
flt,
fmt_money,
formatdate,
get_last_day,
get_link_to_form,
getdate,
nowdate,
@@ -48,10 +42,7 @@ from erpnext.accounts.party import (
validate_party_frozen_disabled,
)
from erpnext.accounts.utils import (
create_gain_loss_journal,
get_account_currency,
get_currency_precision,
get_fiscal_years,
validate_fiscal_year,
)
from erpnext.accounts.utils import (
@@ -1407,239 +1398,30 @@ class AccountsController(TransactionBase):
set_advance_gain_or_loss(self)
def make_precision_loss_gl_entry(self, gl_entries):
(
round_off_account,
round_off_cost_center,
round_off_for_opening,
) = get_round_off_account_and_cost_center(
self.company, "Purchase Invoice", self.name, self.use_company_roundoff_cost_center
)
from erpnext.accounts.services.exchange_gain_loss import make_precision_loss_gl_entry
precision_loss = self.get("base_net_total") - flt(
self.get("net_total") * self.conversion_rate, self.precision("net_total")
)
credit_or_debit = "credit" if self.doctype == "Purchase Invoice" else "debit"
against = self.supplier if self.doctype == "Purchase Invoice" else self.customer
if precision_loss:
gl_entries.append(
self.get_gl_dict(
{
"account": round_off_account,
"against": against,
credit_or_debit: precision_loss,
"cost_center": round_off_cost_center
if self.use_company_roundoff_cost_center
else self.cost_center or round_off_cost_center,
"remarks": _("Net total calculation precision loss"),
}
)
)
make_precision_loss_gl_entry(self, gl_entries)
def gain_loss_journal_already_booked(
self,
gain_loss_account,
exc_gain_loss,
ref2_dt,
ref2_dn,
ref2_detail_no,
self, gain_loss_account, exc_gain_loss, ref2_dt, ref2_dn, ref2_detail_no
) -> bool:
"""
Check if gain/loss is booked
"""
if res := frappe.db.get_all(
"Journal Entry Account",
filters={
"docstatus": 1,
"account": gain_loss_account,
"reference_type": ref2_dt, # this will be Journal Entry
"reference_name": ref2_dn,
"reference_detail_no": ref2_detail_no,
},
pluck="parent",
):
# deduplicate
res = list({x for x in res})
if exc_vouchers := frappe.db.get_all(
"Journal Entry",
filters={"name": ["in", res], "voucher_type": "Exchange Gain Or Loss"},
fields=["voucher_type", "total_debit", "total_credit"],
):
booked_voucher = exc_vouchers[0]
if (
booked_voucher.total_debit == exc_gain_loss
and booked_voucher.total_credit == exc_gain_loss
and booked_voucher.voucher_type == "Exchange Gain Or Loss"
):
return True
return False
from erpnext.accounts.services.exchange_gain_loss import gain_loss_journal_already_booked
return gain_loss_journal_already_booked(
gain_loss_account, exc_gain_loss, ref2_dt, ref2_dn, ref2_detail_no
)
def make_exchange_gain_loss_journal(
self, args: dict | None = None, dimensions_dict: dict | None = None
) -> None:
"""
Make Exchange Gain/Loss journal for Invoices and Payments
"""
# Cancelling existing exchange gain/loss journals is handled during the `on_cancel` event.
# see accounts/utils.py:cancel_exchange_gain_loss_journal()
if self.docstatus == 1:
if dimensions_dict is None:
dimensions_dict = frappe._dict()
active_dimensions = get_dimensions()[0]
for dim in active_dimensions:
dimensions_dict[dim.fieldname] = self.get(dim.fieldname)
from erpnext.accounts.services.exchange_gain_loss import make_exchange_gain_loss_journal
if self.get("doctype") == "Journal Entry":
# 'args' is populated with exchange gain/loss account and the amount to be booked.
# These are generated by Sales/Purchase Invoice during reconciliation and advance allocation.
# and below logic is only for such scenarios
if args:
precision = get_currency_precision()
for arg in args:
# Advance section uses `exchange_gain_loss` and reconciliation uses `difference_amount`
if (
flt(arg.get("difference_amount", 0), precision) != 0
or flt(arg.get("exchange_gain_loss", 0), precision) != 0
) and arg.get("difference_account"):
party_account = arg.get("account")
gain_loss_account = arg.get("difference_account")
difference_amount = arg.get("difference_amount") or arg.get("exchange_gain_loss")
if difference_amount > 0:
dr_or_cr = "debit" if arg.get("party_type") == "Customer" else "credit"
else:
dr_or_cr = "credit" if arg.get("party_type") == "Customer" else "debit"
reverse_dr_or_cr = "debit" if dr_or_cr == "credit" else "credit"
if not self.gain_loss_journal_already_booked(
gain_loss_account,
difference_amount,
self.doctype,
self.name,
arg.get("referenced_row"),
):
posting_date = arg.get("difference_posting_date") or frappe.db.get_value(
arg.voucher_type, arg.voucher_no, "posting_date"
)
je = create_gain_loss_journal(
self.company,
posting_date,
arg.get("party_type"),
arg.get("party"),
party_account,
gain_loss_account,
difference_amount,
dr_or_cr,
reverse_dr_or_cr,
arg.get("against_voucher_type"),
arg.get("against_voucher"),
arg.get("idx"),
self.doctype,
self.name,
arg.get("referenced_row"),
arg.get("cost_center"),
dimensions_dict,
arg.get("project"),
)
frappe.msgprint(
_("Exchange Gain/Loss amount has been booked through {0}").format(
get_link_to_form("Journal Entry", je)
)
)
if self.get("doctype") == "Payment Entry":
# For Payment Entry, exchange_gain_loss field in the `references` table is the trigger for journal creation
gain_loss_to_book = [x for x in self.references if x.exchange_gain_loss != 0]
booked = []
if gain_loss_to_book:
[x.reference_doctype for x in gain_loss_to_book]
[x.reference_name for x in gain_loss_to_book]
je = qb.DocType("Journal Entry")
jea = qb.DocType("Journal Entry Account")
parents = (
qb.from_(jea)
.select(jea.parent)
.where(
(jea.reference_type == "Payment Entry")
& (jea.reference_name == self.name)
& (jea.docstatus == 1)
)
.run()
)
booked = []
if parents:
booked = (
qb.from_(je)
.inner_join(jea)
.on(je.name == jea.parent)
.select(jea.reference_type, jea.reference_name, jea.reference_detail_no)
.where(
(je.docstatus == 1)
& (je.name.isin(parents))
& (je.voucher_type == "Exchange Gain or Loss")
)
.run()
)
for d in gain_loss_to_book:
# Filter out References for which Gain/Loss is already booked
if d.exchange_gain_loss and (
(d.reference_doctype, d.reference_name, str(d.idx)) not in booked
):
if self.book_advance_payments_in_separate_party_account:
party_account = d.account
else:
if self.payment_type == "Receive":
party_account = self.paid_from
elif self.payment_type == "Pay":
party_account = self.paid_to
dr_or_cr = "debit" if d.exchange_gain_loss > 0 else "credit"
# Inverse debit/credit for payable accounts
if self.is_payable_account(d.reference_doctype, party_account):
dr_or_cr = "debit" if dr_or_cr == "credit" else "credit"
reverse_dr_or_cr = "debit" if dr_or_cr == "credit" else "credit"
gain_loss_account = frappe.get_cached_value(
"Company", self.company, "exchange_gain_loss_account"
)
je = create_gain_loss_journal(
self.company,
args.get("difference_posting_date") if args else self.posting_date,
self.party_type,
self.party,
party_account,
gain_loss_account,
d.exchange_gain_loss,
dr_or_cr,
reverse_dr_or_cr,
d.reference_doctype,
d.reference_name,
d.idx,
self.doctype,
self.name,
d.idx,
self.cost_center,
dimensions_dict,
self.project,
)
frappe.msgprint(
_("Exchange Gain/Loss amount has been booked through {0}").format(
get_link_to_form("Journal Entry", je)
)
)
make_exchange_gain_loss_journal(self, args, dimensions_dict)
def is_payable_account(self, reference_doctype, account):
if reference_doctype == "Purchase Invoice" or (
reference_doctype == "Journal Entry"
and frappe.get_cached_value("Account", account, "account_type") == "Payable"
):
return True
return False
from erpnext.accounts.services.exchange_gain_loss import is_payable_account
return is_payable_account(reference_doctype, account)
def update_against_document_in_jv(self):
"""
@@ -1907,147 +1689,34 @@ class AccountsController(TransactionBase):
)
)
def validate_multiple_billing(self, ref_dt, item_ref_dn, based_on):
from erpnext.controllers.status_updater import get_allowance_for
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
ref_wise_billed_amount = self.get_reference_wise_billed_amt(ref_dt, item_ref_dn, based_on)
validate_multiple_billing(self, ref_dt, item_ref_dn, based_on)
if not ref_wise_billed_amount:
return
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
total_overbilled_amt = 0.0
overbilled_items = []
precision = self.precision(based_on, "items")
precision_allowance = 1 / (10**precision)
return get_billing_reference_details(self, reference_names, reference_doctype, based_on)
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()
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
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]
return get_reference_wise_billed_amt(self, ref_dt, item_ref_dn, based_on)
max_allowed_amt = flt(row.ref_amt * (100 + allowance) / 100)
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
if total_billed_amt < 0 and max_allowed_amt < 0:
# while making debit note against purchase return entry(purchase receipt) getting overbill error
total_billed_amt, max_allowed_amt = abs(total_billed_amt), abs(max_allowed_amt)
return get_already_billed_amount(self, reference_names, item_ref_dn, based_on)
overbill_amt = total_billed_amt - max_allowed_amt
row["max_allowed_amt"] = max_allowed_amt
total_overbilled_amt += overbill_amt
def throw_overbill_exception(self, overbilled_items: list, precision: int) -> None:
from erpnext.accounts.services.billing_validation import throw_overbill_exception
if overbill_amt > precision_allowance and not is_overbilling_allowed:
if self.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 overbilled_items:
self.throw_overbill_exception(overbilled_items, precision)
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_billing_reference_details(self, reference_names, reference_doctype, based_on):
return frappe._dict(
frappe.get_all(
reference_doctype,
filters={"name": ("in", reference_names)},
fields=["name", based_on],
as_list=1,
)
)
def get_reference_wise_billed_amt(self, ref_dt, item_ref_dn, based_on):
"""
Returns Sum of Amount of
Sales/Purchase Invoice Items
that are linked to `item_ref_dn` (`dn_detail` / `pr_detail`)
that are submitted OR not submitted but are under current invoice
"""
reference_names = [d.get(item_ref_dn) for d in self.items if d.get(item_ref_dn)]
if not reference_names:
return
ref_wise_billed_amount = {}
precision = self.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)
for item in self.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: # Skip warning for free items
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_already_billed_amount(self, reference_names, item_ref_dn, based_on):
item_doctype = frappe.qb.DocType(self.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.name))
.groupby(join_field)
).run()
)
def throw_overbill_exception(self, overbilled_items, precision):
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.currency)),
)
for item in overbilled_items
)
+ "</ul>"
)
message += _("<p>To allow over-billing, please set allowance in Accounts Settings.</p>")
frappe.throw(_(message))
throw_overbill_exception(self, overbilled_items, precision)
def get_company_default(self, fieldname, ignore_validation=False):
from erpnext.accounts.utils import get_company_default
@@ -2235,285 +1904,64 @@ class AccountsController(TransactionBase):
for item in duplicate_list:
self.remove(item)
def set_payment_schedule(self):
if (self.doctype == "Sales Invoice" and self.is_pos) or self.get("is_opening") == "Yes":
self.payment_terms_template = ""
return
def set_payment_schedule(self) -> None:
from erpnext.accounts.services.payment_schedule import set_payment_schedule
party_account_currency = self.get("party_account_currency")
if not party_account_currency:
party_type, party = self.get_party()
set_payment_schedule(self)
if party_type and party:
party_account_currency = get_party_account_currency(party_type, party, self.company)
def get_order_details(self) -> tuple:
from erpnext.accounts.services.payment_schedule import get_order_details
posting_date = self.get("bill_date") or self.get("posting_date") or self.get("transaction_date")
date = self.get("due_date")
due_date = date or posting_date
return get_order_details(self)
base_grand_total = flt(self.get("base_rounded_total") or self.base_grand_total)
grand_total = flt(self.get("rounded_total") or self.grand_total)
automatically_fetch_payment_terms = 0
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
if self.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 self.doctype != "Sales Order":
base_grand_total = base_grand_total - flt(self.base_write_off_amount)
grand_total = grand_total - flt(self.write_off_amount)
return linked_order_has_payment_terms(self, po_or_so, fieldname, doctype)
if self.get("total_advance"):
if party_account_currency == self.company_currency:
base_grand_total -= self.get("total_advance")
grand_total = flt(
base_grand_total / self.get("conversion_rate"), self.precision("grand_total")
)
else:
grand_total -= self.get("total_advance")
base_grand_total = flt(
grand_total * self.get("conversion_rate"), self.precision("base_grand_total")
)
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
if not self.get("payment_schedule"):
if (
self.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 self.get("payment_terms_template"):
self.ignore_default_payment_terms_template = 1
elif self.get("payment_terms_template"):
data = get_payment_terms(
self.payment_terms_template, posting_date, grand_total, base_grand_total
)
for item in data:
self.append("payment_schedule", item)
elif self.doctype not in ["Purchase Receipt"]:
data = dict(
due_date=due_date,
invoice_portion=100,
payment_amount=grand_total,
base_payment_amount=base_grand_total,
)
self.append("payment_schedule", data)
return all_items_have_same_po_or_so(self, po_or_so, fieldname)
allocate_payment_based_on_payment_terms = frappe.db.get_value(
"Payment Terms Template", self.payment_terms_template, "allocate_payment_based_on_payment_terms"
)
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
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)
):
for d in self.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 * self.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
)
self.ignore_default_payment_terms_template = 1
return linked_order_has_payment_terms_template(po_or_so, doctype)
def get_order_details(self):
if not self.get("items"):
return None, None, None
if self.doctype == "Sales Invoice":
prev_doc = self.get("items")[0].get("sales_order")
prev_doctype = "Sales Order"
prev_doctype_name = "sales_order"
elif self.doctype == "Purchase Invoice":
prev_doc = self.get("items")[0].get("purchase_order")
prev_doctype = "Purchase Order"
prev_doctype_name = "purchase_order"
else:
prev_doc = self.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_schedule(self, po_or_so) -> list:
from erpnext.accounts.services.payment_schedule import linked_order_has_payment_schedule
def linked_order_has_payment_terms(self, po_or_so, fieldname, doctype):
if po_or_so and self.all_items_have_same_po_or_so(po_or_so, fieldname):
if self.linked_order_has_payment_terms_template(po_or_so, doctype):
return True
elif self.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):
for item in self.get("items"):
if item.get(fieldname) != po_or_so:
return False
return True
def linked_order_has_payment_terms_template(self, po_or_so, doctype):
return frappe.get_value(doctype, po_or_so, "payment_terms_template")
def linked_order_has_payment_schedule(self, po_or_so):
return frappe.get_all("Payment Schedule", filters={"parent": po_or_so})
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
):
"""
Fetch Payment Terms from Purchase/Sales Order on creating a new Purchase/Sales Invoice.
"""
po_or_so = frappe.get_cached_doc(po_or_so_doctype, po_or_so)
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
self.payment_schedule = []
self.payment_terms_template = po_or_so.payment_terms_template
posting_date = self.get("bill_date") or self.get("posting_date") or self.get("transaction_date")
fetch_payment_terms_from_order(
self, po_or_so, po_or_so_doctype, grand_total, base_grand_total, automatically_fetch_payment_terms
)
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,
}
def set_due_date(self) -> None:
from erpnext.accounts.services.payment_schedule import set_due_date
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)
set_due_date(self)
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)
def validate_payment_schedule_dates(self) -> None:
from erpnext.accounts.services.payment_schedule import validate_payment_schedule_dates
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 * self.get("conversion_rate"),
schedule.precision("base_payment_amount"),
)
validate_payment_schedule_dates(self)
if schedule.discount_type == "Percentage":
payment_schedule["discount_type"] = schedule.discount_type
payment_schedule["discount"] = schedule.discount
def validate_payment_schedule_amount(self) -> None:
from erpnext.accounts.services.payment_schedule import validate_payment_schedule_amount
if not schedule.invoice_portion:
payment_schedule["payment_amount"] = schedule.payment_amount
self.append("payment_schedule", payment_schedule)
def set_due_date(self):
due_dates = [d.due_date for d in self.get("payment_schedule") if d.due_date]
if due_dates:
self.due_date = max(due_dates)
def validate_payment_schedule_dates(self):
dates = []
li = []
if self.doctype == "Sales Invoice" and self.is_pos:
return
for d in self.get("payment_schedule"):
d.validate_from_to_dates("discount_date", "due_date")
if self.doctype in ["Sales Order", "Quotation"] and getdate(d.due_date) < getdate(
self.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:
duplicates = "<br>" + "<br>".join(li)
frappe.throw(
_("Rows with duplicate due dates in other rows were found: {0}").format(duplicates),
title=_("Payment Schedule"),
)
def validate_payment_schedule_amount(self):
if (self.doctype == "Sales Invoice" and self.is_pos) or self.get("is_opening") == "Yes":
return
party_account_currency = self.get("party_account_currency")
if not party_account_currency:
party_type, party = self.get_party()
if party_type and party:
party_account_currency = get_party_account_currency(party_type, party, self.company)
if self.get("payment_schedule"):
total = 0
base_total = 0
for d in self.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(self.get("base_rounded_total") or self.base_grand_total)
grand_total = flt(self.get("rounded_total") or self.grand_total)
if self.doctype in ("Sales Invoice", "Purchase Invoice"):
base_grand_total = base_grand_total - flt(self.base_write_off_amount)
grand_total = grand_total - flt(self.write_off_amount)
if self.get("total_advance"):
if party_account_currency == self.company_currency:
base_grand_total -= self.get("total_advance")
grand_total = flt(
base_grand_total / self.get("conversion_rate"), self.precision("grand_total")
)
else:
grand_total -= self.get("total_advance")
base_grand_total = flt(
grand_total * self.get("conversion_rate"), self.precision("base_grand_total")
)
if (
abs(
flt(total, self.precision("grand_total"))
- flt(grand_total, self.precision("grand_total"))
)
> 0.1
or abs(
flt(base_total, self.precision("base_grand_total"))
- flt(base_grand_total, self.precision("base_grand_total"))
)
> 0.1
):
frappe.throw(
_("Total Payment Amount in Payment Schedule must be equal to Grand / Rounded Total")
)
validate_payment_schedule_amount(self)
def is_rounded_total_disabled(self):
if self.meta.get_field("disable_rounded_total"):
@@ -2641,10 +2089,10 @@ class AccountsController(TransactionBase):
def get_advance_payment_doctypes(self, payment_type=None) -> list:
return _get_advance_payment_doctypes(payment_type=payment_type)
def set_transaction_currency_and_rate_in_gl_map(self, gl_entries):
for x in gl_entries:
x["transaction_currency"] = self.currency
x["transaction_exchange_rate"] = self.get("conversion_rate") or 1
def set_transaction_currency_and_rate_in_gl_map(self, gl_entries: list) -> None:
from erpnext.accounts.services.exchange_gain_loss import set_transaction_currency_and_rate_in_gl_map
set_transaction_currency_and_rate_in_gl_map(self, gl_entries)
def after_mapping(self, source_doc):
self.set_discount_amount_after_mapping(source_doc)
@@ -2823,98 +2271,12 @@ def update_invoice_status():
frappe.qb.update(invoice).set("status", status).where(conditions).run()
@frappe.whitelist()
def get_payment_terms(
terms_template: str,
posting_date: DateTimeLikeObject | None = None,
grand_total: float | None = None,
base_grand_total: float | None = None,
bill_date: DateTimeLikeObject | None = None,
):
if not terms_template:
return
terms_doc = frappe.get_doc("Payment Terms Template", terms_template)
schedule = []
for d in terms_doc.get("terms"):
d = frappe._dict(d.as_dict())
term_details = get_payment_term_details(d, posting_date, grand_total, base_grand_total, bill_date)
schedule.append(term_details)
return schedule
@frappe.whitelist()
def get_payment_term_details(
term: str | frappe._dict,
posting_date: DateTimeLikeObject | None = None,
grand_total: float | None = None,
base_grand_total: float | None = None,
bill_date: DateTimeLikeObject | None = None,
):
term_details = frappe._dict()
if isinstance(term, str):
term = frappe.get_doc("Payment Term", term)
else:
term_details.payment_term = term.payment_term
fields_to_copy = [
"description",
"invoice_portion",
"discount_type",
"discount",
"mode_of_payment",
"due_date_based_on",
"credit_days",
"credit_months",
"discount_validity_based_on",
"discount_validity",
]
for field in fields_to_copy:
term_details[field] = term.get(field)
term_details.payment_amount = flt(term.invoice_portion) * flt(grand_total) / 100
term_details.base_payment_amount = flt(term.invoice_portion) * flt(base_grand_total) / 100
term_details.outstanding = term_details.payment_amount
term_details.base_outstanding = term_details.base_payment_amount
if bill_date:
term_details.due_date = get_due_date(term, bill_date)
term_details.discount_date = get_discount_date(term, bill_date)
elif posting_date:
term_details.due_date = get_due_date(term, posting_date)
term_details.discount_date = get_discount_date(term, posting_date)
if posting_date and getdate(term_details.due_date) < getdate(posting_date):
term_details.due_date = posting_date
return term_details
def get_due_date(term, posting_date=None, bill_date=None):
due_date = None
date = bill_date or posting_date
if term.due_date_based_on == "Day(s) after invoice date":
due_date = add_days(date, cint(term.credit_days))
elif term.due_date_based_on == "Day(s) after the end of the invoice month":
due_date = add_days(get_last_day(date), cint(term.credit_days))
elif term.due_date_based_on == "Month(s) after the end of the invoice month":
due_date = get_last_day(add_months(date, cint(term.credit_months)))
return due_date
def get_discount_date(term, posting_date=None, bill_date=None):
discount_validity = None
date = bill_date or posting_date
if term.discount_validity_based_on == "Day(s) after invoice date":
discount_validity = add_days(date, cint(term.discount_validity))
elif term.discount_validity_based_on == "Day(s) after the end of the invoice month":
discount_validity = add_days(get_last_day(date), cint(term.discount_validity))
elif term.discount_validity_based_on == "Month(s) after the end of the invoice month":
discount_validity = get_last_day(add_months(date, cint(term.discount_validity)))
return discount_validity
from erpnext.accounts.services.payment_schedule import (
get_discount_date,
get_due_date,
get_payment_term_details,
get_payment_terms,
)
def get_supplier_block_status(party_name):