# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and contributors # For license information, please see license.txt import json from datetime import date from functools import reduce import frappe from frappe import ValidationError, _, qb, scrub, throw from frappe.model.meta import get_field_precision from frappe.query_builder import Tuple from frappe.query_builder.functions import Count from frappe.utils import cint, comma_or, flt, getdate, nowdate from frappe.utils.data import comma_and, fmt_money, get_link_to_form from pypika import Case from pypika.functions import Coalesce, Sum import erpnext from erpnext.accounts.doctype.accounting_dimension.accounting_dimension import get_dimensions from erpnext.accounts.doctype.bank_account.bank_account import ( get_bank_account_details, get_default_company_bank_account, get_party_bank_account, ) from erpnext.accounts.doctype.invoice_discounting.invoice_discounting import ( get_party_account_based_on_invoice_discounting, ) from erpnext.accounts.doctype.journal_entry.journal_entry import get_default_bank_cash_account from erpnext.accounts.doctype.mode_of_payment.mode_of_payment import validate_enabled_mode_of_payment from erpnext.accounts.doctype.repost_accounting_ledger.repost_accounting_ledger import ( validate_docs_for_deferred_accounting, validate_docs_for_voucher_types, ) from erpnext.accounts.doctype.tax_withholding_category.tax_withholding_category import ( get_party_tax_withholding_details, ) from erpnext.accounts.general_ledger import ( make_gl_entries, make_reverse_gl_entries, process_gl_map, ) from erpnext.accounts.party import ( complete_contact_details, get_default_contact, get_party_account, ) from erpnext.accounts.utils import ( cancel_exchange_gain_loss_journal, get_account_currency, get_advance_payment_doctypes, get_balance_on, get_outstanding_invoices, get_reconciliation_effect_date, ) from erpnext.controllers.accounts_controller import ( AccountsController, validate_taxes_and_charges, ) from erpnext.controllers.queries import has_ignored_field from erpnext.setup.utils import get_exchange_rate class InvalidPaymentEntry(ValidationError): pass class PaymentEntry(AccountsController): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) if not self.is_new(): self.setup_party_account_field() def setup_party_account_field(self): self.party_account_field = None self.party_account = None self.party_account_currency = None if self.payment_type == "Receive": self.party_account_field = "paid_from" self.party_account = self.paid_from self.party_account_currency = self.paid_from_account_currency elif self.payment_type == "Pay": self.party_account_field = "paid_to" self.party_account = self.paid_to self.party_account_currency = self.paid_to_account_currency def validate(self): validate_enabled_mode_of_payment(self.mode_of_payment) self.setup_party_account_field() self.set_missing_values() self.set_liability_account() self.set_missing_ref_details(force=True) self.validate_payment_type() self.validate_internal_transfer_accounts() self.validate_party_details() self.set_exchange_rate() self.validate_mandatory() self.validate_reference_documents() self.validate_dunning_deductions() self.set_amounts() self.validate_amounts() self.apply_taxes() self.set_amounts_after_tax() self.clear_unallocated_reference_document_rows() self.validate_transaction_reference() self.set_title() self.set_remarks() self.validate_duplicate_entry() self.validate_payment_type_with_outstanding() self.validate_allocated_amount() self.validate_paid_invoices() self.ensure_supplier_is_not_blocked() self.set_tax_withholding() self.set_status() self.set_total_in_words() def validate_dunning_deductions(self): dunning_names = {row.dunning for row in self.deductions if row.dunning} if not dunning_names: return dunnings = { row.name: row for row in frappe.get_all( "Dunning", filters={"name": ("in", list(dunning_names))}, fields=["name", "docstatus", "customer", "company", "income_account"], ) } for row in self.deductions: if not row.dunning: continue dunning = dunnings.get(row.dunning) if not dunning or dunning.docstatus != 1: frappe.throw(_("Row #{0}: Dunning {1} must be submitted.").format(row.idx, row.dunning)) if self.payment_type != "Receive" or self.party_type != "Customer": frappe.throw( _("Row #{0}: Dunning charges require a Receive payment from a Customer.").format(row.idx) ) if self.party != dunning.customer: frappe.throw( _("Row #{0}: Dunning {1} belongs to a different customer.").format(row.idx, row.dunning) ) if self.company != dunning.company: frappe.throw( _("Row #{0}: Dunning {1} belongs to a different company.").format(row.idx, row.dunning) ) if row.account != dunning.income_account: frappe.throw( _( "Row #{0}: Dunning charges must use the income account {1} configured on Dunning {2}." ).format(row.idx, dunning.income_account, row.dunning) ) if flt(row.amount) >= 0: frappe.throw( _("Row #{0}: Dunning charges must have a negative deduction amount.").format(row.idx) ) def before_save(self): self.set_matched_unset_payment_requests_to_response() def on_submit(self): if self.difference_amount: frappe.throw(_("Difference Amount must be zero")) self.update_payment_requests() self.update_payment_schedule() self.make_gl_entries() self.update_outstanding_amounts() self.update_linked_dunnings() self.set_status() def update_linked_dunnings(self): from erpnext.accounts.doctype.dunning.dunning import update_dunnings_linked_to_payment update_dunnings_linked_to_payment(self) def validate_for_repost(self): validate_docs_for_voucher_types(["Payment Entry"]) validate_docs_for_deferred_accounting([self.name], []) def on_update_after_submit(self): # Flag will be set on Reconciliation # Reconciliation tool will anyways repost ledger entries. So, no need to check and do implicit repost. if self.flags.get("ignore_reposting_on_reconciliation"): return self.needs_repost = self.check_if_fields_updated( fields_to_check=[], child_tables={"references": [], "taxes": [], "deductions": []} ) if self.needs_repost: self.validate_for_repost() self.repost_accounting_entries() def set_liability_account(self): # Auto setting liability account should only be done during 'draft' status if self.docstatus > 0 or self.payment_type == "Internal Transfer": return self.book_advance_payments_in_separate_party_account = False if self.party_type not in ("Customer", "Supplier"): self.is_opening = "No" return if not frappe.db.get_value( "Company", self.company, "book_advance_payments_in_separate_party_account" ): self.is_opening = "No" return # Important to set this flag for the gl building logic to work properly self.book_advance_payments_in_separate_party_account = True account_type = frappe.get_value( "Account", {"name": self.party_account, "company": self.company}, "account_type" ) if (account_type == "Payable" and self.party_type == "Customer") or ( account_type == "Receivable" and self.party_type == "Supplier" ): self.is_opening = "No" return if self.references: allowed_types = frozenset(["Sales Order", "Purchase Order"]) reference_types = set([x.reference_doctype for x in self.references]) # If there are referencers other than `allowed_types`, treat this as a normal payment entry if reference_types - allowed_types: self.book_advance_payments_in_separate_party_account = False self.is_opening = "No" return accounts = get_party_account(self.party_type, self.party, self.company, include_advance=True) liability_account = accounts[1] if len(accounts) > 1 else None fieldname = ( "default_advance_received_account" if self.party_type == "Customer" else "default_advance_paid_account" ) if not liability_account: throw( _("Please set default {0} in Company {1}").format( frappe.bold(frappe.get_meta("Company").get_label(fieldname)), frappe.bold(self.company) ) ) self.set(self.party_account_field, liability_account) frappe.msgprint( _( "Book Advance Payments as Liability option is chosen. Paid From account changed from {0} to {1}." ).format( frappe.bold(self.party_account), frappe.bold(liability_account), ), alert=True, ) def on_cancel(self): self.ignore_linked_doctypes = ( "GL Entry", "Stock Ledger Entry", "Payment Ledger Entry", "Repost Payment Ledger", "Repost Payment Ledger Items", "Repost Accounting Ledger", "Repost Accounting Ledger Items", "Unreconcile Payment", "Unreconcile Payment Entries", "Advance Payment Ledger Entry", ) super().on_cancel() self.update_payment_requests(cancel=True) self.update_payment_schedule(cancel=1) self.make_gl_entries(cancel=1) self.update_outstanding_amounts() self.update_linked_dunnings() self.delink_advance_entry_references() self.set_status() def update_payment_requests(self, cancel=False): from erpnext.accounts.doctype.payment_request.payment_request import ( update_payment_requests_as_per_pe_references, ) update_payment_requests_as_per_pe_references(self.references, cancel=cancel) def update_outstanding_amounts(self): self.set_missing_ref_details(force=True) def validate_duplicate_entry(self): reference_names = set() for d in self.get("references"): key = (d.reference_doctype, d.reference_name, d.payment_term, d.payment_request) if key in reference_names: frappe.throw( _("Row #{0}: Duplicate entry in References {1} {2}").format( d.idx, d.reference_doctype, d.reference_name ) ) reference_names.add(key) def set_bank_account_data(self): if not self.bank_account: return bank_data = get_bank_account_details(self.bank_account) field = "paid_from" if self.payment_type == "Pay" else "paid_to" self.bank = bank_data.bank self.bank_account_no = bank_data.bank_account_no if not self.get(field): self.set(field, bank_data.account) def validate_payment_type_with_outstanding(self): total_outstanding = sum(d.allocated_amount for d in self.get("references")) if total_outstanding < 0 and self.party_type == "Customer" and self.payment_type == "Receive": frappe.throw( _("Cannot receive from customer against negative outstanding"), title=_("Incorrect Payment Type"), ) def validate_allocated_amount(self): if self.payment_type == "Internal Transfer": return self.validate_allocated_amount_as_per_payment_request() if self.party_type in ("Customer", "Supplier"): self.validate_allocated_amount_with_latest_data() return fail_message = _("Row #{0}: Allocated Amount cannot be greater than outstanding amount.") for d in self.get("references"): if (flt(d.allocated_amount)) > 0 and flt(d.allocated_amount) > flt(d.outstanding_amount): frappe.throw(fail_message.format(d.idx)) # Check for negative outstanding invoices as well if flt(d.allocated_amount) < 0 and flt(d.allocated_amount) < flt(d.outstanding_amount): frappe.throw(fail_message.format(d.idx)) def validate_allocated_amount_as_per_payment_request(self): """ Allocated amount should not be greater than the outstanding amount of the Payment Request. """ if not self.references: return pr_outstanding_amounts = get_payment_request_outstanding_set_in_references(self.references) if not pr_outstanding_amounts: return for ref in self.references: if ref.payment_request and ref.allocated_amount > pr_outstanding_amounts[ref.payment_request]: frappe.throw( msg=_( "Row #{0}: Allocated Amount cannot be greater than Outstanding Amount of Payment Request {1}" ).format(ref.idx, get_link_to_form("Payment Request", ref.payment_request)), title=_("Invalid Allocated Amount"), ) def term_based_allocation_enabled_for_reference( self, reference_doctype: str, reference_name: str ) -> bool: if ( reference_doctype and reference_doctype in ["Sales Invoice", "Sales Order", "Purchase Order", "Purchase Invoice"] and reference_name ): if template := frappe.db.get_value(reference_doctype, reference_name, "payment_terms_template"): return frappe.db.get_value( "Payment Terms Template", template, "allocate_payment_based_on_payment_terms" ) return False def validate_allocated_amount_with_latest_data(self): if not self.references: return uniq_vouchers = {(x.reference_doctype, x.reference_name) for x in self.references} vouchers = [frappe._dict({"voucher_type": x[0], "voucher_no": x[1]}) for x in uniq_vouchers] latest_references = get_outstanding_reference_documents( { "posting_date": self.posting_date, "company": self.company, "party_type": self.party_type, "payment_type": self.payment_type, "party": self.party, "party_account": self.paid_from if self.payment_type == "Receive" else self.paid_to, "get_outstanding_invoices": True, "get_orders_to_be_billed": True, "vouchers": vouchers, "book_advance_payments_in_separate_party_account": self.book_advance_payments_in_separate_party_account, }, validate=True, ) # Group latest_references by (voucher_type, voucher_no) latest_lookup = {} for d in latest_references: d = frappe._dict(d) latest_lookup.setdefault((d.voucher_type, d.voucher_no), frappe._dict())[d.payment_term] = d for idx, d in enumerate(self.get("references"), start=1): latest = latest_lookup.get((d.reference_doctype, d.reference_name)) or frappe._dict() # If term based allocation is enabled, throw if ( d.payment_term is None or d.payment_term == "" ) and self.term_based_allocation_enabled_for_reference(d.reference_doctype, d.reference_name): frappe.throw( _( "{0} has Payment Term based allocation enabled. Select a Payment Term for Row #{1} in Payment References section" ).format(frappe.bold(d.reference_name), frappe.bold(idx)) ) # if no payment template is used by invoice and has a custom term(no `payment_term`), then invoice outstanding will be in 'None' key latest = latest.get(d.payment_term) or latest.get(None) # The reference has already been fully paid if not latest: frappe.throw( _("{0} {1} has already been fully paid.").format(_(d.reference_doctype), d.reference_name) ) # The reference has already been partly paid elif ( latest.outstanding_amount < latest.invoice_amount and flt(d.outstanding_amount, d.precision("outstanding_amount")) != flt(latest.outstanding_amount, d.precision("outstanding_amount")) and d.payment_term == "" ): frappe.throw( _( "{0} {1} has already been partly paid. Please use the 'Get Outstanding Invoice' or the 'Get Outstanding Orders' button to get the latest outstanding amounts." ).format(_(d.reference_doctype), d.reference_name) ) fail_message = _("Row #{0}: Allocated Amount cannot be greater than outstanding amount.") if ( d.payment_term and ( (flt(d.allocated_amount)) > 0 and latest.payment_term_outstanding and (flt(d.allocated_amount) > flt(latest.payment_term_outstanding)) ) and self.term_based_allocation_enabled_for_reference(d.reference_doctype, d.reference_name) ): frappe.throw( _( "Row #{0}: Allocated amount:{1} is greater than outstanding amount:{2} for Payment Term {3}" ).format(d.idx, d.allocated_amount, latest.payment_term_outstanding, d.payment_term) ) if (flt(d.allocated_amount)) > 0 and flt(d.allocated_amount) > flt(latest.outstanding_amount): frappe.throw(fail_message.format(d.idx)) # Check for negative outstanding invoices as well if flt(d.allocated_amount) < 0 and flt(d.allocated_amount) < flt(latest.outstanding_amount): frappe.throw(fail_message.format(d.idx)) def delink_advance_entry_references(self): for reference in self.references: if reference.reference_doctype in ("Sales Invoice", "Purchase Invoice"): doc = frappe.get_doc(reference.reference_doctype, reference.reference_name) doc.delink_advance_entries(self.name) def set_missing_values(self): if self.payment_type == "Internal Transfer": for field in ( "party", "party_balance", "total_allocated_amount", "base_total_allocated_amount", "unallocated_amount", ): self.set(field, None) self.references = [] else: if not self.party_type: frappe.throw(_("Party Type is mandatory")) if not self.party: frappe.throw(_("Party is mandatory")) _party_name = "title" if self.party_type == "Shareholder" else self.party_type.lower() + "_name" if frappe.db.has_column(self.party_type, _party_name): self.party_name = frappe.db.get_value(self.party_type, self.party, _party_name) else: self.party_name = frappe.db.get_value(self.party_type, self.party, "name") if self.party: if self.party_type == "Employee": self.contact_person = None elif not self.contact_person: self.contact_person = get_default_contact(self.party_type, self.party) complete_contact_details(self) if not self.party_balance and frappe.get_single_value("Accounts Settings", "show_party_balance"): self.party_balance = get_balance_on( party_type=self.party_type, party=self.party, date=self.posting_date, company=self.company ) if not self.party_account: party_account = get_party_account(self.party_type, self.party, self.company) self.set(self.party_account_field, party_account) self.party_account = party_account if self.paid_from and ( not self.paid_from_account_currency or not self.paid_from_account_balance or not self.paid_from_account_type ): acc = get_account_details(self.paid_from, self.posting_date, self.cost_center) self.paid_from_account_currency = acc.account_currency self.paid_from_account_balance = acc.account_balance self.paid_from_account_type = acc.account_type if self.paid_to and ( not self.paid_to_account_currency or not self.paid_to_account_balance or not self.paid_to_account_type ): acc = get_account_details(self.paid_to, self.posting_date, self.cost_center) self.paid_to_account_currency = acc.account_currency self.paid_to_account_balance = acc.account_balance self.paid_to_account_type = acc.account_type self.party_account_currency = ( self.paid_from_account_currency if self.payment_type == "Receive" else self.paid_to_account_currency ) def set_missing_ref_details( self, force: bool = False, update_ref_details_only_for: list | None = None, reference_exchange_details: dict | None = None, ) -> None: for d in self.get("references"): if not d.allocated_amount: continue if update_ref_details_only_for and ( (d.reference_doctype, d.reference_name) not in update_ref_details_only_for ): continue ref_details = get_reference_details( d.reference_doctype, d.reference_name, self.party_account_currency, self.party_type, self.party, ) # Only update exchange rate when the reference is Journal Entry if ( reference_exchange_details and d.reference_doctype == reference_exchange_details.reference_doctype and d.reference_name == reference_exchange_details.reference_name ): ref_details.update({"exchange_rate": reference_exchange_details.exchange_rate}) for field, value in ref_details.items(): if d.exchange_gain_loss: # for cases where gain/loss is booked into invoice # exchange_gain_loss is calculated from invoice & populated # and row.exchange_rate is already set to payment entry's exchange rate # refer -> `update_reference_in_payment_entry()` in utils.py continue if field == "exchange_rate" or not d.get(field) or force: d.db_set(field, value) def validate_payment_type(self): if self.payment_type not in ("Receive", "Pay", "Internal Transfer"): frappe.throw(_("Payment Type must be one of Receive, Pay and Internal Transfer")) def validate_internal_transfer_accounts(self): if self.payment_type == "Internal Transfer" and self.paid_from and self.paid_from == self.paid_to: frappe.throw(_("Paid From and Paid To accounts must be different for an Internal Transfer.")) def validate_party_details(self): if self.party and not frappe.db.exists(self.party_type, self.party): frappe.throw(_("{0} {1} does not exist").format(_(self.party_type), self.party)) def set_exchange_rate(self, ref_doc=None): self.set_source_exchange_rate(ref_doc) self.set_target_exchange_rate(ref_doc) def set_source_exchange_rate(self, ref_doc=None): if self.paid_from: if self.paid_from_account_currency == self.company_currency: self.source_exchange_rate = 1 elif ref_doc and self.paid_from_account_currency == ref_doc.currency: self.source_exchange_rate = ref_doc.get("exchange_rate") or ref_doc.get("conversion_rate") if not self.source_exchange_rate: self.source_exchange_rate = get_exchange_rate( self.paid_from_account_currency, self.company_currency, self.posting_date ) def set_target_exchange_rate(self, ref_doc=None): if self.paid_from_account_currency == self.paid_to_account_currency: self.target_exchange_rate = self.source_exchange_rate elif self.paid_to and not self.target_exchange_rate: if ref_doc and self.paid_to_account_currency == ref_doc.currency: self.target_exchange_rate = ref_doc.get("exchange_rate") or ref_doc.get("conversion_rate") if not self.target_exchange_rate: self.target_exchange_rate = get_exchange_rate( self.paid_to_account_currency, self.company_currency, self.posting_date ) def validate_mandatory(self): for field in ("paid_amount", "received_amount", "source_exchange_rate", "target_exchange_rate"): self.set(field, flt(self.get(field))) if not self.get(field): frappe.throw(_("{0} is mandatory").format(_(self.meta.get_label(field)))) if self.get(field) < 0: frappe.throw(_("{0} must be greater than zero").format(_(self.meta.get_label(field)))) def validate_reference_documents(self): valid_reference_doctypes = self.get_valid_reference_doctypes() if not valid_reference_doctypes: return for d in self.get("references"): if not d.allocated_amount: continue if d.reference_doctype not in valid_reference_doctypes: frappe.throw( _("Reference Doctype must be one of {0}").format( comma_or([_(d) for d in valid_reference_doctypes]) ) ) elif d.reference_name: if not frappe.db.exists(d.reference_doctype, d.reference_name): frappe.throw(_("{0} {1} does not exist").format(d.reference_doctype, d.reference_name)) ref_doc = frappe.get_doc(d.reference_doctype, d.reference_name) if d.reference_doctype != "Journal Entry": if self.party != ref_doc.get(scrub(self.party_type)): frappe.throw( _("{0} {1} is not associated with {2} {3}").format( _(d.reference_doctype), d.reference_name, _(self.party_type), self.party ) ) else: self.validate_journal_entry() if d.reference_doctype in frappe.get_hooks("invoice_doctypes"): if self.party_type == "Customer": ref_party_account = ( get_party_account_based_on_invoice_discounting(d.reference_name) or ref_doc.debit_to ) elif self.party_type == "Supplier": ref_party_account = ref_doc.credit_to elif self.party_type == "Employee": ref_party_account = ref_doc.payable_account if ( ref_party_account != self.party_account and not self.book_advance_payments_in_separate_party_account ): frappe.throw( _("{0} {1} is associated with {2}, but Party Account is {3}").format( _(d.reference_doctype), d.reference_name, ref_party_account, self.party_account, ) ) if ref_doc.doctype == "Purchase Invoice" and ref_doc.get("on_hold"): frappe.throw( _("{0} {1} is on hold").format(_(d.reference_doctype), d.reference_name), title=_("Invalid Purchase Invoice"), ) if ref_doc.docstatus != 1: frappe.throw( _("{0} {1} must be submitted").format(_(d.reference_doctype), d.reference_name) ) def get_valid_reference_doctypes(self): if self.party_type == "Customer": return ("Sales Order", "Sales Invoice", "Journal Entry", "Dunning", "Payment Entry") elif self.party_type in ["Shareholder", "Employee"]: return ("Journal Entry",) elif self.party_type == "Supplier": return ("Purchase Order", "Purchase Invoice", "Journal Entry", "Payment Entry") def validate_paid_invoices(self): no_oustanding_refs = {} for d in self.get("references"): if not d.allocated_amount: continue if d.reference_doctype in ("Sales Invoice", "Purchase Invoice"): outstanding_amount, is_return = frappe.get_cached_value( d.reference_doctype, d.reference_name, ["outstanding_amount", "is_return"] ) if outstanding_amount <= 0 and not is_return: no_oustanding_refs.setdefault(d.reference_doctype, []).append(d) for reference_doctype, references in no_oustanding_refs.items(): frappe.msgprint( _( "References {0} of type {1} had no outstanding amount left before submitting the Payment Entry. Now they have a negative outstanding amount." ).format( frappe.bold(comma_and([d.reference_name for d in references])), _(reference_doctype), ) + "

" + _("If this is undesirable please cancel the corresponding Payment Entry."), title=_("Warning"), indicator="orange", ) def validate_journal_entry(self): for d in self.get("references"): if d.allocated_amount and d.reference_doctype == "Journal Entry": je_accounts = frappe.db.sql( """select debit, credit from `tabJournal Entry Account` where account = %s and party=%s and docstatus = 1 and parent = %s and (reference_type is null or reference_type in ("", "Sales Order", "Purchase Order")) """, (self.party_account, self.party, d.reference_name), as_dict=True, ) if not je_accounts: frappe.throw( _( "Row #{0}: Journal Entry {1} does not have account {2} or already matched against another voucher" ).format(d.idx, d.reference_name, self.party_account) ) else: dr_or_cr = "debit" if self.payment_type == "Receive" else "credit" valid = False for jvd in je_accounts: if flt(jvd[dr_or_cr]) > 0: valid = True if not valid: frappe.throw( _("Against Journal Entry {0} does not have any unmatched {1} entry").format( d.reference_name, _(dr_or_cr) ) ) def update_payment_schedule(self, cancel=0): invoice_payment_amount_map = {} invoice_paid_amount_map = {} for ref in self.get("references"): if not ref.payment_term or not ref.reference_name: continue key = (ref.payment_term, ref.reference_name, ref.reference_doctype) invoice_payment_amount_map.setdefault(key, 0.0) invoice_payment_amount_map[key] += ref.allocated_amount if not invoice_paid_amount_map.get(key): payment_schedule = frappe.get_all( "Payment Schedule", filters={"parent": ref.reference_name}, fields=[ "paid_amount", "payment_amount", "payment_term", "discount", "outstanding", "discount_type", ], ) for term in payment_schedule: invoice_key = (term.payment_term, ref.reference_name, ref.reference_doctype) invoice_paid_amount_map.setdefault(invoice_key, {}) invoice_paid_amount_map[invoice_key]["outstanding"] = term.outstanding if not (term.discount_type and term.discount): continue if term.discount_type == "Percentage": invoice_paid_amount_map[invoice_key]["discounted_amt"] = ref.total_amount * ( term.discount / 100 ) else: invoice_paid_amount_map[invoice_key]["discounted_amt"] = term.discount for idx, (key, allocated_amount) in enumerate(invoice_payment_amount_map.items(), 1): if not invoice_paid_amount_map.get(key): frappe.throw(_("Payment term {0} not used in {1}").format(key[0], key[1])) allocated_amount = self.get_allocated_amount_in_transaction_currency( allocated_amount, key[2], key[1] ) outstanding = flt(invoice_paid_amount_map.get(key, {}).get("outstanding")) discounted_amt = flt(invoice_paid_amount_map.get(key, {}).get("discounted_amt")) conversion_rate = frappe.db.get_value(key[2], {"name": key[1]}, "conversion_rate") base_paid_amount_precision = get_field_precision( frappe.get_meta("Payment Schedule").get_field("base_paid_amount") ) base_outstanding_precision = get_field_precision( frappe.get_meta("Payment Schedule").get_field("base_outstanding") ) base_paid_amount = flt( (allocated_amount - discounted_amt) * conversion_rate, base_paid_amount_precision ) base_outstanding = flt(allocated_amount * conversion_rate, base_outstanding_precision) if cancel: frappe.db.sql( """ UPDATE `tabPayment Schedule` SET paid_amount = `paid_amount` - %s, base_paid_amount = `base_paid_amount` - %s, discounted_amount = `discounted_amount` - %s, outstanding = `outstanding` + %s, base_outstanding = `base_outstanding` + %s WHERE parent = %s and payment_term = %s""", ( allocated_amount - discounted_amt, base_paid_amount, discounted_amt, allocated_amount, base_outstanding, key[1], key[0], ), ) else: if allocated_amount > outstanding: frappe.throw( _("Row #{0}: Cannot allocate more than {1} against payment term {2}").format( idx, fmt_money(outstanding), key[0] ) ) if allocated_amount and outstanding: frappe.db.sql( """ UPDATE `tabPayment Schedule` SET paid_amount = `paid_amount` + %s, base_paid_amount = `base_paid_amount` + %s, discounted_amount = `discounted_amount` + %s, outstanding = `outstanding` - %s, base_outstanding = `base_outstanding` - %s WHERE parent = %s and payment_term = %s""", ( allocated_amount - discounted_amt, base_paid_amount, discounted_amt, allocated_amount, base_outstanding, key[1], key[0], ), ) def get_allocated_amount_in_transaction_currency( self, allocated_amount, reference_doctype, reference_docname ): """ Payment Entry could be in base currency while reference's payment schedule is always in transaction currency. E.g. * SI with base=INR and currency=USD * SI with payment schedule in USD * PE in INR (accounting done in base currency) """ ref_currency, ref_exchange_rate = frappe.db.get_value( reference_doctype, reference_docname, ["currency", "conversion_rate"] ) is_single_currency = self.paid_from_account_currency == self.paid_to_account_currency # PE in different currency reference_is_multi_currency = self.paid_from_account_currency != ref_currency if not (is_single_currency and reference_is_multi_currency): return allocated_amount allocated_amount = flt(allocated_amount / ref_exchange_rate, self.precision("total_allocated_amount")) return allocated_amount def set_status(self): if self.docstatus == 2: self.status = "Cancelled" elif self.docstatus == 1: self.status = "Submitted" else: self.status = "Draft" self.db_set("status", self.status, update_modified=True) def set_total_in_words(self): from frappe.utils import money_in_words if self.payment_type in ("Pay", "Internal Transfer"): base_amount = abs(self.base_paid_amount) amount = abs(self.paid_amount) currency = self.paid_from_account_currency elif self.payment_type == "Receive": base_amount = abs(self.base_received_amount) amount = abs(self.received_amount) currency = self.paid_to_account_currency self.base_in_words = money_in_words(base_amount, self.company_currency) self.in_words = money_in_words(amount, currency) def set_tax_withholding(self): if not self.party_type == "Supplier": return if not self.apply_tax_withholding_amount: return net_total = self.calculate_tax_withholding_net_total() # Adding args as purchase invoice to get TDS amount args = frappe._dict( { "company": self.company, "doctype": "Payment Entry", "supplier": self.party, "posting_date": self.posting_date, "net_total": net_total, } ) tax_withholding_details = get_party_tax_withholding_details(args, self.tax_withholding_category) if not tax_withholding_details: return tax_withholding_details.update( {"cost_center": self.cost_center or erpnext.get_default_cost_center(self.company)} ) accounts = [] for d in self.taxes: if d.account_head == tax_withholding_details.get("account_head"): # Preserve user updated included in paid amount if d.included_in_paid_amount: tax_withholding_details.update({"included_in_paid_amount": d.included_in_paid_amount}) d.update(tax_withholding_details) accounts.append(d.account_head) if not accounts or tax_withholding_details.get("account_head") not in accounts: self.append("taxes", tax_withholding_details) to_remove = [ d for d in self.taxes if not d.tax_amount and d.account_head == tax_withholding_details.get("account_head") ] for d in to_remove: self.remove(d) def calculate_tax_withholding_net_total(self): net_total = 0 order_details = self.get_order_wise_tax_withholding_net_total() for d in self.references: tax_withholding_net_total = order_details.get(d.reference_name) if not tax_withholding_net_total: continue net_taxable_outstanding = max( 0, d.outstanding_amount - (d.total_amount - tax_withholding_net_total) ) net_total += min(net_taxable_outstanding, d.allocated_amount) net_total += self.unallocated_amount return net_total def get_order_wise_tax_withholding_net_total(self): if self.party_type == "Supplier": doctype = "Purchase Order" else: doctype = "Sales Order" docnames = [d.reference_name for d in self.references if d.reference_doctype == doctype] return frappe._dict( frappe.db.get_all( doctype, filters={"name": ["in", docnames]}, fields=["name", "base_tax_withholding_net_total"], as_list=True, ) ) def apply_taxes(self): self.initialize_taxes() self.determine_exclusive_rate() self.calculate_taxes() def set_amounts(self): self.set_received_amount() self.set_amounts_in_company_currency() self.set_total_allocated_amount() self.set_unallocated_amount() self.set_exchange_gain_loss() self.set_difference_amount() def validate_amounts(self): self.validate_received_amount() def validate_received_amount(self): if self.paid_from_account_currency == self.paid_to_account_currency: if self.paid_amount < self.received_amount: frappe.throw(_("Received Amount cannot be greater than Paid Amount")) def set_received_amount(self): self.base_received_amount = self.base_paid_amount if ( self.paid_from_account_currency == self.paid_to_account_currency and not self.payment_type == "Internal Transfer" ): self.received_amount = self.paid_amount def set_amounts_after_tax(self): applicable_tax = 0 base_applicable_tax = 0 for tax in self.get("taxes"): if tax.included_in_paid_amount: continue amount = -1 * tax.tax_amount if tax.add_deduct_tax == "Deduct" else tax.tax_amount base_amount = -1 * tax.base_tax_amount if tax.add_deduct_tax == "Deduct" else tax.base_tax_amount applicable_tax += amount base_applicable_tax += base_amount self.paid_amount_after_tax = flt( flt(self.paid_amount) + flt(applicable_tax), self.precision("paid_amount_after_tax") ) self.base_paid_amount_after_tax = flt( flt(self.paid_amount_after_tax) * flt(self.source_exchange_rate), self.precision("base_paid_amount_after_tax"), ) self.received_amount_after_tax = flt( flt(self.received_amount) + flt(applicable_tax), self.precision("paid_amount_after_tax") ) self.base_received_amount_after_tax = flt( flt(self.received_amount_after_tax) * flt(self.target_exchange_rate), self.precision("base_paid_amount_after_tax"), ) def set_amounts_in_company_currency(self): self.base_paid_amount, self.base_received_amount, self.difference_amount = 0, 0, 0 if self.paid_amount: self.base_paid_amount = flt( flt(self.paid_amount) * flt(self.source_exchange_rate), self.precision("base_paid_amount") ) if self.received_amount: self.base_received_amount = flt( flt(self.received_amount) * flt(self.target_exchange_rate), self.precision("base_received_amount"), ) def calculate_base_allocated_amount_for_reference(self, d) -> float: base_allocated_amount = 0 if d.reference_doctype in get_advance_payment_doctypes(): # When referencing Sales/Purchase Order, use the source/target exchange rate depending on payment type. # This is so there are no Exchange Gain/Loss generated for such doctypes exchange_rate = 1 if self.payment_type == "Receive": exchange_rate = self.source_exchange_rate elif self.payment_type == "Pay": exchange_rate = self.target_exchange_rate base_allocated_amount += flt( flt(d.allocated_amount) * flt(exchange_rate), self.precision("base_paid_amount") ) else: # Use source/target exchange rate, so no difference amount is calculated. # then update exchange gain/loss amount in reference table # if there is an exchange gain/loss amount in reference table, submit a JE for that exchange_rate = 1 if self.payment_type == "Receive": exchange_rate = self.source_exchange_rate elif self.payment_type == "Pay": exchange_rate = self.target_exchange_rate base_allocated_amount += flt( flt(d.allocated_amount) * flt(exchange_rate), self.precision("base_paid_amount") ) # on rare case, when `exchange_rate` is unset, gain/loss amount is incorrectly calculated # for base currency transactions if d.exchange_rate is None: d.exchange_rate = 1 allocated_amount_in_ref_exchange_rate = flt( flt(d.allocated_amount) * flt(d.exchange_rate), self.precision("base_paid_amount") ) d.exchange_gain_loss = base_allocated_amount - allocated_amount_in_ref_exchange_rate return base_allocated_amount def set_total_allocated_amount(self): if self.payment_type == "Internal Transfer": return total_allocated_amount, base_total_allocated_amount = 0, 0 for d in self.get("references"): if d.allocated_amount: total_allocated_amount += flt(d.allocated_amount) base_total_allocated_amount += self.calculate_base_allocated_amount_for_reference(d) self.total_allocated_amount = abs(total_allocated_amount) self.base_total_allocated_amount = abs(base_total_allocated_amount) def set_unallocated_amount(self): self.unallocated_amount = 0 if not self.party: return deductions_to_consider = sum( flt(d.amount) for d in self.get("deductions") if not d.is_exchange_gain_loss ) included_taxes = self.get_included_taxes() if self.payment_type == "Receive" and self.base_total_allocated_amount < ( self.base_paid_amount + deductions_to_consider ): self.unallocated_amount = ( flt( ( self.base_paid_amount + deductions_to_consider - self.base_total_allocated_amount - included_taxes ), self.precision("unallocated_amount"), ) / self.source_exchange_rate ) elif self.payment_type == "Pay" and self.base_total_allocated_amount < ( self.base_received_amount - deductions_to_consider ): self.unallocated_amount = ( flt( ( self.base_received_amount - deductions_to_consider - self.base_total_allocated_amount - included_taxes ), self.precision("unallocated_amount"), ) / self.target_exchange_rate ) def set_exchange_gain_loss(self): exchange_gain_loss = flt( self.base_paid_amount - self.base_received_amount, self.precision("amount", "deductions"), ) exchange_gain_loss_rows = [row for row in self.get("deductions") if row.is_exchange_gain_loss] exchange_gain_loss_row = exchange_gain_loss_rows.pop(0) if exchange_gain_loss_rows else None for row in exchange_gain_loss_rows: self.remove(row) if not exchange_gain_loss: if exchange_gain_loss_row: self.remove(exchange_gain_loss_row) return if not exchange_gain_loss_row: values = frappe.get_cached_value( "Company", self.company, ("exchange_gain_loss_account", "cost_center"), as_dict=True ) for fieldname, value in values.items(): if value: continue label = _(frappe.get_meta("Company").get_label(fieldname)) return frappe.msgprint( _("Please set {0} in Company {1} to account for Exchange Gain / Loss").format( label, get_link_to_form("Company", self.company) ), title=_("Missing Default in Company"), indicator="red" if self.docstatus.is_submitted() else "yellow", raise_exception=self.docstatus.is_submitted(), ) exchange_gain_loss_row = self.append( "deductions", { "account": values.exchange_gain_loss_account, "cost_center": values.cost_center, "is_exchange_gain_loss": 1, }, ) exchange_gain_loss_row.amount = exchange_gain_loss def set_difference_amount(self): base_unallocated_amount = flt(self.unallocated_amount) * ( flt(self.source_exchange_rate) if self.payment_type == "Receive" else flt(self.target_exchange_rate) ) base_party_amount = flt(self.base_total_allocated_amount) + flt(base_unallocated_amount) included_taxes = self.get_included_taxes() if self.payment_type == "Receive": self.difference_amount = base_party_amount - self.base_received_amount + included_taxes elif self.payment_type == "Pay": self.difference_amount = self.base_paid_amount - base_party_amount - included_taxes else: self.difference_amount = self.base_paid_amount - flt(self.base_received_amount) - included_taxes total_deductions = sum(flt(d.amount) for d in self.get("deductions")) self.difference_amount = flt( self.difference_amount - total_deductions, self.precision("difference_amount") ) def get_included_taxes(self): included_taxes = 0 for tax in self.get("taxes"): if not tax.included_in_paid_amount: continue if tax.add_deduct_tax == "Add": included_taxes += flt(tax.base_tax_amount) else: included_taxes -= flt(tax.base_tax_amount) return included_taxes # Paid amount is auto allocated in the reference document by default. # Clear the reference document which doesn't have allocated amount on validate so that form can be loaded fast def clear_unallocated_reference_document_rows(self): self.set("references", self.get("references", {"allocated_amount": ["not in", [0, None, ""]]})) frappe.db.sql( """delete from `tabPayment Entry Reference` where parent = %s and allocated_amount = 0""", self.name, ) def set_title(self): if frappe.flags.in_import and self.title: # do not set title dynamically if title exists during data import. return if self.payment_type in ("Receive", "Pay"): self.title = self.party else: self.title = self.paid_from + " - " + self.paid_to def validate_transaction_reference(self): bank_account = self.paid_to if self.payment_type == "Receive" else self.paid_from bank_account_type = frappe.get_cached_value("Account", bank_account, "account_type") if bank_account_type == "Bank": if not self.reference_no or not self.reference_date: frappe.throw(_("Reference No and Reference Date is mandatory for Bank transaction")) def set_remarks(self): if self.custom_remarks: return if self.payment_type == "Internal Transfer": remarks = [ _("Amount {0} {1} transferred from {2} to {3}").format( _(self.paid_from_account_currency), self.paid_amount, self.paid_from, self.paid_to ) ] else: remarks = [ _("Amount {0} {1} {2} {3}").format( _(self.party_account_currency), self.paid_amount if self.payment_type == "Receive" else self.received_amount, _("received from") if self.payment_type == "Receive" else _("to"), self.party, ) ] if self.reference_no: remarks.append( _("Transaction reference no {0} dated {1}").format(self.reference_no, self.reference_date) ) if self.payment_type in ["Receive", "Pay"]: for d in self.get("references"): if d.allocated_amount: remarks.append( _("Amount {0} {1} against {2} {3}").format( _(self.party_account_currency), d.allocated_amount, d.reference_doctype, d.reference_name, ) ) for d in self.get("deductions"): if d.amount: remarks.append( _("Amount {0} {1} deducted against {2}").format( _(self.company_currency), d.amount, d.account ) ) self.set("remarks", "\n".join(remarks)) def set_transaction_currency_and_rate(self): company_currency = erpnext.get_company_currency(self.company) self.transaction_currency = company_currency self.transaction_exchange_rate = 1 if self.paid_from_account_currency != company_currency: self.transaction_currency = self.paid_from_account_currency self.transaction_exchange_rate = self.source_exchange_rate elif self.paid_to_account_currency != company_currency: self.transaction_currency = self.paid_to_account_currency self.transaction_exchange_rate = self.target_exchange_rate def build_gl_map(self): if self.payment_type in ("Receive", "Pay") and not self.get("party_account_field"): self.setup_party_account_field() self.set_transaction_currency_and_rate() gl_entries = [] self.add_party_gl_entries(gl_entries) self.add_bank_gl_entries(gl_entries) self.add_deductions_gl_entries(gl_entries) self.add_tax_gl_entries(gl_entries) add_regional_gl_entries(gl_entries, self) self.set_transaction_currency_and_rate_in_gl_map(gl_entries) return gl_entries def set_transaction_currency_and_rate_in_gl_map(self, gl_entries): for gle in gl_entries: gle.setdefault("transaction_currency", self.transaction_currency) gle.setdefault("transaction_exchange_rate", self.transaction_exchange_rate) def make_gl_entries(self, cancel=0, adv_adj=0): gl_entries = self.build_gl_map() gl_entries = process_gl_map(gl_entries) make_gl_entries(gl_entries, cancel=cancel, adv_adj=adv_adj) if cancel: cancel_exchange_gain_loss_journal(frappe._dict(doctype=self.doctype, name=self.name)) else: self.make_exchange_gain_loss_journal() self.make_advance_gl_entries(cancel=cancel) def add_party_gl_entries(self, gl_entries): if not self.party_account: return advance_payment_doctypes = get_advance_payment_doctypes() if self.payment_type == "Receive": against_account = self.paid_to else: against_account = self.paid_from party_account_type = frappe.db.get_value("Party Type", self.party_type, "account_type") party_gl_dict = self.get_gl_dict( { "account": self.party_account, "party_type": self.party_type, "party": self.party, "against": against_account, "account_currency": self.party_account_currency, "cost_center": self.cost_center, }, item=self, ) for d in self.get("references"): # re-defining dr_or_cr for every reference in order to avoid the last value affecting calculation of reverse dr_or_cr = "credit" if self.payment_type == "Receive" else "debit" cost_center = self.cost_center if d.reference_doctype == "Sales Invoice" and not cost_center: cost_center = frappe.db.get_value(d.reference_doctype, d.reference_name, "cost_center") gle = party_gl_dict.copy() allocated_amount_in_company_currency = self.calculate_base_allocated_amount_for_reference(d) if ( d.reference_doctype in ["Sales Invoice", "Purchase Invoice"] and d.allocated_amount < 0 and ( (party_account_type == "Receivable" and self.payment_type == "Pay") or (party_account_type == "Payable" and self.payment_type == "Receive") ) ): # reversing dr_cr because because it will get reversed in gl processing due to negative amount dr_or_cr = "debit" if dr_or_cr == "credit" else "credit" gle.update( self.get_gl_dict( { "account": self.party_account, "party_type": self.party_type, "party": self.party, "against": against_account, "account_currency": self.party_account_currency, "cost_center": cost_center, dr_or_cr + "_in_account_currency": d.allocated_amount, dr_or_cr: allocated_amount_in_company_currency, dr_or_cr + "_in_transaction_currency": d.allocated_amount if self.transaction_currency == self.party_account_currency else allocated_amount_in_company_currency / self.transaction_exchange_rate, "advance_voucher_type": d.advance_voucher_type, "advance_voucher_no": d.advance_voucher_no, }, item=self, ) ) if d.reference_doctype in advance_payment_doctypes: # advance reference gle.update( { "against_voucher_type": self.doctype, "against_voucher": self.name, "advance_voucher_type": d.reference_doctype, "advance_voucher_no": d.reference_name, } ) elif self.book_advance_payments_in_separate_party_account: # Do not reference Invoices while Advance is in separate party account gle.update({"against_voucher_type": self.doctype, "against_voucher": self.name}) else: gle.update( { "against_voucher_type": d.reference_doctype, "against_voucher": d.reference_name, } ) gl_entries.append(gle) if self.unallocated_amount: dr_or_cr = "credit" if self.payment_type == "Receive" else "debit" exchange_rate = self.get_exchange_rate() base_unallocated_amount = self.unallocated_amount * exchange_rate gle = party_gl_dict.copy() gle.update( self.get_gl_dict( { "account": self.party_account, "party_type": self.party_type, "party": self.party, "against": against_account, "account_currency": self.party_account_currency, "cost_center": self.cost_center, dr_or_cr + "_in_account_currency": self.unallocated_amount, dr_or_cr + "_in_transaction_currency": self.unallocated_amount if self.party_account_currency == self.transaction_currency else base_unallocated_amount / self.transaction_exchange_rate, dr_or_cr: base_unallocated_amount, }, item=self, ) ) if self.book_advance_payments_in_separate_party_account: gle.update( { "against_voucher_type": "Payment Entry", "against_voucher": self.name, } ) gl_entries.append(gle) def make_advance_gl_entries( self, entry: object | dict = None, cancel: bool = 0, update_outstanding: str = "Yes" ): self.set_transaction_currency_and_rate() gl_entries = [] self.add_advance_gl_entries(gl_entries, entry) if cancel: make_reverse_gl_entries(gl_entries, partial_cancel=True) else: make_gl_entries(gl_entries, update_outstanding=update_outstanding) def add_advance_gl_entries(self, gl_entries: list, entry: object | dict | None): """ If 'entry' is passed, GL entries only for that reference is added. """ if self.book_advance_payments_in_separate_party_account: references = [x for x in self.get("references")] if entry: references = [x for x in self.get("references") if x.name == entry.name] for ref in references: if ref.reference_doctype in ( "Sales Invoice", "Purchase Invoice", "Journal Entry", "Payment Entry", ): self.add_advance_gl_for_reference(gl_entries, ref) def get_dr_and_account_for_advances(self, reference): if reference.reference_doctype == "Sales Invoice": return "credit", reference.account if reference.reference_doctype == "Purchase Invoice": return "debit", reference.account if reference.reference_doctype == "Payment Entry": # reference.account_type and reference.payment_type is only available for Reverse payments if reference.account_type == "Receivable" and reference.payment_type == "Pay": return "credit", self.party_account else: return "debit", self.party_account if reference.reference_doctype == "Journal Entry": if self.party_type == "Customer" and self.payment_type == "Receive": return "credit", reference.account else: return "debit", reference.account def add_advance_gl_for_reference(self, gl_entries, invoice): args_dict = { "party_type": self.party_type, "party": self.party, "account_currency": self.party_account_currency, "cost_center": self.cost_center, "voucher_type": "Payment Entry", "voucher_no": self.name, "voucher_detail_no": invoice.name, } if invoice.reconcile_effect_on: posting_date = invoice.reconcile_effect_on else: # For backwards compatibility # Supporting reposting on payment entries reconciled before select field introduction posting_date = get_reconciliation_effect_date( invoice.reference_doctype, invoice.reference_name, self.company, self.posting_date ) frappe.db.set_value("Payment Entry Reference", invoice.name, "reconcile_effect_on", posting_date) dr_or_cr, account = self.get_dr_and_account_for_advances(invoice) base_allocated_amount = self.calculate_base_allocated_amount_for_reference(invoice) args_dict["account"] = account args_dict[dr_or_cr] = base_allocated_amount args_dict[dr_or_cr + "_in_account_currency"] = invoice.allocated_amount args_dict[dr_or_cr + "_in_transaction_currency"] = ( invoice.allocated_amount if self.party_account_currency == self.transaction_currency else base_allocated_amount / self.transaction_exchange_rate ) args_dict.update( { "against_voucher_type": invoice.reference_doctype, "against_voucher": invoice.reference_name, "advance_voucher_type": invoice.advance_voucher_type, "advance_voucher_no": invoice.advance_voucher_no, "posting_date": posting_date, } ) gle = self.get_gl_dict( args_dict, item=self, ) gl_entries.append(gle) args_dict[dr_or_cr] = 0 args_dict[dr_or_cr + "_in_account_currency"] = 0 dr_or_cr = "debit" if dr_or_cr == "credit" else "credit" args_dict["account"] = self.party_account args_dict[dr_or_cr] = base_allocated_amount args_dict[dr_or_cr + "_in_account_currency"] = invoice.allocated_amount args_dict[dr_or_cr + "_in_transaction_currency"] = ( invoice.allocated_amount if self.party_account_currency == self.transaction_currency else base_allocated_amount / self.transaction_exchange_rate ) args_dict.update( { "against_voucher_type": "Payment Entry", "against_voucher": self.name, "advance_voucher_type": invoice.advance_voucher_type, "advance_voucher_no": invoice.advance_voucher_no, } ) gle = self.get_gl_dict( args_dict, item=self, ) gl_entries.append(gle) def add_bank_gl_entries(self, gl_entries): if self.payment_type in ("Pay", "Internal Transfer"): gl_entries.append( self.get_gl_dict( { "account": self.paid_from, "account_currency": self.paid_from_account_currency, "against": self.party if self.payment_type == "Pay" else self.paid_to, "credit_in_account_currency": self.paid_amount, "credit_in_transaction_currency": self.paid_amount if self.paid_from_account_currency == self.transaction_currency else self.base_paid_amount / self.transaction_exchange_rate, "credit": self.base_paid_amount, "cost_center": self.cost_center, "post_net_value": True, }, item=self, ) ) if self.payment_type in ("Receive", "Internal Transfer"): gl_entries.append( self.get_gl_dict( { "account": self.paid_to, "account_currency": self.paid_to_account_currency, "against": self.party if self.payment_type == "Receive" else self.paid_from, "debit_in_account_currency": self.received_amount, "debit_in_transaction_currency": self.received_amount if self.paid_to_account_currency == self.transaction_currency else self.base_received_amount / self.transaction_exchange_rate, "debit": self.base_received_amount, "cost_center": self.cost_center, }, item=self, ) ) def add_tax_gl_entries(self, gl_entries): for d in self.get("taxes"): account_currency = get_account_currency(d.account_head) if account_currency != self.company_currency: frappe.throw(_("Currency for {0} must be {1}").format(d.account_head, self.company_currency)) if self.payment_type in ("Pay", "Internal Transfer"): dr_or_cr = "debit" if d.add_deduct_tax == "Add" else "credit" rev_dr_or_cr = "credit" if dr_or_cr == "debit" else "debit" against = self.party or self.paid_from elif self.payment_type == "Receive": dr_or_cr = "credit" if d.add_deduct_tax == "Add" else "debit" rev_dr_or_cr = "credit" if dr_or_cr == "debit" else "debit" against = self.party or self.paid_to payment_account = self.get_party_account_for_taxes() tax_amount = d.tax_amount base_tax_amount = d.base_tax_amount gl_entries.append( self.get_gl_dict( { "account": d.account_head, "against": against, dr_or_cr: tax_amount, dr_or_cr + "_in_account_currency": base_tax_amount if account_currency == self.company_currency else d.tax_amount, dr_or_cr + "_in_transaction_currency": base_tax_amount / self.transaction_exchange_rate, "cost_center": d.cost_center, "post_net_value": True, }, account_currency, item=d, ) ) if not d.included_in_paid_amount: if get_account_currency(payment_account) != self.company_currency: if self.payment_type == "Receive": exchange_rate = self.target_exchange_rate elif self.payment_type in ["Pay", "Internal Transfer"]: exchange_rate = self.source_exchange_rate base_tax_amount = flt((tax_amount / exchange_rate), self.precision("paid_amount")) gl_entries.append( self.get_gl_dict( { "account": payment_account, "against": against, rev_dr_or_cr: tax_amount, rev_dr_or_cr + "_in_account_currency": base_tax_amount if account_currency == self.company_currency else d.tax_amount, rev_dr_or_cr + "_in_transaction_currency": base_tax_amount / self.transaction_exchange_rate, "cost_center": self.cost_center, "post_net_value": True, }, account_currency, item=d, ) ) def add_deductions_gl_entries(self, gl_entries): for d in self.get("deductions"): if not d.amount: continue account_currency = get_account_currency(d.account) if account_currency != self.company_currency: frappe.throw(_("Currency for {0} must be {1}").format(d.account, self.company_currency)) gl_entries.append( self.get_gl_dict( { "account": d.account, "account_currency": account_currency, "against": self.party or self.paid_from, "debit_in_account_currency": d.amount, "debit_in_transaction_currency": d.amount / self.transaction_exchange_rate, "debit": d.amount, "cost_center": d.cost_center, }, item=d, ) ) def get_party_account_for_taxes(self): if self.payment_type == "Receive": return self.paid_to elif self.payment_type in ("Pay", "Internal Transfer"): return self.paid_from def get_value_in_transaction_currency(self, account_currency, gl_dict, field): company_currency = erpnext.get_company_currency(self.company) conversion_rate = self.target_exchange_rate if self.paid_from_account_currency != company_currency: conversion_rate = self.source_exchange_rate return flt(gl_dict.get(field, 0) / (conversion_rate or 1)) def on_recurring(self, reference_doc, auto_repeat_doc): self.reference_no = reference_doc.name self.reference_date = nowdate() def calculate_deductions(self, tax_details): return { "account": tax_details["tax"]["account_head"], "cost_center": frappe.get_cached_value("Company", self.company, "cost_center"), "amount": self.total_allocated_amount * (tax_details["tax"]["rate"] / 100), } def set_gain_or_loss(self, account_details=None): if not self.difference_amount: self.set_difference_amount() row = {"amount": self.difference_amount} if account_details: row.update(account_details) if not row.get("amount"): # if no difference amount return self.append("deductions", row) self.set_unallocated_amount() def get_exchange_rate(self): return self.source_exchange_rate if self.payment_type == "Receive" else self.target_exchange_rate def initialize_taxes(self): for tax in self.get("taxes"): validate_taxes_and_charges(tax) validate_inclusive_tax(tax, self) tax_fields = ["total", "tax_fraction_for_current_item", "grand_total_fraction_for_current_item"] if tax.charge_type != "Actual": tax_fields.append("tax_amount") for fieldname in tax_fields: tax.set(fieldname, 0.0) self.paid_amount_after_tax = self.base_paid_amount def determine_exclusive_rate(self): if not any(cint(tax.included_in_paid_amount) for tax in self.get("taxes")): return cumulated_tax_fraction = 0 for i, tax in enumerate(self.get("taxes")): tax.tax_fraction_for_current_item = self.get_current_tax_fraction(tax) if i == 0: tax.grand_total_fraction_for_current_item = 1 + tax.tax_fraction_for_current_item else: tax.grand_total_fraction_for_current_item = ( self.get("taxes")[i - 1].grand_total_fraction_for_current_item + tax.tax_fraction_for_current_item ) cumulated_tax_fraction += tax.tax_fraction_for_current_item self.paid_amount_after_tax = flt(self.base_paid_amount / (1 + cumulated_tax_fraction)) def calculate_taxes(self): self.total_taxes_and_charges = 0.0 self.base_total_taxes_and_charges = 0.0 actual_tax_dict = dict( [ [tax.idx, flt(tax.tax_amount, tax.precision("tax_amount"))] for tax in self.get("taxes") if tax.charge_type == "Actual" ] ) for i, tax in enumerate(self.get("taxes")): current_tax_amount = self.get_current_tax_amount(tax) if tax.charge_type == "Actual": actual_tax_dict[tax.idx] -= current_tax_amount if i == len(self.get("taxes")) - 1: current_tax_amount += actual_tax_dict[tax.idx] tax.tax_amount = current_tax_amount tax.base_tax_amount = current_tax_amount if tax.add_deduct_tax == "Deduct": current_tax_amount *= -1.0 else: current_tax_amount *= 1.0 if i == 0: tax.total = flt(self.paid_amount_after_tax + current_tax_amount, self.precision("total", tax)) else: tax.total = flt( self.get("taxes")[i - 1].total + current_tax_amount, self.precision("total", tax) ) tax.base_total = tax.total if self.payment_type == "Pay": if tax.currency != self.paid_to_account_currency: self.total_taxes_and_charges += flt(current_tax_amount / self.target_exchange_rate) else: self.total_taxes_and_charges += current_tax_amount elif self.payment_type == "Receive": if tax.currency != self.paid_from_account_currency: self.total_taxes_and_charges += flt(current_tax_amount / self.source_exchange_rate) else: self.total_taxes_and_charges += current_tax_amount self.base_total_taxes_and_charges += current_tax_amount if self.get("taxes"): self.paid_amount_after_tax = self.get("taxes")[-1].base_total def get_current_tax_amount(self, tax): tax_rate = tax.rate # To set row_id by default as previous row. if tax.charge_type in ["On Previous Row Amount", "On Previous Row Total"]: if tax.idx == 1: frappe.throw( _( "Cannot select charge type as 'On Previous Row Amount' or 'On Previous Row Total' for first row" ) ) if not tax.row_id: tax.row_id = tax.idx - 1 if tax.charge_type == "Actual": current_tax_amount = flt(tax.tax_amount, self.precision("tax_amount", tax)) elif tax.charge_type == "On Paid Amount": current_tax_amount = (tax_rate / 100.0) * self.paid_amount_after_tax elif tax.charge_type == "On Previous Row Amount": current_tax_amount = (tax_rate / 100.0) * self.get("taxes")[cint(tax.row_id) - 1].tax_amount elif tax.charge_type == "On Previous Row Total": current_tax_amount = (tax_rate / 100.0) * self.get("taxes")[cint(tax.row_id) - 1].total return current_tax_amount def get_current_tax_fraction(self, tax): current_tax_fraction = 0 if cint(tax.included_in_paid_amount): tax_rate = tax.rate if tax.charge_type == "On Paid Amount": current_tax_fraction = tax_rate / 100.0 elif tax.charge_type == "On Previous Row Amount": current_tax_fraction = (tax_rate / 100.0) * self.get("taxes")[ cint(tax.row_id) - 1 ].tax_fraction_for_current_item elif tax.charge_type == "On Previous Row Total": current_tax_fraction = (tax_rate / 100.0) * self.get("taxes")[ cint(tax.row_id) - 1 ].grand_total_fraction_for_current_item if getattr(tax, "add_deduct_tax", None) and tax.add_deduct_tax == "Deduct": current_tax_fraction *= -1.0 return current_tax_fraction def set_matched_unset_payment_requests_to_response(self): """ Find matched Payment Requests for those references which have no Payment Request set.\n And set to `frappe.response` to show in the frontend for allocation. """ if not self.references: return matched_payment_requests = get_matched_payment_request_of_references( [row for row in self.references if not row.payment_request] ) if not matched_payment_requests: return frappe.response["matched_payment_requests"] = matched_payment_requests @frappe.whitelist() def allocate_amount_to_references(self, paid_amount, paid_amount_change, allocate_payment_amount): """ Allocate `Allocated Amount` and `Payment Request` against `Reference` based on `Paid Amount` and `Outstanding Amount`.\n :param paid_amount: Paid Amount / Received Amount. :param paid_amount_change: Flag to check if `Paid Amount` is changed or not. :param allocate_payment_amount: Flag to allocate amount or not. (Payment Request is also dependent on this flag) """ if not self.references: return if not allocate_payment_amount: for ref in self.references: ref.allocated_amount = 0 return # calculating outstanding amounts precision = self.precision("paid_amount") total_positive_outstanding_including_order = 0 total_negative_outstanding = 0 paid_amount -= sum(flt(d.amount, precision) for d in self.deductions) for ref in self.references: reference_outstanding_amount = flt(ref.outstanding_amount) abs_outstanding_amount = abs(reference_outstanding_amount) if reference_outstanding_amount > 0: total_positive_outstanding_including_order += abs_outstanding_amount else: total_negative_outstanding += abs_outstanding_amount # calculating allocated outstanding amounts allocated_negative_outstanding = 0 allocated_positive_outstanding = 0 # checking party type and payment type if (self.payment_type == "Receive" and self.party_type == "Customer") or ( self.payment_type == "Pay" and self.party_type in ("Supplier", "Employee") ): if total_positive_outstanding_including_order > paid_amount: remaining_outstanding = flt( total_positive_outstanding_including_order - paid_amount, precision ) allocated_negative_outstanding = min(remaining_outstanding, total_negative_outstanding) allocated_positive_outstanding = paid_amount + allocated_negative_outstanding elif self.party_type in ("Supplier", "Customer"): if paid_amount > total_negative_outstanding: if total_negative_outstanding == 0: frappe.msgprint( _("Cannot {0} from {1} without any negative outstanding invoice").format( self.payment_type, self.party_type, ) ) else: frappe.msgprint( _("Paid Amount cannot be greater than total negative outstanding amount {0}").format( total_negative_outstanding ) ) return else: allocated_positive_outstanding = flt(total_negative_outstanding - paid_amount, precision) allocated_negative_outstanding = paid_amount + min( total_positive_outstanding_including_order, allocated_positive_outstanding ) # inner function to set `allocated_amount` to those row which have no PR def _allocation_to_unset_pr_row( row, outstanding_amount, allocated_positive_outstanding, allocated_negative_outstanding ): if outstanding_amount > 0 and allocated_positive_outstanding >= 0: row.allocated_amount = min(allocated_positive_outstanding, outstanding_amount) allocated_positive_outstanding = flt( allocated_positive_outstanding - row.allocated_amount, precision ) elif outstanding_amount < 0 and allocated_negative_outstanding: row.allocated_amount = min(allocated_negative_outstanding, abs(outstanding_amount)) * -1 allocated_negative_outstanding = flt( allocated_negative_outstanding - abs(row.allocated_amount), precision ) return allocated_positive_outstanding, allocated_negative_outstanding # allocate amount based on `paid_amount` is changed or not if not paid_amount_change: for ref in self.references: allocated_positive_outstanding, allocated_negative_outstanding = _allocation_to_unset_pr_row( ref, ref.outstanding_amount, allocated_positive_outstanding, allocated_negative_outstanding, ) allocate_open_payment_requests_to_references(self.references, self.precision("paid_amount")) else: payment_request_outstanding_amounts = ( get_payment_request_outstanding_set_in_references(self.references) or {} ) references_outstanding_amounts = get_references_outstanding_amount(self.references) or {} remaining_references_allocated_amounts = references_outstanding_amounts.copy() # Re allocate amount to those references which have PR set (Higher priority) for ref in self.references: if not (ref.reference_doctype and ref.reference_name and ref.payment_request): continue # fetch outstanding_amount of `Reference` (Payment Term) and `Payment Request` to allocate new amount key = (ref.reference_doctype, ref.reference_name, ref.get("payment_term")) reference_outstanding_amount = references_outstanding_amounts[key] pr_outstanding_amount = payment_request_outstanding_amounts[ref.payment_request] if reference_outstanding_amount > 0 and allocated_positive_outstanding >= 0: # allocate amount according to outstanding amounts outstanding_amounts = ( allocated_positive_outstanding, reference_outstanding_amount, pr_outstanding_amount, ) ref.allocated_amount = min(outstanding_amounts) # update amounts to track allocation allocated_amount = ref.allocated_amount allocated_positive_outstanding = flt( allocated_positive_outstanding - allocated_amount, precision ) remaining_references_allocated_amounts[key] = flt( remaining_references_allocated_amounts[key] - allocated_amount, precision ) payment_request_outstanding_amounts[ref.payment_request] = flt( payment_request_outstanding_amounts[ref.payment_request] - allocated_amount, precision ) elif reference_outstanding_amount < 0 and allocated_negative_outstanding: # allocate amount according to outstanding amounts outstanding_amounts = ( allocated_negative_outstanding, abs(reference_outstanding_amount), pr_outstanding_amount, ) ref.allocated_amount = min(outstanding_amounts) * -1 # update amounts to track allocation allocated_amount = abs(ref.allocated_amount) allocated_negative_outstanding = flt( allocated_negative_outstanding - allocated_amount, precision ) remaining_references_allocated_amounts[key] += allocated_amount # negative amount payment_request_outstanding_amounts[ref.payment_request] = flt( payment_request_outstanding_amounts[ref.payment_request] - allocated_amount, precision ) # Re allocate amount to those references which have no PR (Lower priority) for ref in self.references: if ref.payment_request or not (ref.reference_doctype and ref.reference_name): continue key = (ref.reference_doctype, ref.reference_name, ref.get("payment_term")) reference_outstanding_amount = remaining_references_allocated_amounts[key] allocated_positive_outstanding, allocated_negative_outstanding = _allocation_to_unset_pr_row( ref, reference_outstanding_amount, allocated_positive_outstanding, allocated_negative_outstanding, ) @frappe.whitelist() def set_matched_payment_requests(self, matched_payment_requests): """ Set `Payment Request` against `Reference` based on `matched_payment_requests`.\n :param matched_payment_requests: List of tuple of matched Payment Requests. --- Example: [(reference_doctype, reference_name, allocated_amount, payment_request), ...] """ if not self.references or not matched_payment_requests: return if isinstance(matched_payment_requests, str): matched_payment_requests = json.loads(matched_payment_requests) # modify matched_payment_requests # like (reference_doctype, reference_name, allocated_amount): payment_request payment_requests = {} for row in matched_payment_requests: key = tuple(row[:3]) payment_requests[key] = row[3] for ref in self.references: if ref.payment_request: continue key = (ref.reference_doctype, ref.reference_name, ref.allocated_amount) if key in payment_requests: ref.payment_request = payment_requests[key] del payment_requests[key] # to avoid duplicate allocation def get_matched_payment_request_of_references(references=None): """ Get those `Payment Requests` which are matched with `References`.\n - Amount must be same. - Only single `Payment Request` available for this amount. Example: [(reference_doctype, reference_name, allocated_amount, payment_request), ...] """ if not references: return # to fetch matched rows refs = { (row.reference_doctype, row.reference_name, row.allocated_amount) for row in references if row.reference_doctype and row.reference_name and row.allocated_amount } if not refs: return PR = frappe.qb.DocType("Payment Request") # query to group by reference_doctype, reference_name, outstanding_amount subquery = ( frappe.qb.from_(PR) .select( PR.reference_doctype, PR.reference_name, PR.outstanding_amount.as_("allocated_amount"), PR.name.as_("payment_request"), Count("*").as_("count"), ) .where(Tuple(PR.reference_doctype, PR.reference_name, PR.outstanding_amount).isin(refs)) .where(PR.status != "Paid") .where(PR.docstatus == 1) .groupby(PR.reference_doctype, PR.reference_name, PR.outstanding_amount) ) # query to fetch matched rows which are single matched_prs = ( frappe.qb.from_(subquery) .select( subquery.reference_doctype, subquery.reference_name, subquery.allocated_amount, subquery.payment_request, ) .where(subquery.count == 1) .run() ) return matched_prs if matched_prs else None def get_references_outstanding_amount(references=None): """ Fetch accurate outstanding amount of `References`.\n - If `Payment Term` is set, then fetch outstanding amount from `Payment Schedule`. - If `Payment Term` is not set, then fetch outstanding amount from `References` it self. Example: {(reference_doctype, reference_name, payment_term): outstanding_amount, ...} """ if not references: return refs_with_payment_term = get_outstanding_of_references_with_payment_term(references) or {} refs_without_payment_term = get_outstanding_of_references_with_no_payment_term(references) or {} return {**refs_with_payment_term, **refs_without_payment_term} def get_outstanding_of_references_with_payment_term(references=None): """ Fetch outstanding amount of `References` which have `Payment Term` set.\n Example: {(reference_doctype, reference_name, payment_term): outstanding_amount, ...} """ if not references: return refs = { (row.reference_doctype, row.reference_name, row.payment_term) for row in references if row.reference_doctype and row.reference_name and row.payment_term } if not refs: return PS = frappe.qb.DocType("Payment Schedule") response = ( frappe.qb.from_(PS) .select(PS.parenttype, PS.parent, PS.payment_term, PS.outstanding) .where(Tuple(PS.parenttype, PS.parent, PS.payment_term).isin(refs)) ).run(as_dict=True) if not response: return return {(row.parenttype, row.parent, row.payment_term): row.outstanding for row in response} def get_outstanding_of_references_with_no_payment_term(references): """ Fetch outstanding amount of `References` which have no `Payment Term` set.\n - Fetch outstanding amount from `References` it self. Note: `None` is used for allocation of `Payment Request` Example: {(reference_doctype, reference_name, None): outstanding_amount, ...} """ if not references: return outstanding_amounts = {} for ref in references: if ref.payment_term: continue key = (ref.reference_doctype, ref.reference_name, None) if key not in outstanding_amounts: outstanding_amounts[key] = ref.outstanding_amount return outstanding_amounts def get_payment_request_outstanding_set_in_references(references=None): """ Fetch outstanding amount of `Payment Request` which are set in `References`.\n Example: {payment_request: outstanding_amount, ...} """ if not references: return referenced_payment_requests = {row.payment_request for row in references if row.payment_request} if not referenced_payment_requests: return PR = frappe.qb.DocType("Payment Request") response = ( frappe.qb.from_(PR) .select(PR.name, PR.outstanding_amount) .where(PR.name.isin(referenced_payment_requests)) ).run() return dict(response) if response else None def validate_inclusive_tax(tax, doc): def _on_previous_row_error(row_range): throw( _("To include tax in row {0} in Item rate, taxes in rows {1} must also be included").format( tax.idx, row_range ) ) if cint(getattr(tax, "included_in_paid_amount", None)): if tax.charge_type == "Actual": # inclusive tax cannot be of type Actual throw( _("Charge of type 'Actual' in row {0} cannot be included in Item Rate or Paid Amount").format( tax.idx ) ) elif tax.charge_type == "On Previous Row Amount" and not cint( doc.get("taxes")[cint(tax.row_id) - 1].included_in_paid_amount ): # referred row should also be inclusive _on_previous_row_error(tax.row_id) elif tax.charge_type == "On Previous Row Total" and not all( [cint(t.included_in_paid_amount for t in doc.get("taxes")[: cint(tax.row_id) - 1])] ): # all rows about the referred tax should be inclusive _on_previous_row_error("1 - %d" % (cint(tax.row_id),)) elif tax.get("category") == "Valuation": frappe.throw(_("Valuation type charges can not be marked as Inclusive")) @frappe.whitelist() def get_outstanding_reference_documents(args, validate=False): if isinstance(args, str): args = json.loads(args) if args.get("party_type") == "Member": return if args.get("party_type") and args.get("party"): party_to_check = None if has_ignored_field("Payment Entry", args["party_type"]) else args["party"] frappe.has_permission(args["party_type"], "read", party_to_check, throw=True) if not args.get("get_outstanding_invoices") and not args.get("get_orders_to_be_billed"): args["get_outstanding_invoices"] = True ple = qb.DocType("Payment Ledger Entry") common_filter = [] accounting_dimensions_filter = [] posting_and_due_date = [] if args.get("party_type") == "Supplier" and frappe.get_lazy_doc("Supplier", args["party"]).is_blocked_for( "Payments" ): return [] party_account_currency = get_account_currency(args.get("party_account")) company_currency = frappe.get_cached_value("Company", args.get("company"), "default_currency") # Get positive outstanding sales /purchase invoices condition = "" if args.get("voucher_type") and args.get("voucher_no"): condition = f" and voucher_type={frappe.db.escape(args['voucher_type'])} and voucher_no={frappe.db.escape(args['voucher_no'])}" common_filter.append(ple.voucher_type == args["voucher_type"]) common_filter.append(ple.voucher_no == args["voucher_no"]) # Add cost center condition if args.get("cost_center"): condition += f" and cost_center={frappe.db.escape(args.get('cost_center'))}" accounting_dimensions_filter.append(ple.cost_center == args.get("cost_center")) # dynamic dimension filters active_dimensions = get_dimensions()[0] for dim in active_dimensions: if args.get(dim.fieldname): condition += f" and {dim.fieldname}={frappe.db.escape(args.get(dim.fieldname))}" accounting_dimensions_filter.append(ple[dim.fieldname] == args.get(dim.fieldname)) date_fields_dict = { "posting_date": ["from_posting_date", "to_posting_date"], "due_date": ["from_due_date", "to_due_date"], } for fieldname, date_fields in date_fields_dict.items(): from_date = frappe.db.escape(str(args.get(date_fields[0]))) if args.get(date_fields[0]) else None to_date = frappe.db.escape(str(args.get(date_fields[1]))) if args.get(date_fields[1]) else None if args.get(date_fields[0]) and args.get(date_fields[1]): condition += f" and {fieldname} between {from_date} and {to_date}" posting_and_due_date.append(ple[fieldname][args.get(date_fields[0]) : args.get(date_fields[1])]) elif args.get(date_fields[0]): # if only from date is supplied condition += f" and {fieldname} >= {from_date}" posting_and_due_date.append(ple[fieldname].gte(args.get(date_fields[0]))) elif args.get(date_fields[1]): # if only to date is supplied condition += f" and {fieldname} <= {to_date}" posting_and_due_date.append(ple[fieldname].lte(args.get(date_fields[1]))) if args.get("company"): condition += " and company = {}".format(frappe.db.escape(args.get("company"))) common_filter.append(ple.company == args.get("company")) outstanding_invoices = [] negative_outstanding_invoices = [] party_account = args.get("party_account") # get party account if advance account is set. if args.get("book_advance_payments_in_separate_party_account"): accounts = get_party_account( args.get("party_type"), args.get("party"), args.get("company"), include_advance=True ) advance_account = accounts[1] if len(accounts) > 1 else None if party_account == advance_account: party_account = accounts[0] if args.get("get_outstanding_invoices"): outstanding_invoices = get_outstanding_invoices( args.get("party_type"), args.get("party"), [party_account], common_filter=common_filter, posting_date=posting_and_due_date, min_outstanding=args.get("outstanding_amt_greater_than"), max_outstanding=args.get("outstanding_amt_less_than"), accounting_dimensions=accounting_dimensions_filter, vouchers=args.get("vouchers") or None, ) outstanding_invoices = split_invoices_based_on_payment_terms( outstanding_invoices, args.get("company") ) for d in outstanding_invoices: d["exchange_rate"] = 1 if party_account_currency != company_currency: if d.voucher_type in frappe.get_hooks("invoice_doctypes"): d["exchange_rate"] = frappe.db.get_value(d.voucher_type, d.voucher_no, "conversion_rate") elif d.voucher_type == "Journal Entry": d["exchange_rate"] = get_exchange_rate( party_account_currency, company_currency, d.posting_date ) if d.voucher_type in ("Purchase Invoice"): d["bill_no"] = frappe.db.get_value(d.voucher_type, d.voucher_no, "bill_no") # Get negative outstanding sales /purchase invoices if args.get("party_type") != "Employee": negative_outstanding_invoices = get_negative_outstanding_invoices( args.get("party_type"), args.get("party"), args.get("party_account"), party_account_currency, company_currency, condition=condition, ) # Get all SO / PO which are not fully billed or against which full advance not paid orders_to_be_billed = [] if args.get("get_orders_to_be_billed"): orders_to_be_billed = get_orders_to_be_billed( args.get("posting_date"), args.get("party_type"), args.get("party"), args.get("company"), party_account_currency, company_currency, filters=args, ) orders_to_be_billed = split_invoices_based_on_payment_terms(orders_to_be_billed, args.get("company")) data = negative_outstanding_invoices + outstanding_invoices + orders_to_be_billed if not data: if args.get("get_outstanding_invoices") and args.get("get_orders_to_be_billed"): ref_document_type = "invoices or orders" elif args.get("get_outstanding_invoices"): ref_document_type = "invoices" elif args.get("get_orders_to_be_billed"): ref_document_type = "orders" if not validate: frappe.msgprint( _( "No outstanding {0} found for the {1} {2} which qualify the filters you have specified." ).format( _(ref_document_type), _(args.get("party_type")).lower(), frappe.bold(args.get("party")) ) ) return data def split_invoices_based_on_payment_terms(outstanding_invoices, company) -> list: """Split a list of invoices based on their payment terms.""" exc_rates = get_currency_data(outstanding_invoices, company) outstanding_invoices_after_split = [] for entry in outstanding_invoices: if entry.voucher_type in ["Sales Invoice", "Purchase Invoice", "Sales Order", "Purchase Order"]: if payment_term_template := frappe.db.get_value( entry.voucher_type, entry.voucher_no, "payment_terms_template" ): split_rows = get_split_invoice_rows(entry, payment_term_template, exc_rates) if not split_rows: continue if len(split_rows) > 1: frappe.msgprint( _("Splitting {0} {1} into {2} rows as per Payment Terms").format( _(entry.voucher_type), frappe.bold(entry.voucher_no), len(split_rows) ), alert=True, ) outstanding_invoices_after_split += split_rows continue # If not an invoice or no payment terms template, add as it is outstanding_invoices_after_split.append(entry) return outstanding_invoices_after_split def get_currency_data(outstanding_invoices: list, company: str | None = None) -> dict: """Get currency and conversion data for a list of invoices.""" exc_rates = frappe._dict() company_currency = frappe.db.get_value("Company", company, "default_currency") if company else None for doctype in ["Sales Invoice", "Purchase Invoice", "Sales Order", "Purchase Order"]: invoices = [x.voucher_no for x in outstanding_invoices if x.voucher_type == doctype] for x in frappe.db.get_all( doctype, filters={"name": ["in", invoices]}, fields=["name", "currency", "conversion_rate", "party_account_currency"], ): exc_rates[x.name] = frappe._dict( conversion_rate=x.conversion_rate, currency=x.currency, party_account_currency=x.party_account_currency, company_currency=company_currency, ) return exc_rates def get_split_invoice_rows(invoice: dict, payment_term_template: str, exc_rates: dict) -> list: """Split invoice based on its payment schedule table.""" split_rows = [] allocate_payment_based_on_payment_terms = frappe.db.get_value( "Payment Terms Template", payment_term_template, "allocate_payment_based_on_payment_terms" ) if not allocate_payment_based_on_payment_terms: return [invoice] payment_schedule = frappe.get_all( "Payment Schedule", filters={"parent": invoice.voucher_no}, fields=["*"], order_by="due_date" ) for payment_term in payment_schedule: if not payment_term.outstanding > 0.1: continue doc_details = exc_rates.get(payment_term.parent, None) is_multi_currency_acc = (doc_details.currency != doc_details.company_currency) and ( doc_details.party_account_currency != doc_details.company_currency ) payment_term_outstanding = flt(payment_term.outstanding) if not is_multi_currency_acc: payment_term_outstanding = doc_details.conversion_rate * flt(payment_term.outstanding) split_rows.append( frappe._dict( { "due_date": invoice.due_date, "currency": invoice.currency, "voucher_no": invoice.voucher_no, "voucher_type": invoice.voucher_type, "posting_date": invoice.posting_date, "invoice_amount": flt(invoice.invoice_amount), "outstanding_amount": payment_term_outstanding if payment_term_outstanding else invoice.outstanding_amount, "payment_term_outstanding": payment_term_outstanding, "payment_amount": payment_term.payment_amount, "payment_term": payment_term.payment_term, } ) ) return split_rows def get_orders_to_be_billed( posting_date, party_type, party, company, party_account_currency, company_currency, cost_center=None, filters=None, ): voucher_type = None if party_type == "Customer": voucher_type = "Sales Order" elif party_type == "Supplier": voucher_type = "Purchase Order" if not voucher_type: return [] # dynamic dimension filters condition = "" active_dimensions = get_dimensions(True)[0] for dim in active_dimensions: if filters.get(dim.fieldname): condition += f" and {dim.fieldname}={frappe.db.escape(filters.get(dim.fieldname))}" if party_account_currency == company_currency: grand_total_field = "base_grand_total" rounded_total_field = "base_rounded_total" else: grand_total_field = "grand_total" rounded_total_field = "rounded_total" orders = frappe.db.sql( """ select name as voucher_no, if({rounded_total_field}, {rounded_total_field}, {grand_total_field}) as invoice_amount, (if({rounded_total_field}, {rounded_total_field}, {grand_total_field}) - advance_paid) as outstanding_amount, transaction_date as posting_date from `tab{voucher_type}` where {party_type} = %s and docstatus = 1 and company = %s and status != "Closed" and if({rounded_total_field}, {rounded_total_field}, {grand_total_field}) > advance_paid and abs(100 - per_billed) > 0.01 {condition} order by transaction_date, name """.format( **{ "rounded_total_field": rounded_total_field, "grand_total_field": grand_total_field, "voucher_type": voucher_type, "party_type": scrub(party_type), "condition": condition, } ), (party, company), as_dict=True, ) order_list = [] for d in orders: if ( filters and filters.get("outstanding_amt_greater_than") and filters.get("outstanding_amt_less_than") and not ( flt(filters.get("outstanding_amt_greater_than")) <= flt(d.outstanding_amount) <= flt(filters.get("outstanding_amt_less_than")) ) ): continue d["voucher_type"] = voucher_type # This assumes that the exchange rate required is the one in the SO d["exchange_rate"] = get_exchange_rate(party_account_currency, company_currency, posting_date) order_list.append(d) return order_list def get_negative_outstanding_invoices( party_type, party, party_account, party_account_currency, company_currency, cost_center=None, condition=None, ): if party_type not in ["Customer", "Supplier"]: return [] voucher_type = "Sales Invoice" if party_type == "Customer" else "Purchase Invoice" account = "debit_to" if voucher_type == "Sales Invoice" else "credit_to" supplier_condition = "" if voucher_type == "Purchase Invoice": supplier_condition = "and (release_date is null or release_date <= CURRENT_DATE)" if party_account_currency == company_currency: grand_total_field = "base_grand_total" rounded_total_field = "base_rounded_total" else: grand_total_field = "grand_total" rounded_total_field = "rounded_total" return frappe.db.sql( """ select "{voucher_type}" as voucher_type, name as voucher_no, {account} as account, if({rounded_total_field}, {rounded_total_field}, {grand_total_field}) as invoice_amount, outstanding_amount, posting_date, due_date, conversion_rate as exchange_rate from `tab{voucher_type}` where {party_type} = %s and {party_account} = %s and docstatus = 1 and outstanding_amount < 0 {supplier_condition} {condition} order by posting_date, name """.format( **{ "supplier_condition": supplier_condition, "condition": condition, "rounded_total_field": rounded_total_field, "grand_total_field": grand_total_field, "voucher_type": voucher_type, "party_type": scrub(party_type), "party_account": "debit_to" if party_type == "Customer" else "credit_to", "cost_center": cost_center, "account": account, } ), (party, party_account), as_dict=True, ) @frappe.whitelist() def get_party_details(company, party_type, party, date, cost_center=None): bank_account = "" party_bank_account = "" if not frappe.db.exists(party_type, party): frappe.throw(_("{0} {1} does not exist").format(_(party_type), party)) ptype = "select" if frappe.only_has_select_perm(party_type) else "read" party_to_check = None if has_ignored_field("Payment Entry", party_type) else party frappe.has_permission(party_type, ptype, party_to_check, throw=True) party_account = get_party_account(party_type, party, company) account_currency = get_account_currency(party_account) account_balance = ( get_balance_on(party_account, date, cost_center=cost_center) if frappe.get_single_value("Accounts Settings", "show_account_balance") else 0 ) _party_name = "title" if party_type == "Shareholder" else party_type.lower() + "_name" party_name = frappe.db.get_value(party_type, party, _party_name) party_balance = ( get_balance_on(party_type=party_type, party=party, company=company, cost_center=cost_center) if frappe.get_single_value("Accounts Settings", "show_party_balance") else 0 ) if party_type in ["Customer", "Supplier"]: party_bank_account = get_party_bank_account(party_type, party) bank_account = get_default_company_bank_account(company, party_type, party, ignore_permissions=False) return { "party_account": party_account, "party_name": party_name, "party_account_currency": account_currency, "party_balance": party_balance, "account_balance": account_balance, "party_bank_account": party_bank_account, "bank_account": bank_account, } @frappe.whitelist() def get_account_details(account, date, cost_center=None): frappe.has_permission("Payment Entry", throw=True) # to check if the passed account is accessible under reference doctype Payment Entry account_list = frappe.get_list("Account", {"name": account}, reference_doctype="Payment Entry", limit=1) # There might be some user permissions which will allow account under certain doctypes # except for Payment Entry, only in such case we should throw permission error if not account_list: frappe.throw(_("Account: {0} is not permitted under Payment Entry").format(account)) account_balance = ( get_balance_on(account, date, cost_center=cost_center, ignore_account_permission=True) if frappe.get_single_value("Accounts Settings", "show_account_balance") else 0 ) return frappe._dict( { "account_currency": get_account_currency(account), "account_balance": account_balance, "account_type": frappe.get_cached_value("Account", account, "account_type"), } ) @frappe.whitelist() def get_company_defaults(company): frappe.has_permission("Company", doc=company, throw=True) fields = ["write_off_account", "exchange_gain_loss_account", "cost_center"] return frappe.get_cached_value("Company", company, fields, as_dict=1) def get_outstanding_on_journal_entry(voucher_no, party_type, party): ple = frappe.qb.DocType("Payment Ledger Entry") outstanding = ( frappe.qb.from_(ple) .select(Sum(ple.amount_in_account_currency)) .where( (ple.against_voucher_no == voucher_no) & (ple.party_type == party_type) & (ple.party == party) & (ple.delinked == 0) ) ).run() outstanding_amount = outstanding[0][0] if outstanding else 0 total = ( frappe.qb.from_(ple) .select(Sum(ple.amount_in_account_currency)) .where( (ple.voucher_no == voucher_no) & (ple.party_type == party_type) & (ple.party == party) & (ple.delinked == 0) ) ).run() total_amount = total[0][0] if total else 0 return outstanding_amount, total_amount @frappe.whitelist() def get_reference_details( reference_doctype, reference_name, party_account_currency, party_type=None, party=None ): total_amount = outstanding_amount = exchange_rate = account = None frappe.has_permission(reference_doctype, "read", reference_name, throw=True) ref_doc = frappe.get_lazy_doc(reference_doctype, reference_name) company_currency = ref_doc.get("company_currency") or erpnext.get_company_currency(ref_doc.company) # Only applies for Reverse Payment Entries account_type = None payment_type = None if reference_doctype == "Dunning": total_amount = outstanding_amount = ref_doc.get("dunning_amount") exchange_rate = 1 elif reference_doctype == "Journal Entry" and ref_doc.docstatus == 1: if ref_doc.multi_currency: exchange_rate = get_exchange_rate(party_account_currency, company_currency, ref_doc.posting_date) else: exchange_rate = 1 outstanding_amount, total_amount = get_outstanding_on_journal_entry(reference_name, party_type, party) elif reference_doctype == "Payment Entry": if reverse_payment_details := frappe.db.get_all( "Payment Entry", filters={"name": reference_name}, fields=["payment_type", "party_type"], )[0]: payment_type = reverse_payment_details.payment_type account_type = frappe.db.get_value( "Party Type", reverse_payment_details.party_type, "account_type" ) exchange_rate = 1 elif reference_doctype != "Journal Entry": if not total_amount: if party_account_currency == company_currency: # for handling cases that don't have multi-currency (base field) total_amount = ( ref_doc.get("base_rounded_total") or ref_doc.get("rounded_total") or ref_doc.get("base_grand_total") or ref_doc.get("grand_total") ) exchange_rate = 1 else: total_amount = ref_doc.get("rounded_total") or ref_doc.get("grand_total") if not exchange_rate: # Get the exchange rate from the original ref doc # or get it based on the posting date of the ref doc. exchange_rate = ref_doc.get("conversion_rate") or get_exchange_rate( party_account_currency, company_currency, ref_doc.posting_date ) if reference_doctype in ("Sales Invoice", "Purchase Invoice"): outstanding_amount = ref_doc.get("outstanding_amount") account = ( ref_doc.get("debit_to") if reference_doctype == "Sales Invoice" else ref_doc.get("credit_to") ) else: outstanding_amount = flt(total_amount) - flt(ref_doc.get("advance_paid")) if reference_doctype in ["Sales Order", "Purchase Order"]: party_type = "Customer" if reference_doctype == "Sales Order" else "Supplier" party_field = "customer" if reference_doctype == "Sales Order" else "supplier" party = ref_doc.get(party_field) account = get_party_account(party_type, party, ref_doc.company) else: # Get the exchange rate based on the posting date of the ref doc. exchange_rate = get_exchange_rate(party_account_currency, company_currency, ref_doc.posting_date) res = frappe._dict( { "due_date": ref_doc.get("due_date"), "total_amount": flt(total_amount), "outstanding_amount": flt(outstanding_amount), "exchange_rate": flt(exchange_rate), "bill_no": ref_doc.get("bill_no"), "account_type": account_type, "payment_type": payment_type, } ) if account: res.update({"account": account}) return res @frappe.whitelist() def get_payment_entry( dt: str, dn: str, party_amount: int | float | None = None, bank_account: str | None = None, bank_amount: int | float | None = None, party_type: str | None = None, payment_type: str | None = None, reference_date: str | date | None = None, created_from_payment_request: bool | None = False, ): frappe.has_permission("Payment Entry", ptype="create", throw=True) doc = frappe.get_doc(dt, dn) doc.check_permission() over_billing_allowance = frappe.db.get_single_value("Accounts Settings", "over_billing_allowance") if dt in ("Sales Order", "Purchase Order") and flt(doc.per_billed, 2) >= (100.0 + over_billing_allowance): frappe.throw(_("Can only make payment against unbilled {0}").format(_(dt))) if not party_type: party_type = set_party_type(dt) party_account = set_party_account(dt, dn, doc, party_type) party_account_currency = set_party_account_currency(dt, party_account, doc) if not payment_type: payment_type = set_payment_type(dt, doc) grand_total, outstanding_amount = set_grand_total_and_outstanding_amount( party_amount, dt, party_account_currency, doc ) # bank or cash bank = get_bank_cash_account(doc, bank_account) # if default bank or cash account is not set in company master and party has default company bank account, fetch it if party_type in ["Customer", "Supplier"] and not bank: party_bank_account = get_party_bank_account(party_type, doc.get(scrub(party_type))) if party_bank_account: account = frappe.db.get_value("Bank Account", party_bank_account, "account") bank = get_bank_cash_account(doc, account) paid_amount, received_amount = set_paid_amount_and_received_amount( dt, party_account_currency, bank, outstanding_amount, payment_type, bank_amount, doc ) reference_date = getdate(reference_date) paid_amount, received_amount, discount_amount, valid_discounts = apply_early_payment_discount( paid_amount, received_amount, doc, party_account_currency, reference_date ) pe = frappe.new_doc("Payment Entry") pe.payment_type = payment_type pe.company = doc.company pe.cost_center = doc.get("cost_center") pe.posting_date = nowdate() pe.reference_date = reference_date pe.mode_of_payment = doc.get("mode_of_payment") pe.party_type = party_type pe.party = doc.get(scrub(party_type)) pe.contact_person = doc.get("contact_person") complete_contact_details(pe) pe.ensure_supplier_is_not_blocked() pe.paid_from = party_account if payment_type == "Receive" else bank.account pe.paid_to = party_account if payment_type == "Pay" else bank.account pe.paid_from_account_currency = ( party_account_currency if payment_type == "Receive" else bank.account_currency ) pe.paid_to_account_currency = party_account_currency if payment_type == "Pay" else bank.account_currency pe.paid_from_account_type = frappe.db.get_value("Account", pe.paid_from, "account_type") pe.paid_to_account_type = frappe.db.get_value("Account", pe.paid_to, "account_type") pe.paid_amount = paid_amount pe.received_amount = received_amount pe.letter_head = doc.get("letter_head") pe.bank_account = frappe.db.get_value( "Bank Account", {"is_company_account": 1, "is_default": 1, "company": doc.company}, "name" ) if dt in ["Purchase Order", "Sales Order", "Sales Invoice", "Purchase Invoice"]: pe.project = doc.get("project") or reduce( lambda prev, cur: prev or cur, [x.get("project") for x in doc.get("items")], None ) # get first non-empty project from items if pe.party_type in ["Customer", "Supplier"]: bank_account = get_party_bank_account(pe.party_type, pe.party) pe.set("party_bank_account", bank_account) pe.set_bank_account_data() # only Purchase Invoice can be blocked individually if doc.doctype == "Purchase Invoice" and doc.invoice_is_blocked(): frappe.msgprint(_("{0} is on hold till {1}").format(doc.name, doc.release_date)) else: if doc.doctype in ( "Sales Invoice", "Purchase Invoice", "Purchase Order", "Sales Order", ) and frappe.get_cached_value( "Payment Terms Template", doc.payment_terms_template, "allocate_payment_based_on_payment_terms", ): for reference in get_reference_as_per_payment_terms( doc.payment_schedule, dt, dn, doc, grand_total, outstanding_amount, party_account_currency ): pe.append("references", reference) else: if dt == "Dunning": for overdue_payment, outstanding in doc.get_unpaid_overdue_payments(): pe.append( "references", { "reference_doctype": "Sales Invoice", "reference_name": overdue_payment.sales_invoice, "payment_term": overdue_payment.payment_term, "due_date": overdue_payment.due_date, "total_amount": outstanding, "outstanding_amount": outstanding, "allocated_amount": outstanding, }, ) if (unpaid_dunning_amount := doc.get_unpaid_base_dunning_amount()) > 0: pe.append( "deductions", { "account": doc.income_account, "cost_center": doc.cost_center, "amount": -1 * unpaid_dunning_amount, "description": _("Interest and/or dunning fee"), "dunning": doc.name, }, ) else: pe.append( "references", { "reference_doctype": dt, "reference_name": dn, "bill_no": doc.get("bill_no"), "due_date": doc.get("due_date"), "total_amount": grand_total, "outstanding_amount": outstanding_amount, "allocated_amount": outstanding_amount, }, ) pe.setup_party_account_field() pe.set_missing_values() pe.set_missing_ref_details() update_accounting_dimensions(pe, doc) if party_account and bank: if discount_amount: base_total_discount_loss = 0 if frappe.db.get_single_value("Accounts Settings", "book_tax_discount_loss"): base_total_discount_loss = split_early_payment_discount_loss(pe, doc, valid_discounts) set_pending_discount_loss( pe, doc, discount_amount, base_total_discount_loss, party_account_currency ) pe.set_exchange_rate() pe.set_amounts() # If PE is created from PR directly, then no need to find open PRs for the references if not created_from_payment_request: allocate_open_payment_requests_to_references(pe.references, pe.precision("paid_amount")) return pe def get_open_payment_requests_for_references(references=None): """ Fetch all unpaid Payment Requests for the references. \n - Each reference can have multiple Payment Requests. \n Example: {("Sales Invoice", "SINV-00001"): {"PREQ-00001": 1000, "PREQ-00002": 2000}} """ if not references: return refs = { (row.reference_doctype, row.reference_name) for row in references if row.reference_doctype and row.reference_name and row.allocated_amount } if not refs: return PR = frappe.qb.DocType("Payment Request") response = ( frappe.qb.from_(PR) .select(PR.name, PR.reference_doctype, PR.reference_name, PR.outstanding_amount) .where(Tuple(PR.reference_doctype, PR.reference_name).isin(list(refs))) .where(PR.status != "Paid") .where(PR.docstatus == 1) .where(PR.outstanding_amount > 0) # to avoid old PRs with 0 outstanding amount .orderby(Coalesce(PR.transaction_date, PR.creation), order=frappe.qb.asc) ).run(as_dict=True) if not response: return reference_payment_requests = {} for row in response: key = (row.reference_doctype, row.reference_name) if key not in reference_payment_requests: reference_payment_requests[key] = {row.name: row.outstanding_amount} else: reference_payment_requests[key][row.name] = row.outstanding_amount return reference_payment_requests def allocate_open_payment_requests_to_references(references=None, precision=None): """ Allocate unpaid Payment Requests to the references. \n --- - Allocation based on below factors - Reference Allocated Amount - Reference Outstanding Amount (With Payment Terms or without Payment Terms) - Reference Payment Request's outstanding amount --- - Allocation based on below scenarios - Reference's Allocated Amount == Payment Request's Outstanding Amount - Allocate the Payment Request to the reference - This PR will not be allocated further - Reference's Allocated Amount < Payment Request's Outstanding Amount - Allocate the Payment Request to the reference - Reduce the PR's outstanding amount by the allocated amount - This PR can be allocated further - Reference's Allocated Amount > Payment Request's Outstanding Amount - Allocate the Payment Request to the reference - Reduce Allocated Amount of the reference by the PR's outstanding amount - Create a new row for the remaining amount until the Allocated Amount is 0 - Allocate PR if available --- - Note: - Priority is given to the first Payment Request of respective references. - Single Reference can have multiple rows. - With Payment Terms or without Payment Terms - With Payment Request or without Payment Request """ if not references: return # get all unpaid payment requests for the references references_open_payment_requests = get_open_payment_requests_for_references(references) if not references_open_payment_requests: return if not precision: precision = references[0].precision("allocated_amount") # to manage new rows row_number = 1 MOVE_TO_NEXT_ROW = 1 TO_SKIP_NEW_ROW = 2 while row_number <= len(references): row = references[row_number - 1] reference_key = (row.reference_doctype, row.reference_name) # update the idx to maintain the order row.idx = row_number # unpaid payment requests for the reference reference_payment_requests = references_open_payment_requests.get(reference_key) if not reference_payment_requests: row_number += MOVE_TO_NEXT_ROW # to move to next reference row continue # get the first payment request and its outstanding amount payment_request, pr_outstanding_amount = next(iter(reference_payment_requests.items())) allocated_amount = row.allocated_amount # allocate the payment request to the reference and PR's outstanding amount row.payment_request = payment_request if pr_outstanding_amount == allocated_amount: del reference_payment_requests[payment_request] row_number += MOVE_TO_NEXT_ROW elif pr_outstanding_amount > allocated_amount: # reduce the outstanding amount of the payment request reference_payment_requests[payment_request] -= allocated_amount row_number += MOVE_TO_NEXT_ROW else: # split the reference row to allocate the remaining amount del reference_payment_requests[payment_request] row.allocated_amount = pr_outstanding_amount allocated_amount = flt(allocated_amount - pr_outstanding_amount, precision) # set the remaining amount to the next row while allocated_amount: # create a new row for the remaining amount new_row = frappe.copy_doc(row) references.insert(row_number, new_row) # get the first payment request and its outstanding amount payment_request, pr_outstanding_amount = next( iter(reference_payment_requests.items()), (None, None) ) # update new row new_row.idx = row_number + 1 new_row.payment_request = payment_request new_row.allocated_amount = min( pr_outstanding_amount if pr_outstanding_amount else allocated_amount, allocated_amount ) if not payment_request or not pr_outstanding_amount: row_number += TO_SKIP_NEW_ROW break elif pr_outstanding_amount == allocated_amount: del reference_payment_requests[payment_request] row_number += TO_SKIP_NEW_ROW break elif pr_outstanding_amount > allocated_amount: reference_payment_requests[payment_request] -= allocated_amount row_number += TO_SKIP_NEW_ROW break else: allocated_amount = flt(allocated_amount - pr_outstanding_amount, precision) del reference_payment_requests[payment_request] row_number += MOVE_TO_NEXT_ROW def update_accounting_dimensions(pe, doc): """ Updates accounting dimensions in Payment Entry based on the accounting dimensions in the reference document """ from erpnext.accounts.doctype.accounting_dimension.accounting_dimension import ( get_accounting_dimensions, ) for dimension in get_accounting_dimensions(): pe.set(dimension, doc.get(dimension)) def get_bank_cash_account(doc, bank_account): bank = get_default_bank_cash_account( doc.company, "Bank", mode_of_payment=doc.get("mode_of_payment"), account=bank_account ) if not bank: bank = get_default_bank_cash_account( doc.company, "Cash", mode_of_payment=doc.get("mode_of_payment"), account=bank_account ) return bank def set_party_type(dt): if dt in ("Sales Invoice", "Sales Order", "Dunning"): party_type = "Customer" elif dt in ("Purchase Invoice", "Purchase Order"): party_type = "Supplier" return party_type def set_party_account(dt, dn, doc, party_type): if dt == "Sales Invoice": party_account = get_party_account_based_on_invoice_discounting(dn) or doc.debit_to elif dt == "Purchase Invoice": party_account = doc.credit_to else: party_account = get_party_account(party_type, doc.get(party_type.lower()), doc.company) return party_account def set_party_account_currency(dt, party_account, doc): if dt not in ("Sales Invoice", "Purchase Invoice"): party_account_currency = get_account_currency(party_account) else: party_account_currency = doc.get("party_account_currency") or get_account_currency(party_account) return party_account_currency def set_payment_type(dt, doc): if ( (dt == "Sales Order" or (dt == "Sales Invoice" and doc.outstanding_amount > 0)) or (dt == "Purchase Invoice" and doc.outstanding_amount < 0) or dt == "Dunning" ): payment_type = "Receive" else: payment_type = "Pay" return payment_type def set_grand_total_and_outstanding_amount(party_amount, dt, party_account_currency, doc): grand_total = outstanding_amount = 0 if party_amount: grand_total = outstanding_amount = party_amount elif dt in ("Sales Invoice", "Purchase Invoice"): if party_account_currency == doc.company_currency: grand_total = doc.base_rounded_total or doc.base_grand_total else: grand_total = doc.rounded_total or doc.grand_total outstanding_amount = doc.outstanding_amount elif dt == "Dunning": # only what is left to collect, the totals on the dunning are the ones it was raised with grand_total = sum(outstanding for _row, outstanding in doc.get_unpaid_overdue_payments()) grand_total += doc.get_unpaid_dunning_amount() outstanding_amount = grand_total else: if party_account_currency == doc.company_currency: grand_total = flt(doc.get("base_rounded_total") or doc.get("base_grand_total")) else: grand_total = flt(doc.get("rounded_total") or doc.get("grand_total")) outstanding_amount = doc.get("outstanding_amount") or (grand_total - flt(doc.advance_paid)) return grand_total, outstanding_amount def set_paid_amount_and_received_amount( dt, party_account_currency, bank, outstanding_amount, payment_type, bank_amount, doc ): paid_amount = received_amount = 0 if party_account_currency == bank.account_currency: paid_amount = received_amount = abs(outstanding_amount) else: # settings if it is for receive paid_amount = abs(outstanding_amount) if bank_amount: received_amount = bank_amount else: company_currency = frappe.get_cached_value("Company", doc.get("company"), "default_currency") if bank and company_currency != bank.account_currency: # doc currency can be different from bank currency conversion_rate = get_exchange_rate(bank.account_currency, party_account_currency) received_amount = paid_amount / conversion_rate else: conversion_rate = get_exchange_rate(doc.get("currency", company_currency), company_currency) received_amount = paid_amount * conversion_rate # if payment type is pay, then paid amount and received amount are swapped if payment_type == "Pay": paid_amount, received_amount = received_amount, paid_amount return paid_amount, received_amount def apply_early_payment_discount(paid_amount, received_amount, doc, party_account_currency, reference_date): total_discount = 0 valid_discounts = [] eligible_for_payments = ["Sales Order", "Sales Invoice", "Purchase Order", "Purchase Invoice"] has_payment_schedule = hasattr(doc, "payment_schedule") and doc.payment_schedule is_multi_currency = party_account_currency != doc.company_currency if doc.doctype in eligible_for_payments and has_payment_schedule: for term in doc.payment_schedule: if not term.discounted_amount and term.discount and reference_date <= term.discount_date: if term.discount_type == "Percentage": grand_total = doc.get("grand_total") if is_multi_currency else doc.get("base_grand_total") discount_amount = flt(grand_total) * (term.discount / 100) else: discount_amount = term.discount # if accounting is done in the same currency, paid_amount = received_amount conversion_rate = doc.get("conversion_rate", 1) if is_multi_currency else 1 discount_amount_in_foreign_currency = discount_amount * conversion_rate if doc.doctype == "Sales Invoice": paid_amount -= discount_amount received_amount -= discount_amount_in_foreign_currency else: received_amount -= discount_amount paid_amount -= discount_amount_in_foreign_currency valid_discounts.append({"type": term.discount_type, "discount": term.discount}) total_discount += discount_amount if total_discount: currency = doc.get("currency") if is_multi_currency else doc.company_currency money = frappe.utils.fmt_money(total_discount, currency=currency) frappe.msgprint(_("Discount of {} applied as per Payment Term").format(money), alert=1) return paid_amount, received_amount, total_discount, valid_discounts def set_pending_discount_loss(pe, doc, discount_amount, base_total_discount_loss, party_account_currency): # If multi-currency, get base discount amount to adjust with base currency deductions/losses if party_account_currency != doc.company_currency: discount_amount = discount_amount * doc.get("conversion_rate", 1) # Avoid considering miniscule losses discount_amount = flt(discount_amount - base_total_discount_loss, doc.precision("grand_total")) # Set base discount amount (discount loss/pending rounding loss) in deductions if discount_amount > 0.0: positive_negative = -1 if pe.payment_type == "Pay" else 1 # If tax loss booking is enabled, pending loss will be rounding loss. # Otherwise it will be the total discount loss. book_tax_loss = frappe.db.get_single_value("Accounts Settings", "book_tax_discount_loss") account_type = "round_off_account" if book_tax_loss else "default_discount_account" pe.append( "deductions", { "account": frappe.get_cached_value("Company", pe.company, account_type), "cost_center": pe.cost_center or frappe.get_cached_value("Company", pe.company, "cost_center"), "amount": discount_amount * positive_negative, }, ) def split_early_payment_discount_loss(pe, doc, valid_discounts) -> float: """Split early payment discount into Income Loss & Tax Loss.""" total_discount_percent = get_total_discount_percent(doc, valid_discounts) if not total_discount_percent: return 0.0 base_loss_on_income = add_income_discount_loss(pe, doc, total_discount_percent) base_loss_on_taxes = add_tax_discount_loss(pe, doc, total_discount_percent) # Round off total loss rather than individual losses to reduce rounding error return flt(base_loss_on_income + base_loss_on_taxes, doc.precision("grand_total")) def get_total_discount_percent(doc, valid_discounts) -> float: """Get total percentage and amount discount applied as a percentage.""" total_discount_percent = ( sum(discount.get("discount") for discount in valid_discounts if discount.get("type") == "Percentage") or 0.0 ) # Operate in percentages only as it makes the income & tax split easier total_discount_amount = ( sum(discount.get("discount") for discount in valid_discounts if discount.get("type") == "Amount") or 0.0 ) if total_discount_amount: discount_percentage = (total_discount_amount / doc.get("grand_total")) * 100 total_discount_percent += discount_percentage return total_discount_percent return total_discount_percent def add_income_discount_loss(pe, doc, total_discount_percent) -> float: """Add loss on income discount in base currency.""" precision = doc.precision("total") base_loss_on_income = doc.get("base_total") * (total_discount_percent / 100) positive_negative = -1 if pe.payment_type == "Pay" else 1 pe.append( "deductions", { "account": frappe.get_cached_value("Company", pe.company, "default_discount_account"), "cost_center": pe.cost_center or frappe.get_cached_value("Company", pe.company, "cost_center"), "amount": flt(base_loss_on_income, precision) * positive_negative, }, ) return base_loss_on_income # Return loss without rounding def add_tax_discount_loss(pe, doc, total_discount_percentage) -> float: """Add loss on tax discount in base currency.""" tax_discount_loss = {} base_total_tax_loss = 0 precision = doc.precision("tax_amount_after_discount_amount", "taxes") positive_negative = -1 if pe.payment_type == "Pay" else 1 # The same account head could be used more than once for tax in doc.get("taxes", []): base_tax_loss = tax.get("base_tax_amount_after_discount_amount") * (total_discount_percentage / 100) account = tax.get("account_head") if not tax_discount_loss.get(account): tax_discount_loss[account] = base_tax_loss else: tax_discount_loss[account] += base_tax_loss for account, loss in tax_discount_loss.items(): base_total_tax_loss += loss if loss == 0.0: continue pe.append( "deductions", { "account": account, "cost_center": pe.cost_center or frappe.get_cached_value("Company", pe.company, "cost_center"), "amount": flt(loss, precision) * positive_negative, }, ) return base_total_tax_loss # Return loss without rounding def get_reference_as_per_payment_terms( payment_schedule, dt, dn, doc, grand_total, outstanding_amount, party_account_currency ): references = [] is_multi_currency_acc = (doc.currency != doc.company_currency) and ( party_account_currency != doc.company_currency ) for payment_term in payment_schedule: payment_term_outstanding = flt( payment_term.payment_amount - payment_term.paid_amount, payment_term.precision("payment_amount") ) if not is_multi_currency_acc: # If accounting is done in company currency for multi-currency transaction payment_term_outstanding = flt( payment_term_outstanding * doc.get("conversion_rate"), payment_term.precision("payment_amount"), ) if payment_term_outstanding: references.append( { "reference_doctype": dt, "reference_name": dn, "bill_no": doc.get("bill_no"), "due_date": doc.get("due_date"), "total_amount": grand_total, "outstanding_amount": outstanding_amount, "payment_term_outstanding": payment_term_outstanding, "payment_term": payment_term.payment_term, "allocated_amount": payment_term_outstanding, } ) return references def get_paid_amount(dt, dn, party_type, party, account, due_date): if party_type == "Customer": dr_or_cr = "credit_in_account_currency - debit_in_account_currency" else: dr_or_cr = "debit_in_account_currency - credit_in_account_currency" paid_amount = frappe.db.sql( f""" select ifnull(sum({dr_or_cr}), 0) as paid_amount from `tabGL Entry` where against_voucher_type = %s and against_voucher = %s and party_type = %s and party = %s and account = %s and due_date = %s and {dr_or_cr} > 0 """, (dt, dn, party_type, party, account, due_date), ) return paid_amount[0][0] if paid_amount else 0 @frappe.whitelist() def get_party_and_account_balance( company, date, paid_from=None, paid_to=None, ptype=None, pty=None, cost_center=None ): show_account_balance = frappe.get_single_value("Accounts Settings", "show_account_balance") return frappe._dict( { "party_balance": get_balance_on(party_type=ptype, party=pty, cost_center=cost_center) if frappe.get_single_value("Accounts Settings", "show_party_balance") else 0, "paid_from_account_balance": get_balance_on(paid_from, date, cost_center=cost_center) if show_account_balance else 0, "paid_to_account_balance": get_balance_on(paid_to, date=date, cost_center=cost_center) if show_account_balance else 0, } ) @frappe.whitelist() def make_payment_order(source_name, target_doc=None): from frappe.model.mapper import get_mapped_doc def set_missing_values(source, target): target.payment_order_type = "Payment Entry" target.company = target.company or source.company target.append( "references", dict( reference_doctype="Payment Entry", reference_name=source.name, bank_account=source.party_bank_account, amount=source.paid_amount, account=source.paid_to, supplier=source.party, mode_of_payment=source.mode_of_payment, ), ) doclist = get_mapped_doc( "Payment Entry", source_name, { "Payment Entry": { "doctype": "Payment Order", "validation": {"docstatus": ["=", 1]}, # keep the company of an existing order; references are validated against it "field_no_map": ["company"] if target_doc else [], } }, target_doc, set_missing_values, ) return doclist @erpnext.allow_regional def add_regional_gl_entries(gl_entries, doc): return @frappe.whitelist() def get_linked_bank_transactions(payment_entry: str) -> list: frappe.has_permission("Payment Entry", ptype="read", doc=payment_entry, throw=True) return frappe.get_all( "Bank Transaction Payments", filters={ "payment_document": "Payment Entry", "payment_entry": payment_entry, }, pluck="parent", )