diff --git a/erpnext/assets/doctype/asset_capitalization/services/gl_composer.py b/erpnext/assets/doctype/asset_capitalization/services/gl_composer.py index 2b13bddd5ad..5e1f08edad0 100644 --- a/erpnext/assets/doctype/asset_capitalization/services/gl_composer.py +++ b/erpnext/assets/doctype/asset_capitalization/services/gl_composer.py @@ -30,7 +30,7 @@ class AssetCapitalizationGLComposer(BaseStockGLComposer): gl_entries = [] self.inventory_account_map = inventory_account_map or doc.get_inventory_account_map() - self.precision = doc.get_debit_field_precision() + self.precision = self.get_debit_field_precision() self.sle_map = doc.get_stock_ledger_details() target_account = doc.get_target_account() diff --git a/erpnext/controllers/buying_controller.py b/erpnext/controllers/buying_controller.py index 1fac4f8b216..1b5574b764f 100644 --- a/erpnext/controllers/buying_controller.py +++ b/erpnext/controllers/buying_controller.py @@ -35,6 +35,10 @@ class BuyingController(SubcontractingController): self.flags.ignore_permlevel_for_fields = ["buying_price_list", "price_list_currency"] def validate(self): + from erpnext.stock.doctype.landed_cost_voucher.landed_cost_voucher import ( + set_landed_cost_voucher_amount, + ) + self.set_rate_for_standalone_debit_note() super().validate() @@ -59,7 +63,7 @@ class BuyingController(SubcontractingController): self.validate_rejected_warehouse() self.validate_accepted_rejected_qty() validate_for_items(self) - self.set_landed_cost_voucher_amount() + set_landed_cost_voucher_amount(self) if self.doctype in ("Purchase Receipt", "Purchase Invoice"): self.update_valuation_rate() diff --git a/erpnext/controllers/ledger_preview.py b/erpnext/controllers/ledger_preview.py new file mode 100644 index 00000000000..ce074903aaf --- /dev/null +++ b/erpnext/controllers/ledger_preview.py @@ -0,0 +1,142 @@ +# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors +# License: GNU General Public License v3. See license.txt + +"""Read-side GL / Stock Ledger preview helpers. + +A dry-run consumer of the posting path, shared across accounts and stock vouchers +(Sales/Purchase Invoice, Payment Entry, Delivery Note, Purchase Receipt, Stock +Entry): it submits-in-memory, reads the resulting GL/SLE entries and formats them +for the datatable preview, then rolls back. Lives separately from the posting +services it orchestrates. The whitelisted ``show_*_preview`` entry points stay on +``stock_controller`` (their dotted path is referenced from client JS). +""" + +import frappe + + +def get_accounting_ledger_preview(doc, filters): + from erpnext.accounts.report.general_ledger.general_ledger import get_columns as get_gl_columns + + gl_columns, gl_data = [], [] + fields = [ + "posting_date", + "account", + "debit", + "credit", + "against", + "party_type", + "party", + "cost_center", + "against_voucher_type", + "against_voucher", + ] + + # Dry run: submit in memory to materialise GL entries, read them, then roll back + # to the savepoint so the preview never persists anything, regardless of caller. + frappe.db.savepoint("ledger_preview") + try: + doc.docstatus = 1 + + if doc.get("update_stock") or doc.doctype in ("Purchase Receipt", "Delivery Note", "Stock Entry"): + doc.update_stock_ledger() + + doc.make_gl_entries() + columns = get_gl_columns(filters) + gl_entries = get_gl_entries_for_preview(doc.doctype, doc.name, fields) + + gl_columns = get_columns(columns, fields) + gl_data = get_data(fields, gl_entries) + finally: + frappe.db.rollback(save_point="ledger_preview") + + return gl_columns, gl_data + + +def get_stock_ledger_preview(doc, filters): + from erpnext.stock.report.stock_ledger.stock_ledger import get_columns as get_sl_columns + + sl_columns, sl_data = [], [] + fields = [ + "item_code", + "stock_uom", + "actual_qty", + "qty_after_transaction", + "warehouse", + "incoming_rate", + "valuation_rate", + "stock_value", + "stock_value_difference", + ] + columns_fields = [ + "item_code", + "stock_uom", + "in_qty", + "out_qty", + "qty_after_transaction", + "warehouse", + "incoming_rate", + "in_out_rate", + "stock_value", + "stock_value_difference", + ] + + if doc.get("update_stock") or doc.doctype in ("Purchase Receipt", "Delivery Note", "Stock Entry"): + # Dry run: submit in memory to materialise SLEs, read them, then roll back to + # the savepoint so the preview never persists anything, regardless of caller. + frappe.db.savepoint("ledger_preview") + try: + doc.docstatus = 1 + doc.make_bundle_using_old_serial_batch_fields() + doc.update_stock_ledger() + + columns = get_sl_columns(filters) + sl_entries = get_sl_entries_for_preview(doc.doctype, doc.name, fields) + + sl_columns = get_columns(columns, columns_fields) + sl_data = get_data(columns_fields, sl_entries) + finally: + frappe.db.rollback(save_point="ledger_preview") + + return sl_columns, sl_data + + +def get_sl_entries_for_preview(doctype, docname, fields): + sl_entries = frappe.get_all( + "Stock Ledger Entry", filters={"voucher_type": doctype, "voucher_no": docname}, fields=fields + ) + + for entry in sl_entries: + if entry.actual_qty > 0: + entry["in_qty"] = entry.actual_qty + entry["out_qty"] = 0 + else: + entry["out_qty"] = abs(entry.actual_qty) + entry["in_qty"] = 0 + + entry["in_out_rate"] = entry["valuation_rate"] + + return sl_entries + + +def get_gl_entries_for_preview(doctype, docname, fields): + return frappe.get_all("GL Entry", filters={"voucher_type": doctype, "voucher_no": docname}, fields=fields) + + +def get_columns(raw_columns, fields): + return [ + {"name": d.get("label"), "editable": False, "width": 110, "fieldtype": d.get("fieldtype")} + for d in raw_columns + if not d.get("hidden") and d.get("fieldname") in fields + ] + + +def get_data(raw_columns, raw_data): + datatable_data = [] + for row in raw_data: + data_row = [] + for column in raw_columns: + data_row.append(row.get(column) or "") + + datatable_data.append(data_row) + + return datatable_data diff --git a/erpnext/controllers/stock_controller.py b/erpnext/controllers/stock_controller.py index ce7422ed2d5..904d600c9a8 100644 --- a/erpnext/controllers/stock_controller.py +++ b/erpnext/controllers/stock_controller.py @@ -2,11 +2,9 @@ # License: GNU General Public License v3. See license.txt import json -from collections import defaultdict import frappe from frappe import _, bold -from frappe.query_builder.functions import Sum from frappe.utils import cint, cstr, flt, get_link_to_form, getdate import erpnext @@ -14,107 +12,58 @@ from erpnext.accounts.general_ledger import ( make_gl_entries, make_reverse_gl_entries, ) -from erpnext.accounts.utils import cancel_exchange_gain_loss_journal, get_fiscal_year +from erpnext.accounts.utils import cancel_exchange_gain_loss_journal from erpnext.controllers.accounts_controller import AccountsController from erpnext.controllers.sales_and_purchase_return import ( available_serial_batch_for_return, filter_serial_batches, make_serial_batch_bundle_for_return, ) + +# Re-exported for backward compatibility; canonical home is erpnext.exceptions. +from erpnext.exceptions import ( + BatchExpiredError, + QualityInspectionNotSubmittedError, + QualityInspectionRejectedError, + QualityInspectionRequiredError, +) from erpnext.setup.doctype.brand.brand import get_brand_defaults from erpnext.setup.doctype.item_group.item_group import get_item_group_defaults from erpnext.stock import get_warehouse_account_map -from erpnext.stock.doctype.batch.batch import get_batch_qty -from erpnext.stock.doctype.inventory_dimension.inventory_dimension import ( - get_evaluated_inventory_dimension, -) from erpnext.stock.doctype.item.item import get_item_defaults -from erpnext.stock.doctype.serial_and_batch_bundle.serial_and_batch_bundle import ( - combine_datetime, - get_type_of_transaction, -) +from erpnext.stock.services.internal_transfer import StockInternalTransferService from erpnext.stock.stock_ledger import get_items_to_be_repost -class QualityInspectionRequiredError(frappe.ValidationError): - pass - - -class QualityInspectionRejectedError(frappe.ValidationError): - pass - - -class QualityInspectionNotSubmittedError(frappe.ValidationError): - pass - - -class BatchExpiredError(frappe.ValidationError): - pass - - class StockController(AccountsController): def validate(self): + from erpnext.stock.doctype.putaway_rule.putaway_rule import validate_putaway_capacity + from erpnext.stock.services.serial_batch_bundle_service import SerialBatchBundleService + + sbb = SerialBatchBundleService(self) + super().validate() if self.docstatus == 0: for table_name in ["items", "packed_items", "supplied_items"]: - self.validate_duplicate_serial_and_batch_bundle(table_name) + sbb.validate_duplicate_serial_and_batch_bundle(table_name) if not self.get("is_return"): self.validate_inspection() - self.validate_warehouse_of_sabb() - self.validate_serialized_batch() - self.clean_serial_nos() + sbb.validate_warehouse_of_sabb() + sbb.validate_serialized_batch() + sbb.clean_serial_nos() self.validate_customer_provided_item() self.set_rate_of_stock_uom() - self.validate_internal_transfer() - self.validate_putaway_capacity() + StockInternalTransferService(self).validate_internal_transfer() + validate_putaway_capacity(self) self.reset_conversion_factor() def on_update(self): super().on_update() self.check_zero_rate() - def validate_warehouse_of_sabb(self): - if self.is_internal_transfer(): - return - - doc_before_save = self.get_doc_before_save() - - for row in self.items: - if not row.get("serial_and_batch_bundle"): - continue - - sabb_details = frappe.db.get_value( - "Serial and Batch Bundle", - row.serial_and_batch_bundle, - ["type_of_transaction", "warehouse", "has_serial_no"], - as_dict=True, - ) - if not sabb_details: - continue - - if sabb_details.type_of_transaction != "Outward": - continue - - warehouse = row.get("warehouse") or row.get("s_warehouse") - if sabb_details.warehouse != warehouse: - frappe.throw( - _( - "Row #{0}: Warehouse {1} does not match with the warehouse {2} in Serial and Batch Bundle {3}." - ).format(row.idx, warehouse, sabb_details.warehouse, row.serial_and_batch_bundle) - ) - - if self.doctype == "Stock Reconciliation": - continue - - if sabb_details.has_serial_no and doc_before_save and doc_before_save.get("items"): - prev_row = doc_before_save.get("items", {"idx": row.idx}) - if prev_row and prev_row[0].serial_and_batch_bundle != row.serial_and_batch_bundle: - sabb_doc = frappe.get_doc("Serial and Batch Bundle", row.serial_and_batch_bundle) - sabb_doc.validate_serial_no_status() - def reset_conversion_factor(self): for row in self.get("items"): if row.uom != row.stock_uom: @@ -169,39 +118,6 @@ class StockController(AccountsController): if non_exists_items: frappe.throw(_("Items {0} do not exist in the Item master.").format(", ".join(non_exists_items))) - def validate_duplicate_serial_and_batch_bundle(self, table_name): - if not self.get(table_name): - return - - sbb_list = [] - for item in self.get(table_name): - if item.get("serial_and_batch_bundle"): - sbb_list.append(item.get("serial_and_batch_bundle")) - - if item.get("rejected_serial_and_batch_bundle"): - sbb_list.append(item.get("rejected_serial_and_batch_bundle")) - - if sbb_list: - SLE = frappe.qb.DocType("Stock Ledger Entry") - data = ( - frappe.qb.from_(SLE) - .select(SLE.voucher_type, SLE.voucher_no, SLE.serial_and_batch_bundle) - .where( - (SLE.docstatus == 1) - & (SLE.serial_and_batch_bundle.notnull()) - & (SLE.serial_and_batch_bundle.isin(sbb_list)) - ) - .limit(1) - ).run(as_dict=True) - - if data: - data = data[0] - frappe.throw( - _("Serial and Batch Bundle {0} is already used in {1} {2}.").format( - frappe.bold(data.serial_and_batch_bundle), data.voucher_type, data.voucher_no - ) - ) - def get_item_wise_inventory_account_map(self): inventory_account_map = frappe._dict() for table in ["items", "packed_items", "supplied_items"]: @@ -286,406 +202,22 @@ class StockController(AccountsController): ) make_gl_entries(gl_entries, from_repost=from_repost) - def validate_serialized_batch(self): - from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos - - is_material_issue = False - if self.doctype == "Stock Entry" and self.purpose in ["Material Issue", "Material Transfer"]: - is_material_issue = True - - for d in self.get("items"): - if hasattr(d, "serial_no") and hasattr(d, "batch_no") and d.serial_no and d.batch_no: - serial_nos = frappe.get_all( - "Serial No", - fields=["batch_no", "name", "warehouse"], - filters={"name": ("in", get_serial_nos(d.serial_no))}, - ) - - for row in serial_nos: - if row.warehouse and row.batch_no != d.batch_no: - frappe.throw( - _("Row #{0}: Serial No {1} does not belong to Batch {2}").format( - d.idx, row.name, d.batch_no - ) - ) - - if is_material_issue: - continue - - if flt(d.qty) > 0.0 and d.get("batch_no") and self.get("posting_date") and self.docstatus < 2: - expiry_date = frappe.get_cached_value("Batch", d.get("batch_no"), "expiry_date") - - if expiry_date and getdate(expiry_date) < getdate(self.posting_date): - frappe.throw( - _("Row #{0}: The batch {1} has already expired.").format( - d.idx, get_link_to_form("Batch", d.get("batch_no")) - ), - BatchExpiredError, - ) - - def clean_serial_nos(self): - from erpnext.stock.doctype.serial_no.serial_no import clean_serial_no_string - - for row in self.get("items"): - if hasattr(row, "serial_no") and row.serial_no: - # remove extra whitespace and store one serial no on each line - row.serial_no = clean_serial_no_string(row.serial_no) - - for row in self.get("packed_items") or []: - if hasattr(row, "serial_no") and row.serial_no: - # remove extra whitespace and store one serial no on each line - row.serial_no = clean_serial_no_string(row.serial_no) - def make_bundle_using_old_serial_batch_fields(self, table_name=None, via_landed_cost_voucher=False): - if self.get("_action") == "update_after_submit": - return + from erpnext.stock.services.serial_batch_bundle_service import SerialBatchBundleService - # To handle test cases - if frappe.in_test and frappe.flags.use_serial_and_batch_fields: - return - - if not table_name: - table_name = "items" - - if self.doctype == "Asset Capitalization": - table_name = "stock_items" - - parent_details = frappe._dict() - if table_name == "packed_items": - parent_details = self.get_parent_details_for_packed_items() - - for row in self.get(table_name): - if ( - not via_landed_cost_voucher - and row.serial_and_batch_bundle - and (row.serial_no or row.batch_no) - ): - self.validate_serial_nos_and_batches_with_bundle(row) - - if not row.serial_no and not row.batch_no and not row.get("rejected_serial_no"): - continue - - if not row.use_serial_batch_fields and ( - row.serial_no or row.batch_no or row.get("rejected_serial_no") - ): - row.use_serial_batch_fields = 1 - - if row.use_serial_batch_fields and ( - not row.serial_and_batch_bundle and not row.get("rejected_serial_and_batch_bundle") - ): - bundle_details = { - "item_code": row.get("rm_item_code") or row.item_code, - "posting_datetime": combine_datetime(self.posting_date, self.posting_time), - "voucher_type": self.doctype, - "voucher_no": self.name, - "voucher_detail_no": row.name, - "company": self.company, - "is_rejected": 1 if row.get("rejected_warehouse") else 0, - "use_serial_batch_fields": row.use_serial_batch_fields, - "via_landed_cost_voucher": via_landed_cost_voucher, - "do_not_submit": True if not via_landed_cost_voucher else False, - } - - if self.is_internal_transfer() and row.get("from_warehouse") and not self.is_return: - self.update_bundle_details(bundle_details, table_name, row) - bundle_details["type_of_transaction"] = "Outward" - bundle_details["warehouse"] = row.get("from_warehouse") - bundle_details["qty"] = row.get("stock_qty") or row.get("qty") - self.create_serial_batch_bundle(bundle_details, row) - continue - - if row.get("qty") or row.get("consumed_qty") or row.get("stock_qty"): - self.update_bundle_details(bundle_details, table_name, row, parent_details=parent_details) - self.create_serial_batch_bundle(bundle_details, row) - - if row.get("rejected_qty"): - self.update_bundle_details(bundle_details, table_name, row, is_rejected=True) - self.create_serial_batch_bundle(bundle_details, row) - - def get_parent_details_for_packed_items(self): - parent_details = frappe._dict() - for row in self.get("items"): - parent_details[row.name] = row - - return parent_details + return SerialBatchBundleService(self).make_bundle_using_old_serial_batch_fields( + table_name, via_landed_cost_voucher + ) def make_bundle_for_sales_purchase_return(self, table_name=None): - if not self.get("is_return"): - return + from erpnext.stock.services.serial_batch_bundle_service import SerialBatchBundleService - if not table_name: - table_name = "items" - - self.make_bundle_for_non_rejected_qty(table_name) - - if self.doctype in ["Purchase Invoice", "Purchase Receipt"]: - self.make_bundle_for_rejected_qty(table_name) - - def make_bundle_for_rejected_qty(self, table_name=None): - field, reference_ids = self.get_reference_ids( - table_name, "rejected_qty", "rejected_serial_and_batch_bundle" - ) - - if not reference_ids: - return - - child_doctype = self.doctype + " Item" - available_dict = available_serial_batch_for_return( - field, child_doctype, reference_ids, is_rejected=True - ) - - for row in self.get(table_name): - if data := available_dict.get(row.get(field)): - qty_field = "rejected_qty" - warehouse_field = "rejected_warehouse" - if row.get("return_qty_from_rejected_warehouse"): - qty_field = "qty" - warehouse_field = "warehouse" - - if not data.get("qty"): - frappe.throw( - _("For the {0}, no stock is available for the return in the warehouse {1}.").format( - frappe.bold(row.item_code), row.get(warehouse_field) - ) - ) - - data = filter_serial_batches( - self, data, row, warehouse_field=warehouse_field, qty_field=qty_field - ) - bundle = make_serial_batch_bundle_for_return(data, row, self, warehouse_field, qty_field) - if row.get("return_qty_from_rejected_warehouse"): - row.db_set( - { - "serial_and_batch_bundle": bundle, - "batch_no": "", - "serial_no": "", - } - ) - else: - row.db_set( - { - "rejected_serial_and_batch_bundle": bundle, - "batch_no": "", - "rejected_serial_no": "", - } - ) - - def make_bundle_for_non_rejected_qty(self, table_name): - field, reference_ids = self.get_reference_ids(table_name) - if not reference_ids: - return - - child_doctype = self.doctype + " Item" - if table_name == "packed_items": - field = "parent_detail_docname" - child_doctype = "Packed Item" - - available_dict = available_serial_batch_for_return(field, child_doctype, reference_ids) - - for row in self.get(table_name): - value = row.get(field) - if table_name == "packed_items" and row.get("parent_detail_docname"): - value = self.get_value_for_packed_item(row) - if not value: - continue - - if data := available_dict.get(value): - data = filter_serial_batches(self, data, row) - bundle = make_serial_batch_bundle_for_return(data, row, self) - row.db_set( - { - "serial_and_batch_bundle": bundle, - "batch_no": "", - "serial_no": "", - } - ) - - if self.doctype in ["Sales Invoice", "Delivery Note"]: - row.db_set( - "incoming_rate", frappe.db.get_value("Serial and Batch Bundle", bundle, "avg_rate") - ) - - def get_value_for_packed_item(self, row): - parent_items = self.get("items", {"name": row.parent_detail_docname}) - if parent_items: - ref = parent_items[0].get("dn_detail") - return (row.item_code, ref) - - return None - - def get_reference_ids(self, table_name, qty_field=None, bundle_field=None) -> tuple[str, list[str]]: - field = { - "Sales Invoice": "sales_invoice_item", - "Delivery Note": "dn_detail", - "Purchase Receipt": "purchase_receipt_item", - "Purchase Invoice": "purchase_invoice_item", - "POS Invoice": "pos_invoice_item", - }.get(self.doctype) - - if not bundle_field: - bundle_field = "serial_and_batch_bundle" - - if not qty_field: - qty_field = "qty" - - reference_ids = [] - - for row in self.get(table_name): - if not self.is_serial_batch_item(row.item_code): - continue - - if ( - row.get(field) - and ( - qty_field == "qty" - and not row.get("return_qty_from_rejected_warehouse") - or qty_field == "rejected_qty" - and (row.get("return_qty_from_rejected_warehouse") or row.get("rejected_warehouse")) - ) - and not row.get("use_serial_batch_fields") - and not row.get(bundle_field) - ): - reference_ids.append(row.get(field)) - - if table_name == "packed_items" and row.get("parent_detail_docname"): - parent_rows = self.get("items", {"name": row.parent_detail_docname}) or [] - for d in parent_rows: - if d.get(field) and not d.get(bundle_field): - reference_ids.append(d.get(field)) - - return field, reference_ids - - @frappe.request_cache - def is_serial_batch_item(self, item_code) -> bool: - if not frappe.db.exists("Item", item_code): - frappe.throw(_("Item {0} does not exist.").format(bold(item_code))) - - item_details = frappe.db.get_value("Item", item_code, ["has_serial_no", "has_batch_no"], as_dict=1) - - if item_details.has_serial_no or item_details.has_batch_no: - return True - - return False - - def update_bundle_details(self, bundle_details, table_name, row, is_rejected=False, parent_details=None): - from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos - - # Since qty field is different for different doctypes - qty = row.get("qty") - warehouse = row.get("warehouse") - - if table_name == "packed_items": - type_of_transaction = "Inward" - if not self.is_return: - type_of_transaction = "Outward" - elif table_name == "supplied_items": - qty = row.consumed_qty - warehouse = self.supplier_warehouse - type_of_transaction = "Outward" - if self.is_return: - type_of_transaction = "Inward" - else: - type_of_transaction = get_type_of_transaction(self, row) - - if hasattr(row, "stock_qty"): - qty = row.stock_qty - - if self.doctype == "Stock Entry": - qty = row.transfer_qty - warehouse = row.s_warehouse or row.t_warehouse - - serial_nos = row.serial_no - if is_rejected: - serial_nos = row.get("rejected_serial_no") - type_of_transaction = "Inward" if not self.is_return else "Outward" - qty = flt( - row.get("rejected_qty") * row.get("conversion_factor", 1.0), - frappe.get_precision("Serial and Batch Entry", "qty"), - ) - warehouse = row.get("rejected_warehouse") - - if ( - self.is_internal_transfer() - and self.doctype in ["Sales Invoice", "Delivery Note"] - and self.is_return - ): - warehouse = row.get("target_warehouse") or row.get("warehouse") - type_of_transaction = "Outward" - - if table_name == "packed_items": - if not warehouse: - warehouse = parent_details[row.parent_detail_docname].warehouse - bundle_details["voucher_detail_no"] = parent_details[row.parent_detail_docname].name - - bundle_details.update( - { - "qty": qty, - "is_rejected": is_rejected, - "type_of_transaction": type_of_transaction, - "warehouse": warehouse, - "batches": frappe._dict({row.batch_no: qty}) if row.batch_no else None, - "serial_nos": get_serial_nos(serial_nos) if serial_nos else None, - "batch_no": row.batch_no, - } - ) - - def create_serial_batch_bundle(self, bundle_details, row): - from erpnext.stock.serial_batch_bundle import SerialBatchCreation - - sn_doc = SerialBatchCreation(bundle_details).make_serial_and_batch_bundle() - - field = "serial_and_batch_bundle" - if bundle_details.get("is_rejected"): - field = "rejected_serial_and_batch_bundle" - - row.set(field, sn_doc.name) - row.db_set({field: sn_doc.name}) - - def validate_serial_nos_and_batches_with_bundle(self, row): - from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos - - throw_error = False - if row.serial_no: - serial_nos = frappe.get_all( - "Serial and Batch Entry", - fields=["serial_no"], - filters={"parent": row.serial_and_batch_bundle}, - ) - serial_nos = sorted([cstr(d.serial_no) for d in serial_nos]) - parsed_serial_nos = get_serial_nos(row.serial_no) - - if len(serial_nos) != len(parsed_serial_nos): - throw_error = True - elif serial_nos != parsed_serial_nos: - for serial_no in serial_nos: - if serial_no not in parsed_serial_nos: - throw_error = True - break - - elif row.batch_no: - batches = sorted( - frappe.get_all( - "Serial and Batch Entry", - filters={"parent": row.serial_and_batch_bundle}, - pluck="batch_no", - distinct=True, - ) - ) - - if batches != [row.batch_no]: - throw_error = True - - if throw_error: - frappe.throw( - _( - "At row {0}: Serial and Batch Bundle {1} has already created. Please remove the values from the serial no or batch no fields." - ).format(row.idx, row.serial_and_batch_bundle) - ) + return SerialBatchBundleService(self).make_bundle_for_sales_purchase_return(table_name) def set_use_serial_batch_fields(self): - if frappe.get_single_value("Stock Settings", "use_serial_batch_fields"): - for row in self.items: - row.use_serial_batch_fields = 1 + from erpnext.stock.services.serial_batch_bundle_service import SerialBatchBundleService + + return SerialBatchBundleService(self).set_use_serial_batch_fields() def get_gl_entries( self, inventory_account_map=None, default_expense_account=None, default_cost_center=None @@ -696,527 +228,54 @@ class StockController(AccountsController): inventory_account_map, default_expense_account, default_cost_center ) - def get_debit_field_precision(self): - if not frappe.flags.debit_field_precision: - frappe.flags.debit_field_precision = frappe.get_precision("GL Entry", "debit_in_account_currency") - - return frappe.flags.debit_field_precision - - def get_voucher_details(self, default_expense_account, default_cost_center, sle_map): - if self.doctype == "Stock Reconciliation": - reconciliation_purpose = frappe.db.get_value(self.doctype, self.name, "purpose") - is_opening = "Yes" if reconciliation_purpose == "Opening Stock" else "No" - details = [] - for voucher_detail_no in sle_map: - details.append( - frappe._dict( - { - "name": voucher_detail_no, - "expense_account": default_expense_account, - "cost_center": default_cost_center, - "is_opening": is_opening, - } - ) - ) - return details - else: - details = self.get("items") - - if default_expense_account or default_cost_center: - for d in details: - if default_expense_account and not d.get("expense_account"): - d.expense_account = default_expense_account - if default_cost_center and not d.get("cost_center"): - d.cost_center = default_cost_center - - return details - def get_items_and_warehouses(self) -> tuple[list[str], list[str]]: - """Get list of items and warehouses affected by a transaction""" + from erpnext.stock.services.stock_ledger_service import StockLedgerService - if not (hasattr(self, "items") or hasattr(self, "packed_items")): - return [], [] - - item_rows = (self.get("items") or []) + (self.get("packed_items") or []) - - items = {d.item_code for d in item_rows if d.item_code} - - warehouses = set() - for d in item_rows: - if d.get("warehouse"): - warehouses.add(d.warehouse) - - if self.doctype == "Stock Entry": - if d.get("s_warehouse"): - warehouses.add(d.s_warehouse) - if d.get("t_warehouse"): - warehouses.add(d.t_warehouse) - - return list(items), list(warehouses) + return StockLedgerService(self).get_items_and_warehouses() def get_stock_ledger_details(self): - stock_ledger = {} + from erpnext.stock.services.stock_ledger_service import StockLedgerService - table = frappe.qb.DocType("Stock Ledger Entry") - - stock_ledger_entries = ( - frappe.qb.from_(table) - .select( - table.name, - table.warehouse, - table.stock_value_difference, - table.valuation_rate, - table.voucher_detail_no, - table.item_code, - table.posting_date, - table.posting_time, - table.actual_qty, - table.qty_after_transaction, - table.project, - ) - .where( - (table.voucher_type == self.doctype) - & (table.voucher_no == self.name) - & (table.is_cancelled == 0) - ) - ).run(as_dict=True) - - for sle in stock_ledger_entries: - stock_ledger.setdefault(sle.voucher_detail_no, []).append(sle) - - return stock_ledger - - def check_expense_account(self, item): - if not item.get("expense_account"): - msg = _("Please set an Expense Account in the Items table") - frappe.throw( - _("Row #{0}: Expense Account not set for the Item {1}. {2}").format( - item.idx, frappe.bold(item.item_code), msg - ), - title=_("Expense Account Missing"), - ) - - else: - is_expense_account = ( - frappe.get_cached_value("Account", item.get("expense_account"), "report_type") - == "Profit and Loss" - ) - if ( - self.doctype - not in ( - "Purchase Receipt", - "Purchase Invoice", - "Stock Reconciliation", - "Stock Entry", - "Subcontracting Receipt", - "Delivery Note", - ) - and not is_expense_account - ): - frappe.throw( - _("Expense / Difference account ({0}) must be a 'Profit or Loss' account").format( - item.get("expense_account") - ) - ) - if is_expense_account and not item.get("cost_center"): - frappe.throw( - _("{0} {1}: Cost Center is mandatory for Item {2}").format( - _(self.doctype), self.name, item.get("item_code") - ) - ) + return StockLedgerService(self).get_stock_ledger_details() def delete_auto_created_batches(self): - for table_name in ["items", "packed_items", "supplied_items"]: - if not self.get(table_name): - continue + from erpnext.stock.services.serial_batch_bundle_service import SerialBatchBundleService - for row in self.get(table_name): - update_values = {} - if row.get("batch_no"): - update_values["batch_no"] = None - - if row.get("serial_and_batch_bundle"): - update_values["serial_and_batch_bundle"] = None - frappe.db.set_value( - "Serial and Batch Bundle", row.serial_and_batch_bundle, {"is_cancelled": 1} - ) - - frappe.db.set_value( - "Serial and Batch Entry", {"parent": row.serial_and_batch_bundle}, {"is_cancelled": 1} - ) - - if update_values: - row.db_set(update_values) - - if table_name == "items" and row.get("rejected_serial_and_batch_bundle"): - frappe.db.set_value( - "Serial and Batch Bundle", row.rejected_serial_and_batch_bundle, {"is_cancelled": 1} - ) - - frappe.db.set_value( - "Serial and Batch Entry", - {"parent": row.rejected_serial_and_batch_bundle}, - {"is_cancelled": 1}, - ) - - row.db_set("rejected_serial_and_batch_bundle", None) - - if row.get("current_serial_and_batch_bundle"): - row.db_set("current_serial_and_batch_bundle", None) + return SerialBatchBundleService(self).delete_auto_created_batches() def set_serial_and_batch_bundle(self, table_name=None, ignore_validate=False): - if not table_name: - table_name = "items" + from erpnext.stock.services.serial_batch_bundle_service import SerialBatchBundleService - QTY_FIELD = { - "serial_and_batch_bundle": "qty", - "current_serial_and_batch_bundle": "current_qty", - "rejected_serial_and_batch_bundle": "rejected_qty", - } - - for row in self.get(table_name): - for field in QTY_FIELD.keys(): - if row.get(field): - frappe.get_doc("Serial and Batch Bundle", row.get(field)).set_serial_and_batch_values( - self, row, qty_field=QTY_FIELD[field] - ) + return SerialBatchBundleService(self).set_serial_and_batch_bundle(table_name, ignore_validate) def make_package_for_transfer( self, serial_and_batch_bundle, warehouse, type_of_transaction=None, do_not_submit=None, qty=0 ): - return make_bundle_for_material_transfer( - is_new=self.is_new(), - docstatus=self.docstatus, - voucher_type=self.doctype, - voucher_no=self.name, - serial_and_batch_bundle=serial_and_batch_bundle, - warehouse=warehouse, - type_of_transaction=type_of_transaction, - do_not_submit=do_not_submit, - qty=qty, + from erpnext.stock.services.serial_batch_bundle_service import SerialBatchBundleService + + return SerialBatchBundleService(self).make_package_for_transfer( + serial_and_batch_bundle, warehouse, type_of_transaction, do_not_submit, qty ) def get_sl_entries(self, d, args): - sl_dict = frappe._dict( - { - "item_code": d.get("item_code", None), - "warehouse": d.get("warehouse", None), - "serial_and_batch_bundle": d.get("serial_and_batch_bundle"), - "posting_date": self.posting_date, - "posting_time": self.posting_time, - "fiscal_year": get_fiscal_year(self.posting_date, company=self.company)[0], - "voucher_type": self.doctype, - "voucher_no": self.name, - "voucher_detail_no": d.name, - "actual_qty": (self.docstatus == 1 and 1 or -1) * flt(d.get("stock_qty")), - "stock_uom": frappe.get_cached_value( - "Item", args.get("item_code") or d.get("item_code"), "stock_uom" - ), - "incoming_rate": 0, - "company": self.company, - "project": d.get("project") or self.get("project"), - "is_cancelled": 1 if self.docstatus == 2 else 0, - } - ) + from erpnext.stock.services.stock_ledger_service import StockLedgerService - sl_dict.update(args) - self.update_inventory_dimensions(d, sl_dict) - - if self.docstatus == 2: - from erpnext.deprecation_dumpster import deprecation_warning - - deprecation_warning("unknown", "v16", "No instructions.") - # To handle denormalized serial no records, will br deprecated in v16 - for field in ["serial_no", "batch_no"]: - if d.get(field): - sl_dict[field] = d.get(field) - - return sl_dict - - def set_landed_cost_voucher_amount(self): - for d in self.get("items"): - lcv_item = frappe.qb.DocType("Landed Cost Item") - query = ( - frappe.qb.from_(lcv_item) - .select(Sum(lcv_item.applicable_charges), lcv_item.cost_center) - .where((lcv_item.docstatus == 1) & (lcv_item.receipt_document == self.name)) - ) - - if self.doctype == "Stock Entry": - query = query.where(lcv_item.stock_entry_item == d.name) - else: - query = query.where(lcv_item.purchase_receipt_item == d.name) - - lc_voucher_data = query.run(as_list=True) - - d.landed_cost_voucher_amount = lc_voucher_data[0][0] if lc_voucher_data else 0.0 - if not d.cost_center and lc_voucher_data and lc_voucher_data[0][1]: - d.db_set("cost_center", lc_voucher_data[0][1]) - - def has_landed_cost_amount(self): - for row in self.items: - if row.get("landed_cost_voucher_amount"): - return True - - return False + return StockLedgerService(self).get_sl_entries(d, args) def get_item_account_wise_lcv_entries(self): - if not self.has_landed_cost_amount(): - return - - landed_cost_vouchers = frappe.get_all( - "Landed Cost Purchase Receipt", - fields=["parent"], - filters={"receipt_document": self.name, "docstatus": 1}, + from erpnext.stock.doctype.landed_cost_voucher.landed_cost_voucher import ( + get_item_account_wise_lcv_entries, ) - if not landed_cost_vouchers: - return - - item_account_wise_cost = {} - - row_fieldname = "purchase_receipt_item" - if self.doctype == "Stock Entry": - row_fieldname = "stock_entry_item" - - for lcv in landed_cost_vouchers: - landed_cost_voucher_doc = frappe.get_doc("Landed Cost Voucher", lcv.parent) - - based_on_field = "applicable_charges" - # Use amount field for total item cost for manually cost distributed LCVs - if landed_cost_voucher_doc.distribute_charges_based_on != "Distribute Manually": - based_on_field = frappe.scrub(landed_cost_voucher_doc.distribute_charges_based_on) - - total_item_cost = 0 - - if based_on_field: - for item in landed_cost_voucher_doc.items: - total_item_cost += item.get(based_on_field) - - for item in landed_cost_voucher_doc.items: - if item.receipt_document == self.name: - for account in landed_cost_voucher_doc.taxes: - exchange_rate = account.exchange_rate or 1 - item_account_wise_cost.setdefault((item.item_code, item.get(row_fieldname)), {}) - item_account_wise_cost[(item.item_code, item.get(row_fieldname))].setdefault( - account.expense_account, {"amount": 0.0, "base_amount": 0.0} - ) - - item_row = item_account_wise_cost[(item.item_code, item.get(row_fieldname))][ - account.expense_account - ] - - if total_item_cost > 0: - item_row["amount"] += account.amount * item.get(based_on_field) / total_item_cost - - item_row["base_amount"] += ( - account.base_amount * item.get(based_on_field) / total_item_cost - ) - else: - item_row["amount"] += item.applicable_charges / exchange_rate - item_row["base_amount"] += item.applicable_charges - - return item_account_wise_cost - - def update_inventory_dimensions(self, row, sl_dict) -> None: - # To handle delivery note and sales invoice - if row.get("item_row"): - row = row.get("item_row") - - dimensions = get_evaluated_inventory_dimension(row, sl_dict, parent_doc=self) - for dimension in dimensions: - if not dimension: - continue - - if ( - self.doctype in ["Purchase Invoice", "Purchase Receipt"] - and row.get("rejected_warehouse") - and sl_dict.get("warehouse") == row.get("rejected_warehouse") - ): - fieldname = f"rejected_{dimension.source_fieldname}" - sl_dict[dimension.target_fieldname] = row.get(fieldname) - continue - - if self.doctype in [ - "Purchase Invoice", - "Purchase Receipt", - "Sales Invoice", - "Delivery Note", - "Stock Entry", - ]: - if ( - ( - sl_dict.actual_qty > 0 - and not self.get("is_return") - or sl_dict.actual_qty < 0 - and self.get("is_return") - ) - and self.doctype in ["Purchase Invoice", "Purchase Receipt", "Stock Entry"] - ) or ( - ( - sl_dict.actual_qty < 0 - and not self.get("is_return") - or sl_dict.actual_qty > 0 - and self.get("is_return") - ) - and self.doctype in ["Sales Invoice", "Delivery Note", "Stock Entry"] - ): - if self.doctype == "Stock Entry": - if row.get("t_warehouse") == sl_dict.warehouse and sl_dict.get("actual_qty") > 0: - fieldname = f"to_{dimension.source_fieldname}" - if dimension.source_fieldname.startswith("to_"): - fieldname = f"{dimension.source_fieldname}" - - sl_dict[dimension.target_fieldname] = row.get(fieldname) - continue - - sl_dict[dimension.target_fieldname] = row.get(dimension.source_fieldname) - else: - fieldname_start_with = "to" - if self.doctype in ["Purchase Invoice", "Purchase Receipt"]: - fieldname_start_with = "from" - - fieldname = f"{fieldname_start_with}_{dimension.source_fieldname}" - sl_dict[dimension.target_fieldname] = row.get(fieldname) - - if not sl_dict.get(dimension.target_fieldname): - sl_dict[dimension.target_fieldname] = row.get(dimension.source_fieldname) - - elif row.get(dimension.source_fieldname): - sl_dict[dimension.target_fieldname] = row.get(dimension.source_fieldname) - - if not sl_dict.get(dimension.target_fieldname) and dimension.fetch_from_parent: - sl_dict[dimension.target_fieldname] = self.get(dimension.fetch_from_parent) - - # Get value based on doctype name - if not sl_dict.get(dimension.target_fieldname): - fieldname = next( - ( - field.fieldname - for field in frappe.get_meta(self.doctype).fields - if field.options == dimension.fetch_from_parent - ), - None, - ) - - if fieldname and self.get(fieldname): - sl_dict[dimension.target_fieldname] = self.get(fieldname) - - if sl_dict[dimension.target_fieldname] and self.docstatus == 1: - row.db_set(dimension.source_fieldname, sl_dict[dimension.target_fieldname]) + return get_item_account_wise_lcv_entries(self) def make_sl_entries(self, sl_entries, allow_negative_stock=False, via_landed_cost_voucher=False): - from erpnext.stock.serial_batch_bundle import update_batch_qty - from erpnext.stock.stock_ledger import make_sl_entries + from erpnext.stock.services.stock_ledger_service import StockLedgerService - make_sl_entries(sl_entries, allow_negative_stock, via_landed_cost_voucher) - update_batch_qty( - self.doctype, self.name, self.docstatus, via_landed_cost_voucher=via_landed_cost_voucher + return StockLedgerService(self).make_sl_entries( + sl_entries, allow_negative_stock, via_landed_cost_voucher ) - self.validate_reserved_batches() - - def validate_reserved_batches(self): - if not frappe.db.get_single_value("Stock Settings", "enable_stock_reservation"): - return - - if self.doctype not in ["Delivery Note", "Sales Invoice", "Stock Entry"]: - return - - batches = frappe.get_all( - "Serial and Batch Entry", - filters={ - "voucher_type": self.doctype, - "voucher_no": self.name, - "docstatus": 1, - "batch_no": ("is", "set"), - "qty": ("<", 0), - }, - pluck="batch_no", - ) - - if not batches: - return - - field_mapper = { - "Sales Invoice": [["Sales Order", "sales_order"]], - "Delivery Note": [["Sales Order", "against_sales_order"]], - "Stock Entry": [ - ["Work Order", "work_order"], - ["Subcontracting Inward Order", "subcontracting_inward_order"], - ], - }.get(self.doctype) - - qty_field = { - "Sales Invoice": "qty", - "Delivery Note": "qty", - "Stock Entry": "fg_completed_qty", - }.get(self.doctype) - - reserved_batches_data = self.get_reserved_batches(batches) - items = self.items - if self.doctype == "Stock Entry": - items = [self] - - for item in items: - for field in field_mapper: - if not item.get(field[1]): - continue - - value = item.get(field[1]) - for row in reserved_batches_data: - if self.doctype in ["Sales Invoice", "Delivery Note"] and row.item_code != item.get( - "item_code" - ): - continue - - if row.voucher_no == value: - continue - - batch_qty = get_batch_qty( - row.batch_no, - row.warehouse, - posting_date=self.posting_date, - posting_time=self.posting_time, - consider_negative_batches=True, - ) - - if item.get(qty_field) < batch_qty: - continue - - frappe.throw( - _( - "The batch {0} is already reserved in {1} {2}. So, cannot proceed with the {3} {4}, which is created against the {5} {6}." - ).format( - frappe.bold(row.batch_no), - frappe.bold(row.voucher_type), - frappe.bold(row.voucher_no), - frappe.bold(self.doctype), - frappe.bold(self.name), - frappe.bold(field[0]), - frappe.bold(value), - ), - title=_("Reserved Batch Conflict"), - ) - - def get_reserved_batches(self, batches): - doctype = frappe.qb.DocType("Stock Reservation Entry") - child_doc = frappe.qb.DocType("Serial and Batch Entry") - - return ( - frappe.qb.from_(doctype) - .join(child_doc) - .on(doctype.name == child_doc.parent) - .select( - child_doc.batch_no, - doctype.voucher_type, - doctype.voucher_no, - doctype.item_code, - doctype.warehouse, - ) - .where((doctype.docstatus == 1) & (child_doc.batch_no.isin(batches))) - ).run(as_dict=True) - def make_gl_entries_on_cancel(self, from_repost=False): if not from_repost: cancel_exchange_gain_loss_journal(frappe._dict(doctype=self.doctype, name=self.name)) @@ -1227,18 +286,6 @@ class StockController(AccountsController): ): self.make_gl_entries() - def get_serialized_items(self): - serialized_items = [] - item_codes = list(set(d.item_code for d in self.get("items"))) - if item_codes: - serialized_items = frappe.db.sql_list( - """select name from `tabItem` - where has_serial_no=1 and name in ({})""".format(", ".join(["%s"] * len(item_codes))), - tuple(item_codes), - ) - - return serialized_items - def validate_warehouse(self): from erpnext.stock.utils import validate_disabled_warehouse, validate_warehouse_company @@ -1284,92 +331,9 @@ class StockController(AccountsController): ) def validate_inspection(self): - """Checks if quality inspection is set/ is valid for Items that require inspection.""" - inspection_fieldname_map = { - "Purchase Receipt": "inspection_required_before_purchase", - "Purchase Invoice": "inspection_required_before_purchase", - "Subcontracting Receipt": "inspection_required_before_purchase", - "Sales Invoice": "inspection_required_before_delivery", - "Delivery Note": "inspection_required_before_delivery", - } - inspection_required_fieldname = inspection_fieldname_map.get(self.doctype) + from erpnext.stock.services.quality_inspection_service import QualityInspectionService - # return if inspection is not required on document level - if ( - (not inspection_required_fieldname and self.doctype != "Stock Entry") - or (self.doctype == "Stock Entry" and not self.inspection_required) - or (self.doctype in ["Sales Invoice", "Purchase Invoice"] and not self.update_stock) - ): - return - - for row in self.get("items"): - qi_required = False - if inspection_required_fieldname and frappe.get_cached_value( - "Item", row.item_code, inspection_required_fieldname - ): - qi_required = True - elif self.doctype == "Stock Entry" and row.t_warehouse: - qi_required = True # inward stock needs inspection - - if row.get("secondary_item_type") or row.get("is_legacy_scrap_item"): - continue - - if qi_required: # validate row only if inspection is required on item level - if self.doctype in [ - "Purchase Receipt", - "Purchase Invoice", - "Sales Invoice", - "Delivery Note", - ] and frappe.get_single_value( - "Stock Settings", "allow_to_make_quality_inspection_after_purchase_or_delivery" - ): - return - - self.validate_qi_presence(row) - if self.docstatus == 1: - self.validate_qi_submission(row) - self.validate_qi_rejection(row) - - def validate_qi_presence(self, row): - """Check if QI is present on row level. Warn on save and stop on submit if missing.""" - if not row.quality_inspection: - msg = _("Row #{0}: Quality Inspection is required for Item {1}").format( - row.idx, frappe.bold(row.item_code) - ) - if self.docstatus == 1: - frappe.throw(msg, title=_("Inspection Required"), exc=QualityInspectionRequiredError) - else: - frappe.msgprint(msg, title=_("Inspection Required"), indicator="blue") - - def validate_qi_submission(self, row): - """Check if QI is submitted on row level, during submission""" - action = frappe.get_single_value("Stock Settings", "action_if_quality_inspection_is_not_submitted") - qa_docstatus = frappe.db.get_value("Quality Inspection", row.quality_inspection, "docstatus") - - if qa_docstatus != 1: - link = frappe.utils.get_link_to_form("Quality Inspection", row.quality_inspection) - msg = _("Row #{0}: Quality Inspection {1} is not submitted for the item: {2}").format( - row.idx, link, row.item_code - ) - if action == "Stop": - frappe.throw(msg, title=_("Inspection Submission"), exc=QualityInspectionNotSubmittedError) - else: - frappe.msgprint(msg, alert=True, indicator="orange") - - def validate_qi_rejection(self, row): - """Check if QI is rejected on row level, during submission""" - action = frappe.get_single_value("Stock Settings", "action_if_quality_inspection_is_rejected") - qa_status = frappe.db.get_value("Quality Inspection", row.quality_inspection, "status") - - if qa_status == "Rejected": - link = frappe.utils.get_link_to_form("Quality Inspection", row.quality_inspection) - msg = _("Row #{0}: Quality Inspection {1} was rejected for item {2}").format( - row.idx, link, row.item_code - ) - if action == "Stop": - frappe.throw(msg, title=_("Inspection Rejected"), exc=QualityInspectionRejectedError) - else: - frappe.msgprint(msg, alert=True, indicator="orange") + return QualityInspectionService(self).validate_inspection() def update_blanket_order(self): blanket_orders = list(set([d.blanket_order for d in self.items if d.blanket_order])) @@ -1395,254 +359,10 @@ class StockController(AccountsController): for d in self.get("items"): d.stock_uom_rate = d.rate / (d.conversion_factor or 1) - def validate_internal_transfer(self): - if self.doctype in ("Sales Invoice", "Delivery Note", "Purchase Invoice", "Purchase Receipt"): - if self.is_internal_transfer(): - self.validate_in_transit_warehouses() - self.validate_multi_currency() - self.validate_packed_items() - - if self.get("is_internal_supplier") and self.docstatus == 1: - self.validate_internal_transfer_qty() - else: - self.validate_internal_transfer_warehouse() - - def validate_internal_transfer_warehouse(self): - for row in self.items: - if row.get("target_warehouse"): - row.target_warehouse = None - - if row.get("from_warehouse"): - row.from_warehouse = None - - def validate_in_transit_warehouses(self): - if (self.doctype == "Sales Invoice" and self.get("update_stock")) or self.doctype == "Delivery Note": - for item in self.get("items"): - if not item.target_warehouse: - frappe.throw( - _("Row {0}: Target Warehouse is mandatory for internal transfers").format(item.idx) - ) - - if ( - self.doctype == "Purchase Invoice" and self.get("update_stock") - ) or self.doctype == "Purchase Receipt": - for item in self.get("items"): - if not item.from_warehouse: - frappe.throw( - _("Row {0}: From Warehouse is mandatory for internal transfers").format(item.idx) - ) - - def validate_multi_currency(self): - if self.currency != self.company_currency: - frappe.throw(_("Internal transfers can only be done in company's default currency")) - - def validate_packed_items(self): - if self.doctype in ("Sales Invoice", "Delivery Note Item") and self.get("packed_items"): - frappe.throw(_("Packed Items cannot be transferred internally")) - - def validate_internal_transfer_qty(self): - if self.doctype not in ["Purchase Invoice", "Purchase Receipt"]: - return - - self.__inter_company_reference = ( - self.get("inter_company_reference") - if self.doctype == "Purchase Invoice" - else self.get("inter_company_invoice_reference") - ) - - item_wise_transfer_qty = self.get_item_wise_inter_transfer_qty() - if not item_wise_transfer_qty: - return - - item_wise_received_qty = self.get_item_wise_inter_received_qty() - precision = frappe.get_precision(self.doctype + " Item", "qty") - - over_receipt_allowance = frappe.get_single_value("Stock Settings", "over_delivery_receipt_allowance") - - parent_doctype = { - "Purchase Receipt": "Delivery Note", - "Purchase Invoice": "Sales Invoice", - }.get(self.doctype) - - for key, transferred_qty in item_wise_transfer_qty.items(): - recevied_qty = flt(item_wise_received_qty.get(key), precision) - if over_receipt_allowance: - transferred_qty = transferred_qty + flt( - transferred_qty * over_receipt_allowance / 100, precision - ) - - if recevied_qty > flt(transferred_qty, precision): - frappe.throw( - _("For Item {0} cannot be received more than {1} qty against the {2} {3}").format( - bold(key[1]), - bold(flt(transferred_qty, precision)), - bold(parent_doctype), - get_link_to_form(parent_doctype, self.__inter_company_reference), - ) - ) - - def get_item_wise_inter_transfer_qty(self): - parent_doctype = { - "Purchase Receipt": "Delivery Note", - "Purchase Invoice": "Sales Invoice", - }.get(self.doctype) - - child_doctype = parent_doctype + " Item" - - parent_tab = frappe.qb.DocType(parent_doctype) - child_tab = frappe.qb.DocType(child_doctype) - - query = ( - frappe.qb.from_(parent_doctype) - .inner_join(child_tab) - .on(child_tab.parent == parent_tab.name) - .select( - child_tab.name, - child_tab.item_code, - child_tab.qty, - ) - .where((parent_tab.name == self.__inter_company_reference) & (parent_tab.docstatus == 1)) - ) - - data = query.run(as_dict=True) - item_wise_transfer_qty = defaultdict(float) - for row in data: - item_wise_transfer_qty[(row.name, row.item_code)] += flt(row.qty) - - return item_wise_transfer_qty - - def get_item_wise_inter_received_qty(self): - child_doctype = self.doctype + " Item" - - parent_tab = frappe.qb.DocType(self.doctype) - child_tab = frappe.qb.DocType(child_doctype) - - query = ( - frappe.qb.from_(self.doctype) - .inner_join(child_tab) - .on(child_tab.parent == parent_tab.name) - .select( - child_tab.item_code, - child_tab.qty, - ) - .where(parent_tab.docstatus == 1) - ) - - if self.doctype == "Purchase Invoice": - query = query.select( - child_tab.sales_invoice_item.as_("name"), - ) - - query = query.where( - parent_tab.inter_company_invoice_reference == self.inter_company_invoice_reference - ) - else: - query = query.select( - child_tab.delivery_note_item.as_("name"), - ) - - query = query.where(parent_tab.inter_company_reference == self.inter_company_reference) - - data = query.run(as_dict=True) - item_wise_transfer_qty = defaultdict(float) - for row in data: - item_wise_transfer_qty[(row.name, row.item_code)] += flt(row.qty) - - return item_wise_transfer_qty - - def validate_putaway_capacity(self): - # if over receipt is attempted while 'apply putaway rule' is disabled - # and if rule was applied on the transaction, validate it. - from erpnext.stock.doctype.putaway_rule.putaway_rule import get_available_putaway_capacity - - valid_doctype = self.doctype in ( - "Purchase Receipt", - "Stock Entry", - "Purchase Invoice", - "Stock Reconciliation", - ) - - if not frappe.get_all("Putaway Rule", limit=1): - return - - if self.doctype == "Purchase Invoice" and self.get("update_stock") == 0: - valid_doctype = False - - if valid_doctype: - rule_map = defaultdict(dict) - for item in self.get("items"): - warehouse_field = "t_warehouse" if self.doctype == "Stock Entry" else "warehouse" - rule = frappe.db.get_value( - "Putaway Rule", - {"item_code": item.get("item_code"), "warehouse": item.get(warehouse_field)}, - ["stock_capacity", "name", "disable"], - as_dict=True, - ) - if rule: - if rule.get("disabled"): - continue # dont validate for disabled rule - - if self.doctype == "Stock Reconciliation": - stock_qty = flt(item.qty) - else: - stock_qty = ( - flt(item.transfer_qty) if self.doctype == "Stock Entry" else flt(item.stock_qty) - ) - - rule_name = rule.get("name") - if not rule_map[rule_name]: - rule_map[rule_name]["warehouse"] = item.get(warehouse_field) - rule_map[rule_name]["item"] = item.get("item_code") - rule_map[rule_name]["qty_put"] = 0 - rule_map[rule_name]["capacity"] = ( - rule.stock_capacity - if self.doctype == "Stock Reconciliation" - else get_available_putaway_capacity(rule_name) - ) - rule_map[rule_name]["qty_put"] += flt(stock_qty) - - for rule, values in rule_map.items(): - if flt(values["qty_put"]) > flt(values["capacity"]): - message = self.prepare_over_receipt_message(rule, values) - frappe.throw(msg=message, title=_("Over Receipt")) - - def prepare_over_receipt_message(self, rule, values): - message = _("{0} qty of Item {1} is being received into Warehouse {2} with capacity {3}.").format( - frappe.bold(values["qty_put"]), - frappe.bold(values["item"]), - frappe.bold(values["warehouse"]), - frappe.bold(values["capacity"]), - ) - message += "

" - rule_link = frappe.utils.get_link_to_form("Putaway Rule", rule) - message += _("Please adjust the qty or edit {0} to proceed.").format(rule_link) - return message - def repost_future_sle_and_gle(self, force=False, via_landed_cost_voucher=False): - args = frappe._dict( - { - "posting_date": self.posting_date, - "posting_time": self.posting_time, - "voucher_type": self.doctype, - "voucher_no": self.name, - "company": self.company, - "via_landed_cost_voucher": via_landed_cost_voucher, - } - ) + from erpnext.stock.services.stock_ledger_service import StockLedgerService - if self.docstatus == 2: - force = True - - if force or future_sle_exists(args) or repost_required_for_queue(self): - item_based_reposting = frappe.get_single_value("Stock Reposting Settings", "item_based_reposting") - if item_based_reposting: - create_item_wise_repost_entries( - voucher_type=self.doctype, - voucher_no=self.name, - via_landed_cost_voucher=via_landed_cost_voucher, - ) - else: - create_repost_item_valuation_entry(args) + return StockLedgerService(self).repost_future_sle_and_gle(force, via_landed_cost_voucher) def add_gl_entry( self, @@ -1843,6 +563,8 @@ class StockController(AccountsController): @frappe.whitelist() def show_accounting_ledger_preview(company: str, doctype: str, docname: str): + from erpnext.controllers.ledger_preview import get_accounting_ledger_preview + filters = frappe._dict(company=company, include_dimensions=1) doc = frappe.get_lazy_doc(doctype, docname) doc.run_method("before_gl_preview") @@ -1856,6 +578,8 @@ def show_accounting_ledger_preview(company: str, doctype: str, docname: str): @frappe.whitelist() def show_stock_ledger_preview(company: str, doctype: str, docname: str): + from erpnext.controllers.ledger_preview import get_stock_ledger_preview + filters = frappe._dict(company=company) doc = frappe.get_lazy_doc(doctype, docname) doc.run_method("before_sl_preview") @@ -1870,122 +594,6 @@ def show_stock_ledger_preview(company: str, doctype: str, docname: str): } -def get_accounting_ledger_preview(doc, filters): - from erpnext.accounts.report.general_ledger.general_ledger import get_columns as get_gl_columns - - gl_columns, gl_data = [], [] - fields = [ - "posting_date", - "account", - "debit", - "credit", - "against", - "party_type", - "party", - "cost_center", - "against_voucher_type", - "against_voucher", - ] - - doc.docstatus = 1 - - if doc.get("update_stock") or doc.doctype in ("Purchase Receipt", "Delivery Note", "Stock Entry"): - doc.update_stock_ledger() - - doc.make_gl_entries() - columns = get_gl_columns(filters) - gl_entries = get_gl_entries_for_preview(doc.doctype, doc.name, fields) - - gl_columns = get_columns(columns, fields) - gl_data = get_data(fields, gl_entries) - - return gl_columns, gl_data - - -def get_stock_ledger_preview(doc, filters): - from erpnext.stock.report.stock_ledger.stock_ledger import get_columns as get_sl_columns - - sl_columns, sl_data = [], [] - fields = [ - "item_code", - "stock_uom", - "actual_qty", - "qty_after_transaction", - "warehouse", - "incoming_rate", - "valuation_rate", - "stock_value", - "stock_value_difference", - ] - columns_fields = [ - "item_code", - "stock_uom", - "in_qty", - "out_qty", - "qty_after_transaction", - "warehouse", - "incoming_rate", - "in_out_rate", - "stock_value", - "stock_value_difference", - ] - - if doc.get("update_stock") or doc.doctype in ("Purchase Receipt", "Delivery Note", "Stock Entry"): - doc.docstatus = 1 - doc.make_bundle_using_old_serial_batch_fields() - doc.update_stock_ledger() - - columns = get_sl_columns(filters) - sl_entries = get_sl_entries_for_preview(doc.doctype, doc.name, fields) - - sl_columns = get_columns(columns, columns_fields) - sl_data = get_data(columns_fields, sl_entries) - - return sl_columns, sl_data - - -def get_sl_entries_for_preview(doctype, docname, fields): - sl_entries = frappe.get_all( - "Stock Ledger Entry", filters={"voucher_type": doctype, "voucher_no": docname}, fields=fields - ) - - for entry in sl_entries: - if entry.actual_qty > 0: - entry["in_qty"] = entry.actual_qty - entry["out_qty"] = 0 - else: - entry["out_qty"] = abs(entry.actual_qty) - entry["in_qty"] = 0 - - entry["in_out_rate"] = entry["valuation_rate"] - - return sl_entries - - -def get_gl_entries_for_preview(doctype, docname, fields): - return frappe.get_all("GL Entry", filters={"voucher_type": doctype, "voucher_no": docname}, fields=fields) - - -def get_columns(raw_columns, fields): - return [ - {"name": d.get("label"), "editable": False, "width": 110, "fieldtype": d.get("fieldtype")} - for d in raw_columns - if not d.get("hidden") and d.get("fieldname") in fields - ] - - -def get_data(raw_columns, raw_data): - datatable_data = [] - for row in raw_data: - data_row = [] - for column in raw_columns: - data_row.append(row.get(column) or "") - - datatable_data.append(data_row) - - return datatable_data - - def repost_required_for_queue(doc: StockController) -> bool: """check if stock document contains repeated item-warehouse with queue based valuation. @@ -2017,18 +625,12 @@ def repost_required_for_queue(doc: StockController) -> bool: @frappe.whitelist() def check_item_quality_inspection(doctype: str, docstatus: str | int, items: str | list[dict]): + from erpnext.stock.services.quality_inspection_service import INSPECTION_FIELDNAME_MAP + if isinstance(items, str): items = json.loads(items) - inspection_fieldname_map = { - "Purchase Receipt": "inspection_required_before_purchase", - "Purchase Invoice": "inspection_required_before_purchase", - "Subcontracting Receipt": "inspection_required_before_purchase", - "Sales Invoice": "inspection_required_before_delivery", - "Delivery Note": "inspection_required_before_delivery", - } - - inspection_fieldname = inspection_fieldname_map.get(doctype) + inspection_fieldname = INSPECTION_FIELDNAME_MAP.get(doctype) if inspection_fieldname is None: return [] diff --git a/erpnext/exceptions.py b/erpnext/exceptions.py index e12c69757e0..c8d2d790d1b 100644 --- a/erpnext/exceptions.py +++ b/erpnext/exceptions.py @@ -28,3 +28,20 @@ class MandatoryAccountDimensionError(frappe.ValidationError): class ReportingCurrencyExchangeNotFoundError(frappe.ValidationError): pass + + +# stock +class QualityInspectionRequiredError(frappe.ValidationError): + pass + + +class QualityInspectionRejectedError(frappe.ValidationError): + pass + + +class QualityInspectionNotSubmittedError(frappe.ValidationError): + pass + + +class BatchExpiredError(frappe.ValidationError): + pass diff --git a/erpnext/stock/doctype/landed_cost_voucher/landed_cost_voucher.py b/erpnext/stock/doctype/landed_cost_voucher/landed_cost_voucher.py index 932ce242b01..6576380e862 100644 --- a/erpnext/stock/doctype/landed_cost_voucher/landed_cost_voucher.py +++ b/erpnext/stock/doctype/landed_cost_voucher/landed_cost_voucher.py @@ -9,6 +9,7 @@ from frappe import _ from frappe.model.document import Document from frappe.model.meta import get_field_precision from frappe.query_builder.custom import ConstantColumn +from frappe.query_builder.functions import Sum from frappe.utils import cint, flt import erpnext @@ -314,7 +315,7 @@ class LandedCostVoucher(Document): self.validate_asset_qty_and_status(d.receipt_document_type, doc) # set landed cost voucher amount in pr item - doc.set_landed_cost_voucher_amount() + set_landed_cost_voucher_amount(doc) if d.receipt_document_type == "Subcontracting Receipt": doc.calculate_items_qty_and_amount() @@ -523,3 +524,93 @@ def get_vendor_invoice_query(filters): query = query.where(doctype.name == filters.get("name")) return query + + +def set_landed_cost_voucher_amount(doc): + """Set landed_cost_voucher_amount on the receipt document's items from submitted LCVs.""" + for d in doc.get("items"): + lcv_item = frappe.qb.DocType("Landed Cost Item") + query = ( + frappe.qb.from_(lcv_item) + .select(Sum(lcv_item.applicable_charges), lcv_item.cost_center) + .where((lcv_item.docstatus == 1) & (lcv_item.receipt_document == doc.name)) + ) + + if doc.doctype == "Stock Entry": + query = query.where(lcv_item.stock_entry_item == d.name) + else: + query = query.where(lcv_item.purchase_receipt_item == d.name) + + lc_voucher_data = query.run(as_list=True) + + d.landed_cost_voucher_amount = lc_voucher_data[0][0] if lc_voucher_data else 0.0 + if not d.cost_center and lc_voucher_data and lc_voucher_data[0][1]: + d.db_set("cost_center", lc_voucher_data[0][1]) + + +def has_landed_cost_amount(doc): + for row in doc.items: + if row.get("landed_cost_voucher_amount"): + return True + + return False + + +def get_item_account_wise_lcv_entries(doc): + """Account-wise landed-cost map for a receipt document, consumed by the GL composers.""" + if not has_landed_cost_amount(doc): + return + + landed_cost_vouchers = frappe.get_all( + "Landed Cost Purchase Receipt", + fields=["parent"], + filters={"receipt_document": doc.name, "docstatus": 1}, + ) + + if not landed_cost_vouchers: + return + + item_account_wise_cost = {} + + row_fieldname = "purchase_receipt_item" + if doc.doctype == "Stock Entry": + row_fieldname = "stock_entry_item" + + for lcv in landed_cost_vouchers: + landed_cost_voucher_doc = frappe.get_doc("Landed Cost Voucher", lcv.parent) + + based_on_field = "applicable_charges" + # Use amount field for total item cost for manually cost distributed LCVs + if landed_cost_voucher_doc.distribute_charges_based_on != "Distribute Manually": + based_on_field = frappe.scrub(landed_cost_voucher_doc.distribute_charges_based_on) + + total_item_cost = 0 + + if based_on_field: + for item in landed_cost_voucher_doc.items: + total_item_cost += item.get(based_on_field) + + for item in landed_cost_voucher_doc.items: + if item.receipt_document == doc.name: + for account in landed_cost_voucher_doc.taxes: + exchange_rate = account.exchange_rate or 1 + item_account_wise_cost.setdefault((item.item_code, item.get(row_fieldname)), {}) + item_account_wise_cost[(item.item_code, item.get(row_fieldname))].setdefault( + account.expense_account, {"amount": 0.0, "base_amount": 0.0} + ) + + item_row = item_account_wise_cost[(item.item_code, item.get(row_fieldname))][ + account.expense_account + ] + + if total_item_cost > 0: + item_row["amount"] += account.amount * item.get(based_on_field) / total_item_cost + + item_row["base_amount"] += ( + account.base_amount * item.get(based_on_field) / total_item_cost + ) + else: + item_row["amount"] += item.applicable_charges / exchange_rate + item_row["base_amount"] += item.applicable_charges + + return item_account_wise_cost diff --git a/erpnext/stock/doctype/putaway_rule/putaway_rule.py b/erpnext/stock/doctype/putaway_rule/putaway_rule.py index 382b7106552..b7dacb9c230 100644 --- a/erpnext/stock/doctype/putaway_rule/putaway_rule.py +++ b/erpnext/stock/doctype/putaway_rule/putaway_rule.py @@ -334,3 +334,71 @@ def get_serial_nos_to_allocate(serial_nos, to_allocate): return "\n".join(allocated_serial_nos) if allocated_serial_nos else "" else: return "" + + +def validate_putaway_capacity(doc): + # if over receipt is attempted while 'apply putaway rule' is disabled + # and if rule was applied on the transaction, validate it. + valid_doctype = doc.doctype in ( + "Purchase Receipt", + "Stock Entry", + "Purchase Invoice", + "Stock Reconciliation", + ) + + if not frappe.get_all("Putaway Rule", limit=1): + return + + if doc.doctype == "Purchase Invoice" and doc.get("update_stock") == 0: + valid_doctype = False + + if valid_doctype: + rule_map = defaultdict(dict) + for item in doc.get("items"): + warehouse_field = "t_warehouse" if doc.doctype == "Stock Entry" else "warehouse" + rule = frappe.db.get_value( + "Putaway Rule", + {"item_code": item.get("item_code"), "warehouse": item.get(warehouse_field)}, + ["stock_capacity", "name", "disable"], + as_dict=True, + ) + if rule: + if rule.get("disable"): + continue # dont validate for disabled rule + + if doc.doctype == "Stock Reconciliation": + stock_qty = flt(item.qty) + else: + stock_qty = ( + flt(item.transfer_qty) if doc.doctype == "Stock Entry" else flt(item.stock_qty) + ) + + rule_name = rule.get("name") + if not rule_map[rule_name]: + rule_map[rule_name]["warehouse"] = item.get(warehouse_field) + rule_map[rule_name]["item"] = item.get("item_code") + rule_map[rule_name]["qty_put"] = 0 + rule_map[rule_name]["capacity"] = ( + rule.stock_capacity + if doc.doctype == "Stock Reconciliation" + else get_available_putaway_capacity(rule_name) + ) + rule_map[rule_name]["qty_put"] += flt(stock_qty) + + for rule, values in rule_map.items(): + if flt(values["qty_put"]) > flt(values["capacity"]): + message = _prepare_over_receipt_message(rule, values) + frappe.throw(msg=message, title=_("Over Receipt")) + + +def _prepare_over_receipt_message(rule, values): + message = _("{0} qty of Item {1} is being received into Warehouse {2} with capacity {3}.").format( + frappe.bold(values["qty_put"]), + frappe.bold(values["item"]), + frappe.bold(values["warehouse"]), + frappe.bold(values["capacity"]), + ) + message += "

" + rule_link = frappe.utils.get_link_to_form("Putaway Rule", rule) + message += _("Please adjust the qty or edit {0} to proceed.").format(rule_link) + return message diff --git a/erpnext/stock/doctype/stock_entry/stock_entry.py b/erpnext/stock/doctype/stock_entry/stock_entry.py index 1574efa67e9..9f78d3295d6 100644 --- a/erpnext/stock/doctype/stock_entry/stock_entry.py +++ b/erpnext/stock/doctype/stock_entry/stock_entry.py @@ -265,17 +265,22 @@ class StockEntry(StockController, SubcontractingInwardController): ) def validate(self): + from erpnext.stock.doctype.putaway_rule.putaway_rule import validate_putaway_capacity + from erpnext.stock.services.serial_batch_bundle_service import SerialBatchBundleService + + sbb = SerialBatchBundleService(self) + if self.purpose_cls: self.purpose_cls(self).validate() - self.validate_duplicate_serial_and_batch_bundle("items") + sbb.validate_duplicate_serial_and_batch_bundle("items") self.validate_posting_time() self.validate_item() self.validate_customer_provided_item() self.set_transfer_qty() self.validate_uom_is_integer("uom", "qty") self.validate_uom_is_integer("stock_uom", "transfer_qty") - self.validate_warehouse_of_sabb() + sbb.validate_warehouse_of_sabb() self.validate_source_stock_entry() self.validate_bom() self.set_process_loss_qty() @@ -294,11 +299,11 @@ class StockEntry(StockController, SubcontractingInwardController): self.validate_difference_account() self.validate_job_card_item() self.set_purpose_for_stock_entry() - self.clean_serial_nos() + sbb.clean_serial_nos() self.remove_fg_completed_qty() - self.validate_serialized_batch() + sbb.validate_serialized_batch() self.calculate_rate_and_amount() - self.validate_putaway_capacity() + validate_putaway_capacity(self) self.validate_closed_subcontracting_order() super().validate_subcontracting_inward() diff --git a/erpnext/stock/doctype/stock_reconciliation/stock_reconciliation.py b/erpnext/stock/doctype/stock_reconciliation/stock_reconciliation.py index 388ab6793b8..13aa0225568 100644 --- a/erpnext/stock/doctype/stock_reconciliation/stock_reconciliation.py +++ b/erpnext/stock/doctype/stock_reconciliation/stock_reconciliation.py @@ -65,6 +65,11 @@ class StockReconciliation(StockController): self.head_row = ["Item Code", "Warehouse", "Quantity", "Valuation Rate"] def validate(self): + from erpnext.stock.doctype.putaway_rule.putaway_rule import validate_putaway_capacity + from erpnext.stock.services.serial_batch_bundle_service import SerialBatchBundleService + + sbb = SerialBatchBundleService(self) + self.validate_items_exist() if not self.expense_account: self.expense_account = frappe.get_cached_value( @@ -75,16 +80,16 @@ class StockReconciliation(StockController): self.validate_posting_time() self.set_current_serial_and_batch_bundle() self.set_new_serial_and_batch_bundle() - self.validate_duplicate_serial_and_batch_bundle("items") + sbb.validate_duplicate_serial_and_batch_bundle("items") self.remove_items_with_no_change() self.validate_data() self.change_row_indexes() self.validate_expense_account() self.validate_customer_provided_item() self.set_zero_value_for_customer_provided_items() - self.clean_serial_nos() + sbb.clean_serial_nos() self.set_total_qty_and_amount() - self.validate_putaway_capacity() + validate_putaway_capacity(self) self.validate_inventory_dimension() self.validate_uom_is_integer("stock_uom", "qty") @@ -925,7 +930,9 @@ class StockReconciliation(StockController): data.qty_after_transaction = 0.0 data.incoming_rate = flt(row.valuation_rate) - self.update_inventory_dimensions(row, data) + from erpnext.stock.services.stock_ledger_service import StockLedgerService + + StockLedgerService(self).update_inventory_dimensions(row, data) return data diff --git a/erpnext/stock/services/base_stock_gl_composer.py b/erpnext/stock/services/base_stock_gl_composer.py index 89837db9909..8f9d207d8a3 100644 --- a/erpnext/stock/services/base_stock_gl_composer.py +++ b/erpnext/stock/services/base_stock_gl_composer.py @@ -29,11 +29,11 @@ class BaseStockGLComposer(BaseGLComposer): inventory_account_map = doc.get_inventory_account_map() sle_map = doc.get_stock_ledger_details() - voucher_details = doc.get_voucher_details(default_expense_account, default_cost_center, sle_map) + voucher_details = self.get_voucher_details(default_expense_account, default_cost_center, sle_map) gl_list = [] warehouse_with_no_account = [] - precision = doc.get_debit_field_precision() + precision = self.get_debit_field_precision() for item_row in voucher_details: sle_list = sle_map.get(item_row.name) @@ -45,7 +45,7 @@ class BaseStockGLComposer(BaseGLComposer): if _inv_dict.get("account"): sle_rounding_diff += flt(sle.stock_value_difference) - doc.check_expense_account(item_row) + self.check_expense_account(item_row) if item_row.get("target_warehouse"): _target_wh_inv_dict = doc.get_inventory_account_dict( @@ -152,3 +152,78 @@ class BaseStockGLComposer(BaseGLComposer): return process_gl_map( gl_list, precision=precision, from_repost=frappe.flags.through_repost_item_valuation ) + + def get_debit_field_precision(self): + if not frappe.flags.debit_field_precision: + frappe.flags.debit_field_precision = frappe.get_precision("GL Entry", "debit_in_account_currency") + + return frappe.flags.debit_field_precision + + def get_voucher_details(self, default_expense_account, default_cost_center, sle_map): + doc = self.doc + if doc.doctype == "Stock Reconciliation": + reconciliation_purpose = frappe.db.get_value(doc.doctype, doc.name, "purpose") + is_opening = "Yes" if reconciliation_purpose == "Opening Stock" else "No" + details = [] + for voucher_detail_no in sle_map: + details.append( + frappe._dict( + { + "name": voucher_detail_no, + "expense_account": default_expense_account, + "cost_center": default_cost_center, + "is_opening": is_opening, + } + ) + ) + return details + else: + details = doc.get("items") + + if default_expense_account or default_cost_center: + for d in details: + if default_expense_account and not d.get("expense_account"): + d.expense_account = default_expense_account + if default_cost_center and not d.get("cost_center"): + d.cost_center = default_cost_center + + return details + + def check_expense_account(self, item): + if not item.get("expense_account"): + msg = _("Please set an Expense Account in the Items table") + frappe.throw( + _("Row #{0}: Expense Account not set for the Item {1}. {2}").format( + item.idx, frappe.bold(item.item_code), msg + ), + title=_("Expense Account Missing"), + ) + + else: + is_expense_account = ( + frappe.get_cached_value("Account", item.get("expense_account"), "report_type") + == "Profit and Loss" + ) + if ( + self.doc.doctype + not in ( + "Purchase Receipt", + "Purchase Invoice", + "Stock Reconciliation", + "Stock Entry", + "Subcontracting Receipt", + "Delivery Note", + ) + and not is_expense_account + ): + frappe.throw( + _("Expense / Difference account ({0}) must be a 'Profit or Loss' account").format( + item.get("expense_account") + ) + ) + if is_expense_account and not item.get("cost_center"): + frappe.throw( + _("{0} {1}: Cost Center is mandatory for Item {2}").format( + _(self.doc.doctype), self.doc.name, item.get("item_code") + ) + ) diff --git a/erpnext/stock/services/internal_transfer.py b/erpnext/stock/services/internal_transfer.py new file mode 100644 index 00000000000..62fec7ff95c --- /dev/null +++ b/erpnext/stock/services/internal_transfer.py @@ -0,0 +1,179 @@ +# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors +# License: GNU General Public License v3. See license.txt + +"""Internal (inter-company) transfer validation for stock transactions. + +Extracted from ``StockController``. Validates warehouses, currency, packed items +and over-receipt quantities for internal-transfer stock vouchers. This is the +stock-side counterpart to ``accounts/services/internal_transfer.py`` (which owns +the party / rate / pricing / account side). The ``is_internal_transfer()`` +predicate lives on ``AccountsController`` (delegating to the accounts service). +""" + +from collections import defaultdict + +import frappe +from frappe import _, bold +from frappe.utils import flt, get_link_to_form + + +class StockInternalTransferService: + def __init__(self, doc) -> None: + self.doc = doc + + def validate_internal_transfer(self): + if self.doc.doctype in ("Sales Invoice", "Delivery Note", "Purchase Invoice", "Purchase Receipt"): + if self.doc.is_internal_transfer(): + self.validate_in_transit_warehouses() + self.validate_multi_currency() + self.validate_packed_items() + + if self.doc.get("is_internal_supplier") and self.doc.docstatus == 1: + self.validate_internal_transfer_qty() + else: + self.validate_internal_transfer_warehouse() + + def validate_internal_transfer_warehouse(self): + for row in self.doc.items: + if row.get("target_warehouse"): + row.target_warehouse = None + + if row.get("from_warehouse"): + row.from_warehouse = None + + def validate_in_transit_warehouses(self): + if ( + self.doc.doctype == "Sales Invoice" and self.doc.get("update_stock") + ) or self.doc.doctype == "Delivery Note": + for item in self.doc.get("items"): + if not item.target_warehouse: + frappe.throw( + _("Row {0}: Target Warehouse is mandatory for internal transfers").format(item.idx) + ) + + if ( + self.doc.doctype == "Purchase Invoice" and self.doc.get("update_stock") + ) or self.doc.doctype == "Purchase Receipt": + for item in self.doc.get("items"): + if not item.from_warehouse: + frappe.throw( + _("Row {0}: From Warehouse is mandatory for internal transfers").format(item.idx) + ) + + def validate_multi_currency(self): + if self.doc.currency != self.doc.company_currency: + frappe.throw(_("Internal transfers can only be done in company's default currency")) + + def validate_packed_items(self): + if self.doc.doctype in ("Sales Invoice", "Delivery Note Item") and self.doc.get("packed_items"): + frappe.throw(_("Packed Items cannot be transferred internally")) + + def validate_internal_transfer_qty(self): + if self.doc.doctype not in ["Purchase Invoice", "Purchase Receipt"]: + return + + inter_company_reference = ( + self.doc.get("inter_company_reference") + if self.doc.doctype == "Purchase Invoice" + else self.doc.get("inter_company_invoice_reference") + ) + + item_wise_transfer_qty = self.get_item_wise_inter_transfer_qty(inter_company_reference) + if not item_wise_transfer_qty: + return + + item_wise_received_qty = self.get_item_wise_inter_received_qty() + precision = frappe.get_precision(self.doc.doctype + " Item", "qty") + + over_receipt_allowance = frappe.get_single_value("Stock Settings", "over_delivery_receipt_allowance") + + parent_doctype = { + "Purchase Receipt": "Delivery Note", + "Purchase Invoice": "Sales Invoice", + }.get(self.doc.doctype) + + for key, transferred_qty in item_wise_transfer_qty.items(): + recevied_qty = flt(item_wise_received_qty.get(key), precision) + if over_receipt_allowance: + transferred_qty = transferred_qty + flt( + transferred_qty * over_receipt_allowance / 100, precision + ) + + if recevied_qty > flt(transferred_qty, precision): + frappe.throw( + _("For Item {0} cannot be received more than {1} qty against the {2} {3}").format( + bold(key[1]), + bold(flt(transferred_qty, precision)), + bold(parent_doctype), + get_link_to_form(parent_doctype, inter_company_reference), + ) + ) + + def get_item_wise_inter_transfer_qty(self, inter_company_reference): + parent_doctype = { + "Purchase Receipt": "Delivery Note", + "Purchase Invoice": "Sales Invoice", + }.get(self.doc.doctype) + + child_doctype = parent_doctype + " Item" + + parent_tab = frappe.qb.DocType(parent_doctype) + child_tab = frappe.qb.DocType(child_doctype) + + query = ( + frappe.qb.from_(parent_doctype) + .inner_join(child_tab) + .on(child_tab.parent == parent_tab.name) + .select( + child_tab.name, + child_tab.item_code, + child_tab.qty, + ) + .where((parent_tab.name == inter_company_reference) & (parent_tab.docstatus == 1)) + ) + + data = query.run(as_dict=True) + item_wise_transfer_qty = defaultdict(float) + for row in data: + item_wise_transfer_qty[(row.name, row.item_code)] += flt(row.qty) + + return item_wise_transfer_qty + + def get_item_wise_inter_received_qty(self): + child_doctype = self.doc.doctype + " Item" + + parent_tab = frappe.qb.DocType(self.doc.doctype) + child_tab = frappe.qb.DocType(child_doctype) + + query = ( + frappe.qb.from_(self.doc.doctype) + .inner_join(child_tab) + .on(child_tab.parent == parent_tab.name) + .select( + child_tab.item_code, + child_tab.qty, + ) + .where(parent_tab.docstatus == 1) + ) + + if self.doc.doctype == "Purchase Invoice": + query = query.select( + child_tab.sales_invoice_item.as_("name"), + ) + + query = query.where( + parent_tab.inter_company_invoice_reference == self.doc.inter_company_invoice_reference + ) + else: + query = query.select( + child_tab.delivery_note_item.as_("name"), + ) + + query = query.where(parent_tab.inter_company_reference == self.doc.inter_company_reference) + + data = query.run(as_dict=True) + item_wise_transfer_qty = defaultdict(float) + for row in data: + item_wise_transfer_qty[(row.name, row.item_code)] += flt(row.qty) + + return item_wise_transfer_qty diff --git a/erpnext/stock/services/quality_inspection_service.py b/erpnext/stock/services/quality_inspection_service.py new file mode 100644 index 00000000000..7e7fc4ba078 --- /dev/null +++ b/erpnext/stock/services/quality_inspection_service.py @@ -0,0 +1,113 @@ +# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors +# License: GNU General Public License v3. See license.txt + +"""Quality inspection validation for stock transactions. + +Extracted from ``StockController``. Validates that items requiring quality +inspection have a present / submitted / non-rejected Quality Inspection. +""" + +import frappe +from frappe import _ + +from erpnext.exceptions import ( + QualityInspectionNotSubmittedError, + QualityInspectionRejectedError, + QualityInspectionRequiredError, +) + +# Doctype -> the document-level "inspection required" flag. Shared with +# check_item_quality_inspection in stock_controller so the two stay in sync. +INSPECTION_FIELDNAME_MAP = { + "Purchase Receipt": "inspection_required_before_purchase", + "Purchase Invoice": "inspection_required_before_purchase", + "Subcontracting Receipt": "inspection_required_before_purchase", + "Sales Invoice": "inspection_required_before_delivery", + "Delivery Note": "inspection_required_before_delivery", +} + + +class QualityInspectionService: + def __init__(self, doc) -> None: + self.doc = doc + + def validate_inspection(self): + """Checks if quality inspection is set/ is valid for Items that require inspection.""" + inspection_required_fieldname = INSPECTION_FIELDNAME_MAP.get(self.doc.doctype) + + # return if inspection is not required on document level + if ( + (not inspection_required_fieldname and self.doc.doctype != "Stock Entry") + or (self.doc.doctype == "Stock Entry" and not self.doc.inspection_required) + or (self.doc.doctype in ["Sales Invoice", "Purchase Invoice"] and not self.doc.update_stock) + ): + return + + for row in self.doc.get("items"): + qi_required = False + if inspection_required_fieldname and frappe.get_cached_value( + "Item", row.item_code, inspection_required_fieldname + ): + qi_required = True + elif self.doc.doctype == "Stock Entry" and row.t_warehouse: + qi_required = True # inward stock needs inspection + + if row.get("secondary_item_type") or row.get("is_legacy_scrap_item"): + continue + + if qi_required: # validate row only if inspection is required on item level + if self.doc.doctype in [ + "Purchase Receipt", + "Purchase Invoice", + "Sales Invoice", + "Delivery Note", + ] and frappe.get_single_value( + "Stock Settings", "allow_to_make_quality_inspection_after_purchase_or_delivery" + ): + return + + self.validate_qi_presence(row) + if self.doc.docstatus == 1: + self.validate_qi_submission(row) + self.validate_qi_rejection(row) + + def validate_qi_presence(self, row): + """Check if QI is present on row level. Warn on save and stop on submit if missing.""" + if not row.quality_inspection: + msg = _("Row #{0}: Quality Inspection is required for Item {1}").format( + row.idx, frappe.bold(row.item_code) + ) + if self.doc.docstatus == 1: + frappe.throw(msg, title=_("Inspection Required"), exc=QualityInspectionRequiredError) + else: + frappe.msgprint(msg, title=_("Inspection Required"), indicator="blue") + + def validate_qi_submission(self, row): + """Check if QI is submitted on row level, during submission""" + action = frappe.get_single_value("Stock Settings", "action_if_quality_inspection_is_not_submitted") + qa_docstatus = frappe.db.get_value("Quality Inspection", row.quality_inspection, "docstatus") + + if qa_docstatus != 1: + link = frappe.utils.get_link_to_form("Quality Inspection", row.quality_inspection) + msg = _("Row #{0}: Quality Inspection {1} is not submitted for the item: {2}").format( + row.idx, link, row.item_code + ) + if action == "Stop": + frappe.throw(msg, title=_("Inspection Submission"), exc=QualityInspectionNotSubmittedError) + else: + frappe.msgprint(msg, alert=True, indicator="orange") + + def validate_qi_rejection(self, row): + """Check if QI is rejected on row level, during submission""" + action = frappe.get_single_value("Stock Settings", "action_if_quality_inspection_is_rejected") + qa_status = frappe.db.get_value("Quality Inspection", row.quality_inspection, "status") + + if qa_status == "Rejected": + link = frappe.utils.get_link_to_form("Quality Inspection", row.quality_inspection) + msg = _("Row #{0}: Quality Inspection {1} was rejected for item {2}").format( + row.idx, link, row.item_code + ) + if action == "Stop": + frappe.throw(msg, title=_("Inspection Rejected"), exc=QualityInspectionRejectedError) + else: + frappe.msgprint(msg, alert=True, indicator="orange") diff --git a/erpnext/stock/services/serial_batch_bundle_service.py b/erpnext/stock/services/serial_batch_bundle_service.py new file mode 100644 index 00000000000..17b3af32fd7 --- /dev/null +++ b/erpnext/stock/services/serial_batch_bundle_service.py @@ -0,0 +1,680 @@ +# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors +# License: GNU General Public License v3. See license.txt + +"""Serial & Batch Bundle handling for stock transactions. + +Extracted from ``StockController``. Owns creation, validation and teardown of +Serial and Batch Bundles for a stock voucher. The controller keeps thin +delegators for methods reached from other doctypes / ``run_method``; internal +helpers live here only. +""" + +import frappe +from frappe import _, bold +from frappe.utils import cstr, flt, get_link_to_form, getdate + +from erpnext.controllers.sales_and_purchase_return import ( + available_serial_batch_for_return, + filter_serial_batches, + make_serial_batch_bundle_for_return, +) +from erpnext.stock.doctype.batch.batch import get_batch_qty +from erpnext.stock.doctype.serial_and_batch_bundle.serial_and_batch_bundle import ( + combine_datetime, + get_type_of_transaction, +) + + +class SerialBatchBundleService: + def __init__(self, doc) -> None: + self.doc = doc + + def validate_warehouse_of_sabb(self): + if self.doc.is_internal_transfer(): + return + + doc_before_save = self.doc.get_doc_before_save() + + for row in self.doc.items: + if not row.get("serial_and_batch_bundle"): + continue + + sabb_details = frappe.db.get_value( + "Serial and Batch Bundle", + row.serial_and_batch_bundle, + ["type_of_transaction", "warehouse", "has_serial_no"], + as_dict=True, + ) + if not sabb_details: + continue + + if sabb_details.type_of_transaction != "Outward": + continue + + warehouse = row.get("warehouse") or row.get("s_warehouse") + if sabb_details.warehouse != warehouse: + frappe.throw( + _( + "Row #{0}: Warehouse {1} does not match with the warehouse {2} in Serial and Batch Bundle {3}." + ).format(row.idx, warehouse, sabb_details.warehouse, row.serial_and_batch_bundle) + ) + + if self.doc.doctype == "Stock Reconciliation": + continue + + if sabb_details.has_serial_no and doc_before_save and doc_before_save.get("items"): + prev_row = doc_before_save.get("items", {"idx": row.idx}) + if prev_row and prev_row[0].serial_and_batch_bundle != row.serial_and_batch_bundle: + sabb_doc = frappe.get_doc("Serial and Batch Bundle", row.serial_and_batch_bundle) + sabb_doc.validate_serial_no_status() + + def validate_duplicate_serial_and_batch_bundle(self, table_name): + if not self.doc.get(table_name): + return + + sbb_list = [] + for item in self.doc.get(table_name): + if item.get("serial_and_batch_bundle"): + sbb_list.append(item.get("serial_and_batch_bundle")) + + if item.get("rejected_serial_and_batch_bundle"): + sbb_list.append(item.get("rejected_serial_and_batch_bundle")) + + if sbb_list: + SLE = frappe.qb.DocType("Stock Ledger Entry") + data = ( + frappe.qb.from_(SLE) + .select(SLE.voucher_type, SLE.voucher_no, SLE.serial_and_batch_bundle) + .where( + (SLE.docstatus == 1) + & (SLE.serial_and_batch_bundle.notnull()) + & (SLE.serial_and_batch_bundle.isin(sbb_list)) + ) + .limit(1) + ).run(as_dict=True) + + if data: + data = data[0] + frappe.throw( + _("Serial and Batch Bundle {0} is already used in {1} {2}.").format( + frappe.bold(data.serial_and_batch_bundle), data.voucher_type, data.voucher_no + ) + ) + + def validate_serialized_batch(self): + from erpnext.exceptions import BatchExpiredError + from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos + + is_material_issue = False + if self.doc.doctype == "Stock Entry" and self.doc.purpose in ["Material Issue", "Material Transfer"]: + is_material_issue = True + + for d in self.doc.get("items"): + if hasattr(d, "serial_no") and hasattr(d, "batch_no") and d.serial_no and d.batch_no: + serial_nos = frappe.get_all( + "Serial No", + fields=["batch_no", "name", "warehouse"], + filters={"name": ("in", get_serial_nos(d.serial_no))}, + ) + + for row in serial_nos: + if row.warehouse and row.batch_no != d.batch_no: + frappe.throw( + _("Row #{0}: Serial No {1} does not belong to Batch {2}").format( + d.idx, row.name, d.batch_no + ) + ) + + if is_material_issue: + continue + + if ( + flt(d.qty) > 0.0 + and d.get("batch_no") + and self.doc.get("posting_date") + and self.doc.docstatus < 2 + ): + expiry_date = frappe.get_cached_value("Batch", d.get("batch_no"), "expiry_date") + + if expiry_date and getdate(expiry_date) < getdate(self.doc.posting_date): + frappe.throw( + _("Row #{0}: The batch {1} has already expired.").format( + d.idx, get_link_to_form("Batch", d.get("batch_no")) + ), + BatchExpiredError, + ) + + def clean_serial_nos(self): + from erpnext.stock.doctype.serial_no.serial_no import clean_serial_no_string + + for row in self.doc.get("items"): + if hasattr(row, "serial_no") and row.serial_no: + # remove extra whitespace and store one serial no on each line + row.serial_no = clean_serial_no_string(row.serial_no) + + for row in self.doc.get("packed_items") or []: + if hasattr(row, "serial_no") and row.serial_no: + # remove extra whitespace and store one serial no on each line + row.serial_no = clean_serial_no_string(row.serial_no) + + def make_bundle_using_old_serial_batch_fields(self, table_name=None, via_landed_cost_voucher=False): + if self.doc.get("_action") == "update_after_submit": + return + + # To handle test cases + if frappe.in_test and frappe.flags.use_serial_and_batch_fields: + return + + if not table_name: + table_name = "items" + + if self.doc.doctype == "Asset Capitalization": + table_name = "stock_items" + + parent_details = frappe._dict() + if table_name == "packed_items": + parent_details = self.get_parent_details_for_packed_items() + + for row in self.doc.get(table_name): + if ( + not via_landed_cost_voucher + and row.serial_and_batch_bundle + and (row.serial_no or row.batch_no) + ): + self.validate_serial_nos_and_batches_with_bundle(row) + + if not row.serial_no and not row.batch_no and not row.get("rejected_serial_no"): + continue + + if not row.use_serial_batch_fields and ( + row.serial_no or row.batch_no or row.get("rejected_serial_no") + ): + row.use_serial_batch_fields = 1 + + if row.use_serial_batch_fields and ( + not row.serial_and_batch_bundle and not row.get("rejected_serial_and_batch_bundle") + ): + bundle_details = { + "item_code": row.get("rm_item_code") or row.item_code, + "posting_datetime": combine_datetime(self.doc.posting_date, self.doc.posting_time), + "voucher_type": self.doc.doctype, + "voucher_no": self.doc.name, + "voucher_detail_no": row.name, + "company": self.doc.company, + "is_rejected": 1 if row.get("rejected_warehouse") else 0, + "use_serial_batch_fields": row.use_serial_batch_fields, + "via_landed_cost_voucher": via_landed_cost_voucher, + "do_not_submit": True if not via_landed_cost_voucher else False, + } + + if self.doc.is_internal_transfer() and row.get("from_warehouse") and not self.doc.is_return: + self.update_bundle_details(bundle_details, table_name, row) + bundle_details["type_of_transaction"] = "Outward" + bundle_details["warehouse"] = row.get("from_warehouse") + bundle_details["qty"] = row.get("stock_qty") or row.get("qty") + self.create_serial_batch_bundle(bundle_details, row) + continue + + if row.get("qty") or row.get("consumed_qty") or row.get("stock_qty"): + self.update_bundle_details(bundle_details, table_name, row, parent_details=parent_details) + self.create_serial_batch_bundle(bundle_details, row) + + if row.get("rejected_qty"): + self.update_bundle_details(bundle_details, table_name, row, is_rejected=True) + self.create_serial_batch_bundle(bundle_details, row) + + def get_parent_details_for_packed_items(self): + parent_details = frappe._dict() + for row in self.doc.get("items"): + parent_details[row.name] = row + + return parent_details + + def make_bundle_for_sales_purchase_return(self, table_name=None): + if not self.doc.get("is_return"): + return + + if not table_name: + table_name = "items" + + self.make_bundle_for_non_rejected_qty(table_name) + + if self.doc.doctype in ["Purchase Invoice", "Purchase Receipt"]: + self.make_bundle_for_rejected_qty(table_name) + + def make_bundle_for_rejected_qty(self, table_name=None): + field, reference_ids = self.get_reference_ids( + table_name, "rejected_qty", "rejected_serial_and_batch_bundle" + ) + + if not reference_ids: + return + + child_doctype = self.doc.doctype + " Item" + available_dict = available_serial_batch_for_return( + field, child_doctype, reference_ids, is_rejected=True + ) + + for row in self.doc.get(table_name): + if data := available_dict.get(row.get(field)): + qty_field = "rejected_qty" + warehouse_field = "rejected_warehouse" + if row.get("return_qty_from_rejected_warehouse"): + qty_field = "qty" + warehouse_field = "warehouse" + + if not data.get("qty"): + frappe.throw( + _("For the {0}, no stock is available for the return in the warehouse {1}.").format( + frappe.bold(row.item_code), row.get(warehouse_field) + ) + ) + + data = filter_serial_batches( + self.doc, data, row, warehouse_field=warehouse_field, qty_field=qty_field + ) + bundle = make_serial_batch_bundle_for_return(data, row, self.doc, warehouse_field, qty_field) + if row.get("return_qty_from_rejected_warehouse"): + row.db_set( + { + "serial_and_batch_bundle": bundle, + "batch_no": "", + "serial_no": "", + } + ) + else: + row.db_set( + { + "rejected_serial_and_batch_bundle": bundle, + "batch_no": "", + "rejected_serial_no": "", + } + ) + + def make_bundle_for_non_rejected_qty(self, table_name): + field, reference_ids = self.get_reference_ids(table_name) + if not reference_ids: + return + + child_doctype = self.doc.doctype + " Item" + if table_name == "packed_items": + field = "parent_detail_docname" + child_doctype = "Packed Item" + + available_dict = available_serial_batch_for_return(field, child_doctype, reference_ids) + + for row in self.doc.get(table_name): + value = row.get(field) + if table_name == "packed_items" and row.get("parent_detail_docname"): + value = self.get_value_for_packed_item(row) + if not value: + continue + + if data := available_dict.get(value): + data = filter_serial_batches(self.doc, data, row) + bundle = make_serial_batch_bundle_for_return(data, row, self.doc) + row.db_set( + { + "serial_and_batch_bundle": bundle, + "batch_no": "", + "serial_no": "", + } + ) + + if self.doc.doctype in ["Sales Invoice", "Delivery Note"]: + row.db_set( + "incoming_rate", frappe.db.get_value("Serial and Batch Bundle", bundle, "avg_rate") + ) + + def get_value_for_packed_item(self, row): + parent_items = self.doc.get("items", {"name": row.parent_detail_docname}) + if parent_items: + ref = parent_items[0].get("dn_detail") + return (row.item_code, ref) + + return None + + def get_reference_ids(self, table_name, qty_field=None, bundle_field=None) -> tuple[str, list[str]]: + field = { + "Sales Invoice": "sales_invoice_item", + "Delivery Note": "dn_detail", + "Purchase Receipt": "purchase_receipt_item", + "Purchase Invoice": "purchase_invoice_item", + "POS Invoice": "pos_invoice_item", + }.get(self.doc.doctype) + + if not bundle_field: + bundle_field = "serial_and_batch_bundle" + + if not qty_field: + qty_field = "qty" + + reference_ids = [] + + for row in self.doc.get(table_name): + if not self.is_serial_batch_item(row.item_code): + continue + + if ( + row.get(field) + and ( + qty_field == "qty" + and not row.get("return_qty_from_rejected_warehouse") + or qty_field == "rejected_qty" + and (row.get("return_qty_from_rejected_warehouse") or row.get("rejected_warehouse")) + ) + and not row.get("use_serial_batch_fields") + and not row.get(bundle_field) + ): + reference_ids.append(row.get(field)) + + if table_name == "packed_items" and row.get("parent_detail_docname"): + parent_rows = self.doc.get("items", {"name": row.parent_detail_docname}) or [] + for d in parent_rows: + if d.get(field) and not d.get(bundle_field): + reference_ids.append(d.get(field)) + + return field, reference_ids + + def is_serial_batch_item(self, item_code) -> bool: + item_details = frappe.get_cached_value( + "Item", item_code, ["has_serial_no", "has_batch_no"], as_dict=True + ) + if not item_details: + frappe.throw(_("Item {0} does not exist.").format(bold(item_code))) + + return bool(item_details.has_serial_no or item_details.has_batch_no) + + def update_bundle_details(self, bundle_details, table_name, row, is_rejected=False, parent_details=None): + from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos + + # Since qty field is different for different doctypes + qty = row.get("qty") + warehouse = row.get("warehouse") + + if table_name == "packed_items": + type_of_transaction = "Inward" + if not self.doc.is_return: + type_of_transaction = "Outward" + elif table_name == "supplied_items": + qty = row.consumed_qty + warehouse = self.doc.supplier_warehouse + type_of_transaction = "Outward" + if self.doc.is_return: + type_of_transaction = "Inward" + else: + type_of_transaction = get_type_of_transaction(self.doc, row) + + if hasattr(row, "stock_qty"): + qty = row.stock_qty + + if self.doc.doctype == "Stock Entry": + qty = row.transfer_qty + warehouse = row.s_warehouse or row.t_warehouse + + serial_nos = row.serial_no + if is_rejected: + serial_nos = row.get("rejected_serial_no") + type_of_transaction = "Inward" if not self.doc.is_return else "Outward" + qty = flt( + row.get("rejected_qty") * row.get("conversion_factor", 1.0), + frappe.get_precision("Serial and Batch Entry", "qty"), + ) + warehouse = row.get("rejected_warehouse") + + if ( + self.doc.is_internal_transfer() + and self.doc.doctype in ["Sales Invoice", "Delivery Note"] + and self.doc.is_return + ): + warehouse = row.get("target_warehouse") or row.get("warehouse") + type_of_transaction = "Outward" + + if table_name == "packed_items": + if not warehouse: + warehouse = parent_details[row.parent_detail_docname].warehouse + bundle_details["voucher_detail_no"] = parent_details[row.parent_detail_docname].name + + bundle_details.update( + { + "qty": qty, + "is_rejected": is_rejected, + "type_of_transaction": type_of_transaction, + "warehouse": warehouse, + "batches": frappe._dict({row.batch_no: qty}) if row.batch_no else None, + "serial_nos": get_serial_nos(serial_nos) if serial_nos else None, + "batch_no": row.batch_no, + } + ) + + def create_serial_batch_bundle(self, bundle_details, row): + from erpnext.stock.serial_batch_bundle import SerialBatchCreation + + sn_doc = SerialBatchCreation(bundle_details).make_serial_and_batch_bundle() + + field = "serial_and_batch_bundle" + if bundle_details.get("is_rejected"): + field = "rejected_serial_and_batch_bundle" + + row.set(field, sn_doc.name) + row.db_set({field: sn_doc.name}) + + def validate_serial_nos_and_batches_with_bundle(self, row): + from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos + + throw_error = False + if row.serial_no: + serial_nos = frappe.get_all( + "Serial and Batch Entry", + fields=["serial_no"], + filters={"parent": row.serial_and_batch_bundle}, + ) + serial_nos = sorted([cstr(d.serial_no) for d in serial_nos]) + parsed_serial_nos = get_serial_nos(row.serial_no) + + if len(serial_nos) != len(parsed_serial_nos): + throw_error = True + elif serial_nos != parsed_serial_nos: + for serial_no in serial_nos: + if serial_no not in parsed_serial_nos: + throw_error = True + break + + elif row.batch_no: + batches = sorted( + frappe.get_all( + "Serial and Batch Entry", + filters={"parent": row.serial_and_batch_bundle}, + pluck="batch_no", + distinct=True, + ) + ) + + if batches != [row.batch_no]: + throw_error = True + + if throw_error: + frappe.throw( + _( + "At row {0}: Serial and Batch Bundle {1} has already created. Please remove the values from the serial no or batch no fields." + ).format(row.idx, row.serial_and_batch_bundle) + ) + + def set_use_serial_batch_fields(self): + if frappe.get_single_value("Stock Settings", "use_serial_batch_fields"): + for row in self.doc.items: + row.use_serial_batch_fields = 1 + + def delete_auto_created_batches(self): + for table_name in ["items", "packed_items", "supplied_items"]: + if not self.doc.get(table_name): + continue + + for row in self.doc.get(table_name): + update_values = {} + if row.get("batch_no"): + update_values["batch_no"] = None + + if row.get("serial_and_batch_bundle"): + update_values["serial_and_batch_bundle"] = None + frappe.db.set_value( + "Serial and Batch Bundle", row.serial_and_batch_bundle, {"is_cancelled": 1} + ) + + frappe.db.set_value( + "Serial and Batch Entry", {"parent": row.serial_and_batch_bundle}, {"is_cancelled": 1} + ) + + if update_values: + row.db_set(update_values) + + if table_name == "items" and row.get("rejected_serial_and_batch_bundle"): + frappe.db.set_value( + "Serial and Batch Bundle", row.rejected_serial_and_batch_bundle, {"is_cancelled": 1} + ) + + frappe.db.set_value( + "Serial and Batch Entry", + {"parent": row.rejected_serial_and_batch_bundle}, + {"is_cancelled": 1}, + ) + + row.db_set("rejected_serial_and_batch_bundle", None) + + if row.get("current_serial_and_batch_bundle"): + row.db_set("current_serial_and_batch_bundle", None) + + def set_serial_and_batch_bundle(self, table_name=None, ignore_validate=False): + if not table_name: + table_name = "items" + + QTY_FIELD = { + "serial_and_batch_bundle": "qty", + "current_serial_and_batch_bundle": "current_qty", + "rejected_serial_and_batch_bundle": "rejected_qty", + } + + for row in self.doc.get(table_name): + for field in QTY_FIELD.keys(): + if row.get(field): + frappe.get_doc("Serial and Batch Bundle", row.get(field)).set_serial_and_batch_values( + self.doc, row, qty_field=QTY_FIELD[field] + ) + + def make_package_for_transfer( + self, serial_and_batch_bundle, warehouse, type_of_transaction=None, do_not_submit=None, qty=0 + ): + from erpnext.controllers.stock_controller import make_bundle_for_material_transfer + + return make_bundle_for_material_transfer( + is_new=self.doc.is_new(), + docstatus=self.doc.docstatus, + voucher_type=self.doc.doctype, + voucher_no=self.doc.name, + serial_and_batch_bundle=serial_and_batch_bundle, + warehouse=warehouse, + type_of_transaction=type_of_transaction, + do_not_submit=do_not_submit, + qty=qty, + ) + + def validate_reserved_batches(self): + if not frappe.db.get_single_value("Stock Settings", "enable_stock_reservation"): + return + + if self.doc.doctype not in ["Delivery Note", "Sales Invoice", "Stock Entry"]: + return + + batches = frappe.get_all( + "Serial and Batch Entry", + filters={ + "voucher_type": self.doc.doctype, + "voucher_no": self.doc.name, + "docstatus": 1, + "batch_no": ("is", "set"), + "qty": ("<", 0), + }, + pluck="batch_no", + ) + + if not batches: + return + + field_mapper = { + "Sales Invoice": [["Sales Order", "sales_order"]], + "Delivery Note": [["Sales Order", "against_sales_order"]], + "Stock Entry": [ + ["Work Order", "work_order"], + ["Subcontracting Inward Order", "subcontracting_inward_order"], + ], + }.get(self.doc.doctype) + + qty_field = { + "Sales Invoice": "qty", + "Delivery Note": "qty", + "Stock Entry": "fg_completed_qty", + }.get(self.doc.doctype) + + reserved_batches_data = self.get_reserved_batches(batches) + items = self.doc.items + if self.doc.doctype == "Stock Entry": + items = [self.doc] + + for item in items: + for field in field_mapper: + if not item.get(field[1]): + continue + + value = item.get(field[1]) + for row in reserved_batches_data: + if self.doc.doctype in ["Sales Invoice", "Delivery Note"] and row.item_code != item.get( + "item_code" + ): + continue + + if row.voucher_no == value: + continue + + batch_qty = get_batch_qty( + row.batch_no, + row.warehouse, + posting_date=self.doc.posting_date, + posting_time=self.doc.posting_time, + consider_negative_batches=True, + ) + + if item.get(qty_field) < batch_qty: + continue + + frappe.throw( + _( + "The batch {0} is already reserved in {1} {2}. So, cannot proceed with the {3} {4}, which is created against the {5} {6}." + ).format( + frappe.bold(row.batch_no), + frappe.bold(row.voucher_type), + frappe.bold(row.voucher_no), + frappe.bold(self.doc.doctype), + frappe.bold(self.doc.name), + frappe.bold(field[0]), + frappe.bold(value), + ), + title=_("Reserved Batch Conflict"), + ) + + def get_reserved_batches(self, batches): + doctype = frappe.qb.DocType("Stock Reservation Entry") + child_doc = frappe.qb.DocType("Serial and Batch Entry") + + return ( + frappe.qb.from_(doctype) + .join(child_doc) + .on(doctype.name == child_doc.parent) + .select( + child_doc.batch_no, + doctype.voucher_type, + doctype.voucher_no, + doctype.item_code, + doctype.warehouse, + ) + .where((doctype.docstatus == 1) & (child_doc.batch_no.isin(batches))) + ).run(as_dict=True) diff --git a/erpnext/stock/services/stock_ledger_service.py b/erpnext/stock/services/stock_ledger_service.py new file mode 100644 index 00000000000..275f27b9652 --- /dev/null +++ b/erpnext/stock/services/stock_ledger_service.py @@ -0,0 +1,250 @@ +# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors +# License: GNU General Public License v3. See license.txt + +"""Stock Ledger Entry building and reposting for stock transactions. + +Extracted from ``StockController``. Builds the SLE dicts for a voucher, writes +them, and triggers future SLE/GL reposting. The repost helper *functions* remain +module-level in ``stock_controller`` (imported widely); this service owns the +instance-level logic. +""" + +import frappe +from frappe.utils import flt + +from erpnext.accounts.utils import get_fiscal_year +from erpnext.stock.doctype.inventory_dimension.inventory_dimension import ( + get_evaluated_inventory_dimension, +) + + +class StockLedgerService: + def __init__(self, doc) -> None: + self.doc = doc + + def get_items_and_warehouses(self) -> tuple[list[str], list[str]]: + """Get list of items and warehouses affected by a transaction""" + + if not (hasattr(self.doc, "items") or hasattr(self.doc, "packed_items")): + return [], [] + + item_rows = (self.doc.get("items") or []) + (self.doc.get("packed_items") or []) + + items = {d.item_code for d in item_rows if d.item_code} + + warehouses = set() + for d in item_rows: + if d.get("warehouse"): + warehouses.add(d.warehouse) + + if self.doc.doctype == "Stock Entry": + if d.get("s_warehouse"): + warehouses.add(d.s_warehouse) + if d.get("t_warehouse"): + warehouses.add(d.t_warehouse) + + return list(items), list(warehouses) + + def get_stock_ledger_details(self): + stock_ledger = {} + + table = frappe.qb.DocType("Stock Ledger Entry") + + stock_ledger_entries = ( + frappe.qb.from_(table) + .select( + table.name, + table.warehouse, + table.stock_value_difference, + table.valuation_rate, + table.voucher_detail_no, + table.item_code, + table.posting_date, + table.posting_time, + table.actual_qty, + table.qty_after_transaction, + table.project, + ) + .where( + (table.voucher_type == self.doc.doctype) + & (table.voucher_no == self.doc.name) + & (table.is_cancelled == 0) + ) + ).run(as_dict=True) + + for sle in stock_ledger_entries: + stock_ledger.setdefault(sle.voucher_detail_no, []).append(sle) + + return stock_ledger + + def get_sl_entries(self, d, args): + sl_dict = frappe._dict( + { + "item_code": d.get("item_code", None), + "warehouse": d.get("warehouse", None), + "serial_and_batch_bundle": d.get("serial_and_batch_bundle"), + "posting_date": self.doc.posting_date, + "posting_time": self.doc.posting_time, + "fiscal_year": get_fiscal_year(self.doc.posting_date, company=self.doc.company)[0], + "voucher_type": self.doc.doctype, + "voucher_no": self.doc.name, + "voucher_detail_no": d.name, + "actual_qty": (self.doc.docstatus == 1 and 1 or -1) * flt(d.get("stock_qty")), + "stock_uom": frappe.get_cached_value( + "Item", args.get("item_code") or d.get("item_code"), "stock_uom" + ), + "incoming_rate": 0, + "company": self.doc.company, + "project": d.get("project") or self.doc.get("project"), + "is_cancelled": 1 if self.doc.docstatus == 2 else 0, + } + ) + + sl_dict.update(args) + self.update_inventory_dimensions(d, sl_dict) + + if self.doc.docstatus == 2: + from erpnext.deprecation_dumpster import deprecation_warning + + deprecation_warning("unknown", "v16", "No instructions.") + # To handle denormalized serial no records, will br deprecated in v16 + for field in ["serial_no", "batch_no"]: + if d.get(field): + sl_dict[field] = d.get(field) + + return sl_dict + + def update_inventory_dimensions(self, row, sl_dict) -> None: + # To handle delivery note and sales invoice + if row.get("item_row"): + row = row.get("item_row") + + dimensions = get_evaluated_inventory_dimension(row, sl_dict, parent_doc=self.doc) + for dimension in dimensions: + if not dimension: + continue + + if ( + self.doc.doctype in ["Purchase Invoice", "Purchase Receipt"] + and row.get("rejected_warehouse") + and sl_dict.get("warehouse") == row.get("rejected_warehouse") + ): + fieldname = f"rejected_{dimension.source_fieldname}" + sl_dict[dimension.target_fieldname] = row.get(fieldname) + continue + + if self.doc.doctype in [ + "Purchase Invoice", + "Purchase Receipt", + "Sales Invoice", + "Delivery Note", + "Stock Entry", + ]: + if ( + ( + sl_dict.actual_qty > 0 + and not self.doc.get("is_return") + or sl_dict.actual_qty < 0 + and self.doc.get("is_return") + ) + and self.doc.doctype in ["Purchase Invoice", "Purchase Receipt", "Stock Entry"] + ) or ( + ( + sl_dict.actual_qty < 0 + and not self.doc.get("is_return") + or sl_dict.actual_qty > 0 + and self.doc.get("is_return") + ) + and self.doc.doctype in ["Sales Invoice", "Delivery Note", "Stock Entry"] + ): + if self.doc.doctype == "Stock Entry": + if row.get("t_warehouse") == sl_dict.warehouse and sl_dict.get("actual_qty") > 0: + fieldname = f"to_{dimension.source_fieldname}" + if dimension.source_fieldname.startswith("to_"): + fieldname = f"{dimension.source_fieldname}" + + sl_dict[dimension.target_fieldname] = row.get(fieldname) + continue + + sl_dict[dimension.target_fieldname] = row.get(dimension.source_fieldname) + else: + fieldname_start_with = "to" + if self.doc.doctype in ["Purchase Invoice", "Purchase Receipt"]: + fieldname_start_with = "from" + + fieldname = f"{fieldname_start_with}_{dimension.source_fieldname}" + sl_dict[dimension.target_fieldname] = row.get(fieldname) + + if not sl_dict.get(dimension.target_fieldname): + sl_dict[dimension.target_fieldname] = row.get(dimension.source_fieldname) + + elif row.get(dimension.source_fieldname): + sl_dict[dimension.target_fieldname] = row.get(dimension.source_fieldname) + + if not sl_dict.get(dimension.target_fieldname) and dimension.fetch_from_parent: + sl_dict[dimension.target_fieldname] = self.doc.get(dimension.fetch_from_parent) + + # Get value based on doctype name + if not sl_dict.get(dimension.target_fieldname): + fieldname = next( + ( + field.fieldname + for field in frappe.get_meta(self.doc.doctype).fields + if field.options == dimension.fetch_from_parent + ), + None, + ) + + if fieldname and self.doc.get(fieldname): + sl_dict[dimension.target_fieldname] = self.doc.get(fieldname) + + if sl_dict[dimension.target_fieldname] and self.doc.docstatus == 1: + row.db_set(dimension.source_fieldname, sl_dict[dimension.target_fieldname]) + + def make_sl_entries(self, sl_entries, allow_negative_stock=False, via_landed_cost_voucher=False): + from erpnext.stock.serial_batch_bundle import update_batch_qty + from erpnext.stock.services.serial_batch_bundle_service import SerialBatchBundleService + from erpnext.stock.stock_ledger import make_sl_entries + + make_sl_entries(sl_entries, allow_negative_stock, via_landed_cost_voucher) + update_batch_qty( + self.doc.doctype, + self.doc.name, + self.doc.docstatus, + via_landed_cost_voucher=via_landed_cost_voucher, + ) + + SerialBatchBundleService(self.doc).validate_reserved_batches() + + def repost_future_sle_and_gle(self, force=False, via_landed_cost_voucher=False): + from erpnext.controllers.stock_controller import ( + create_item_wise_repost_entries, + create_repost_item_valuation_entry, + future_sle_exists, + repost_required_for_queue, + ) + + args = frappe._dict( + { + "posting_date": self.doc.posting_date, + "posting_time": self.doc.posting_time, + "voucher_type": self.doc.doctype, + "voucher_no": self.doc.name, + "company": self.doc.company, + "via_landed_cost_voucher": via_landed_cost_voucher, + } + ) + + if self.doc.docstatus == 2: + force = True + + if force or future_sle_exists(args) or repost_required_for_queue(self.doc): + item_based_reposting = frappe.get_single_value("Stock Reposting Settings", "item_based_reposting") + if item_based_reposting: + create_item_wise_repost_entries( + voucher_type=self.doc.doctype, + voucher_no=self.doc.name, + via_landed_cost_voucher=via_landed_cost_voucher, + ) + else: + create_repost_item_valuation_entry(args)