From 18d1a88a64e37998048a965e7caf4a8d707e0182 Mon Sep 17 00:00:00 2001 From: Nabin Hait Date: Fri, 29 May 2026 12:22:17 +0530 Subject: [PATCH] refactor(purchase_order): move mapping functions to mapper.py --- .../buying/doctype/purchase_order/mapper.py | 315 +++++++++++++++++ .../doctype/purchase_order/purchase_order.py | 317 +----------------- 2 files changed, 326 insertions(+), 306 deletions(-) create mode 100644 erpnext/buying/doctype/purchase_order/mapper.py diff --git a/erpnext/buying/doctype/purchase_order/mapper.py b/erpnext/buying/doctype/purchase_order/mapper.py new file mode 100644 index 00000000000..23aa32f4410 --- /dev/null +++ b/erpnext/buying/doctype/purchase_order/mapper.py @@ -0,0 +1,315 @@ +# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors +# License: GNU General Public License v3. See license.txt + +import json + +import frappe +from frappe import _ +from frappe.model.document import Document +from frappe.model.mapper import get_mapped_doc +from frappe.utils import flt, get_link_to_form + +from erpnext.accounts.party import get_party_account +from erpnext.setup.doctype.item_group.item_group import get_item_group_defaults +from erpnext.stock.doctype.item.item import get_item_defaults + + +def set_missing_values(source, target): + target.run_method("set_missing_values") + target.run_method("calculate_taxes_and_totals") + target.run_method("set_use_serial_batch_fields") + + +@frappe.whitelist() +def make_purchase_receipt( + source_name: str, target_doc: str | Document | None = None, args: str | dict | None = None +): + if args is None: + args = {} + if isinstance(args, str): + args = json.loads(args) + + has_unit_price_items = frappe.db.get_value("Purchase Order", source_name, "has_unit_price_items") + + def is_unit_price_row(source): + return has_unit_price_items and source.qty == 0 + + def update_item(obj, target, source_parent): + target.qty = flt(obj.qty) if is_unit_price_row(obj) else flt(obj.qty) - flt(obj.received_qty) + target.stock_qty = (flt(obj.qty) - flt(obj.received_qty)) * flt(obj.conversion_factor) + target.amount = (flt(obj.qty) - flt(obj.received_qty)) * flt(obj.rate) + target.base_amount = ( + (flt(obj.qty) - flt(obj.received_qty)) * flt(obj.rate) * flt(source_parent.conversion_rate) + ) + + def select_item(d): + filtered_items = args.get("filtered_children", []) + child_filter = d.name in filtered_items if filtered_items else True + return child_filter + + doc = get_mapped_doc( + "Purchase Order", + source_name, + { + "Purchase Order": { + "doctype": "Purchase Receipt", + "field_map": {"supplier_warehouse": "supplier_warehouse"}, + "validation": { + "docstatus": ["=", 1], + }, + }, + "Purchase Order Item": { + "doctype": "Purchase Receipt Item", + "field_map": { + "name": "purchase_order_item", + "parent": "purchase_order", + "bom": "bom", + "material_request": "material_request", + "material_request_item": "material_request_item", + "sales_order": "sales_order", + "sales_order_item": "sales_order_item", + "wip_composite_asset": "wip_composite_asset", + }, + "postprocess": update_item, + "condition": lambda doc: ( + True if is_unit_price_row(doc) else abs(doc.received_qty) < abs(doc.qty) + ) + and doc.delivered_by_supplier != 1 + and select_item(doc), + }, + "Purchase Taxes and Charges": {"doctype": "Purchase Taxes and Charges", "reset_value": True}, + }, + target_doc, + set_missing_values, + ) + + return doc + + +@frappe.whitelist() +def make_purchase_invoice( + source_name: str, target_doc: str | Document | None = None, args: str | dict | None = None +): + return get_mapped_purchase_invoice(source_name, target_doc, args=args) + + +@frappe.whitelist() +def make_purchase_invoice_from_portal(purchase_order_name: str): + doc = get_mapped_purchase_invoice(purchase_order_name, ignore_permissions=True) + if frappe.session.user not in frappe.get_all("Portal User", {"parent": doc.supplier}, pluck="user"): + frappe.throw(_("Not Permitted"), frappe.PermissionError) + doc.save() + if not frappe.in_test: + frappe.db.commit() + frappe.response["type"] = "redirect" + frappe.response.location = "/purchase-invoices/" + doc.name + + +def get_mapped_purchase_invoice(source_name, target_doc=None, ignore_permissions=False, args=None): + if args is None: + args = {} + if isinstance(args, str): + args = json.loads(args) + + def postprocess(source, target): + target.flags.ignore_permissions = ignore_permissions + set_missing_values(source, target) + + # Get the advance paid Journal Entries in Purchase Invoice Advance + if target.get("allocate_advances_automatically"): + target.set_advances() + + from erpnext.accounts.services.payment_schedule import PaymentScheduleService + + PaymentScheduleService(target).set_payment_schedule() + target.credit_to = get_party_account("Supplier", source.supplier, source.company) + + def get_billed_qty(po_item_name): + from frappe.query_builder.functions import Sum + + table = frappe.qb.DocType("Purchase Invoice Item") + query = ( + frappe.qb.from_(table) + .select(Sum(table.qty).as_("qty")) + .where((table.docstatus == 1) & (table.po_detail == po_item_name)) + ) + return query.run(pluck="qty")[0] or 0 + + def update_item(obj, target, source_parent): + billed_qty = flt(get_billed_qty(obj.name)) + target.qty = flt(obj.qty) - billed_qty + + item = get_item_defaults(target.item_code, source_parent.company) + item_group = get_item_group_defaults(target.item_code, source_parent.company) + target.cost_center = ( + obj.cost_center + or frappe.db.get_value("Project", obj.project, "cost_center") + or item.get("buying_cost_center") + or item_group.get("buying_cost_center") + ) + + def select_item(d): + filtered_items = args.get("filtered_children", []) + child_filter = d.name in filtered_items if filtered_items else True + return child_filter + + fields = { + "Purchase Order": { + "doctype": "Purchase Invoice", + "field_map": { + "party_account_currency": "party_account_currency", + "supplier_warehouse": "supplier_warehouse", + }, + "field_no_map": ["payment_terms_template"], + "validation": { + "docstatus": ["=", 1], + }, + }, + "Purchase Order Item": { + "doctype": "Purchase Invoice Item", + "field_map": { + "name": "po_detail", + "parent": "purchase_order", + "material_request": "material_request", + "material_request_item": "material_request_item", + "wip_composite_asset": "wip_composite_asset", + }, + "postprocess": update_item, + "condition": lambda doc: ( + doc.base_amount == 0 + or abs(doc.billed_amt) < abs(doc.amount) + or doc.qty > flt(get_billed_qty(doc.name)) + ) + and select_item(doc), + }, + "Purchase Taxes and Charges": {"doctype": "Purchase Taxes and Charges", "reset_value": True}, + } + + doc = get_mapped_doc( + "Purchase Order", + source_name, + fields, + target_doc, + postprocess, + ignore_permissions=ignore_permissions, + ) + + return doc + + +@frappe.whitelist() +def make_inter_company_sales_order(source_name: str, target_doc: str | Document | None = None): + from erpnext.accounts.doctype.sales_invoice.sales_invoice import make_inter_company_transaction + + return make_inter_company_transaction("Purchase Order", source_name, target_doc) + + +@frappe.whitelist() +def make_subcontracting_order( + source_name: str, + target_doc: str | Document | None = None, + save: bool = False, + submit: bool = False, + notify: bool = False, +): + if not is_po_fully_subcontracted(source_name): + target_doc = get_mapped_subcontracting_order(source_name, target_doc) + + if (save or submit) and frappe.has_permission(target_doc.doctype, "create"): + target_doc.save() + + if submit and frappe.has_permission(target_doc.doctype, "submit", target_doc): + try: + target_doc.submit() + except Exception as e: + target_doc.add_comment("Comment", _("Submit Action Failed") + "

" + str(e)) + + if notify: + frappe.msgprint( + _("Subcontracting Order {0} created.").format( + get_link_to_form(target_doc.doctype, target_doc.name) + ), + indicator="green", + alert=True, + ) + + return target_doc + else: + frappe.throw(_("This Purchase Order has been fully subcontracted.")) + + +def is_po_fully_subcontracted(po_name: str) -> bool: + table = frappe.qb.DocType("Purchase Order Item") + query = ( + frappe.qb.from_(table) + .select(table.name) + .where((table.parent == po_name) & (table.qty != table.subcontracted_qty)) + ) + return not query.run(as_dict=True) + + +def get_mapped_subcontracting_order(source_name: str, target_doc: str | Document | None = None) -> Document: + def post_process(source_doc, target_doc): + target_doc.populate_items_table() + + if target_doc.set_warehouse: + for item in target_doc.items: + item.warehouse = target_doc.set_warehouse + else: + if source_doc.set_warehouse: + for item in target_doc.items: + item.warehouse = source_doc.set_warehouse + else: + for idx, item in enumerate(target_doc.items): + item.warehouse = source_doc.items[idx].warehouse + + for idx, item in enumerate(target_doc.items): + item.job_card = source_doc.items[idx].job_card + if not target_doc.supplier_warehouse: + # WIP warehouse is set as Supplier Warehouse in Job Card + target_doc.supplier_warehouse = frappe.get_cached_value( + "Job Card", item.job_card, "wip_warehouse" + ) + + production_plan = set([item.production_plan for item in source_doc.items if item.production_plan]) + if production_plan: + target_doc.production_plan = production_plan.pop() + target_doc.reserve_stock = frappe.get_single_value( + "Stock Settings", "auto_reserve_stock" + ) or frappe.get_value("Production Plan", target_doc.production_plan, "reserve_stock") + + if target_doc and isinstance(target_doc, str): + target_doc = json.loads(target_doc) + for key in ["service_items", "items", "supplied_items"]: + if key in target_doc: + del target_doc[key] + target_doc = json.dumps(target_doc) + + target_doc = get_mapped_doc( + "Purchase Order", + source_name, + { + "Purchase Order": { + "doctype": "Subcontracting Order", + "field_map": {}, + "field_no_map": ["total_qty", "total", "net_total"], + "validation": { + "docstatus": ["=", 1], + }, + }, + "Purchase Order Item": { + "doctype": "Subcontracting Order Service Item", + "field_map": { + "name": "purchase_order_item", + "material_request": "material_request", + "material_request_item": "material_request_item", + }, + "field_no_map": ["qty", "fg_item_qty", "amount"], + "condition": lambda item: item.qty != item.subcontracted_qty, + }, + }, + target_doc, + post_process, + ) + + return target_doc diff --git a/erpnext/buying/doctype/purchase_order/purchase_order.py b/erpnext/buying/doctype/purchase_order/purchase_order.py index 7ff80d31d58..c57666643ff 100644 --- a/erpnext/buying/doctype/purchase_order/purchase_order.py +++ b/erpnext/buying/doctype/purchase_order/purchase_order.py @@ -8,28 +8,34 @@ import frappe from frappe import _, msgprint from frappe.desk.notifications import clear_doctype_notifications from frappe.model.document import Document -from frappe.model.mapper import get_mapped_doc -from frappe.utils import cint, cstr, flt, get_link_to_form +from frappe.utils import cint, cstr, flt from erpnext.accounts.doctype.sales_invoice.sales_invoice import ( unlink_inter_company_doc, update_linked_doc, validate_inter_company_party, ) -from erpnext.accounts.party import get_party_account, get_party_account_currency +from erpnext.accounts.party import get_party_account_currency from erpnext.buying.utils import check_on_hold_or_closed_status, validate_for_items from erpnext.controllers.buying_controller import BuyingController from erpnext.manufacturing.doctype.blanket_order.blanket_order import ( validate_against_blanket_order, ) -from erpnext.setup.doctype.item_group.item_group import get_item_group_defaults -from erpnext.stock.doctype.item.item import get_item_defaults, get_last_purchase_details +from erpnext.stock.doctype.item.item import get_last_purchase_details from erpnext.stock.stock_balance import get_ordered_qty, update_bin_qty from erpnext.stock.utils import get_bin from erpnext.subcontracting.doctype.subcontracting_bom.subcontracting_bom import ( get_subcontracting_boms_for_finished_goods, ) +from .mapper import ( + make_inter_company_sales_order, + make_purchase_invoice, + make_purchase_invoice_from_portal, + make_purchase_receipt, + make_subcontracting_order, +) + form_grid_templates = {"items": "templates/form_grid/item_grid.html"} @@ -737,189 +743,6 @@ def close_or_unclose_purchase_orders(names: str, status: str): frappe.local.message_log = [] -def set_missing_values(source, target): - target.run_method("set_missing_values") - target.run_method("calculate_taxes_and_totals") - target.run_method("set_use_serial_batch_fields") - - -@frappe.whitelist() -def make_purchase_receipt( - source_name: str, target_doc: str | Document | None = None, args: str | dict | None = None -): - if args is None: - args = {} - if isinstance(args, str): - args = json.loads(args) - - has_unit_price_items = frappe.db.get_value("Purchase Order", source_name, "has_unit_price_items") - - def is_unit_price_row(source): - return has_unit_price_items and source.qty == 0 - - def update_item(obj, target, source_parent): - target.qty = flt(obj.qty) if is_unit_price_row(obj) else flt(obj.qty) - flt(obj.received_qty) - target.stock_qty = (flt(obj.qty) - flt(obj.received_qty)) * flt(obj.conversion_factor) - target.amount = (flt(obj.qty) - flt(obj.received_qty)) * flt(obj.rate) - target.base_amount = ( - (flt(obj.qty) - flt(obj.received_qty)) * flt(obj.rate) * flt(source_parent.conversion_rate) - ) - - def select_item(d): - filtered_items = args.get("filtered_children", []) - child_filter = d.name in filtered_items if filtered_items else True - return child_filter - - doc = get_mapped_doc( - "Purchase Order", - source_name, - { - "Purchase Order": { - "doctype": "Purchase Receipt", - "field_map": {"supplier_warehouse": "supplier_warehouse"}, - "validation": { - "docstatus": ["=", 1], - }, - }, - "Purchase Order Item": { - "doctype": "Purchase Receipt Item", - "field_map": { - "name": "purchase_order_item", - "parent": "purchase_order", - "bom": "bom", - "material_request": "material_request", - "material_request_item": "material_request_item", - "sales_order": "sales_order", - "sales_order_item": "sales_order_item", - "wip_composite_asset": "wip_composite_asset", - }, - "postprocess": update_item, - "condition": lambda doc: ( - True if is_unit_price_row(doc) else abs(doc.received_qty) < abs(doc.qty) - ) - and doc.delivered_by_supplier != 1 - and select_item(doc), - }, - "Purchase Taxes and Charges": {"doctype": "Purchase Taxes and Charges", "reset_value": True}, - }, - target_doc, - set_missing_values, - ) - - return doc - - -@frappe.whitelist() -def make_purchase_invoice( - source_name: str, target_doc: str | Document | None = None, args: str | dict | None = None -): - return get_mapped_purchase_invoice(source_name, target_doc, args=args) - - -@frappe.whitelist() -def make_purchase_invoice_from_portal(purchase_order_name: str): - doc = get_mapped_purchase_invoice(purchase_order_name, ignore_permissions=True) - if frappe.session.user not in frappe.get_all("Portal User", {"parent": doc.supplier}, pluck="user"): - frappe.throw(_("Not Permitted"), frappe.PermissionError) - doc.save() - if not frappe.in_test: - frappe.db.commit() - frappe.response["type"] = "redirect" - frappe.response.location = "/purchase-invoices/" + doc.name - - -def get_mapped_purchase_invoice(source_name, target_doc=None, ignore_permissions=False, args=None): - if args is None: - args = {} - if isinstance(args, str): - args = json.loads(args) - - def postprocess(source, target): - target.flags.ignore_permissions = ignore_permissions - set_missing_values(source, target) - - # Get the advance paid Journal Entries in Purchase Invoice Advance - if target.get("allocate_advances_automatically"): - target.set_advances() - - from erpnext.accounts.services.payment_schedule import PaymentScheduleService - - PaymentScheduleService(target).set_payment_schedule() - target.credit_to = get_party_account("Supplier", source.supplier, source.company) - - def get_billed_qty(po_item_name): - from frappe.query_builder.functions import Sum - - table = frappe.qb.DocType("Purchase Invoice Item") - query = ( - frappe.qb.from_(table) - .select(Sum(table.qty).as_("qty")) - .where((table.docstatus == 1) & (table.po_detail == po_item_name)) - ) - return query.run(pluck="qty")[0] or 0 - - def update_item(obj, target, source_parent): - billed_qty = flt(get_billed_qty(obj.name)) - target.qty = flt(obj.qty) - billed_qty - - item = get_item_defaults(target.item_code, source_parent.company) - item_group = get_item_group_defaults(target.item_code, source_parent.company) - target.cost_center = ( - obj.cost_center - or frappe.db.get_value("Project", obj.project, "cost_center") - or item.get("buying_cost_center") - or item_group.get("buying_cost_center") - ) - - def select_item(d): - filtered_items = args.get("filtered_children", []) - child_filter = d.name in filtered_items if filtered_items else True - return child_filter - - fields = { - "Purchase Order": { - "doctype": "Purchase Invoice", - "field_map": { - "party_account_currency": "party_account_currency", - "supplier_warehouse": "supplier_warehouse", - }, - "field_no_map": ["payment_terms_template"], - "validation": { - "docstatus": ["=", 1], - }, - }, - "Purchase Order Item": { - "doctype": "Purchase Invoice Item", - "field_map": { - "name": "po_detail", - "parent": "purchase_order", - "material_request": "material_request", - "material_request_item": "material_request_item", - "wip_composite_asset": "wip_composite_asset", - }, - "postprocess": update_item, - "condition": lambda doc: ( - doc.base_amount == 0 - or abs(doc.billed_amt) < abs(doc.amount) - or doc.qty > flt(get_billed_qty(doc.name)) - ) - and select_item(doc), - }, - "Purchase Taxes and Charges": {"doctype": "Purchase Taxes and Charges", "reset_value": True}, - } - - doc = get_mapped_doc( - "Purchase Order", - source_name, - fields, - target_doc, - postprocess, - ignore_permissions=ignore_permissions, - ) - - return doc - - def get_list_context(context=None): from erpnext.controllers.website_list_for_contact import get_list_context @@ -941,121 +764,3 @@ def update_status(status: str, name: str): po = frappe.get_lazy_doc("Purchase Order", name, check_permission="submit") po.update_status(status) po.update_delivered_qty_in_sales_order() - - -@frappe.whitelist() -def make_inter_company_sales_order(source_name: str, target_doc: str | Document | None = None): - from erpnext.accounts.doctype.sales_invoice.sales_invoice import make_inter_company_transaction - - return make_inter_company_transaction("Purchase Order", source_name, target_doc) - - -@frappe.whitelist() -def make_subcontracting_order( - source_name: str, - target_doc: str | Document | None = None, - save: bool = False, - submit: bool = False, - notify: bool = False, -): - if not is_po_fully_subcontracted(source_name): - target_doc = get_mapped_subcontracting_order(source_name, target_doc) - - if (save or submit) and frappe.has_permission(target_doc.doctype, "create"): - target_doc.save() - - if submit and frappe.has_permission(target_doc.doctype, "submit", target_doc): - try: - target_doc.submit() - except Exception as e: - target_doc.add_comment("Comment", _("Submit Action Failed") + "

" + str(e)) - - if notify: - frappe.msgprint( - _("Subcontracting Order {0} created.").format( - get_link_to_form(target_doc.doctype, target_doc.name) - ), - indicator="green", - alert=True, - ) - - return target_doc - else: - frappe.throw(_("This Purchase Order has been fully subcontracted.")) - - -def is_po_fully_subcontracted(po_name): - table = frappe.qb.DocType("Purchase Order Item") - query = ( - frappe.qb.from_(table) - .select(table.name) - .where((table.parent == po_name) & (table.qty != table.subcontracted_qty)) - ) - return not query.run(as_dict=True) - - -def get_mapped_subcontracting_order(source_name, target_doc=None): - def post_process(source_doc, target_doc): - target_doc.populate_items_table() - - if target_doc.set_warehouse: - for item in target_doc.items: - item.warehouse = target_doc.set_warehouse - else: - if source_doc.set_warehouse: - for item in target_doc.items: - item.warehouse = source_doc.set_warehouse - else: - for idx, item in enumerate(target_doc.items): - item.warehouse = source_doc.items[idx].warehouse - - for idx, item in enumerate(target_doc.items): - item.job_card = source_doc.items[idx].job_card - if not target_doc.supplier_warehouse: - # WIP warehouse is set as Supplier Warehouse in Job Card - target_doc.supplier_warehouse = frappe.get_cached_value( - "Job Card", item.job_card, "wip_warehouse" - ) - - production_plan = set([item.production_plan for item in source_doc.items if item.production_plan]) - if production_plan: - target_doc.production_plan = production_plan.pop() - target_doc.reserve_stock = frappe.get_single_value( - "Stock Settings", "auto_reserve_stock" - ) or frappe.get_value("Production Plan", target_doc.production_plan, "reserve_stock") - - if target_doc and isinstance(target_doc, str): - target_doc = json.loads(target_doc) - for key in ["service_items", "items", "supplied_items"]: - if key in target_doc: - del target_doc[key] - target_doc = json.dumps(target_doc) - - target_doc = get_mapped_doc( - "Purchase Order", - source_name, - { - "Purchase Order": { - "doctype": "Subcontracting Order", - "field_map": {}, - "field_no_map": ["total_qty", "total", "net_total"], - "validation": { - "docstatus": ["=", 1], - }, - }, - "Purchase Order Item": { - "doctype": "Subcontracting Order Service Item", - "field_map": { - "name": "purchase_order_item", - "material_request": "material_request", - "material_request_item": "material_request_item", - }, - "field_no_map": ["qty", "fg_item_qty", "amount"], - "condition": lambda item: item.qty != item.subcontracted_qty, - }, - }, - target_doc, - post_process, - ) - - return target_doc