From 7b456c6405d8a7b55dc76abce8944b56690087bd Mon Sep 17 00:00:00 2001 From: Nabin Hait Date: Fri, 29 May 2026 13:01:38 +0530 Subject: [PATCH] refactor(sales_invoice): move mapping functions to mapper.py --- .../accounts/doctype/sales_invoice/mapper.py | 615 +++++++++++++++++ .../doctype/sales_invoice/sales_invoice.py | 625 +----------------- 2 files changed, 631 insertions(+), 609 deletions(-) create mode 100644 erpnext/accounts/doctype/sales_invoice/mapper.py diff --git a/erpnext/accounts/doctype/sales_invoice/mapper.py b/erpnext/accounts/doctype/sales_invoice/mapper.py new file mode 100644 index 00000000000..cebae93fd1a --- /dev/null +++ b/erpnext/accounts/doctype/sales_invoice/mapper.py @@ -0,0 +1,615 @@ +# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors +# License: GNU General Public License v3. See license.txt + +import frappe +from frappe import _ +from frappe.contacts.doctype.address.address import get_address_display +from frappe.model.document import Document +from frappe.model.mapper import get_mapped_doc +from frappe.model.utils import get_fetch_values +from frappe.utils import flt, get_link_to_form, getdate + +from erpnext.accounts.party import get_party_details + + +@frappe.whitelist() +def make_maintenance_schedule(source_name: str, target_doc: str | Document | None = None): + doclist = get_mapped_doc( + "Sales Invoice", + source_name, + { + "Sales Invoice": {"doctype": "Maintenance Schedule", "validation": {"docstatus": ["=", 1]}}, + "Sales Invoice Item": { + "doctype": "Maintenance Schedule Item", + }, + }, + target_doc, + ) + + return doclist + + +@frappe.whitelist() +def make_delivery_note(source_name: str, target_doc: Document | None = None): + def set_missing_values(source, target): + target.run_method("set_missing_values") + target.run_method("set_po_nos") + target.run_method("calculate_taxes_and_totals") + + def update_item(source_doc, target_doc, source_parent): + target_doc.qty = flt(source_doc.qty) - flt(source_doc.delivered_qty) + target_doc.stock_qty = target_doc.qty * flt(source_doc.conversion_factor) + + target_doc.base_amount = target_doc.qty * flt(source_doc.base_rate) + target_doc.amount = target_doc.qty * flt(source_doc.rate) + + doclist = get_mapped_doc( + "Sales Invoice", + source_name, + { + "Sales Invoice": {"doctype": "Delivery Note", "validation": {"docstatus": ["=", 1]}}, + "Sales Invoice Item": { + "doctype": "Delivery Note Item", + "field_map": { + "name": "si_detail", + "parent": "against_sales_invoice", + "serial_no": "serial_no", + "sales_order": "against_sales_order", + "so_detail": "so_detail", + "cost_center": "cost_center", + }, + "postprocess": update_item, + "condition": lambda doc: doc.delivered_by_supplier != 1 + and not doc.scio_detail + and not doc.dn_detail + and doc.qty - doc.delivered_qty > 0, + }, + "Sales Taxes and Charges": {"doctype": "Sales Taxes and Charges", "reset_value": True}, + "Sales Team": { + "doctype": "Sales Team", + "field_map": {"incentives": "incentives"}, + "add_if_empty": True, + }, + }, + target_doc, + set_missing_values, + ) + + return doclist + + +@frappe.whitelist() +def make_sales_return(source_name: str, target_doc: Document | None = None): + from erpnext.controllers.sales_and_purchase_return import make_return_doc + + return make_return_doc("Sales Invoice", source_name, target_doc) + + +def get_inter_company_details(doc, doctype): + if doctype in ["Sales Invoice", "Sales Order", "Delivery Note"]: + parties = frappe.db.get_all( + "Supplier", + fields=["name"], + filters={"disabled": 0, "is_internal_supplier": 1, "represents_company": doc.company}, + ) + company = frappe.get_cached_value("Customer", doc.customer, "represents_company") + + if not parties: + frappe.throw( + _("No Supplier found for Inter Company Transactions which represents company {0}").format( + frappe.bold(doc.company) + ) + ) + + party = get_internal_party(parties, "Supplier", doc) + else: + parties = frappe.db.get_all( + "Customer", + fields=["name"], + filters={"disabled": 0, "is_internal_customer": 1, "represents_company": doc.company}, + ) + company = frappe.get_cached_value("Supplier", doc.supplier, "represents_company") + + if not parties: + frappe.throw( + _("No Customer found for Inter Company Transactions which represents company {0}").format( + frappe.bold(doc.company) + ) + ) + + party = get_internal_party(parties, "Customer", doc) + + return {"party": party, "company": company} + + +def get_internal_party(parties, link_doctype, doc): + if len(parties) == 1: + party = parties[0].name + else: + # If more than one Internal Supplier/Customer, get supplier/customer on basis of address + if doc.get("company_address") or doc.get("shipping_address"): + party = frappe.db.get_value( + "Dynamic Link", + { + "parent": doc.get("company_address") or doc.get("shipping_address"), + "parenttype": "Address", + "link_doctype": link_doctype, + }, + "link_name", + ) + + if not party: + party = parties[0].name + else: + party = parties[0].name + + return party + + +def validate_inter_company_transaction(doc, doctype): + details = get_inter_company_details(doc, doctype) + price_list = ( + doc.selling_price_list + if doctype in ["Sales Invoice", "Sales Order", "Delivery Note"] + else doc.buying_price_list + ) + valid_price_list = frappe.db.get_value("Price List", {"name": price_list, "buying": 1, "selling": 1}) + if not valid_price_list and not doc.is_internal_transfer(): + frappe.throw(_("Selected Price List should have buying and selling fields checked.")) + + party = details.get("party") + if not party: + partytype = "Supplier" if doctype in ["Sales Invoice", "Sales Order"] else "Customer" + frappe.throw(_("No {0} found for Inter Company Transactions.").format(partytype)) + + company = details.get("company") + default_currency = frappe.get_cached_value("Company", company, "default_currency") + if default_currency != doc.currency: + frappe.throw( + _("Company currencies of both the companies should match for Inter Company Transactions.") + ) + + return + + +@frappe.whitelist() +def make_inter_company_purchase_invoice(source_name: str, target_doc: Document | None = None): + return make_inter_company_transaction("Sales Invoice", source_name, target_doc) + + +def make_inter_company_transaction(doctype, source_name, target_doc=None): + if doctype in ["Sales Invoice", "Sales Order"]: + source_doc = frappe.get_doc(doctype, source_name) + target_doctype = "Purchase Invoice" if doctype == "Sales Invoice" else "Purchase Order" + target_detail_field = "sales_invoice_item" if doctype == "Sales Invoice" else "sales_order_item" + source_document_warehouse_field = "target_warehouse" + target_document_warehouse_field = "from_warehouse" + received_items = get_received_items(source_name, target_doctype, target_detail_field) + else: + source_doc = frappe.get_doc(doctype, source_name) + target_doctype = "Sales Invoice" if doctype == "Purchase Invoice" else "Sales Order" + source_document_warehouse_field = "from_warehouse" + target_document_warehouse_field = "target_warehouse" + received_items = {} + + validate_inter_company_transaction(source_doc, doctype) + details = get_inter_company_details(source_doc, doctype) + + def set_missing_values(source, target): + target.run_method("set_missing_values") + set_purchase_references(target) + + def update_details(source_doc, target_doc, source_parent): + def _validate_address_link(address, link_doctype, link_name): + return frappe.db.get_value( + "Dynamic Link", + { + "parent": address, + "parenttype": "Address", + "link_doctype": link_doctype, + "link_name": link_name, + }, + "parent", + ) + + target_doc.inter_company_invoice_reference = source_doc.name + if target_doc.doctype in ["Purchase Invoice", "Purchase Order"]: + currency = frappe.db.get_value("Supplier", details.get("party"), "default_currency") + target_doc.company = details.get("company") + target_doc.supplier = details.get("party") + target_doc.is_internal_supplier = 1 + target_doc.ignore_pricing_rule = 1 + target_doc.buying_price_list = source_doc.selling_price_list + + # Invert Addresses + if source_doc.company_address and _validate_address_link( + source_doc.company_address, "Supplier", details.get("party") + ): + update_address(target_doc, "supplier_address", "address_display", source_doc.company_address) + if source_doc.dispatch_address_name and _validate_address_link( + source_doc.dispatch_address_name, "Company", details.get("company") + ): + update_address( + target_doc, + "dispatch_address", + "dispatch_address_display", + source_doc.dispatch_address_name, + ) + if source_doc.shipping_address_name and _validate_address_link( + source_doc.shipping_address_name, "Company", details.get("company") + ): + update_address( + target_doc, + "shipping_address", + "shipping_address_display", + source_doc.shipping_address_name, + ) + if source_doc.customer_address and _validate_address_link( + source_doc.customer_address, "Company", details.get("company") + ): + update_address( + target_doc, "billing_address", "billing_address_display", source_doc.customer_address + ) + + if currency: + target_doc.currency = currency + + update_taxes( + target_doc, + party=target_doc.supplier, + party_type="Supplier", + company=target_doc.company, + doctype=target_doc.doctype, + party_address=target_doc.supplier_address, + company_address=target_doc.shipping_address, + ) + + else: + currency = frappe.db.get_value("Customer", details.get("party"), "default_currency") + target_doc.company = details.get("company") + target_doc.customer = details.get("party") + target_doc.selling_price_list = source_doc.buying_price_list + + if source_doc.supplier_address and _validate_address_link( + source_doc.supplier_address, "Company", details.get("company") + ): + update_address( + target_doc, "company_address", "company_address_display", source_doc.supplier_address + ) + if source_doc.shipping_address and _validate_address_link( + source_doc.shipping_address, "Customer", details.get("party") + ): + update_address( + target_doc, "shipping_address_name", "shipping_address", source_doc.shipping_address + ) + if source_doc.shipping_address and _validate_address_link( + source_doc.shipping_address, "Customer", details.get("party") + ): + update_address(target_doc, "customer_address", "address_display", source_doc.shipping_address) + + if currency: + target_doc.currency = currency + + update_taxes( + target_doc, + party=target_doc.customer, + party_type="Customer", + company=target_doc.company, + doctype=target_doc.doctype, + party_address=target_doc.customer_address, + company_address=target_doc.company_address, + shipping_address_name=target_doc.shipping_address_name, + ) + + def update_item(source, target, source_parent): + target.qty = flt(source.qty) - received_items.get(source.name, 0.0) + if source.doctype == "Purchase Order Item" and target.doctype == "Sales Order Item": + target.purchase_order = source.parent + target.purchase_order_item = source.name + target.material_request = source.material_request + target.material_request_item = source.material_request_item + + if ( + source.get("purchase_order") + and source.get("purchase_order_item") + and target.doctype == "Purchase Invoice Item" + ): + target.purchase_order = source.purchase_order + target.po_detail = source.purchase_order_item + + if (source.get("serial_no") or source.get("batch_no")) and not source.get("serial_and_batch_bundle"): + target.use_serial_batch_fields = 1 + + item_field_map = { + "doctype": target_doctype + " Item", + "field_no_map": ["income_account", "expense_account", "cost_center", "warehouse"], + "field_map": { + "rate": "rate", + }, + "postprocess": update_item, + "condition": lambda doc: doc.qty > 0, + } + + if doctype in ["Sales Invoice", "Sales Order"]: + item_field_map["field_map"].update( + { + "name": target_detail_field, + } + ) + + if source_doc.get("update_stock"): + item_field_map["field_map"].update( + { + source_document_warehouse_field: target_document_warehouse_field, + "batch_no": "batch_no", + "serial_no": "serial_no", + } + ) + elif target_doctype == "Sales Order": + item_field_map["field_map"].update( + { + source_document_warehouse_field: "warehouse", + } + ) + + doclist = get_mapped_doc( + doctype, + source_name, + { + doctype: { + "doctype": target_doctype, + "postprocess": update_details, + "set_target_warehouse": "set_from_warehouse", + "field_no_map": ["taxes_and_charges", "set_warehouse", "shipping_address", "cost_center"], + }, + doctype + " Item": item_field_map, + }, + target_doc, + set_missing_values, + ) + + return doclist + + +def get_received_items(reference_name, doctype, reference_fieldname): + reference_field = "inter_company_invoice_reference" + if doctype == "Purchase Order": + reference_field = "inter_company_order_reference" + + filters = { + reference_field: reference_name, + "docstatus": 1, + } + + target_doctypes = frappe.get_all( + doctype, + filters=filters, + as_list=True, + ) + + if target_doctypes: + target_doctypes = list(target_doctypes[0]) + + received_items_map = frappe._dict( + frappe.get_all( + doctype + " Item", + filters={"parent": ("in", target_doctypes)}, + fields=[reference_fieldname, "qty"], + as_list=1, + ) + ) + + return received_items_map + + +def set_purchase_references(doc): + # add internal PO or PR links if any + + if doc.is_internal_transfer(): + if doc.doctype == "Purchase Receipt": + so_item_map = get_delivery_note_details(doc.inter_company_invoice_reference) + + if so_item_map: + pd_item_map, parent_child_map, warehouse_map = get_pd_details( + "Purchase Order Item", so_item_map, "sales_order_item" + ) + + update_pr_items(doc, so_item_map, pd_item_map, parent_child_map, warehouse_map) + + elif doc.doctype == "Purchase Invoice": + dn_item_map, so_item_map = get_sales_invoice_details(doc.inter_company_invoice_reference) + # First check for Purchase receipt + if list(dn_item_map.values()): + pd_item_map, parent_child_map, warehouse_map = get_pd_details( + "Purchase Receipt Item", dn_item_map, "delivery_note_item" + ) + + update_pi_items( + doc, + "pr_detail", + "purchase_receipt", + dn_item_map, + pd_item_map, + parent_child_map, + warehouse_map, + ) + + +def update_pi_items( + doc, + detail_field, + parent_field, + sales_item_map, + purchase_item_map, + parent_child_map, + warehouse_map, +): + for item in doc.get("items"): + item.set(detail_field, purchase_item_map.get(sales_item_map.get(item.sales_invoice_item))) + item.set(parent_field, parent_child_map.get(sales_item_map.get(item.sales_invoice_item))) + if doc.update_stock: + item.warehouse = warehouse_map.get(sales_item_map.get(item.sales_invoice_item)) + if not item.warehouse and item.get("purchase_order") and item.get("purchase_order_item"): + item.warehouse = frappe.db.get_value( + "Purchase Order Item", item.purchase_order_item, "warehouse" + ) + + +def update_pr_items(doc, sales_item_map, purchase_item_map, parent_child_map, warehouse_map): + for item in doc.get("items"): + item.warehouse = warehouse_map.get(sales_item_map.get(item.delivery_note_item)) + if not item.warehouse and item.get("purchase_order") and item.get("purchase_order_item"): + item.warehouse = frappe.db.get_value("Purchase Order Item", item.purchase_order_item, "warehouse") + + +def get_delivery_note_details(internal_reference): + si_item_details = frappe.get_all( + "Delivery Note Item", fields=["name", "so_detail"], filters={"parent": internal_reference} + ) + + return {d.name: d.so_detail for d in si_item_details if d.so_detail} + + +def get_sales_invoice_details(internal_reference): + dn_item_map = {} + so_item_map = {} + + si_item_details = frappe.get_all( + "Sales Invoice Item", + fields=["name", "so_detail", "dn_detail"], + filters={"parent": internal_reference}, + ) + + for d in si_item_details: + if d.dn_detail: + dn_item_map.setdefault(d.name, d.dn_detail) + if d.so_detail: + so_item_map.setdefault(d.name, d.so_detail) + + return dn_item_map, so_item_map + + +def get_pd_details(doctype, sd_detail_map, sd_detail_field): + pd_item_map = {} + accepted_warehouse_map = {} + parent_child_map = {} + + pd_item_details = frappe.get_all( + doctype, + fields=[sd_detail_field, "name", "warehouse", "parent"], + filters={sd_detail_field: ("in", list(sd_detail_map.values()))}, + ) + + for d in pd_item_details: + pd_item_map.setdefault(d.get(sd_detail_field), d.name) + parent_child_map.setdefault(d.get(sd_detail_field), d.parent) + accepted_warehouse_map.setdefault(d.get(sd_detail_field), d.warehouse) + + return pd_item_map, parent_child_map, accepted_warehouse_map + + +def update_taxes( + doc, + party=None, + party_type=None, + company=None, + doctype=None, + party_address=None, + company_address=None, + shipping_address_name=None, + master_doctype=None, +): + # Update Party Details + party_details = get_party_details( + party=party, + party_type=party_type, + company=company, + doctype=doctype, + party_address=party_address, + company_address=company_address, + shipping_address=shipping_address_name, + ) + + # Update taxes and charges if any + doc.taxes_and_charges = party_details.get("taxes_and_charges") + doc.set("taxes", party_details.get("taxes")) + + +def update_address(doc, address_field, address_display_field, address_name): + doc.set(address_field, address_name) + fetch_values = get_fetch_values(doc.doctype, address_field, address_name) + + for key, value in fetch_values.items(): + doc.set(key, value) + + doc.set(address_display_field, get_address_display(doc.get(address_field))) + + +@frappe.whitelist() +def create_invoice_discounting(source_name: str, target_doc: str | Document | None = None): + invoice = frappe.get_doc("Sales Invoice", source_name) + invoice_discounting = frappe.new_doc("Invoice Discounting") + invoice_discounting.company = invoice.company + invoice_discounting.append( + "invoices", + { + "sales_invoice": source_name, + "customer": invoice.customer, + "posting_date": invoice.posting_date, + "outstanding_amount": invoice.outstanding_amount, + }, + ) + + return invoice_discounting + + +@frappe.whitelist() +def create_dunning( + source_name: str, target_doc: str | Document | None = None, ignore_permissions: bool = False +): + def postprocess_dunning(source, target): + from erpnext.accounts.doctype.dunning.dunning import get_dunning_letter_text + + dunning_type = frappe.db.exists("Dunning Type", {"is_default": 1, "company": source.company}) + if dunning_type: + dunning_type = frappe.get_doc("Dunning Type", dunning_type) + target.dunning_type = dunning_type.name + target.rate_of_interest = dunning_type.rate_of_interest + target.dunning_fee = dunning_type.dunning_fee + target.income_account = dunning_type.income_account + target.cost_center = dunning_type.cost_center + letter_text = get_dunning_letter_text( + dunning_type=dunning_type.name, doc=target.as_dict(), language=source.language + ) + + if letter_text: + target.body_text = letter_text.get("body_text") + target.closing_text = letter_text.get("closing_text") + target.language = letter_text.get("language") + + # update outstanding from doc + if source.payment_schedule and len(source.payment_schedule) == 1: + for row in target.overdue_payments: + if row.payment_schedule == source.payment_schedule[0].name: + row.outstanding = source.get("outstanding_amount") + + target.validate() + + return get_mapped_doc( + from_doctype="Sales Invoice", + from_docname=source_name, + target_doc=target_doc, + table_maps={ + "Sales Invoice": { + "doctype": "Dunning", + "field_map": {"customer_address": "customer_address", "parent": "sales_invoice"}, + }, + "Payment Schedule": { + "doctype": "Overdue Payment", + "field_map": {"name": "payment_schedule", "parent": "sales_invoice"}, + "condition": lambda doc: doc.outstanding > 0 and getdate(doc.due_date) < getdate(), + }, + }, + postprocess=postprocess_dunning, + ignore_permissions=ignore_permissions, + ) diff --git a/erpnext/accounts/doctype/sales_invoice/sales_invoice.py b/erpnext/accounts/doctype/sales_invoice/sales_invoice.py index 2950a182e1d..89851d8c16e 100644 --- a/erpnext/accounts/doctype/sales_invoice/sales_invoice.py +++ b/erpnext/accounts/doctype/sales_invoice/sales_invoice.py @@ -5,10 +5,7 @@ import frappe import frappe.utils from frappe import _, msgprint, throw -from frappe.contacts.doctype.address.address import get_address_display from frappe.model.document import Document -from frappe.model.mapper import get_mapped_doc -from frappe.model.utils import get_fetch_values from frappe.query_builder import Case from frappe.utils import add_days, cint, cstr, flt, formatdate, get_link_to_form, getdate, nowdate from frappe.utils.data import comma_and @@ -29,7 +26,7 @@ from erpnext.accounts.doctype.repost_accounting_ledger.repost_accounting_ledger ) from erpnext.accounts.doctype.tax_withholding_entry.tax_withholding_entry import SalesTaxWithholding from erpnext.accounts.general_ledger import get_round_off_account_and_cost_center -from erpnext.accounts.party import get_due_date, get_party_account, get_party_details +from erpnext.accounts.party import get_due_date, get_party_account from erpnext.accounts.utils import ( get_account_currency, update_voucher_outstanding, @@ -49,6 +46,21 @@ from erpnext.projects.doctype.timesheet.timesheet import get_projectwise_timeshe from erpnext.setup.doctype.company.company import update_company_current_month_sales from erpnext.stock.doctype.delivery_note.delivery_note import update_billed_amount_based_on_so +from .mapper import ( + create_dunning, + create_invoice_discounting, + get_inter_company_details, + make_delivery_note, + make_inter_company_purchase_invoice, + make_inter_company_transaction, + make_maintenance_schedule, + make_sales_return, + set_purchase_references, + update_address, + update_taxes, + validate_inter_company_transaction, +) + form_grid_templates = {"items": "templates/form_grid/item_grid.html"} @@ -1995,544 +2007,11 @@ def get_bank_cash_account(mode_of_payment: str, company: str): return {"account": account} -@frappe.whitelist() -def make_maintenance_schedule(source_name: str, target_doc: str | Document | None = None): - doclist = get_mapped_doc( - "Sales Invoice", - source_name, - { - "Sales Invoice": {"doctype": "Maintenance Schedule", "validation": {"docstatus": ["=", 1]}}, - "Sales Invoice Item": { - "doctype": "Maintenance Schedule Item", - }, - }, - target_doc, - ) - - return doclist - - -@frappe.whitelist() -def make_delivery_note(source_name: str, target_doc: Document | None = None): - def set_missing_values(source, target): - target.run_method("set_missing_values") - target.run_method("set_po_nos") - target.run_method("calculate_taxes_and_totals") - - def update_item(source_doc, target_doc, source_parent): - target_doc.qty = flt(source_doc.qty) - flt(source_doc.delivered_qty) - target_doc.stock_qty = target_doc.qty * flt(source_doc.conversion_factor) - - target_doc.base_amount = target_doc.qty * flt(source_doc.base_rate) - target_doc.amount = target_doc.qty * flt(source_doc.rate) - - doclist = get_mapped_doc( - "Sales Invoice", - source_name, - { - "Sales Invoice": {"doctype": "Delivery Note", "validation": {"docstatus": ["=", 1]}}, - "Sales Invoice Item": { - "doctype": "Delivery Note Item", - "field_map": { - "name": "si_detail", - "parent": "against_sales_invoice", - "serial_no": "serial_no", - "sales_order": "against_sales_order", - "so_detail": "so_detail", - "cost_center": "cost_center", - }, - "postprocess": update_item, - "condition": lambda doc: doc.delivered_by_supplier != 1 - and not doc.scio_detail - and not doc.dn_detail - and doc.qty - doc.delivered_qty > 0, - }, - "Sales Taxes and Charges": {"doctype": "Sales Taxes and Charges", "reset_value": True}, - "Sales Team": { - "doctype": "Sales Team", - "field_map": {"incentives": "incentives"}, - "add_if_empty": True, - }, - }, - target_doc, - set_missing_values, - ) - - return doclist - - -@frappe.whitelist() -def make_sales_return(source_name: str, target_doc: Document | None = None): - from erpnext.controllers.sales_and_purchase_return import make_return_doc - - return make_return_doc("Sales Invoice", source_name, target_doc) - - -def get_inter_company_details(doc, doctype): - if doctype in ["Sales Invoice", "Sales Order", "Delivery Note"]: - parties = frappe.db.get_all( - "Supplier", - fields=["name"], - filters={"disabled": 0, "is_internal_supplier": 1, "represents_company": doc.company}, - ) - company = frappe.get_cached_value("Customer", doc.customer, "represents_company") - - if not parties: - frappe.throw( - _("No Supplier found for Inter Company Transactions which represents company {0}").format( - frappe.bold(doc.company) - ) - ) - - party = get_internal_party(parties, "Supplier", doc) - else: - parties = frappe.db.get_all( - "Customer", - fields=["name"], - filters={"disabled": 0, "is_internal_customer": 1, "represents_company": doc.company}, - ) - company = frappe.get_cached_value("Supplier", doc.supplier, "represents_company") - - if not parties: - frappe.throw( - _("No Customer found for Inter Company Transactions which represents company {0}").format( - frappe.bold(doc.company) - ) - ) - - party = get_internal_party(parties, "Customer", doc) - - return {"party": party, "company": company} - - -def get_internal_party(parties, link_doctype, doc): - if len(parties) == 1: - party = parties[0].name - else: - # If more than one Internal Supplier/Customer, get supplier/customer on basis of address - if doc.get("company_address") or doc.get("shipping_address"): - party = frappe.db.get_value( - "Dynamic Link", - { - "parent": doc.get("company_address") or doc.get("shipping_address"), - "parenttype": "Address", - "link_doctype": link_doctype, - }, - "link_name", - ) - - if not party: - party = parties[0].name - else: - party = parties[0].name - - return party - - -def validate_inter_company_transaction(doc, doctype): - details = get_inter_company_details(doc, doctype) - price_list = ( - doc.selling_price_list - if doctype in ["Sales Invoice", "Sales Order", "Delivery Note"] - else doc.buying_price_list - ) - valid_price_list = frappe.db.get_value("Price List", {"name": price_list, "buying": 1, "selling": 1}) - if not valid_price_list and not doc.is_internal_transfer(): - frappe.throw(_("Selected Price List should have buying and selling fields checked.")) - - party = details.get("party") - if not party: - partytype = "Supplier" if doctype in ["Sales Invoice", "Sales Order"] else "Customer" - frappe.throw(_("No {0} found for Inter Company Transactions.").format(partytype)) - - company = details.get("company") - default_currency = frappe.get_cached_value("Company", company, "default_currency") - if default_currency != doc.currency: - frappe.throw( - _("Company currencies of both the companies should match for Inter Company Transactions.") - ) - - return - - -@frappe.whitelist() -def make_inter_company_purchase_invoice(source_name: str, target_doc: Document | None = None): - return make_inter_company_transaction("Sales Invoice", source_name, target_doc) - - @erpnext.allow_regional def make_regional_gl_entries(gl_entries, doc): return gl_entries -def make_inter_company_transaction(doctype, source_name, target_doc=None): - if doctype in ["Sales Invoice", "Sales Order"]: - source_doc = frappe.get_doc(doctype, source_name) - target_doctype = "Purchase Invoice" if doctype == "Sales Invoice" else "Purchase Order" - target_detail_field = "sales_invoice_item" if doctype == "Sales Invoice" else "sales_order_item" - source_document_warehouse_field = "target_warehouse" - target_document_warehouse_field = "from_warehouse" - received_items = get_received_items(source_name, target_doctype, target_detail_field) - else: - source_doc = frappe.get_doc(doctype, source_name) - target_doctype = "Sales Invoice" if doctype == "Purchase Invoice" else "Sales Order" - source_document_warehouse_field = "from_warehouse" - target_document_warehouse_field = "target_warehouse" - received_items = {} - - validate_inter_company_transaction(source_doc, doctype) - details = get_inter_company_details(source_doc, doctype) - - def set_missing_values(source, target): - target.run_method("set_missing_values") - set_purchase_references(target) - - def update_details(source_doc, target_doc, source_parent): - def _validate_address_link(address, link_doctype, link_name): - return frappe.db.get_value( - "Dynamic Link", - { - "parent": address, - "parenttype": "Address", - "link_doctype": link_doctype, - "link_name": link_name, - }, - "parent", - ) - - target_doc.inter_company_invoice_reference = source_doc.name - if target_doc.doctype in ["Purchase Invoice", "Purchase Order"]: - currency = frappe.db.get_value("Supplier", details.get("party"), "default_currency") - target_doc.company = details.get("company") - target_doc.supplier = details.get("party") - target_doc.is_internal_supplier = 1 - target_doc.ignore_pricing_rule = 1 - target_doc.buying_price_list = source_doc.selling_price_list - - # Invert Addresses - if source_doc.company_address and _validate_address_link( - source_doc.company_address, "Supplier", details.get("party") - ): - update_address(target_doc, "supplier_address", "address_display", source_doc.company_address) - if source_doc.dispatch_address_name and _validate_address_link( - source_doc.dispatch_address_name, "Company", details.get("company") - ): - update_address( - target_doc, - "dispatch_address", - "dispatch_address_display", - source_doc.dispatch_address_name, - ) - if source_doc.shipping_address_name and _validate_address_link( - source_doc.shipping_address_name, "Company", details.get("company") - ): - update_address( - target_doc, - "shipping_address", - "shipping_address_display", - source_doc.shipping_address_name, - ) - if source_doc.customer_address and _validate_address_link( - source_doc.customer_address, "Company", details.get("company") - ): - update_address( - target_doc, "billing_address", "billing_address_display", source_doc.customer_address - ) - - if currency: - target_doc.currency = currency - - update_taxes( - target_doc, - party=target_doc.supplier, - party_type="Supplier", - company=target_doc.company, - doctype=target_doc.doctype, - party_address=target_doc.supplier_address, - company_address=target_doc.shipping_address, - ) - - else: - currency = frappe.db.get_value("Customer", details.get("party"), "default_currency") - target_doc.company = details.get("company") - target_doc.customer = details.get("party") - target_doc.selling_price_list = source_doc.buying_price_list - - if source_doc.supplier_address and _validate_address_link( - source_doc.supplier_address, "Company", details.get("company") - ): - update_address( - target_doc, "company_address", "company_address_display", source_doc.supplier_address - ) - if source_doc.shipping_address and _validate_address_link( - source_doc.shipping_address, "Customer", details.get("party") - ): - update_address( - target_doc, "shipping_address_name", "shipping_address", source_doc.shipping_address - ) - if source_doc.shipping_address and _validate_address_link( - source_doc.shipping_address, "Customer", details.get("party") - ): - update_address(target_doc, "customer_address", "address_display", source_doc.shipping_address) - - if currency: - target_doc.currency = currency - - update_taxes( - target_doc, - party=target_doc.customer, - party_type="Customer", - company=target_doc.company, - doctype=target_doc.doctype, - party_address=target_doc.customer_address, - company_address=target_doc.company_address, - shipping_address_name=target_doc.shipping_address_name, - ) - - def update_item(source, target, source_parent): - target.qty = flt(source.qty) - received_items.get(source.name, 0.0) - if source.doctype == "Purchase Order Item" and target.doctype == "Sales Order Item": - target.purchase_order = source.parent - target.purchase_order_item = source.name - target.material_request = source.material_request - target.material_request_item = source.material_request_item - - if ( - source.get("purchase_order") - and source.get("purchase_order_item") - and target.doctype == "Purchase Invoice Item" - ): - target.purchase_order = source.purchase_order - target.po_detail = source.purchase_order_item - - if (source.get("serial_no") or source.get("batch_no")) and not source.get("serial_and_batch_bundle"): - target.use_serial_batch_fields = 1 - - item_field_map = { - "doctype": target_doctype + " Item", - "field_no_map": ["income_account", "expense_account", "cost_center", "warehouse"], - "field_map": { - "rate": "rate", - }, - "postprocess": update_item, - "condition": lambda doc: doc.qty > 0, - } - - if doctype in ["Sales Invoice", "Sales Order"]: - item_field_map["field_map"].update( - { - "name": target_detail_field, - } - ) - - if source_doc.get("update_stock"): - item_field_map["field_map"].update( - { - source_document_warehouse_field: target_document_warehouse_field, - "batch_no": "batch_no", - "serial_no": "serial_no", - } - ) - elif target_doctype == "Sales Order": - item_field_map["field_map"].update( - { - source_document_warehouse_field: "warehouse", - } - ) - - doclist = get_mapped_doc( - doctype, - source_name, - { - doctype: { - "doctype": target_doctype, - "postprocess": update_details, - "set_target_warehouse": "set_from_warehouse", - "field_no_map": ["taxes_and_charges", "set_warehouse", "shipping_address", "cost_center"], - }, - doctype + " Item": item_field_map, - }, - target_doc, - set_missing_values, - ) - - return doclist - - -def get_received_items(reference_name, doctype, reference_fieldname): - reference_field = "inter_company_invoice_reference" - if doctype == "Purchase Order": - reference_field = "inter_company_order_reference" - - filters = { - reference_field: reference_name, - "docstatus": 1, - } - - target_doctypes = frappe.get_all( - doctype, - filters=filters, - as_list=True, - ) - - if target_doctypes: - target_doctypes = list(target_doctypes[0]) - - received_items_map = frappe._dict( - frappe.get_all( - doctype + " Item", - filters={"parent": ("in", target_doctypes)}, - fields=[reference_fieldname, "qty"], - as_list=1, - ) - ) - - return received_items_map - - -def set_purchase_references(doc): - # add internal PO or PR links if any - - if doc.is_internal_transfer(): - if doc.doctype == "Purchase Receipt": - so_item_map = get_delivery_note_details(doc.inter_company_invoice_reference) - - if so_item_map: - pd_item_map, parent_child_map, warehouse_map = get_pd_details( - "Purchase Order Item", so_item_map, "sales_order_item" - ) - - update_pr_items(doc, so_item_map, pd_item_map, parent_child_map, warehouse_map) - - elif doc.doctype == "Purchase Invoice": - dn_item_map, so_item_map = get_sales_invoice_details(doc.inter_company_invoice_reference) - # First check for Purchase receipt - if list(dn_item_map.values()): - pd_item_map, parent_child_map, warehouse_map = get_pd_details( - "Purchase Receipt Item", dn_item_map, "delivery_note_item" - ) - - update_pi_items( - doc, - "pr_detail", - "purchase_receipt", - dn_item_map, - pd_item_map, - parent_child_map, - warehouse_map, - ) - - -def update_pi_items( - doc, - detail_field, - parent_field, - sales_item_map, - purchase_item_map, - parent_child_map, - warehouse_map, -): - for item in doc.get("items"): - item.set(detail_field, purchase_item_map.get(sales_item_map.get(item.sales_invoice_item))) - item.set(parent_field, parent_child_map.get(sales_item_map.get(item.sales_invoice_item))) - if doc.update_stock: - item.warehouse = warehouse_map.get(sales_item_map.get(item.sales_invoice_item)) - if not item.warehouse and item.get("purchase_order") and item.get("purchase_order_item"): - item.warehouse = frappe.db.get_value( - "Purchase Order Item", item.purchase_order_item, "warehouse" - ) - - -def update_pr_items(doc, sales_item_map, purchase_item_map, parent_child_map, warehouse_map): - for item in doc.get("items"): - item.warehouse = warehouse_map.get(sales_item_map.get(item.delivery_note_item)) - if not item.warehouse and item.get("purchase_order") and item.get("purchase_order_item"): - item.warehouse = frappe.db.get_value("Purchase Order Item", item.purchase_order_item, "warehouse") - - -def get_delivery_note_details(internal_reference): - si_item_details = frappe.get_all( - "Delivery Note Item", fields=["name", "so_detail"], filters={"parent": internal_reference} - ) - - return {d.name: d.so_detail for d in si_item_details if d.so_detail} - - -def get_sales_invoice_details(internal_reference): - dn_item_map = {} - so_item_map = {} - - si_item_details = frappe.get_all( - "Sales Invoice Item", - fields=["name", "so_detail", "dn_detail"], - filters={"parent": internal_reference}, - ) - - for d in si_item_details: - if d.dn_detail: - dn_item_map.setdefault(d.name, d.dn_detail) - if d.so_detail: - so_item_map.setdefault(d.name, d.so_detail) - - return dn_item_map, so_item_map - - -def get_pd_details(doctype, sd_detail_map, sd_detail_field): - pd_item_map = {} - accepted_warehouse_map = {} - parent_child_map = {} - - pd_item_details = frappe.get_all( - doctype, - fields=[sd_detail_field, "name", "warehouse", "parent"], - filters={sd_detail_field: ("in", list(sd_detail_map.values()))}, - ) - - for d in pd_item_details: - pd_item_map.setdefault(d.get(sd_detail_field), d.name) - parent_child_map.setdefault(d.get(sd_detail_field), d.parent) - accepted_warehouse_map.setdefault(d.get(sd_detail_field), d.warehouse) - - return pd_item_map, parent_child_map, accepted_warehouse_map - - -def update_taxes( - doc, - party=None, - party_type=None, - company=None, - doctype=None, - party_address=None, - company_address=None, - shipping_address_name=None, - master_doctype=None, -): - # Update Party Details - party_details = get_party_details( - party=party, - party_type=party_type, - company=company, - doctype=doctype, - party_address=party_address, - company_address=company_address, - shipping_address=shipping_address_name, - ) - - # Update taxes and charges if any - doc.taxes_and_charges = party_details.get("taxes_and_charges") - doc.set("taxes", party_details.get("taxes")) - - -def update_address(doc, address_field, address_display_field, address_name): - doc.set(address_field, address_name) - fetch_values = get_fetch_values(doc.doctype, address_field, address_name) - - for key, value in fetch_values.items(): - doc.set(key, value) - - doc.set(address_display_field, get_address_display(doc.get(address_field))) - - @frappe.whitelist() def get_loyalty_programs(customer: str): """sets applicable loyalty program to the customer or returns a list of applicable programs""" @@ -2551,24 +2030,6 @@ def get_loyalty_programs(customer: str): return lp_details -@frappe.whitelist() -def create_invoice_discounting(source_name: str, target_doc: str | Document | None = None): - invoice = frappe.get_doc("Sales Invoice", source_name) - invoice_discounting = frappe.new_doc("Invoice Discounting") - invoice_discounting.company = invoice.company - invoice_discounting.append( - "invoices", - { - "sales_invoice": source_name, - "customer": invoice.customer, - "posting_date": invoice.posting_date, - "outstanding_amount": invoice.outstanding_amount, - }, - ) - - return invoice_discounting - - def update_multi_mode_option(doc, pos_profile): def append_payment(payment_mode): payment = doc.append("payments", {}) @@ -2651,60 +2112,6 @@ def get_mode_of_payment_info(mode_of_payment, company): ) -@frappe.whitelist() -def create_dunning( - source_name: str, target_doc: str | Document | None = None, ignore_permissions: bool = False -): - from frappe.model.mapper import get_mapped_doc - - def postprocess_dunning(source, target): - from erpnext.accounts.doctype.dunning.dunning import get_dunning_letter_text - - dunning_type = frappe.db.exists("Dunning Type", {"is_default": 1, "company": source.company}) - if dunning_type: - dunning_type = frappe.get_doc("Dunning Type", dunning_type) - target.dunning_type = dunning_type.name - target.rate_of_interest = dunning_type.rate_of_interest - target.dunning_fee = dunning_type.dunning_fee - target.income_account = dunning_type.income_account - target.cost_center = dunning_type.cost_center - letter_text = get_dunning_letter_text( - dunning_type=dunning_type.name, doc=target.as_dict(), language=source.language - ) - - if letter_text: - target.body_text = letter_text.get("body_text") - target.closing_text = letter_text.get("closing_text") - target.language = letter_text.get("language") - - # update outstanding from doc - if source.payment_schedule and len(source.payment_schedule) == 1: - for row in target.overdue_payments: - if row.payment_schedule == source.payment_schedule[0].name: - row.outstanding = source.get("outstanding_amount") - - target.validate() - - return get_mapped_doc( - from_doctype="Sales Invoice", - from_docname=source_name, - target_doc=target_doc, - table_maps={ - "Sales Invoice": { - "doctype": "Dunning", - "field_map": {"customer_address": "customer_address", "parent": "sales_invoice"}, - }, - "Payment Schedule": { - "doctype": "Overdue Payment", - "field_map": {"name": "payment_schedule", "parent": "sales_invoice"}, - "condition": lambda doc: doc.outstanding > 0 and getdate(doc.due_date) < getdate(), - }, - }, - postprocess=postprocess_dunning, - ignore_permissions=ignore_permissions, - ) - - def check_if_return_invoice_linked_with_payment_entry(self): # If a Return invoice is linked with payment entry along with other invoices, # the cancellation of the Return causes allocated amount to be greater than paid