393 lines
15 KiB
Python
393 lines
15 KiB
Python
import frappe
|
|
import copy
|
|
from frappe import _
|
|
from frappe.utils import flt, cint, get_link_to_form
|
|
from collections import defaultdict
|
|
from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
|
|
|
|
class Subcontracting():
|
|
def set_materials_for_subcontracted_items(self, raw_material_table):
|
|
if self.doctype == 'Purchase Invoice' and not self.update_stock:
|
|
return
|
|
|
|
self.raw_material_table = raw_material_table
|
|
self.__identify_change_in_item_table()
|
|
self.__prepare_supplied_items()
|
|
self.__validate_supplied_items()
|
|
|
|
def __prepare_supplied_items(self):
|
|
self.initialized_fields()
|
|
self.__get_purchase_orders()
|
|
self.__get_pending_qty_to_receive()
|
|
self.get_available_materials()
|
|
self.__remove_changed_rows()
|
|
self.__set_supplied_items()
|
|
|
|
def initialized_fields(self):
|
|
self.available_materials = frappe._dict()
|
|
self.__transferred_items = frappe._dict()
|
|
self.alternative_item_details = frappe._dict()
|
|
self.__get_backflush_based_on()
|
|
|
|
def __get_backflush_based_on(self):
|
|
self.backflush_based_on = frappe.db.get_single_value("Buying Settings",
|
|
"backflush_raw_materials_of_subcontract_based_on")
|
|
|
|
def __get_purchase_orders(self):
|
|
self.purchase_orders = []
|
|
|
|
if self.doctype == 'Purchase Order':
|
|
return
|
|
|
|
self.purchase_orders = [d.purchase_order for d in self.items if d.purchase_order]
|
|
|
|
def __identify_change_in_item_table(self):
|
|
self.__changed_name = []
|
|
self.__reference_name = []
|
|
|
|
if self.doctype == 'Purchase Order' or self.is_new():
|
|
self.set(self.raw_material_table, [])
|
|
return
|
|
|
|
item_dict = self.__get_data_before_save()
|
|
if not item_dict:
|
|
return True
|
|
|
|
for n_row in self.items:
|
|
self.__reference_name.append(n_row.name)
|
|
if (n_row.name not in item_dict) or (n_row.item_code, n_row.qty) != item_dict[n_row.name]:
|
|
self.__changed_name.append(n_row.name)
|
|
|
|
if item_dict.get(n_row.name):
|
|
del item_dict[n_row.name]
|
|
|
|
self.__changed_name.extend(item_dict.keys())
|
|
|
|
def __get_data_before_save(self):
|
|
item_dict = {}
|
|
if self.doctype in ['Purchase Receipt', 'Purchase Invoice'] and self._doc_before_save:
|
|
for row in self._doc_before_save.get('items'):
|
|
item_dict[row.name] = (row.item_code, row.qty)
|
|
|
|
return item_dict
|
|
|
|
def get_available_materials(self):
|
|
''' Get the available raw materials which has been transferred to the supplier.
|
|
available_materials = {
|
|
(item_code, subcontracted_item, purchase_order): {
|
|
'qty': 1, 'serial_no': [ABC], 'batch_no': {'batch1': 1}, 'data': item_details
|
|
}
|
|
}
|
|
'''
|
|
if not self.purchase_orders:
|
|
return
|
|
|
|
for row in self.__get_transferred_items():
|
|
key = (row.rm_item_code, row.main_item_code, row.purchase_order)
|
|
|
|
if key not in self.available_materials:
|
|
self.available_materials.setdefault(key, frappe._dict({'qty': 0, 'serial_no': [],
|
|
'batch_no': defaultdict(float), 'item_details': row, 'po_details': []})
|
|
)
|
|
|
|
details = self.available_materials[key]
|
|
details.qty += row.qty
|
|
details.po_details.append(row.po_detail)
|
|
|
|
if row.serial_no:
|
|
details.serial_no.extend(get_serial_nos(row.serial_no))
|
|
|
|
if row.batch_no:
|
|
details.batch_no[row.batch_no] += row.qty
|
|
|
|
self.__set_alternative_item_details(row)
|
|
|
|
self.__transferred_items = copy.deepcopy(self.available_materials)
|
|
for doctype in ['Purchase Receipt', 'Purchase Invoice']:
|
|
self.__update_consumed_materials(doctype)
|
|
|
|
def __update_consumed_materials(self, doctype, return_consumed_items=False):
|
|
'''Deduct the consumed materials from the available materials.'''
|
|
|
|
pr_items = self.__get_received_items(doctype)
|
|
if not pr_items:
|
|
return ([], {}) if return_consumed_items else None
|
|
|
|
pr_items = {d.name: d.get(self.get('po_field') or 'purchase_order') for d in pr_items}
|
|
consumed_materials = self.__get_consumed_items(doctype, pr_items.keys())
|
|
|
|
if return_consumed_items:
|
|
return (consumed_materials, pr_items)
|
|
|
|
for row in consumed_materials:
|
|
key = (row.rm_item_code, row.main_item_code, pr_items.get(row.reference_name))
|
|
if not self.available_materials.get(key):
|
|
continue
|
|
|
|
self.available_materials[key]['qty'] -= row.consumed_qty
|
|
if row.serial_no:
|
|
self.available_materials[key]['serial_no'] = list(
|
|
set(self.available_materials[key]['serial_no']) - set(get_serial_nos(row.serial_no))
|
|
)
|
|
|
|
if row.batch_no:
|
|
self.available_materials[key]['batch_no'][row.batch_no] -= row.consumed_qty
|
|
|
|
def __get_transferred_items(self):
|
|
fields = ['`tabStock Entry`.`purchase_order`']
|
|
alias_dict = {'item_code': 'rm_item_code', 'subcontracted_item': 'main_item_code', 'basic_rate': 'rate'}
|
|
|
|
child_table_fields = ['item_code', 'item_name', 'description', 'qty', 'basic_rate', 'amount',
|
|
'serial_no', 'uom', 'subcontracted_item', 'stock_uom', 'batch_no', 'conversion_factor',
|
|
's_warehouse', 't_warehouse', 'item_group', 'po_detail']
|
|
|
|
if self.backflush_based_on == 'BOM':
|
|
child_table_fields.append('original_item')
|
|
|
|
for field in child_table_fields:
|
|
fields.append(f'`tabStock Entry Detail`.`{field}` As {alias_dict.get(field, field)}')
|
|
|
|
filters = [['Stock Entry', 'docstatus', '=', 1], ['Stock Entry', 'purpose', '=', 'Send to Subcontractor'],
|
|
['Stock Entry', 'purchase_order', 'in', self.purchase_orders]]
|
|
|
|
return frappe.get_all('Stock Entry', fields = fields, filters=filters)
|
|
|
|
def __get_received_items(self, doctype):
|
|
fields = []
|
|
self.po_field = 'purchase_order'
|
|
|
|
for field in ['name', self.po_field, 'parent']:
|
|
fields.append(f'`tab{doctype} Item`.`{field}`')
|
|
|
|
filters = [[doctype, 'docstatus', '=', 1], [f'{doctype} Item', self.po_field, 'in', self.purchase_orders]]
|
|
if doctype == 'Purchase Invoice':
|
|
filters.append(['Purchase Invoice', 'update_stock', "=", 1])
|
|
|
|
return frappe.get_all(f'{doctype}', fields = fields, filters = filters)
|
|
|
|
def __get_consumed_items(self, doctype, pr_items):
|
|
return frappe.get_all('Purchase Receipt Item Supplied',
|
|
fields = ['serial_no', 'rm_item_code', 'reference_name', 'batch_no', 'consumed_qty', 'main_item_code'],
|
|
filters = {'docstatus': 1, 'reference_name': ('in', list(pr_items)), 'parenttype': doctype})
|
|
|
|
def __set_alternative_item_details(self, row):
|
|
if row.get('original_item'):
|
|
self.alternative_item_details[row.get('original_item')] = row
|
|
|
|
def __get_pending_qty_to_receive(self):
|
|
'''Get qty to be received against the purchase order.'''
|
|
|
|
self.qty_to_be_received = defaultdict(float)
|
|
|
|
if self.doctype != 'Purchase Order' and self.backflush_based_on != 'BOM' and self.purchase_orders:
|
|
for row in frappe.get_all('Purchase Order Item',
|
|
fields = ['item_code', '(qty - received_qty) as qty', 'parent', 'name'],
|
|
filters = {'docstatus': 1, 'parent': ('in', self.purchase_orders)}):
|
|
|
|
self.qty_to_be_received[(row.item_code, row.parent)] += row.qty
|
|
|
|
def __get_materials_from_bom(self, item_code, bom_no, exploded_item=0):
|
|
doctype = 'BOM Item' if not exploded_item else 'BOM Explosion Item'
|
|
fields = [f'`tab{doctype}`.`stock_qty` / `tabBOM`.`quantity` as qty_consumed_per_unit']
|
|
|
|
alias_dict = {'item_code': 'rm_item_code', 'name': 'bom_detail_no', 'source_warehouse': 'reserve_warehouse'}
|
|
for field in ['item_code', 'name', 'rate', 'stock_uom',
|
|
'source_warehouse', 'description', 'item_name', 'stock_uom']:
|
|
fields.append(f'`tab{doctype}`.`{field}` As {alias_dict.get(field, field)}')
|
|
|
|
filters = [[doctype, 'parent', '=', bom_no], [doctype, 'docstatus', '=', 1],
|
|
['BOM', 'item', '=', item_code], [doctype, 'sourced_by_supplier', '=', 0]]
|
|
|
|
return frappe.get_all('BOM', fields = fields, filters=filters, order_by = f'`tab{doctype}`.`idx`') or []
|
|
|
|
def __remove_changed_rows(self):
|
|
if not self.__changed_name:
|
|
return
|
|
|
|
i=1
|
|
self.set(self.raw_material_table, [])
|
|
for d in self._doc_before_save.supplied_items:
|
|
if d.reference_name in self.__changed_name:
|
|
continue
|
|
|
|
if (d.reference_name not in self.__reference_name):
|
|
continue
|
|
|
|
d.idx = i
|
|
self.append('supplied_items', d)
|
|
|
|
i += 1
|
|
|
|
def __set_supplied_items(self):
|
|
self.bom_items = {}
|
|
|
|
has_supplied_items = True if self.get(self.raw_material_table) else False
|
|
for row in self.items:
|
|
if (self.doctype != 'Purchase Order' and ((self.__changed_name and row.name not in self.__changed_name)
|
|
or (has_supplied_items and not self.__changed_name))):
|
|
continue
|
|
|
|
if self.doctype == 'Purchase Order' or self.backflush_based_on == 'BOM':
|
|
for bom_item in self.__get_materials_from_bom(row.item_code, row.bom, row.get('include_exploded_items')):
|
|
qty = (flt(bom_item.qty_consumed_per_unit) * flt(row.qty) * row.conversion_factor)
|
|
bom_item.main_item_code = row.item_code
|
|
self.__update_reserve_warehouse(bom_item, row)
|
|
self.__set_alternative_item(bom_item)
|
|
self.__add_supplied_item(row, bom_item, qty)
|
|
|
|
elif self.backflush_based_on != 'BOM':
|
|
for key, transfer_item in self.available_materials.items():
|
|
if (key[1], key[2]) == (row.item_code, row.purchase_order) and transfer_item.qty > 0:
|
|
qty = self.__get_qty_based_on_material_transfer(row, transfer_item) or 0
|
|
transfer_item.qty -= qty
|
|
self.__add_supplied_item(row, transfer_item.get('item_details'), qty)
|
|
|
|
if self.qty_to_be_received:
|
|
self.qty_to_be_received[(row.item_code, row.purchase_order)] -= row.qty
|
|
|
|
def __update_reserve_warehouse(self, row, item):
|
|
if self.doctype == 'Purchase Order':
|
|
row.reserve_warehouse = (self.set_reserve_warehouse or item.warehouse)
|
|
|
|
def __get_qty_based_on_material_transfer(self, item_row, transfer_item):
|
|
key = (item_row.item_code, item_row.purchase_order)
|
|
|
|
if self.qty_to_be_received == item_row.qty:
|
|
return transfer_item.qty
|
|
|
|
if self.qty_to_be_received:
|
|
qty = (flt(item_row.qty) * flt(transfer_item.qty)) / flt(self.qty_to_be_received.get(key, 0))
|
|
transfer_item.item_details.required_qty = transfer_item.qty
|
|
|
|
if (transfer_item.serial_no or frappe.get_cached_value('UOM',
|
|
transfer_item.item_details.stock_uom, 'must_be_whole_number')):
|
|
return frappe.utils.ceil(qty)
|
|
|
|
return qty
|
|
|
|
def __set_alternative_item(self, bom_item):
|
|
if self.alternative_item_details.get(bom_item.rm_item_code):
|
|
bom_item.update(self.alternative_item_details[bom_item.rm_item_code])
|
|
|
|
def __add_supplied_item(self, item_row, bom_item, qty):
|
|
bom_item.conversion_factor = item_row.conversion_factor
|
|
rm_obj = self.append(self.raw_material_table, bom_item)
|
|
rm_obj.reference_name = item_row.name
|
|
|
|
if self.doctype == 'Purchase Order':
|
|
rm_obj.required_qty = qty
|
|
else:
|
|
rm_obj.consumed_qty = 0
|
|
rm_obj.purchase_order = item_row.purchase_order
|
|
self.__set_batch_nos(bom_item, item_row, rm_obj, qty)
|
|
|
|
def __set_batch_nos(self, bom_item, item_row, rm_obj, qty):
|
|
key = (rm_obj.rm_item_code, item_row.item_code, item_row.purchase_order)
|
|
|
|
if (self.available_materials.get(key) and self.available_materials[key]['batch_no']):
|
|
new_rm_obj = None
|
|
for batch_no, batch_qty in self.available_materials[key]['batch_no'].items():
|
|
if batch_qty >= qty:
|
|
self.__set_batch_no_as_per_qty(item_row, rm_obj, batch_no, qty)
|
|
self.available_materials[key]['batch_no'][batch_no] -= qty
|
|
return
|
|
|
|
elif qty > 0 and batch_qty > 0:
|
|
qty -= batch_qty
|
|
new_rm_obj = self.append(self.raw_material_table, bom_item)
|
|
new_rm_obj.reference_name = item_row.name
|
|
self.__set_batch_no_as_per_qty(item_row, new_rm_obj, batch_no, batch_qty)
|
|
self.available_materials[key]['batch_no'][batch_no] = 0
|
|
|
|
if abs(qty) > 0 and not new_rm_obj:
|
|
self.__set_consumed_qty(rm_obj, qty)
|
|
else:
|
|
self.__set_consumed_qty(rm_obj, qty, bom_item.required_qty or qty)
|
|
self.__set_serial_nos(item_row, rm_obj)
|
|
|
|
def __set_consumed_qty(self, rm_obj, consumed_qty, required_qty=0):
|
|
rm_obj.required_qty = required_qty
|
|
rm_obj.consumed_qty = consumed_qty
|
|
|
|
def __set_batch_no_as_per_qty(self, item_row, rm_obj, batch_no, qty):
|
|
rm_obj.update({'consumed_qty': qty, 'batch_no': batch_no,
|
|
'required_qty': qty, 'purchase_order': item_row.purchase_order})
|
|
|
|
self.__set_serial_nos(item_row, rm_obj)
|
|
|
|
def __set_serial_nos(self, item_row, rm_obj):
|
|
key = (rm_obj.rm_item_code, item_row.item_code, item_row.purchase_order)
|
|
if (self.available_materials.get(key) and self.available_materials[key]['serial_no']):
|
|
used_serial_nos = self.available_materials[key]['serial_no'][0: cint(rm_obj.consumed_qty)]
|
|
rm_obj.serial_no = '\n'.join(used_serial_nos)
|
|
|
|
# Removed the used serial nos from the list
|
|
for sn in used_serial_nos:
|
|
self.available_materials[key]['serial_no'].remove(sn)
|
|
|
|
def set_consumed_qty_in_po(self):
|
|
# Update consumed qty back in the purchase order
|
|
if self.is_subcontracted != 'Yes':
|
|
return
|
|
|
|
self.__get_purchase_orders()
|
|
itemwise_consumed_qty = defaultdict(float)
|
|
for doctype in ['Purchase Receipt', 'Purchase Invoice']:
|
|
consumed_items, pr_items = self.__update_consumed_materials(doctype, return_consumed_items=True)
|
|
|
|
for row in consumed_items:
|
|
key = (row.rm_item_code, row.main_item_code, pr_items.get(row.reference_name))
|
|
itemwise_consumed_qty[key] += row.consumed_qty
|
|
|
|
self.__update_consumed_qty_in_po(itemwise_consumed_qty)
|
|
|
|
def __update_consumed_qty_in_po(self, itemwise_consumed_qty):
|
|
fields = ['main_item_code', 'rm_item_code', 'parent', 'supplied_qty', 'name']
|
|
filters = {'docstatus': 1, 'parent': ('in', self.purchase_orders)}
|
|
|
|
for row in frappe.get_all('Purchase Order Item Supplied', fields = fields, filters=filters, order_by='idx'):
|
|
key = (row.rm_item_code, row.main_item_code, row.parent)
|
|
consumed_qty = itemwise_consumed_qty.get(key, 0)
|
|
|
|
if row.supplied_qty < consumed_qty:
|
|
consumed_qty = row.supplied_qty
|
|
|
|
itemwise_consumed_qty[key] -= consumed_qty
|
|
frappe.db.set_value('Purchase Order Item Supplied', row.name, 'consumed_qty', consumed_qty)
|
|
|
|
def __validate_supplied_items(self):
|
|
if self.doctype not in ['Purchase Invoice', 'Purchase Receipt']:
|
|
return
|
|
|
|
for row in self.get(self.raw_material_table):
|
|
self.__validate_consumed_qty(row)
|
|
|
|
key = (row.rm_item_code, row.main_item_code, row.purchase_order)
|
|
if not self.__transferred_items or not self.__transferred_items.get(key):
|
|
return
|
|
|
|
self.__validate_batch_no(row, key)
|
|
self.__validate_serial_no(row, key)
|
|
|
|
def __validate_consumed_qty(self, row):
|
|
if self.backflush_based_on != 'BOM' and flt(row.consumed_qty) == 0.0:
|
|
msg = f'Row {row.idx}: the consumed qty cannot be zero for the item {frappe.bold(row.rm_item_code)}'
|
|
|
|
frappe.throw(_(msg),title=_('Consumed Items Qty Check'))
|
|
|
|
def __validate_batch_no(self, row, key):
|
|
if row.get('batch_no') and row.get('batch_no') not in self.__transferred_items.get(key).get('batch_no'):
|
|
link = get_link_to_form('Purchase Order', row.purchase_order)
|
|
msg = f'The Batch No {frappe.bold(row.get("batch_no"))} has not supplied against the Purchase Order {link}'
|
|
frappe.throw(_(msg), title=_("Incorrect Batch Consumed"))
|
|
|
|
def __validate_serial_no(self, row, key):
|
|
if row.get('serial_no'):
|
|
serial_nos = get_serial_nos(row.get('serial_no'))
|
|
incorrect_sn = set(serial_nos).difference(self.__transferred_items.get(key).get('serial_no'))
|
|
|
|
if incorrect_sn:
|
|
incorrect_sn = "\n".join(incorrect_sn)
|
|
link = get_link_to_form('Purchase Order', row.purchase_order)
|
|
msg = f'The Serial Nos {incorrect_sn} has not supplied against the Purchase Order {link}'
|
|
frappe.throw(_(msg), title=_("Incorrect Serial Number Consumed")) |