diff --git a/erpnext/accounts/doctype/pos_invoice/pos_invoice.py b/erpnext/accounts/doctype/pos_invoice/pos_invoice.py index a5437409bcd..e6b1fcb95f8 100644 --- a/erpnext/accounts/doctype/pos_invoice/pos_invoice.py +++ b/erpnext/accounts/doctype/pos_invoice/pos_invoice.py @@ -659,7 +659,6 @@ class POSInvoice(SalesInvoice): def set_pos_fields(self, for_validate=False): """Set retail related fields from POS Profiles""" from erpnext.stock.get_item_details import ( - ItemDetailsCtx, get_pos_profile, get_pos_profile_item_details_, ) @@ -732,7 +731,7 @@ class POSInvoice(SalesInvoice): for item in self.get("items"): if item.get("item_code"): profile_details = get_pos_profile_item_details_( - ItemDetailsCtx(item.as_dict()), profile.get("company"), profile + frappe._dict(item.as_dict()), profile.get("company"), profile ) for fname, val in profile_details.items(): if (not for_validate) or (for_validate and not item.get(fname)): diff --git a/erpnext/accounts/doctype/sales_invoice/services/pos.py b/erpnext/accounts/doctype/sales_invoice/services/pos.py new file mode 100644 index 00000000000..9c7a7c2654c --- /dev/null +++ b/erpnext/accounts/doctype/sales_invoice/services/pos.py @@ -0,0 +1,417 @@ +# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors +# License: GNU General Public License v3. See license.txt + +"""POS helpers for Sales Invoice.""" + +import frappe +from frappe import _ +from frappe.utils import cint, flt, get_link_to_form + + +class PartialPaymentValidationError(frappe.ValidationError): + pass + + +class POSService: + def __init__(self, doc) -> None: + self.doc = doc + + def set_pos_fields(self, for_validate: bool = False) -> frappe.Document | dict | None: + """Populate POS-profile fields on the invoice; return the profile, {} or None.""" + doc = self.doc + if cint(doc.is_pos) != 1: + return None + + self._set_default_change_amount_account() + + if not self._ensure_pos_profile(): + return None + + pos = frappe.get_doc("POS Profile", doc.pos_profile) if doc.pos_profile else {} + if pos: + self._apply_pos_profile(pos, for_validate) + + return pos + + def _set_default_change_amount_account(self) -> None: + doc = self.doc + if not doc.account_for_change_amount: + doc.account_for_change_amount = frappe.get_cached_value( + "Company", doc.company, "default_cash_account" + ) + + def _ensure_pos_profile(self) -> bool: + """Auto-pick a POS Profile for the company; return False if none could be found.""" + doc = self.doc + if doc.pos_profile or doc.flags.ignore_pos_profile: + return True + + from erpnext.stock.get_item_details import get_pos_profile + + pos_profile = get_pos_profile(doc.company) or {} + if not pos_profile: + return False + + doc.pos_profile = pos_profile.get("name") + return True + + def _apply_pos_profile(self, pos, for_validate: bool) -> None: + doc = self.doc + if not for_validate: + self._apply_editable_pos_defaults(pos) + + if pos.get("account_for_change_amount"): + doc.account_for_change_amount = pos.get("account_for_change_amount") + + self._copy_pos_profile_fields(pos, for_validate) + + if pos.get("company_address"): + doc.company_address = pos.get("company_address") + + self._set_selling_price_list(pos) + + if not for_validate: + self._set_update_stock_from_profile(pos) + + self._apply_pos_item_defaults(pos, for_validate) + self._set_terms_and_taxes(pos) + + def _apply_editable_pos_defaults(self, pos) -> None: + """Profile defaults the user may override; only applied outside validation.""" + doc = self.doc + update_multi_mode_option(doc, pos) + doc.tax_category = pos.get("tax_category") + if not doc.customer: + doc.customer = pos.customer + doc.ignore_pricing_rule = pos.ignore_pricing_rule + + def _copy_pos_profile_fields(self, pos, for_validate: bool) -> None: + doc = self.doc + for fieldname in ( + "currency", + "letter_head", + "tc_name", + "company", + "select_print_heading", + "write_off_account", + "taxes_and_charges", + "write_off_cost_center", + "apply_discount_on", + "cost_center", + ): + if (not for_validate) or (for_validate and not doc.get(fieldname)): + doc.set(fieldname, pos.get(fieldname)) + + def _set_selling_price_list(self, pos) -> None: + doc = self.doc + if doc.customer: + customer_price_list, customer_group = frappe.get_value( + "Customer", doc.customer, ["default_price_list", "customer_group"] + ) + customer_group_price_list = frappe.get_value( + "Customer Group", customer_group, "default_price_list" + ) + selling_price_list = ( + customer_price_list or customer_group_price_list or pos.get("selling_price_list") + ) + else: + selling_price_list = pos.get("selling_price_list") + + if selling_price_list: + doc.set("selling_price_list", selling_price_list) + + def _set_update_stock_from_profile(self, pos) -> None: + doc = self.doc + dn_flag = any(d.get("dn_detail") for d in doc.get("items")) + doc.update_stock = 0 if dn_flag else cint(pos.get("update_stock")) + + def _apply_pos_item_defaults(self, pos, for_validate: bool) -> None: + from erpnext.stock.get_item_details import get_pos_profile_item_details_ + + for item in self.doc.get("items"): + if not item.get("item_code"): + continue + profile_details = get_pos_profile_item_details_( + frappe._dict(item.as_dict()), pos, pos, update_data=True + ) + for fname, val in profile_details.items(): + if (not for_validate) or (for_validate and not item.get(fname)): + item.set(fname, val) + + def _set_terms_and_taxes(self, pos) -> None: + doc = self.doc + if doc.tc_name and not doc.terms: + doc.terms = frappe.db.get_value("Terms and Conditions", doc.tc_name, "terms") + + if doc.taxes_and_charges and not len(doc.get("taxes")): + from erpnext.accounts.services.taxes import TaxService + + TaxService(doc).set_taxes() + + def update_paid_amount(self) -> None: + doc = self.doc + paid_amount = 0.0 + base_paid_amount = 0.0 + + if not cint(doc.is_pos) and doc.is_return: + doc.set("payments", []) + doc.paid_amount = paid_amount + doc.base_paid_amount = base_paid_amount + return + + for data in doc.payments: + data.base_amount = flt(data.amount * doc.conversion_rate, doc.precision("base_paid_amount")) + paid_amount += data.amount + base_paid_amount += data.base_amount + doc.paid_amount = paid_amount + doc.base_paid_amount = base_paid_amount + + def set_account_for_mode_of_payment(self) -> None: + for payment in self.doc.payments: + payment.account = get_bank_cash_account(payment.mode_of_payment, self.doc.company).get("account") + + def reset_mode_of_payments(self) -> None: + doc = self.doc + if doc.pos_profile: + pos_profile = frappe.get_cached_doc("POS Profile", doc.pos_profile) + update_multi_mode_option(doc, pos_profile) + doc.paid_amount = 0 + + def validate_pos_return(self) -> None: + """Ensure POS return payments are not less than the (negative) invoice total.""" + doc = self.doc + if doc.is_consolidated: + return + + if doc.is_pos and doc.is_return: + total_amount_in_payments = sum(payment.amount for payment in doc.payments) + invoice_total = doc.rounded_total or doc.grand_total + if total_amount_in_payments < invoice_total: + frappe.throw(_("Total payments amount can't be greater than {0}").format(-invoice_total)) + + def validate_pos_paid_amount(self) -> None: + doc = self.doc + if len(doc.payments) == 0 and doc.is_pos and flt(doc.grand_total) > 0: + frappe.throw(_("At least one mode of payment is required for POS invoice.")) + + def validate_pos(self) -> None: + """On a POS return, paid amount plus write-off cannot exceed the grand total.""" + doc = self.doc + if doc.is_return: + invoice_total = doc.rounded_total or doc.grand_total + if abs(flt(doc.paid_amount)) + abs(flt(doc.write_off_amount)) - abs(flt(invoice_total)) > 1.0 / ( + 10.0 ** (doc.precision("grand_total") + 1.0) + ): + frappe.throw(_("Paid amount + Write Off Amount can not be greater than Grand Total")) + + def validate_created_using_pos(self) -> None: + doc = self.doc + if doc.is_created_using_pos and not doc.pos_profile: + frappe.throw(_("POS Profile is mandatory to mark this invoice as POS Transaction.")) + + doc.invoice_type_in_pos = frappe.db.get_single_value("POS Settings", "invoice_type") + if doc.invoice_type_in_pos == "POS Invoice" and not doc.is_return: + frappe.throw(_("Transactions using Sales Invoice in POS are disabled.")) + + self.validate_pos_opening_entry() + + def validate_full_payment(self) -> None: + """Block partial payment on a submitted POS invoice unless the profile allows it.""" + doc = self.doc + allow_partial_payment = frappe.db.get_value("POS Profile", doc.pos_profile, "allow_partial_payment") + invoice_total = flt(doc.rounded_total) or flt(doc.grand_total) + + if ( + doc.docstatus == 1 + and not doc.is_return + and not allow_partial_payment + and doc.paid_amount < invoice_total + ): + frappe.throw( + msg=_("Partial Payment in POS Transactions are not allowed."), + exc=PartialPaymentValidationError, + ) + + def validate_pos_opening_entry(self) -> None: + """Require exactly one current, open POS Opening Entry for the profile.""" + doc = self.doc + opening_entries = frappe.get_all( + "POS Opening Entry", + fields=["name", "period_start_date"], + filters={"pos_profile": doc.pos_profile, "status": "Open"}, + order_by="period_start_date desc", + ) + if not opening_entries: + frappe.throw( + title=_("POS Opening Entry Missing"), + msg=_("No open POS Opening Entry found for POS Profile {0}.").format( + frappe.bold(doc.pos_profile) + ), + ) + if len(opening_entries) > 1: + frappe.throw( + title=_("Multiple POS Opening Entry"), + msg=_( + "POS Profile - {0} has multiple open POS Opening Entries. Please close or cancel the existing entries before proceeding." + ).format(doc.pos_profile), + ) + if frappe.utils.get_date_str(opening_entries[0].get("period_start_date")) != frappe.utils.today(): + frappe.throw( + title=_("Outdated POS Opening Entry"), + msg=_( + "POS Opening Entry - {0} is outdated. Please close the POS and create a new POS Opening Entry." + ).format(opening_entries[0].get("name")), + ) + + def check_if_consolidated_invoice(self) -> None: + doc = self.doc + if doc.doctype == "Sales Invoice" and doc.is_consolidated: + invoice_or_credit_note = "consolidated_credit_note" if doc.is_return else "consolidated_invoice" + pos_closing_entry = frappe.get_all( + "POS Invoice Merge Log", + filters={invoice_or_credit_note: doc.name}, + pluck="pos_closing_entry", + ) + if pos_closing_entry and pos_closing_entry[0]: + msg = _("To cancel a {0} you need to cancel the POS Closing Entry {1}.").format( + frappe.bold(_("Consolidated Sales Invoice")), + get_link_to_form("POS Closing Entry", pos_closing_entry[0]), + ) + frappe.throw(msg, title=_("Not Allowed")) + + def check_if_created_using_pos_and_pos_closing_entry_generated(self) -> None: + doc = self.doc + if doc.doctype == "Sales Invoice" and doc.is_created_using_pos and doc.pos_closing_entry: + pos_closing_entry_docstatus = frappe.db.get_value( + "POS Closing Entry", doc.pos_closing_entry, "docstatus" + ) + if pos_closing_entry_docstatus == 1: + frappe.throw( + msg=_( + "To cancel this Sales Invoice you need to cancel the POS Closing Entry {0}." + ).format(get_link_to_form("POS Closing Entry", doc.pos_closing_entry)), + title=_("Not Allowed"), + ) + + def cancel_pos_invoice_credit_note_generated_during_sales_invoice_mode(self) -> None: + pos_invoices = frappe.get_all( + "POS Invoice", filters={"consolidated_invoice": self.doc.name}, pluck="name" + ) + for pos_invoice in pos_invoices: + frappe.get_doc("POS Invoice", pos_invoice).cancel() + + def clear_unallocated_mode_of_payments(self) -> None: + doc = self.doc + doc.set("payments", doc.get("payments", {"amount": ["not in", [0, None, ""]]})) + frappe.db.delete("Sales Invoice Payment", filters={"parent": doc.name, "amount": 0}) + + def allow_write_off_only_on_pos(self) -> None: + if not self.doc.is_pos and self.doc.write_off_account: + self.doc.write_off_account = None + + def verify_payment_amount_is_positive(self) -> None: + for entry in self.doc.payments: + if entry.amount < 0: + frappe.throw(_("Row #{0} (Payment Table): Amount must be positive").format(entry.idx)) + + def verify_payment_amount_is_negative(self) -> None: + for entry in self.doc.payments: + if entry.amount > 0: + frappe.throw(_("Row #{0} (Payment Table): Amount must be negative").format(entry.idx)) + + +def get_bank_cash_account(mode_of_payment: str, company: str) -> dict: + account = frappe.db.get_value( + "Mode of Payment Account", + {"parent": mode_of_payment, "company": company}, + "default_account", + ) + if not account: + frappe.throw( + _("Please set default Cash or Bank account in Mode of Payment {0}").format( + get_link_to_form("Mode of Payment", mode_of_payment) + ), + title=_("Missing Account"), + ) + return {"account": account} + + +def update_multi_mode_option(doc, pos_profile) -> None: + def append_payment(payment_mode): + payment = doc.append("payments", {}) + payment.default = payment_mode.default + payment.mode_of_payment = payment_mode.mop + payment.account = payment_mode.default_account + payment.type = payment_mode.type + + mop_refetched = bool(doc.payments) and not doc.is_created_using_pos + + doc.set("payments", []) + invalid_modes = [] + mode_of_payments = [d.mode_of_payment for d in pos_profile.get("payments")] + mode_of_payments_info = get_mode_of_payments_info(mode_of_payments, doc.company) + + for row in pos_profile.get("payments"): + payment_mode = mode_of_payments_info.get(row.mode_of_payment) + if not payment_mode: + invalid_modes.append(get_link_to_form("Mode of Payment", row.mode_of_payment)) + continue + + payment_mode.default = row.default + append_payment(payment_mode) + + if invalid_modes: + if invalid_modes == 1: + msg = _("Please set default Cash or Bank account in Mode of Payment {0}") + else: + msg = _("Please set default Cash or Bank account in Mode of Payments {0}") + frappe.throw(msg.format(", ".join(invalid_modes)), title=_("Missing Account")) + + if mop_refetched: + frappe.toast( + _("Payment methods refreshed. Please review before proceeding."), + indicator="orange", + ) + + +def get_all_mode_of_payments(doc) -> list: + """All enabled modes of payment with their default accounts for the doc's company.""" + query, mopa, mop = _enabled_mode_of_payment_query(doc.company) + return query.select(mopa.default_account, mopa.parent, mop.type.as_("type")).run(as_dict=1) + + +def get_mode_of_payments_info(mode_of_payments: list, company: str) -> dict: + """Map each of the named modes of payment to its account info for the company.""" + query, mopa, mop = _enabled_mode_of_payment_query(company) + data = ( + query.select(mopa.default_account, mopa.parent.as_("mop"), mop.type.as_("type")) + .where(mop.name.isin(mode_of_payments)) + # group by all selected columns so postgres accepts it (one row per mode of payment) + .groupby(mopa.default_account, mopa.parent, mop.type) + .run(as_dict=1) + ) + return {row.get("mop"): row for row in data} + + +def get_mode_of_payment_info(mode_of_payment: str, company: str) -> list: + """Account info for a single mode of payment in the company.""" + query, mopa, mop = _enabled_mode_of_payment_query(company) + return ( + query.select(mopa.default_account, mopa.parent, mop.type.as_("type")) + .where(mop.name == mode_of_payment) + .run(as_dict=1) + ) + + +def _enabled_mode_of_payment_query(company: str): + """Base query joining enabled modes of payment to their accounts for a company.""" + mopa = frappe.qb.DocType("Mode of Payment Account") + mop = frappe.qb.DocType("Mode of Payment") + query = ( + frappe.qb.from_(mopa) + .join(mop) + .on(mopa.parent == mop.name) + .where(mopa.company == company) + .where(mop.enabled == 1) + ) + return query, mopa, mop diff --git a/erpnext/accounts/services/taxes.py b/erpnext/accounts/services/taxes.py new file mode 100644 index 00000000000..4762b520885 --- /dev/null +++ b/erpnext/accounts/services/taxes.py @@ -0,0 +1,446 @@ +# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors +# License: GNU General Public License v3. See license.txt + +"""Tax helpers: TaxService class for doc-mutating operations, free functions for stateless utilities.""" + +import json + +import frappe +from frappe import _, throw +from frappe.utils import cint, flt, parse_json + +import erpnext +from erpnext.stock.get_item_details import ( + NOT_APPLICABLE_TAX, + _get_item_tax_template, + _get_item_tax_template_from_item_group, + get_item_tax_map, +) + + +class TaxService: + def __init__(self, doc): + self.doc = doc + + def set_taxes(self) -> None: + doc = self.doc + if not doc.meta.get_field("taxes"): + return + + tax_master_doctype = doc.meta.get_field("taxes_and_charges").options + + if (doc.is_new() or self.is_pos_profile_changed()) and not doc.get("taxes"): + if doc.company and not doc.get("taxes_and_charges"): + doc.taxes_and_charges = frappe.db.get_value( + tax_master_doctype, {"is_default": 1, "company": doc.company} + ) + self.append_taxes_from_master(tax_master_doctype) + + def is_pos_profile_changed(self) -> bool: + doc = self.doc + if ( + doc.doctype == "Sales Invoice" + and doc.is_pos + and doc.pos_profile != frappe.db.get_value("Sales Invoice", doc.name, "pos_profile") + ): + return True + + def set_taxes_and_charges(self) -> None: + doc = self.doc + if doc.doctype == "Material Request": + return + + if doc.get("taxes") or doc.get("is_pos"): + return + + if frappe.get_single_value( + "Accounts Settings", "add_taxes_from_taxes_and_charges_template" + ) and hasattr(doc, "taxes_and_charges"): + if tax_master_doctype := doc.meta.get_field("taxes_and_charges").options: + self.append_taxes_from_master(tax_master_doctype) + + if frappe.get_single_value("Accounts Settings", "add_taxes_from_item_tax_template"): + self.append_taxes_from_item_tax_template() + + def append_taxes_from_master(self, tax_master_doctype=None) -> None: + doc = self.doc + if doc.get("taxes_and_charges"): + if not tax_master_doctype: + tax_master_doctype = doc.meta.get_field("taxes_and_charges").options + doc.extend("taxes", get_taxes_and_charges(tax_master_doctype, doc.get("taxes_and_charges"))) + + def append_taxes_from_item_tax_template(self) -> None: + doc = self.doc + if not frappe.get_single_value("Accounts Settings", "add_taxes_from_item_tax_template"): + return + + for row in doc.items: + item_tax_rate = row.get("item_tax_rate") + if not item_tax_rate: + continue + + if isinstance(item_tax_rate, str): + item_tax_rate = parse_json(item_tax_rate) + + for account_head, _rate in item_tax_rate.items(): + if not self.get_tax_row(account_head): + doc.append( + "taxes", + { + "charge_type": "On Net Total", + "account_head": account_head, + "rate": 0, + "description": account_head, + "set_by_item_tax_template": 1, + "category": "Total", + "add_deduct_tax": "Add", + }, + ) + + def get_tax_row(self, account_head): + for row in self.doc.taxes: + if row.account_head == account_head: + return row + + def set_other_charges(self) -> None: + self.doc.set("taxes", []) + self.set_taxes() + + def validate_enabled_taxes_and_charges(self) -> None: + doc = self.doc + taxes_and_charges_doctype = doc.meta.get_options("taxes_and_charges") + if doc.taxes_and_charges and frappe.get_cached_value( + taxes_and_charges_doctype, doc.taxes_and_charges, "disabled" + ): + frappe.throw(_("{0} '{1}' is disabled").format(taxes_and_charges_doctype, doc.taxes_and_charges)) + + def validate_tax_account_company(self) -> None: + doc = self.doc + for d in doc.get("taxes"): + if d.account_head: + tax_account_company = frappe.get_cached_value("Account", d.account_head, "company") + if tax_account_company != doc.company: + frappe.throw( + _("Row #{0}: Account {1} does not belong to company {2}").format( + d.idx, d.account_head, doc.company + ) + ) + + def get_tax_map(self) -> dict: + tax_map = {} + for tax in self.doc.get("taxes"): + tax_map.setdefault(tax.account_head, 0.0) + tax_map[tax.account_head] += tax.tax_amount + return tax_map + + def get_amount_and_base_amount(self, item, enable_discount_accounting): + doc = self.doc + amount = item.net_amount + base_amount = item.base_net_amount + + if ( + enable_discount_accounting + and doc.get("discount_amount") + and doc.get("additional_discount_account") + ): + if not hasattr(doc, "__has_distributed_discount_set"): + doc.__has_distributed_discount_set = any( + i.distributed_discount_amount for i in doc.get("items") + ) + + if not doc.__has_distributed_discount_set: + return item.amount, item.base_amount + + amount += item.distributed_discount_amount + base_amount += flt( + item.distributed_discount_amount * doc.get("conversion_rate"), + item.precision("distributed_discount_amount"), + ) + + return amount, base_amount + + def get_tax_amounts(self, tax, enable_discount_accounting): + doc = self.doc + amount = tax.tax_amount_after_discount_amount + base_amount = tax.base_tax_amount_after_discount_amount + + if ( + enable_discount_accounting + and doc.get("discount_amount") + and doc.get("additional_discount_account") + and doc.get("apply_discount_on") == "Grand Total" + ): + amount = tax.tax_amount + base_amount = tax.base_tax_amount + + return amount, base_amount + + +@frappe.whitelist() +def get_tax_rate(account_head: str) -> dict: + return frappe.get_cached_value("Account", account_head, ["tax_rate", "account_name"], as_dict=True) + + +@frappe.whitelist() +def get_default_taxes_and_charges( + master_doctype: str, tax_template: str | None = None, company: str | None = None +) -> dict | None: + if not company: + return {} + + if tax_template and company: + tax_template_company = frappe.get_cached_value(master_doctype, tax_template, "company") + if tax_template_company == company: + return + + default_tax = frappe.db.get_value(master_doctype, {"is_default": 1, "company": company}) + + return { + "taxes_and_charges": default_tax, + "taxes": get_taxes_and_charges(master_doctype, default_tax), + } + + +@frappe.whitelist() +def get_taxes_and_charges(master_doctype: str, master_name: str | None = None) -> list | None: + if not master_name: + return + from frappe.model import child_table_fields, default_fields + + tax_master = frappe.get_doc(master_doctype, master_name) + + taxes_and_charges = [] + for _i, tax in enumerate(tax_master.get("taxes")): + tax = tax.as_dict() + + for fieldname in default_fields + child_table_fields: + if fieldname in tax: + del tax[fieldname] + + taxes_and_charges.append(tax) + + return taxes_and_charges + + +def validate_conversion_rate( + currency: str, conversion_rate: float, conversion_rate_label: str, company: str +) -> None: + """Throw a validation error if conversion_rate is falsy.""" + company_currency = frappe.get_cached_value("Company", company, "default_currency") + + if not conversion_rate: + throw( + _("{0} is mandatory. Maybe Currency Exchange record is not created for {1} to {2}.").format( + conversion_rate_label, currency, company_currency + ) + ) + + +def validate_taxes_and_charges(tax) -> None: + if tax.charge_type in ["Actual", "On Net Total", "On Paid Amount"] and tax.row_id: + frappe.throw( + _("Can refer row only if the charge type is 'On Previous Row Amount' or 'Previous Row Total'") + ) + elif tax.charge_type in ["On Previous Row Amount", "On Previous Row Total"]: + if cint(tax.idx) == 1: + frappe.throw( + _( + "Cannot select charge type as 'On Previous Row Amount' or 'On Previous Row Total' for first row" + ) + ) + elif not tax.row_id: + frappe.throw( + _("Please specify a valid Row ID for row {0} in table {1}").format(tax.idx, _(tax.doctype)) + ) + elif tax.row_id and cint(tax.row_id) >= cint(tax.idx): + frappe.throw( + _("Cannot refer row number greater than or equal to current row number for this Charge type") + ) + + if tax.charge_type == "Actual": + tax.rate = None + + +def validate_account_head(idx: int, account: str, company: str, context: str | None = None) -> None: + """Throw a ValidationError if the account belongs to a different company or is a group account.""" + if company != frappe.get_cached_value("Account", account, "company"): + frappe.throw( + _("Row {0}: The {3} Account {1} does not belong to the company {2}").format( + idx, frappe.bold(account), frappe.bold(company), context or "" + ), + title=_("Invalid Account"), + ) + + if frappe.get_cached_value("Account", account, "is_group"): + frappe.throw( + _( + "You selected the account group {1} as {2} Account in row {0}. Please select a single account." + ).format(idx, frappe.bold(account), context or ""), + title=_("Invalid Account"), + ) + + +def validate_cost_center(tax, doc) -> None: + if not tax.cost_center: + return + + company = frappe.get_cached_value("Cost Center", tax.cost_center, "company") + + if company != doc.company: + frappe.throw( + _("Row {0}: Cost Center {1} does not belong to Company {2}").format( + tax.idx, frappe.bold(tax.cost_center), frappe.bold(doc.company) + ), + title=_("Invalid Cost Center"), + ) + + +def validate_inclusive_tax(tax, doc) -> None: + 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_print_rate", None)): + if tax.charge_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_print_rate + ): + _on_previous_row_error(tax.row_id) + elif tax.charge_type == "On Previous Row Total" and not all( + [cint(t.included_in_print_rate) for t in doc.get("taxes")[: cint(tax.row_id) - 1]] + ): + _on_previous_row_error("1 - %d" % (tax.row_id,)) + elif tax.get("category") == "Valuation": + frappe.throw(_("Valuation type charges can not be marked as Inclusive")) + + +def set_balance_in_account_currency( + gl_dict, + account_currency: str | None = None, + conversion_rate: float | None = None, + company_currency: str | None = None, +) -> None: + if (not conversion_rate) and (account_currency != company_currency): + frappe.throw( + _("Account: {0} with currency: {1} can not be selected").format(gl_dict.account, account_currency) + ) + + gl_dict["account_currency"] = account_currency + + if flt(gl_dict.debit) and not flt(gl_dict.debit_in_account_currency): + gl_dict.debit_in_account_currency = ( + gl_dict.debit if account_currency == company_currency else flt(gl_dict.debit / conversion_rate, 2) + ) + + if flt(gl_dict.credit) and not flt(gl_dict.credit_in_account_currency): + gl_dict.credit_in_account_currency = ( + gl_dict.credit + if account_currency == company_currency + else flt(gl_dict.credit / conversion_rate, 2) + ) + + +def set_child_tax_template_and_map(item, child_item, parent_doc) -> None: + ctx = frappe._dict( + { + "item_code": item.item_code, + "posting_date": parent_doc.transaction_date, + "tax_category": parent_doc.get("tax_category"), + "company": parent_doc.get("company"), + "base_net_rate": item.get("base_net_rate"), + } + ) + + item_tax_template = _get_item_tax_template(ctx, item.taxes) + + if not item_tax_template: + item_tax_template = _get_item_tax_template_from_item_group(ctx, item.item_group) + + child_item.item_tax_template = item_tax_template + child_item.item_tax_rate = get_item_tax_map( + doc=parent_doc, + tax_template=child_item.item_tax_template, + as_json=True, + ) + + +def add_taxes_from_tax_template(child_item, parent_doc, db_insert: bool = True) -> None: + add_taxes_from_item_tax_template = frappe.get_single_value( + "Accounts Settings", "add_taxes_from_item_tax_template" + ) + + if child_item.get("item_tax_rate") and add_taxes_from_item_tax_template: + tax_map = json.loads(child_item.get("item_tax_rate")) + for tax_type, tax_rate in tax_map.items(): + if tax_rate == NOT_APPLICABLE_TAX: + continue + + tax_rate = flt(tax_rate) + taxes = parent_doc.get("taxes") or [] + found = any(tax.account_head == tax_type for tax in taxes) + if not found: + tax_row = parent_doc.append("taxes", {}) + tax_row.update( + { + "description": str(tax_type).split(" - ")[0], + "charge_type": "On Net Total", + "account_head": tax_type, + "rate": tax_rate, + "set_by_item_tax_template": 1, + } + ) + if parent_doc.doctype == "Purchase Order": + tax_row.update({"category": "Total", "add_deduct_tax": "Add"}) + if db_insert: + tax_row.db_insert() + + +def merge_taxes(source_doc, target_doc) -> None: + tax_map = {} + for tax in source_doc.get("taxes") or []: + found = False + for t in target_doc.get("taxes") or []: + if t.account_head == tax.account_head and t.cost_center == tax.cost_center: + t.tax_amount = flt(t.tax_amount) + flt(tax.tax_amount_after_discount_amount) + t.base_tax_amount = flt(t.base_tax_amount) + flt(tax.base_tax_amount_after_discount_amount) + tax_map[tax.name] = t + found = True + + if not found: + tax.charge_type = "Actual" + tax.included_in_print_rate = 0 + tax.dont_recompute_tax = 1 + tax.row_id = None + tax.idx = None + tax.tax_amount = tax.tax_amount_after_discount_amount + tax.base_tax_amount = tax.base_tax_amount_after_discount_amount + tax_map[tax.name] = target_doc.append("taxes", tax) + + item_map = {d._old_name: d for d in target_doc.get("items") if d.get("_old_name")} + + item_tax_details = target_doc.get("_item_wise_tax_details") or [] + for row in source_doc.get("item_wise_tax_details"): + item = item_map.get(row.item_row) + tax = tax_map.get(row.tax_row) + if not (item and tax): + continue + + item_tax_details.append( + frappe._dict( + item=item, + tax=tax, + amount=row.amount, + rate=row.rate, + taxable_amount=row.taxable_amount, + ) + ) + + target_doc._item_wise_tax_details = item_tax_details diff --git a/erpnext/assets/doctype/asset_capitalization/asset_capitalization.py b/erpnext/assets/doctype/asset_capitalization/asset_capitalization.py index 5a7b9995f39..710a7adc89e 100644 --- a/erpnext/assets/doctype/asset_capitalization/asset_capitalization.py +++ b/erpnext/assets/doctype/asset_capitalization/asset_capitalization.py @@ -657,7 +657,7 @@ def get_target_item_details(item_code: str | None = None, company: str | None = item_group_defaults = get_item_group_defaults(item.name, company) brand_defaults = get_brand_defaults(item.name, company) out.cost_center = get_default_cost_center( - ItemDetailsCtx({"item_code": item.name, "company": company}), + frappe._dict({"item_code": item.name, "company": company}), item_defaults, item_group_defaults, brand_defaults, diff --git a/erpnext/buying/doctype/purchase_order/test_purchase_order.py b/erpnext/buying/doctype/purchase_order/test_purchase_order.py index 409e02f9eda..196cc1fe8af 100644 --- a/erpnext/buying/doctype/purchase_order/test_purchase_order.py +++ b/erpnext/buying/doctype/purchase_order/test_purchase_order.py @@ -1564,6 +1564,91 @@ class TestPurchaseOrder(ERPNextTestSuite): pi2 = make_pi_from_po(po.name) self.assertEqual(len(pi2.items), 2) +<<<<<<< HEAD +======= + def test_get_item_details_propagates_drop_ship_flag_to_po(self): + """`get_item_details` should propagate the Item master's + `delivered_by_supplier` flag to Purchase Orders, not only to Sales + Orders/Invoices, so that POs can be created as drop-ship directly + (via the standard item lookup the form uses) without going through + the Sales Order → Purchase Order mapping pipeline. + """ + from erpnext.stock.get_item_details import get_item_details + + item = make_item("_Test Drop Ship From Master", {"is_stock_item": 1, "delivered_by_supplier": 1}) + + ctx = frappe._dict( + { + "item_code": item.item_code, + "doctype": "Purchase Order", + "company": "_Test Company", + "supplier": "_Test Supplier", + "transaction_date": nowdate(), + "currency": "INR", + "conversion_rate": 1.0, + "buying_price_list": "Standard Buying", + "price_list_currency": "INR", + "plc_conversion_rate": 1.0, + "qty": 1, + } + ) + + details = get_item_details(ctx, frappe.new_doc("Purchase Order")) + self.assertEqual(details.get("delivered_by_supplier"), 1) + + def test_drop_ship_po_allows_non_company_shipping_address_without_so(self): + """A PO with a drop-ship item should save with a non-company shipping + address even when there is no linked Sales Order. + Regression test for https://github.com/frappe/erpnext/issues/51629. + """ + from erpnext.crm.doctype.prospect.test_prospect import make_address + + item = make_item("_Test Drop Ship Direct PO", {"is_stock_item": 1, "delivered_by_supplier": 1}) + + customer_shipping = make_address( + address_title="Drop Ship Direct PO", address_type="Shipping", address_line1="1" + ) + customer_shipping.append("links", {"link_doctype": "Customer", "link_name": "_Test Customer"}) + customer_shipping.save() + + po = create_purchase_order(item=item.item_code, qty=1, do_not_save=True) + # In the UI, `get_item_details` propagates the master flag to the row when + # the item is added; here we simulate that step explicitly. + po.items[0].delivered_by_supplier = 1 + po.items[0].warehouse = "" + po.shipping_address = customer_shipping.name + po.save() + + self.assertEqual(po.items[0].delivered_by_supplier, 1) + self.assertFalse(po.items[0].warehouse) + self.assertEqual(po.shipping_address, customer_shipping.name) + + def test_drop_ship_flag_overridable_per_po_line(self): + """The drop-ship default from the Item master should be overridable + on individual PO lines (e.g. ordering a normally drop-shipped item + into the own warehouse for samples or stock). + """ + item = make_item("_Test Drop Ship Override", {"is_stock_item": 1, "delivered_by_supplier": 1}) + + po = create_purchase_order(item=item.item_code, qty=1, do_not_save=True) + po.items[0].delivered_by_supplier = 0 + po.save() + + self.assertEqual(po.items[0].delivered_by_supplier, 0) + self.assertEqual(po.items[0].warehouse, "_Test Warehouse - _TC") + + def test_remove_unlinked_item_from_mixed_po_does_not_crash(self): + """In a PO that mixes SO-linked and freely-added items, removing an + item that has no `sales_order_item` via Update Items must not crash + on the missing reference. + """ + po = create_purchase_order(do_not_submit=True) + # Force the SO codepath without needing a real linked Sales Order: + po.items[0].sales_order = "DUMMY-SO" + + po.update_ordered_qty_in_so_for_removed_items([frappe._dict({"sales_order_item": None, "qty": 1})]) + +>>>>>>> e6f8f8f7e9 (refactor: use frappe._dict in importers of ItemDetailsCtx) def create_po_for_sc_testing(): from erpnext.controllers.tests.test_subcontracting_controller import ( diff --git a/erpnext/controllers/accounts_controller.py b/erpnext/controllers/accounts_controller.py index c626f92bffb..10c5b90add4 100644 --- a/erpnext/controllers/accounts_controller.py +++ b/erpnext/controllers/accounts_controller.py @@ -67,12 +67,15 @@ from erpnext.setup.utils import get_exchange_rate from erpnext.stock.doctype.item.item import get_uom_conv_factor from erpnext.stock.doctype.packed_item.packed_item import make_packing_list from erpnext.stock.get_item_details import ( +<<<<<<< HEAD NOT_APPLICABLE_TAX, ItemDetailsCtx, _get_item_tax_template, _get_item_tax_template_from_item_group, get_bin_details, get_conversion_factor, +======= +>>>>>>> e6f8f8f7e9 (refactor: use frappe._dict in importers of ItemDetailsCtx) get_item_details, get_item_tax_map, get_item_warehouse_, @@ -1087,7 +1090,7 @@ class AccountsController(TransactionBase): for item in self.get("items"): if item.get("item_code"): - ctx: ItemDetailsCtx = ItemDetailsCtx(parent_dict.copy()) + ctx: frappe._dict = frappe._dict(parent_dict.copy()) ctx.update(item.as_dict()) ctx.update( diff --git a/erpnext/controllers/queries.py b/erpnext/controllers/queries.py index ad8f069e1f1..81d41cd2d52 100644 --- a/erpnext/controllers/queries.py +++ b/erpnext/controllers/queries.py @@ -16,7 +16,7 @@ from pypika import Order import erpnext from erpnext.accounts.utils import build_qb_match_conditions -from erpnext.stock.get_item_details import ItemDetailsCtx, _get_item_tax_template +from erpnext.stock.get_item_details import _get_item_tax_template from erpnext.stock.utils import get_combine_datetime @@ -949,7 +949,7 @@ def get_tax_template(doctype, txt, searchfield, start, page_len, filters): valid_from = filters.get("valid_from") valid_from = valid_from[1] if isinstance(valid_from, list) else valid_from - ctx = ItemDetailsCtx( + ctx = frappe._dict( { "item_code": filters.get("item_code"), "posting_date": valid_from, diff --git a/erpnext/controllers/taxes_and_totals.py b/erpnext/controllers/taxes_and_totals.py index e50a34e2f14..d20e4e1e398 100644 --- a/erpnext/controllers/taxes_and_totals.py +++ b/erpnext/controllers/taxes_and_totals.py @@ -22,7 +22,6 @@ from erpnext.controllers.accounts_controller import ( from erpnext.deprecation_dumpster import deprecated from erpnext.stock.get_item_details import ( NOT_APPLICABLE_TAX, - ItemDetailsCtx, _get_item_tax_template, get_item_tax_map, ) @@ -100,7 +99,7 @@ class calculate_taxes_and_totals: for item in self.doc.items: if item.item_code and item.get("item_tax_template"): item_doc = frappe.get_cached_doc("Item", item.item_code) - ctx = ItemDetailsCtx( + ctx = frappe._dict( { "net_rate": item.net_rate or item.rate, "base_net_rate": item.base_net_rate or item.base_rate, diff --git a/erpnext/manufacturing/doctype/bom/bom.py b/erpnext/manufacturing/doctype/bom/bom.py index aa010590bc5..0254537cae0 100644 --- a/erpnext/manufacturing/doctype/bom/bom.py +++ b/erpnext/manufacturing/doctype/bom/bom.py @@ -18,7 +18,7 @@ from frappe.website.website_generator import WebsiteGenerator import erpnext from erpnext.setup.utils import get_exchange_rate from erpnext.stock.doctype.item.item import get_item_details -from erpnext.stock.get_item_details import ItemDetailsCtx, get_conversion_factor, get_price_list_rate +from erpnext.stock.get_item_details import get_conversion_factor, get_price_list_rate form_grid_templates = {"items": "templates/form_grid/item_grid.html"} @@ -1335,6 +1335,35 @@ def get_bom_item_rate(args, bom_doc): return flt(rate) +<<<<<<< HEAD +======= +def _get_price_list_item_rate(args, bom_doc): + if not bom_doc.buying_price_list: + frappe.throw(_("Please select Price List")) + + ctx = frappe._dict( + { + "doctype": "BOM", + "price_list": bom_doc.buying_price_list, + "qty": args.get("qty") or 1, + "uom": args.get("uom") or args.get("stock_uom"), + "stock_uom": args.get("stock_uom"), + "transaction_type": "buying", + "company": bom_doc.company, + "currency": bom_doc.currency, + "conversion_rate": 1, # Passed conversion rate as 1 purposefully, as conversion rate is applied at the end of the function + "conversion_factor": args.get("conversion_factor") or 1, + "plc_conversion_rate": 1, + "ignore_party": True, + "ignore_conversion_rate": True, + } + ) + item_doc = frappe.get_cached_doc("Item", args.get("item_code")) + price_list_data = get_price_list_rate(ctx, item_doc) + return price_list_data.price_list_rate + + +>>>>>>> e6f8f8f7e9 (refactor: use frappe._dict in importers of ItemDetailsCtx) def get_valuation_rate(data): """ 1) Get average valuation rate from all warehouses diff --git a/erpnext/selling/doctype/quotation/test_quotation.py b/erpnext/selling/doctype/quotation/test_quotation.py index 71688bcb2f8..f6bc1b059f8 100644 --- a/erpnext/selling/doctype/quotation/test_quotation.py +++ b/erpnext/selling/doctype/quotation/test_quotation.py @@ -290,7 +290,7 @@ class TestQuotation(ERPNextTestSuite): def test_gross_profit(self): from erpnext.stock.doctype.item.test_item import make_item from erpnext.stock.doctype.stock_entry.stock_entry_utils import make_stock_entry - from erpnext.stock.get_item_details import ItemDetailsCtx, insert_item_price + from erpnext.stock.get_item_details import insert_item_price item_doc = make_item("_Test Item for Gross Profit", {"is_stock_item": 1}) item_code = item_doc.name @@ -299,7 +299,7 @@ class TestQuotation(ERPNextTestSuite): selling_price_list = frappe.get_all("Price List", filters={"selling": 1}, limit=1)[0].name frappe.db.set_single_value("Stock Settings", "auto_insert_price_list_rate_if_missing", 1) insert_item_price( - ItemDetailsCtx( + frappe._dict( { "item_code": item_code, "price_list": selling_price_list, diff --git a/erpnext/selling/doctype/sales_order/mapper.py b/erpnext/selling/doctype/sales_order/mapper.py new file mode 100644 index 00000000000..5cd85b3bd6a --- /dev/null +++ b/erpnext/selling/doctype/sales_order/mapper.py @@ -0,0 +1,1120 @@ +# 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.contacts.doctype.address.address import get_company_address +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.functions import Sum +from frappe.utils import add_days, cint, flt, nowdate, strip_html + +from erpnext.accounts.party import CROSS_PARTY_FIELD_NO_MAP, get_party_account +from erpnext.manufacturing.doctype.production_plan.production_plan import ( + get_items_for_material_requests, + get_sales_orders, +) +from erpnext.selling.doctype.product_bundle.product_bundle import get_active_product_bundle +from erpnext.setup.doctype.item_group.item_group import get_item_group_defaults +from erpnext.stock.doctype.item.item import get_item_defaults +from erpnext.stock.doctype.packed_item.packed_item import is_product_bundle, make_packing_list +from erpnext.stock.doctype.stock_reservation_entry.stock_reservation_entry import ( + get_sre_details_for_voucher, + get_sre_reserved_qty_details_for_voucher, + get_ssb_bundle_for_voucher, +) +from erpnext.stock.get_item_details import get_bin_details, get_price_list_rate + + +def get_requested_item_qty(sales_order: str) -> dict: + result = {} + + so = frappe.get_doc("Sales Order", sales_order) + + for item in so.items: + if is_product_bundle(item.item_code): + for packed_item in so.get("packed_items"): + if ( + packed_item.parent_item == item.item_code + and packed_item.parent_detail_docname == item.name + ): + result[packed_item.name] = frappe._dict({"qty": packed_item.requested_qty}) + else: + result[item.name] = frappe._dict({"qty": item.requested_qty}) + + return result + + +@frappe.whitelist() +def make_material_request(source_name: str, target_doc: str | Document | None = None): + requested_item_qty = get_requested_item_qty(source_name) + + def postprocess(source, target): + if source.tc_name and frappe.db.get_value("Terms and Conditions", source.tc_name, "buying") != 1: + target.tc_name = None + target.terms = None + + def get_remaining_qty(so_item): + return flt( + flt(so_item.qty) + - flt(requested_item_qty.get(so_item.name, {}).get("qty")) + - max( + flt(so_item.get("delivered_qty")), + 0, + ) + ) + + def get_remaining_packed_item_qty(so_item): + delivered_qty = frappe.db.get_value( + "Sales Order Item", {"name": so_item.parent_detail_docname}, ["delivered_qty"] + ) + + bundle_name = get_active_product_bundle(so_item.parent_item) + bundle_item_qty = ( + frappe.db.get_value( + "Product Bundle Item", + {"parent": bundle_name, "item_code": so_item.item_code}, + ["qty"], + ) + if bundle_name + else None + ) + + return flt( + flt(so_item.qty) + - flt(requested_item_qty.get(so_item.name, {}).get("qty")) + - max( + flt(delivered_qty) * flt(bundle_item_qty), + 0, + ) + ) + + def update_item(source, target, source_parent): + # qty is for packed items, because packed items don't have stock_qty field + target.project = source_parent.project + target.qty = ( + get_remaining_packed_item_qty(source) + if source.parentfield == "packed_items" + else get_remaining_qty(source) + ) + target.stock_qty = flt(target.qty) * flt(target.conversion_factor) + target.actual_qty = get_bin_details( + target.item_code, target.warehouse, source_parent.company, True + ).get("actual_qty", 0) + + ctx = frappe._dict(target.as_dict().copy()) + ctx.update( + { + "company": source_parent.get("company"), + "price_list": frappe.db.get_single_value("Buying Settings", "buying_price_list"), + "currency": source_parent.get("currency"), + "conversion_rate": source_parent.get("conversion_rate"), + } + ) + + target.rate = flt( + get_price_list_rate(ctx, item_doc=frappe.get_cached_doc("Item", target.item_code)).get( + "price_list_rate" + ) + ) + target.amount = target.qty * target.rate + + doc = get_mapped_doc( + "Sales Order", + source_name, + { + "Sales Order": {"doctype": "Material Request", "validation": {"docstatus": ["=", 1]}}, + "Packed Item": { + "doctype": "Material Request Item", + "field_map": {"parent": "sales_order", "uom": "stock_uom", "name": "packed_item"}, + "condition": lambda item: get_remaining_packed_item_qty(item) > 0, + "postprocess": update_item, + }, + "Sales Order Item": { + "doctype": "Material Request Item", + "field_map": { + "name": "sales_order_item", + "parent": "sales_order", + "delivery_date": "schedule_date", + "bom_no": "bom_no", + }, + "condition": lambda item: not is_product_bundle(item.item_code) + and get_remaining_qty(item) > 0, + "postprocess": update_item, + }, + }, + target_doc, + postprocess, + ) + if doc and doc.items: + return doc + else: + frappe.throw(_("Material Request already created for the ordered quantity")) + + +@frappe.whitelist() +def make_project(source_name: str, target_doc: str | Document | None = None): + def postprocess(source, doc): + doc.project_type = "External" + doc.project_name = source.name + + doc = get_mapped_doc( + "Sales Order", + source_name, + { + "Sales Order": { + "doctype": "Project", + "validation": {"docstatus": ["=", 1]}, + "field_map": { + "name": "sales_order", + "base_grand_total": "estimated_costing", + "net_total": "total_sales_amount", + }, + }, + }, + target_doc, + postprocess, + ) + + return doc + + +def set_serial_batch_for_bundle_reservation(source, target, use_serial_batch_fields, packed_sre): + for item in source.packed_items: + target_item = next( + ( + d + for d in target.packed_items + if (d.parent_item, d.item_code, d.warehouse) + == (item.parent_item, item.item_code, item.warehouse) + ), + None, + ) + if target_item and (sre := [sre for sre in packed_sre if sre.voucher_detail_no == item.name]): + if sre[0].reservation_based_on == "Serial and Batch": + qty = 0 + serial_nos = [] + batch_nos = [] + if use_serial_batch_fields: + target_item.use_serial_batch_fields = 1 + for item in sre: + qty += item.reserved_qty + if item.has_serial_no: + serial_nos.extend( + frappe.get_all( + "Serial and Batch Entry", + filters={"parent": item.name}, + pluck="serial_no", + ) + ) + if item.has_batch_no: + batch_nos.extend( + frappe.get_all( + "Serial and Batch Entry", + filters={"parent": item.name}, + pluck="batch_no", + ) + ) + + if len(batch_nos) == 1: + target_item.batch_no = batch_nos[0] if batch_nos else None + if serial_nos and len(batch_nos) < 2: + target_item.serial_no = "\n".join(serial_nos) + + if not use_serial_batch_fields or len(batch_nos) > 1: + target_item.serial_and_batch_bundle = get_ssb_bundle_for_voucher(sre).name + + +@frappe.whitelist() +def make_delivery_note( + source_name: str, target_doc: str | Document | None = None, kwargs: dict | None = None +): + if not kwargs: + kwargs = { + "for_reserved_stock": frappe.flags.args and frappe.flags.args.for_reserved_stock, + "skip_item_mapping": frappe.flags.args and frappe.flags.args.skip_item_mapping, + } + + kwargs = frappe._dict(kwargs) + + sre_details = {} + if kwargs.for_reserved_stock: + sre_details = get_sre_reserved_qty_details_for_voucher("Sales Order", source_name) + + mapper = { + "Sales Order": { + "doctype": "Delivery Note", + "validation": {"docstatus": ["=", 1]}, + # commission_rate is no_copy (so it isn't carried on Duplicate), map it explicitly here + "field_map": {"commission_rate": "commission_rate"}, + }, + "Sales Taxes and Charges": {"doctype": "Sales Taxes and Charges", "reset_value": True}, + "Sales Team": {"doctype": "Sales Team", "add_if_empty": True}, + } + + # 0 qty is accepted, as the qty is uncertain for some items + has_unit_price_items = frappe.db.get_value("Sales Order", source_name, "has_unit_price_items") + use_serial_batch_fields = frappe.get_single_value("Stock Settings", "use_serial_batch_fields") + + def is_unit_price_row(source): + return has_unit_price_items and source.qty == 0 + + def select_item(d): + filtered_items = kwargs.get("filtered_children", []) + child_filter = d.name in filtered_items if filtered_items else True + return child_filter + + def set_missing_values(source, target): + if kwargs.get("ignore_pricing_rule"): + # Skip pricing rule when the dn is creating from the pick list + target.ignore_pricing_rule = 1 + + target.run_method("set_missing_values") + target.run_method("set_po_nos") + target.run_method("calculate_taxes_and_totals") + target.run_method("set_use_serial_batch_fields") + + if source.company_address: + target.update({"company_address": source.company_address}) + else: + # set company address + target.update(get_company_address(target.company)) + + if target.company_address: + target.update(get_fetch_values("Delivery Note", "company_address", target.company_address)) + + # if invoked in bulk creation, validations are ignored and thus this method is nerver invoked + if frappe.flags.bulk_transaction: + # set target items names to ensure proper linking with packed_items + target.set_new_name() + + make_packing_list(target) + + def condition(doc): + if doc.name in sre_details: + del sre_details[doc.name] + return False + + # make_mapped_doc sets js `args` into `frappe.flags.args` + if frappe.flags.args and frappe.flags.args.delivery_dates: + if frappe.utils.cstr(doc.delivery_date) not in frappe.flags.args.delivery_dates: + return False + if frappe.flags.args and frappe.flags.args.until_delivery_date: + if frappe.utils.cstr(doc.delivery_date) > frappe.flags.args.until_delivery_date: + return False + + return ( + (abs(doc.delivered_qty) < abs(doc.qty)) or is_unit_price_row(doc) + ) and doc.delivered_by_supplier != 1 + + def update_item(source, target, source_parent): + target.base_amount = (flt(source.qty) - flt(source.delivered_qty)) * flt(source.base_rate) + target.amount = (flt(source.qty) - flt(source.delivered_qty)) * flt(source.rate) + target.qty = ( + flt(source.qty) if is_unit_price_row(source) else flt(source.qty) - flt(source.delivered_qty) + ) + + item = get_item_defaults(target.item_code, source_parent.company) + item_group = get_item_group_defaults(target.item_code, source_parent.company) + + if item: + target.cost_center = ( + frappe.db.get_value("Project", source_parent.project, "cost_center") + or item.get("buying_cost_center") + or item_group.get("buying_cost_center") + ) + + if not kwargs.skip_item_mapping: + mapper["Sales Order Item"] = { + "doctype": "Delivery Note Item", + "field_map": { + "rate": "rate", + "name": "so_detail", + "parent": "against_sales_order", + }, + "condition": lambda d: condition(d) and select_item(d), + "postprocess": update_item, + } + + so = frappe.get_doc("Sales Order", source_name) + target_doc = get_mapped_doc("Sales Order", so.name, mapper, target_doc) + + packed_sre = [] + if not kwargs.skip_item_mapping and kwargs.for_reserved_stock: + sre_list = get_sre_details_for_voucher("Sales Order", source_name) + + if sre_list: + + def update_dn_item(source, target, source_parent): + update_item(source, target, so) + + so_items = {d.name: d for d in so.items if d.stock_reserved_qty} + + for sre in sre_list: + if not so_items.get(sre.voucher_detail_no): + packed_sre.append(sre) + continue + + if not condition(so_items[sre.voucher_detail_no]): + continue + + dn_item = get_mapped_doc( + "Sales Order Item", + sre.voucher_detail_no, + { + "Sales Order Item": { + "doctype": "Delivery Note Item", + "field_map": { + "rate": "rate", + "name": "so_detail", + "parent": "against_sales_order", + }, + "postprocess": update_dn_item, + } + }, + ignore_permissions=True, + ) + + dn_item.qty = flt(sre.reserved_qty) / flt(dn_item.get("conversion_factor", 1)) + dn_item.warehouse = sre.warehouse + + if sre.reservation_based_on == "Serial and Batch" and (sre.has_serial_no or sre.has_batch_no): + if use_serial_batch_fields: + # Carry the reserved serial/batch in the row fields. A single field can't hold + # multiple batches, so fall back to a bundle in that case. + dn_item.use_serial_batch_fields = 1 + sb_entries = frappe.get_all( + "Serial and Batch Entry", + filters={"parent": sre.name}, + fields=["serial_no", "batch_no"], + ) + serial_nos = [d.serial_no for d in sb_entries if d.serial_no] + batch_nos = list({d.batch_no for d in sb_entries if d.batch_no}) + if serial_nos: + dn_item.serial_no = "\n".join(serial_nos) + if len(batch_nos) == 1: + dn_item.batch_no = batch_nos[0] + elif len(batch_nos) > 1: + dn_item.use_serial_batch_fields = 0 + dn_item.serial_and_batch_bundle = get_ssb_bundle_for_voucher([sre]).name + else: + dn_item.serial_and_batch_bundle = get_ssb_bundle_for_voucher([sre]).name + + target_doc.append("items", dn_item) + # Correct rows index. + for idx, item in enumerate(target_doc.items): + item.idx = idx + 1 + + if not kwargs.skip_item_mapping and frappe.flags.bulk_transaction and not target_doc.items: + # the (date) condition filter resulted in an unintendedly created empty DN; remove it + del target_doc + return + + # Should be called after mapping items. + target_doc.packed_items = [] + set_missing_values(so, target_doc) + set_serial_batch_for_bundle_reservation(so, target_doc, use_serial_batch_fields, packed_sre) + + return target_doc + + +@frappe.whitelist() +def make_sales_invoice( + source_name: str, + target_doc: str | Document | None = None, + ignore_permissions: bool = False, + args: str | dict | None = None, +): + if args is None: + args = {} + args = frappe.parse_json(args) + + # 0 qty is accepted, as the qty is uncertain for some items + has_unit_price_items = frappe.db.get_value("Sales Order", source_name, "has_unit_price_items") + + def is_unit_price_row(source): + return has_unit_price_items and source.qty == 0 + + def postprocess(source, target): + set_missing_values(source, target) + # Get the advance paid Journal Entries in Sales Invoice Advance + if target.get("allocate_advances_automatically"): + target.set_advances() + + make_packing_list(target) + set_serial_batch_for_bundle_reservation( + source, + target, + frappe.get_single_value("Stock Settings", "use_serial_batch_fields"), + get_sre_details_for_voucher("Sales Order", source_name), + ) + + def set_missing_values(source, target): + target.flags.ignore_permissions = True + target.run_method("set_missing_values") + target.run_method("set_po_nos") + target.run_method("calculate_taxes_and_totals") + target.run_method("set_use_serial_batch_fields") + + if source.company_address: + target.update({"company_address": source.company_address}) + else: + # set company address + target.update(get_company_address(target.company)) + + if target.company_address: + target.update(get_fetch_values("Sales Invoice", "company_address", target.company_address)) + + # set the redeem loyalty points if provided via shopping cart + if source.loyalty_points and source.order_type == "Shopping Cart": + target.redeem_loyalty_points = 1 + target.loyalty_points = source.loyalty_points + + target.debit_to = get_party_account("Customer", source.customer, source.company) + + def update_item(source, target, source_parent): + def get_billed_qty(so_item_name): + table = frappe.qb.DocType("Sales Invoice Item") + query = ( + frappe.qb.from_(table) + .select(Sum(table.qty).as_("qty")) + .where((table.docstatus == 1) & (table.so_detail == so_item_name)) + ) + return query.run(pluck="qty")[0] or 0 + + if source_parent.has_unit_price_items: + # 0 Amount rows (as seen in Unit Price Items) should be mapped as it is + pending_amount = flt(source.amount) - flt(source.billed_amt) + target.amount = pending_amount if flt(source.amount) else 0 + else: + target.amount = flt(source.amount) - flt(source.billed_amt) + + target.base_amount = target.amount * flt(source_parent.conversion_rate) + target.qty = ( + source.qty - get_billed_qty(source.name) + if (source.qty and source.billed_amt) + else (source.qty if is_unit_price_row(source) else source.qty - source.returned_qty) + ) + + if source_parent.project: + target.cost_center = frappe.db.get_value("Project", source_parent.project, "cost_center") + if target.item_code: + item = get_item_defaults(target.item_code, source_parent.company) + item_group = get_item_group_defaults(target.item_code, source_parent.company) + cost_center = item.get("selling_cost_center") or item_group.get("selling_cost_center") + + if cost_center: + target.cost_center = 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 + + def add_self_rm(doclist): + parent = frappe.qb.DocType("Subcontracting Inward Order") + child = frappe.qb.DocType("Subcontracting Inward Order Received Item") + query = ( + frappe.qb.from_(parent) + .join(child) + .on(parent.name == child.parent) + .select( + child.required_qty, + child.consumed_qty, + child.billed_qty, + child.rm_item_code, + child.stock_uom, + child.name, + ) + .where( + (parent.docstatus == 1) + & (parent.sales_order == source_name) + & (child.is_customer_provided_item == 0) + ) + ) + result = query.run(as_dict=True) + + if result: + idx = len(doclist.items) + 1 + for item in result: + if (qty := max(item.required_qty, item.consumed_qty) - item.billed_qty) > 0: + doclist.append( + "items", + { + "item_code": item.rm_item_code, + "qty": qty, + "uom": item.stock_uom, + "scio_detail": item.name, + }, + ) + doclist.process_item_selection(idx) + idx += 1 + doclist.has_subcontracted = 1 + + doclist = get_mapped_doc( + "Sales Order", + source_name, + { + "Sales Order": { + "doctype": "Sales Invoice", + "field_map": { + "party_account_currency": "party_account_currency", + # commission_rate is no_copy (so it isn't carried on Duplicate), map it explicitly here + "commission_rate": "commission_rate", + }, + "field_no_map": ["payment_terms_template"], + "validation": {"docstatus": ["=", 1]}, + }, + "Sales Order Item": { + "doctype": "Sales Invoice Item", + "field_map": { + "name": "so_detail", + "parent": "sales_order", + }, + "postprocess": update_item, + "condition": lambda doc: ( + True + if is_unit_price_row(doc) + else (doc.qty and (doc.base_amount == 0 or abs(doc.billed_amt) < abs(doc.amount))) + ) + and select_item(doc) + and not args.get("skip_item_mapping"), + }, + "Sales Taxes and Charges": { + "doctype": "Sales Taxes and Charges", + "reset_value": True, + }, + "Sales Team": {"doctype": "Sales Team", "add_if_empty": True}, + }, + target_doc, + postprocess, + ignore_permissions=ignore_permissions, + ) + + if frappe.get_cached_value("Sales Order", source_name, "is_subcontracted"): + add_self_rm(doclist) + + automatically_fetch_payment_terms = cint( + frappe.get_single_value("Accounts Settings", "automatically_fetch_payment_terms") + ) + if automatically_fetch_payment_terms: + from erpnext.accounts.services.payment_schedule import PaymentScheduleService + + PaymentScheduleService(doclist).set_payment_schedule() + + return doclist + + +@frappe.whitelist() +def make_maintenance_schedule(source_name: str, target_doc: str | Document | None = None): + maint_schedule = frappe.db.exists( + "Maintenance Schedule Item", {"sales_order": source_name, "docstatus": 1} + ) + + if not maint_schedule: + doclist = get_mapped_doc( + "Sales Order", + source_name, + { + "Sales Order": {"doctype": "Maintenance Schedule", "validation": {"docstatus": ["=", 1]}}, + "Sales Order Item": { + "doctype": "Maintenance Schedule Item", + "field_map": {"parent": "sales_order"}, + }, + }, + target_doc, + ) + + return doclist + + +@frappe.whitelist() +def make_maintenance_visit(source_name: str, target_doc: str | Document | None = None): + MaintenanceVisit = frappe.qb.DocType("Maintenance Visit") + MaintenanceVisitPurpose = frappe.qb.DocType("Maintenance Visit Purpose") + + query = ( + frappe.qb.from_(MaintenanceVisit) + .join(MaintenanceVisitPurpose) + .on(MaintenanceVisitPurpose.parent == MaintenanceVisit.name) + .select(MaintenanceVisit.name) + .where(MaintenanceVisitPurpose.prevdoc_docname == source_name) + .where(MaintenanceVisit.docstatus == 1) + .where(MaintenanceVisit.completion_status == "Fully Completed") + ) + + if not query.run(): + doclist = get_mapped_doc( + "Sales Order", + source_name, + { + "Sales Order": {"doctype": "Maintenance Visit", "validation": {"docstatus": ["=", 1]}}, + "Sales Order Item": { + "doctype": "Maintenance Visit Purpose", + "field_map": {"parent": "prevdoc_docname", "parenttype": "prevdoc_doctype"}, + }, + }, + target_doc, + ) + + return doclist + + +@frappe.whitelist() +def make_purchase_order( + source_name: str, selected_items: str | list | None = None, target_doc: str | Document | None = None +): + """Creates Purchase Order for each Supplier. Returns a list of doc objects.""" + + from erpnext.setup.utils import get_exchange_rate + + if not selected_items: + return + + selected_items = frappe.parse_json(selected_items) + + def set_missing_values(source, target): + target.supplier = supplier + company_currency = frappe.db.get_value( + "Company", filters={"name": target.company}, fieldname=["default_currency"] + ) + supplier_currency = frappe.db.get_value( + "Supplier", filters={"name": supplier}, fieldname=["default_currency"] + ) + + target.currency = supplier_currency if supplier_currency else company_currency + + target.conversion_rate = get_exchange_rate(target.currency, company_currency, args="for_buying") + + target.apply_discount_on = "" + target.additional_discount_percentage = 0.0 + target.discount_amount = 0.0 + target.inter_company_order_reference = "" + target.shipping_rule = "" + target.tc_name = "" + target.terms = "" + target.payment_schedule = [] + + default_price_list = frappe.get_value("Supplier", supplier, "default_price_list") + if default_price_list: + target.buying_price_list = default_price_list + + default_payment_terms = frappe.get_value("Supplier", supplier, "payment_terms") + if default_payment_terms: + target.payment_terms_template = default_payment_terms + + if any(item.delivered_by_supplier for item in target.items): + if source.shipping_address_name: + target.shipping_address = source.shipping_address_name + target.shipping_address_display = source.shipping_address + else: + target.shipping_address = source.customer_address + target.shipping_address_display = source.address_display + + target.customer_contact_person = source.contact_person + target.customer_contact_display = source.contact_display + target.customer_contact_mobile = source.contact_mobile + target.customer_contact_email = source.contact_email + + else: + target.customer = "" + target.customer_name = "" + + target.run_method("set_missing_values") + target.run_method("calculate_taxes_and_totals") + + def update_item(source, target, source_parent): + target.schedule_date = source.delivery_date + target.qty = flt(source.qty) - (flt(source.ordered_qty) / flt(source.conversion_factor)) + target.stock_qty = flt(source.stock_qty) - flt(source.ordered_qty) + target.project = source_parent.project + + def update_item_for_packed_item(source, target, _): + target.qty = flt(source.qty) - flt(source.ordered_qty) + + def filter_items(item, supplier): + if ( + item.ordered_qty < item.stock_qty + and not is_product_bundle(item.item_code) + and items_to_map.get(item.item_code) == supplier + ): + return True + + return False + + items_to_map = { + item.get("item_code"): item.get("supplier") for item in selected_items if item.get("item_code") + } + item_codes = list(set(items_to_map.keys())) + suppliers = list(set(items_to_map.values())) + + if not suppliers: + suppliers = [None] + + purchase_orders = [] + for supplier in suppliers: + doc = get_mapped_doc( + "Sales Order", + source_name, + { + "Sales Order": { + "doctype": "Purchase Order", + "field_no_map": [*CROSS_PARTY_FIELD_NO_MAP], + "validation": {"docstatus": ["=", 1]}, + }, + "Sales Order Item": { + "doctype": "Purchase Order Item", + "field_map": [ + ["name", "sales_order_item"], + ["parent", "sales_order"], + ["stock_uom", "stock_uom"], + ["uom", "uom"], + ["conversion_factor", "conversion_factor"], + ["delivery_date", "schedule_date"], + ], + "field_no_map": [ + "rate", + "price_list_rate", + "item_tax_template", + "discount_percentage", + "discount_amount", + "pricing_rules", + "margin_type", + "margin_rate_or_amount", + ], + "postprocess": update_item, + "condition": lambda doc, s=supplier: filter_items(doc, s), + }, + "Packed Item": { + "doctype": "Purchase Order Item", + "field_map": [ + ["name", "sales_order_packed_item"], + ["parent", "sales_order"], + ["uom", "uom"], + ["conversion_factor", "conversion_factor"], + ["product_bundle", "product_bundle"], + ["rate", "rate"], + ], + "field_no_map": [ + "price_list_rate", + "item_tax_template", + "discount_percentage", + "discount_amount", + "supplier", + "pricing_rules", + ], + "postprocess": update_item_for_packed_item, + "condition": lambda doc: doc.parent_item in item_codes + and flt(doc.ordered_qty) < flt(doc.qty), + }, + }, + target_doc, + set_missing_values, + ) + + set_delivery_date(doc.items, source_name) + if doc.supplier: + doc.insert() + purchase_orders.append(doc) + + return purchase_orders + + +def set_delivery_date(items: list, sales_order: str) -> None: + # `product_bundle` now holds the Product Bundle *version*, so match the Purchase + # Order rows to their originating Sales Order rows by that version. + delivery_dates = frappe.get_all( + "Sales Order Item", filters={"parent": sales_order}, fields=["delivery_date", "product_bundle"] + ) + + delivery_by_bundle = frappe._dict() + for date in delivery_dates: + if date.product_bundle: + delivery_by_bundle[date.product_bundle] = date.delivery_date + + for item in items: + if item.product_bundle: + item.schedule_date = delivery_by_bundle.get(item.product_bundle) + + +@frappe.whitelist() +def make_work_orders(items: str | dict, sales_order: str, company: str, project: str | None = None): + """Make Work Orders against the given Sales Order for the given `items`""" + items = frappe.parse_json(items).get("items") + out = [] + + for i in items: + if not i.get("bom"): + frappe.throw(_("Please select BOM against item {0}").format(i.get("item_code"))) + if not i.get("pending_qty"): + frappe.throw(_("Please select Qty against item {0}").format(i.get("item_code"))) + + work_order = frappe.get_doc( + doctype="Work Order", + production_item=i["item_code"], + bom_no=i.get("bom"), + qty=i["pending_qty"], + company=company, + sales_order=sales_order, + sales_order_item=i["sales_order_item"], + project=project, + fg_warehouse=i["warehouse"], + description=i["description"], + ).insert() + work_order.set_work_order_operations() + work_order.flags.ignore_mandatory = True + work_order.save() + out.append(work_order) + + return [p.name for p in out] + + +@frappe.whitelist() +def make_production_plan(source_name: str, target_doc: str | Document | None = None): + sales_order = frappe.get_doc("Sales Order", source_name) + + production_plan = frappe.new_doc( + "Production Plan", + company=sales_order.company, + get_items_from="Sales Order", + posting_date=nowdate(), + ) + + open_so = [data.name for data in get_sales_orders(production_plan)] + if sales_order.name not in open_so: + frappe.throw(_("Sales Order {0} is not available for production").format(sales_order.name)) + + production_plan.append( + "sales_orders", + { + "sales_order": sales_order.name, + "sales_order_date": sales_order.transaction_date, + "customer": sales_order.customer, + "grand_total": sales_order.base_grand_total, + }, + ) + production_plan.get_items() + if not production_plan.get("po_items"): + frappe.throw(_("Sales Order {0} is not available for production").format(sales_order.name)) + + return production_plan + + +@frappe.whitelist() +def make_raw_material_request( + items: str | frappe._dict, company: str, sales_order: str, project: str | None = None +): + if not frappe.has_permission("Sales Order", "write"): + frappe.throw(_("Not permitted"), frappe.PermissionError) + + items = frappe._dict(frappe.parse_json(items)) + + for item in items.get("items"): + item["include_exploded_items"] = items.get("include_exploded_items") + item["ignore_existing_ordered_qty"] = items.get("ignore_existing_ordered_qty") + item["include_raw_materials_from_sales_order"] = items.get("include_raw_materials_from_sales_order") + + items.update({"company": company, "sales_order": sales_order}) + + item_wh = {} + for item in items.get("items"): + if item.get("warehouse"): + item_wh[item.get("item_code")] = item.get("warehouse") + + raw_materials = get_items_for_material_requests(items) + if not raw_materials: + frappe.msgprint(_("Material Request not created, as quantity for Raw Materials already available.")) + return + + material_request = frappe.new_doc("Material Request") + material_request.update( + dict( + doctype="Material Request", + transaction_date=nowdate(), + company=company, + material_request_type="Purchase", + ) + ) + for item in raw_materials: + item_doc = frappe.get_cached_doc("Item", item.get("item_code")) + + schedule_date = add_days(nowdate(), cint(item_doc.lead_time_days)) + row = material_request.append( + "items", + { + "item_code": item.get("item_code"), + "qty": item.get("quantity"), + "schedule_date": schedule_date, + "warehouse": item_wh.get(item.get("main_bom_item")) or item.get("warehouse"), + "sales_order": sales_order, + "project": project, + }, + ) + + if not (strip_html(item.get("description")) and strip_html(item_doc.description)): + row.description = item_doc.item_name or item.get("item_code") + + material_request.insert() + material_request.flags.ignore_permissions = 1 + material_request.run_method("set_missing_values") + material_request.submit() + return material_request + + +@frappe.whitelist() +def make_inter_company_purchase_order(source_name: str, target_doc: str | Document | None = None): + from erpnext.accounts.doctype.sales_invoice.mapper import make_inter_company_transaction + + return make_inter_company_transaction("Sales Order", source_name, target_doc) + + +@frappe.whitelist() +def create_pick_list(source_name: str, target_doc: str | Document | None = None): + def validate_sales_order(): + so = frappe.get_doc("Sales Order", source_name) + for item in so.items: + if item.stock_reserved_qty > 0: + frappe.throw( + _( + "Cannot create a pick list for Sales Order {0} because it has reserved stock. Please unreserve the stock in order to create a pick list." + ).format(frappe.bold(source_name)) + ) + + def update_item_quantity(source, target, source_parent) -> None: + picked_qty = flt(source.picked_qty) / (flt(source.conversion_factor) or 1) + qty_to_be_picked = flt(source.qty) - max(picked_qty, flt(source.delivered_qty)) + + target.qty = qty_to_be_picked + target.stock_qty = qty_to_be_picked * flt(source.conversion_factor) + + # update available qty + bin_details = get_bin_details(source.item_code, source.warehouse, source_parent.company) + target.actual_qty = bin_details.get("actual_qty") + target.company_total_stock = bin_details.get("company_total_stock") + + def update_packed_item_qty(source, target, source_parent) -> None: + qty = flt(source.qty) + for item in source_parent.items: + if source.parent_detail_docname == item.name: + picked_qty = flt(item.picked_qty) / (flt(item.conversion_factor) or 1) + pending_percent = (item.qty - max(picked_qty, item.delivered_qty)) / item.qty + target.qty = target.stock_qty = qty * pending_percent + return + + def should_pick_order_item(item) -> bool: + return ( + abs(item.delivered_qty) < abs(item.qty) + and item.delivered_by_supplier != 1 + and not is_product_bundle(item.item_code) + ) + + # Don't allow a Pick List to be created against a Sales Order that has reserved stock. + validate_sales_order() + + doc = get_mapped_doc( + "Sales Order", + source_name, + { + "Sales Order": { + "doctype": "Pick List", + "field_map": {"set_warehouse": "parent_warehouse"}, + "validation": {"docstatus": ["=", 1]}, + }, + "Sales Order Item": { + "doctype": "Pick List Item", + "field_map": {"parent": "sales_order", "name": "sales_order_item"}, + "postprocess": update_item_quantity, + "condition": should_pick_order_item, + }, + "Packed Item": { + "doctype": "Pick List Item", + "field_map": { + "parent": "sales_order", + "parent_detail_docname": "sales_order_item", + "name": "product_bundle_item", + }, + "field_no_map": ["picked_qty"], + "postprocess": update_packed_item_qty, + }, + }, + target_doc, + ) + + doc.purpose = "Delivery" + + doc.set_item_locations() + + return doc + + +@frappe.whitelist() +def make_subcontracting_inward_order(source_name: str, target_doc: str | Document | None = None): + if not is_so_fully_subcontracted(source_name): + return get_mapped_subcontracting_inward_order(source_name, target_doc) + else: + frappe.throw(_("This Sales Order has been fully subcontracted.")) + + +def is_so_fully_subcontracted(so_name: str) -> bool: + table = frappe.qb.DocType("Sales Order Item") + query = ( + frappe.qb.from_(table) + .select(table.name) + .where((table.parent == so_name) & (table.qty != table.subcontracted_qty)) + ) + return not query.run(as_dict=True) + + +def get_mapped_subcontracting_inward_order( + source_name: str, target_doc: str | Document | None = None +) -> Document: + def post_process(source_doc, target_doc): + if ( + frappe.db.count( + "Warehouse", {"customer": source_doc.customer, "disabled": 0, "is_rejected_warehouse": 0} + ) + == 1 + ): + target_doc.customer_warehouse = frappe.get_cached_value( + "Warehouse", + {"customer": source_doc.customer, "disabled": 0, "is_rejected_warehouse": 0}, + "name", + ) + target_doc.populate_items_table() + + if target_doc and isinstance(target_doc, str): + target_doc = frappe.parse_json(target_doc) + for key in ["service_items", "items", "received_items"]: + if key in target_doc: + del target_doc[key] + target_doc = json.dumps(target_doc) + + target_doc = get_mapped_doc( + "Sales Order", + source_name, + { + "Sales Order": { + "doctype": "Subcontracting Inward Order", + "field_map": {}, + "field_no_map": ["total_qty", "total", "net_total"], + "validation": { + "docstatus": ["=", 1], + }, + }, + "Sales Order Item": { + "doctype": "Subcontracting Inward Order Service Item", + "field_map": { + "name": "sales_order_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/stock/doctype/batch/test_batch.py b/erpnext/stock/doctype/batch/test_batch.py index 984de5c32dd..e6d5565ae8e 100644 --- a/erpnext/stock/doctype/batch/test_batch.py +++ b/erpnext/stock/doctype/batch/test_batch.py @@ -19,7 +19,7 @@ from erpnext.stock.doctype.serial_and_batch_bundle.test_serial_and_batch_bundle get_batch_from_bundle, ) from erpnext.stock.doctype.stock_entry.stock_entry_utils import make_stock_entry -from erpnext.stock.get_item_details import ItemDetailsCtx, get_item_details +from erpnext.stock.get_item_details import get_item_details from erpnext.stock.serial_batch_bundle import SerialBatchCreation from erpnext.tests.utils import ERPNextTestSuite @@ -549,7 +549,7 @@ class TestBatch(ERPNextTestSuite): company = "_Test Company with perpetual inventory" currency = frappe.get_cached_value("Company", company, "default_currency") - ctx = ItemDetailsCtx( + ctx = frappe._dict( { "item_code": "_Test Batch Price Item", "company": company, diff --git a/erpnext/stock/doctype/item/test_item.py b/erpnext/stock/doctype/item/test_item.py index 221c45e3591..56a619c0ee1 100644 --- a/erpnext/stock/doctype/item/test_item.py +++ b/erpnext/stock/doctype/item/test_item.py @@ -25,7 +25,7 @@ from erpnext.stock.doctype.item.item import ( validate_is_stock_item, ) from erpnext.stock.doctype.stock_entry.stock_entry_utils import make_stock_entry -from erpnext.stock.get_item_details import ItemDetailsCtx, get_item_details +from erpnext.stock.get_item_details import get_item_details from erpnext.tests.utils import ERPNextTestSuite @@ -146,7 +146,7 @@ class TestItem(ERPNextTestSuite): currency = frappe.get_cached_value("Company", company, "default_currency") details = get_item_details( - ItemDetailsCtx( + frappe._dict( { "item_code": "_Test Item", "company": company, @@ -176,7 +176,7 @@ class TestItem(ERPNextTestSuite): create_fixed_asset_item() details = get_item_details( - ItemDetailsCtx( + frappe._dict( { "item_code": "Macbook Pro", "company": "_Test Company", @@ -189,7 +189,7 @@ class TestItem(ERPNextTestSuite): frappe.db.set_value("Asset Category", "Computers", "enable_cwip_accounting", "1") details = get_item_details( - ItemDetailsCtx( + frappe._dict( { "item_code": "Macbook Pro", "company": "_Test Company", @@ -279,7 +279,7 @@ class TestItem(ERPNextTestSuite): for data in expected_item_tax_template: details = get_item_details( - ItemDetailsCtx( + frappe._dict( { "item_code": data["item_code"], "tax_category": data["tax_category"], @@ -331,7 +331,7 @@ class TestItem(ERPNextTestSuite): "cost_center": "_Test Cost Center 2 - _TC", # from item group } sales_item_details = get_item_details( - ItemDetailsCtx( + frappe._dict( { "item_code": "Test Item With Defaults", "company": "_Test Company", @@ -356,7 +356,7 @@ class TestItem(ERPNextTestSuite): "cost_center": "_Test Write Off Cost Center - _TC", # from item } purchase_item_details = get_item_details( - ItemDetailsCtx( + frappe._dict( { "item_code": "Test Item With Defaults", "company": "_Test Company", diff --git a/erpnext/stock/doctype/item_price/test_item_price.py b/erpnext/stock/doctype/item_price/test_item_price.py index f2f177dd8fe..dd4f1eadcd6 100644 --- a/erpnext/stock/doctype/item_price/test_item_price.py +++ b/erpnext/stock/doctype/item_price/test_item_price.py @@ -6,7 +6,7 @@ import frappe from frappe.tests.utils import make_test_records_for_doctype from erpnext.stock.doctype.item_price.item_price import ItemPriceDuplicateItem -from erpnext.stock.get_item_details import ItemDetailsCtx, get_price_list_rate_for +from erpnext.stock.get_item_details import get_price_list_rate_for from erpnext.tests.utils import ERPNextTestSuite @@ -69,7 +69,7 @@ class TestItemPrice(ERPNextTestSuite): # Check correct price at this quantity doc = frappe.copy_doc(self.globalTestRecords["Item Price"][2]) - ctx = ItemDetailsCtx( + ctx = frappe._dict( { "price_list": doc.price_list, "customer": doc.customer, @@ -85,7 +85,7 @@ class TestItemPrice(ERPNextTestSuite): def test_price_with_no_qty(self): # Check correct price when no quantity doc = frappe.copy_doc(self.globalTestRecords["Item Price"][2]) - ctx = ItemDetailsCtx( + ctx = frappe._dict( { "price_list": doc.price_list, "customer": doc.customer, @@ -101,7 +101,7 @@ class TestItemPrice(ERPNextTestSuite): # Check correct price at first date doc = frappe.copy_doc(self.globalTestRecords["Item Price"][2]) - ctx = ItemDetailsCtx( + ctx = frappe._dict( { "price_list": doc.price_list, "customer": "_Test Customer", @@ -118,7 +118,7 @@ class TestItemPrice(ERPNextTestSuite): # Check correct price at invalid date doc = frappe.copy_doc(self.globalTestRecords["Item Price"][3]) - ctx = ItemDetailsCtx( + ctx = frappe._dict( { "price_list": doc.price_list, "qty": 7, @@ -134,7 +134,7 @@ class TestItemPrice(ERPNextTestSuite): # Check correct price when outside of the date doc = frappe.copy_doc(self.globalTestRecords["Item Price"][4]) - ctx = ItemDetailsCtx( + ctx = frappe._dict( { "price_list": doc.price_list, "customer": "_Test Customer", @@ -151,7 +151,7 @@ class TestItemPrice(ERPNextTestSuite): # Check lowest price when no date provided doc = frappe.copy_doc(self.globalTestRecords["Item Price"][1]) - ctx = ItemDetailsCtx( + ctx = frappe._dict( { "price_list": doc.price_list, "uom": "_Test UOM", @@ -183,7 +183,7 @@ class TestItemPrice(ERPNextTestSuite): doc.price_list_rate = 21 doc.insert() - ctx = ItemDetailsCtx( + ctx = frappe._dict( { "price_list": doc.price_list, "uom": "_Test UOM", diff --git a/erpnext/stock/doctype/packed_item/packed_item.py b/erpnext/stock/doctype/packed_item/packed_item.py index 5a6531f1526..a39f66f9c2d 100644 --- a/erpnext/stock/doctype/packed_item/packed_item.py +++ b/erpnext/stock/doctype/packed_item/packed_item.py @@ -11,7 +11,7 @@ import frappe.defaults from frappe.model.document import Document from frappe.utils import flt -from erpnext.stock.get_item_details import ItemDetailsCtx, get_item_details, get_price_list_rate +from erpnext.stock.get_item_details import get_item_details, get_price_list_rate class PackedItem(Document): @@ -286,7 +286,7 @@ def update_packed_item_price_data(pi_row, item_data, doc): return item_doc = frappe.get_cached_doc("Item", pi_row.item_code) - ctx = ItemDetailsCtx(pi_row.as_dict().copy()) + ctx = frappe._dict(pi_row.as_dict().copy()) ctx.update( { "company": doc.get("company"), @@ -378,7 +378,33 @@ def on_doctype_update(): def get_items_from_product_bundle(row): row, items = ItemDetailsCtx(json.loads(row)), [] +<<<<<<< HEAD bundled_items = get_product_bundle_items(row["item_code"]) +======= + ``row.product_bundle`` selects a specific version by document name (the buying + dialog passes this); ``row.item_code`` is the legacy contract, resolving the + parent item's active version. + """ + from erpnext.selling.doctype.product_bundle.product_bundle import get_active_product_bundle + + row, items = frappe._dict(frappe.parse_json(row)), [] + + if bundle_name := row.get("product_bundle"): + frappe.has_permission("Product Bundle", "read", bundle_name, throw=True) + bundle = frappe.db.get_value("Product Bundle", bundle_name, ["docstatus", "disabled"], as_dict=True) + if not bundle or bundle.docstatus != 1: + frappe.throw(_("Product Bundle {0} is not submitted").format(frappe.bold(bundle_name))) + if bundle.disabled: + frappe.throw( + _("Product Bundle {0} is disabled and cannot be used in transactions.").format( + frappe.bold(bundle_name) + ) + ) + elif bundle_name := get_active_product_bundle(row.get("item_code")): + frappe.has_permission("Product Bundle", "read", bundle_name, throw=True) + + bundled_items = get_product_bundle_items_by_name(bundle_name) if bundle_name else [] +>>>>>>> e6f8f8f7e9 (refactor: use frappe._dict in importers of ItemDetailsCtx) for item in bundled_items: row.update( { diff --git a/erpnext/stock/doctype/purchase_receipt/test_purchase_receipt.py b/erpnext/stock/doctype/purchase_receipt/test_purchase_receipt.py index 4e9dc7f747a..07e83f57582 100644 --- a/erpnext/stock/doctype/purchase_receipt/test_purchase_receipt.py +++ b/erpnext/stock/doctype/purchase_receipt/test_purchase_receipt.py @@ -2385,7 +2385,7 @@ class TestPurchaseReceipt(ERPNextTestSuite): self.assertTrue(return_pi.docstatus == 1) def test_disable_last_purchase_rate(self): - from erpnext.stock.get_item_details import ItemDetailsCtx, get_item_details + from erpnext.stock.get_item_details import get_item_details item = make_item( "_Test Disable Last Purchase Rate", @@ -2400,7 +2400,7 @@ class TestPurchaseReceipt(ERPNextTestSuite): item_code=item.name, ) - ctx = ItemDetailsCtx(pr.items[0].as_dict()) + ctx = frappe._dict(pr.items[0].as_dict()) ctx.update( { "supplier": pr.supplier, diff --git a/erpnext/stock/doctype/stock_entry/stock_entry.py b/erpnext/stock/doctype/stock_entry/stock_entry.py index cd442909b1c..7deac1c49b7 100644 --- a/erpnext/stock/doctype/stock_entry/stock_entry.py +++ b/erpnext/stock/doctype/stock_entry/stock_entry.py @@ -42,7 +42,6 @@ from erpnext.stock.doctype.stock_reconciliation.stock_reconciliation import ( OpeningEntryAccountError, ) from erpnext.stock.get_item_details import ( - ItemDetailsCtx, get_barcode_data, get_bin_details, get_conversion_factor, @@ -2577,8 +2576,34 @@ class StockEntry(StockController, SubcontractingInwardController): return reserved_work_orders @frappe.whitelist() +<<<<<<< HEAD def get_item_details(self, args: ItemDetailsCtx = None, for_update=False): item = frappe.qb.DocType("Item") +======= + def get_item_details(self, args: frappe._dict | None = None, for_update: bool = False): + item = self._fetch_item_data(args) + item_group_defaults = get_item_group_defaults(item.name, self.company) + brand_defaults = get_brand_defaults(item.name, self.company) + + ret = self._build_item_ret(args, item, item_group_defaults, brand_defaults, for_update) + self._apply_account_defaults(ret) + + args["posting_date"] = self.posting_date + args["posting_time"] = self.posting_time + ret.update(get_warehouse_details(args) if args.get("warehouse") else {}) + + if self.purpose == "Send to Subcontractor": + self._resolve_subcontract_item(args, ret) + + barcode_data = get_barcode_data(item_code=item.name) + if barcode_data and len(barcode_data.get(item.name)) == 1: + ret["barcode"] = barcode_data.get(item.name)[0] + + return ret + + def _fetch_item_data(self, args): + item_dt = frappe.qb.DocType("Item") +>>>>>>> e6f8f8f7e9 (refactor: use frappe._dict in importers of ItemDetailsCtx) item_default = frappe.qb.DocType("Item Default") query = (