Shiloh/brotherton-erpnext
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: use stock value diff for calculation

Browse Source
This commit is contained in:
18alantom 2021-06-07 13:52:26 +05:30
parent 672c8bb112
commit 23b907df1a
1 changed files with 71 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

129
erpnext/stock/report/cogs_by_item_group/cogs_by_item_group.py
View File

@ -9,7 +9,6 @@ from erpnext.accounts.report.general_ledger.general_ledger import get_gl_entries
def execute(filters=None): def execute(filters=None):
print(filters)
validate_filters(filters) validate_filters(filters)
columns = get_columns() columns = get_columns()
data = get_data(filters) data = get_data(filters)
@ -17,9 +16,6 @@ def execute(filters=None):
def validate_filters(filters): def validate_filters(filters):
if not filters.get("from_date") and not filters.get("to_date"):
frappe.throw(_("{0} and {1} are mandatory").format(frappe.bold(_("From Date")), frappe.bold(_("To Date"))))
if filters.from_date > filters.to_date: if filters.from_date > filters.to_date:
frappe.throw(_("From Date must be before To Date")) frappe.throw(_("From Date must be before To Date"))
@ -42,110 +38,100 @@ def get_columns():
def get_data(filters): def get_data(filters):
entries = get_filtered_entries(filters) filtered_entries = get_filtered_entries(filters)
item_groups_list = frappe.get_all("Item Group", fields=("name", "is_group", "lft", "rgt")) svd_list = get_stock_value_difference_list(filtered_entries)
item_groups_dict = get_item_groups_dict(item_groups_list) leveled_dict = get_leveled_dict()
levels_dict = get_levels_dict(item_groups_dict)
update_levels_dict(levels_dict) assign_self_values(leveled_dict, svd_list)
assign_self_values(levels_dict, entries) assign_agg_values(leveled_dict)
assign_agg_values(levels_dict)
data = [] data = []
for _, i in levels_dict.items(): for _, i in leveled_dict.items():
if i['agg_value'] == 0: if i['agg_value'] == 0:
continue continue
data.append(get_row(i['name'], i['agg_value'], i['is_group'], i['level'])) data.append(get_row(i['name'], i['agg_value'], i['is_group'], i['level']))
if i['self_value'] < i['agg_value'] and i['self_value'] > 0: if i['self_value'] < i['agg_value'] and i['self_value'] > 0:
data.append(get_row(i['name'], i['self_value'], 0, i['level'] + 1)) data.append(get_row(i['name'], i['self_value'], 0, i['level'] + 1))
# append_blank()
return data return data
def get_filtered_entries(filters): def get_filtered_entries(filters):
gl_entries = get_gl_entries(filters, []) gl_entries = get_gl_entries(filters, [])
entries = [frappe.get_doc(gle.voucher_type, gle.voucher_no)for gle in gl_entries]
filtered_entries = [] filtered_entries = []
for entry in entries: for entry in gl_entries:
posting_date = entry.get("posting_date") posting_date = entry.get('posting_date')
from_date = filters.get("from_date") from_date = filters.get('from_date')
if date_diff(from_date, posting_date) > 0: if date_diff(from_date, posting_date) > 0:
continue continue
filtered_entries.append(entry) filtered_entries.append(entry)
return filtered_entries return filtered_entries
def append_blank(data): def get_stock_value_difference_list(filtered_entries):
if len(data) == 0: voucher_nos = [fe.get('voucher_no') for fe in filtered_entries]
data.append(get_row("", 0, 0, 0)) svd_list = frappe.get_list('Stock Ledger Entry',
fields=['item_code','stock_value_difference'],
filters=[('voucher_no', 'in', voucher_nos)])
assign_item_groups_to_svd_list(svd_list)
return svd_list
def get_item_groups_dict(item_groups_list): def get_leveled_dict():
return { (i['lft'],i['rgt']):{'name':i['name'], 'is_group':i['is_group']} item_groups_dict = get_item_groups_dict()
for i in item_groups_list } lr_list = sorted(item_groups_dict, key=lambda x : int(x[0]))
leveled_dict = OrderedDict()
def get_levels_dict(item_groups_dict):
lr_list = sorted(item_groups_dict, key=lambda x : x[0])
levels = OrderedDict()
current_level = 0 current_level = 0
nesting_r = [] nesting_r = []
for l,r in lr_list: for l, r in lr_list:
while current_level > 0 and nesting_r[-1] < l: while current_level > 0 and nesting_r[-1] < l:
nesting_r.pop() nesting_r.pop()
current_level -= 1 current_level -= 1
levels[(l,r)] = { leveled_dict[(l,r)] = {
'level' : current_level, 'level' : current_level,
'name' : item_groups_dict[(l,r)]['name'], 'name' : item_groups_dict[(l,r)]['name'],
'is_group' : item_groups_dict[(l,r)]['is_group'] 'is_group' : item_groups_dict[(l,r)]['is_group']
} }
if r - l > 1: if int(r) - int(l) > 1:
current_level += 1 current_level += 1
nesting_r.append(r) nesting_r.append(r)
return levels
update_leveled_dict(leveled_dict)
return leveled_dict
def update_levels_dict(levels_dict): def assign_self_values(leveled_dict, svd_list):
for k in levels_dict: levels_dict[k].update({'self_value':0, 'agg_value':0}) key_dict = {v['name']:k for k, v in leveled_dict.items()}
for item in svd_list:
key = key_dict[item.get("item_group")]
leveled_dict[key]['self_value'] += -item.get("stock_value_difference")
def assign_self_values(levels_dict, entries): def assign_agg_values(leveled_dict):
names_dict = {v['name']:k for k, v in levels_dict.items()} keys = list(leveled_dict.keys())[::-1]
for entry in entries: prev_level = leveled_dict[keys[-1]]['level']
items = entry.get("items")
items = [] if items is None else items
for item in items:
qty = item.get("qty")
incoming_rate = item.get("incoming_rate")
item_group = item.get("item_group")
key = names_dict[item_group]
levels_dict[key]['self_value'] += (incoming_rate * qty)
def assign_agg_values(levels_dict):
keys = list(levels_dict.keys())[::-1]
prev_level = levels_dict[keys[-1]]['level']
accu = [0] accu = [0]
for k in keys[:-1]: for k in keys[:-1]:
curr_level = levels_dict[k]['level'] curr_level = leveled_dict[k]['level']
if curr_level == prev_level: if curr_level == prev_level:
accu[-1] += levels_dict[k]['self_value'] accu[-1] += leveled_dict[k]['self_value']
levels_dict[k]['agg_value'] = levels_dict[k]['self_value'] leveled_dict[k]['agg_value'] = leveled_dict[k]['self_value']
elif curr_level > prev_level: elif curr_level > prev_level:
accu.append(levels_dict[k]['self_value']) accu.append(leveled_dict[k]['self_value'])
levels_dict[k]['agg_value'] = accu[-1] leveled_dict[k]['agg_value'] = accu[-1]
elif curr_level < prev_level: elif curr_level < prev_level:
accu[-1] += levels_dict[k]['self_value'] accu[-1] += leveled_dict[k]['self_value']
levels_dict[k]['agg_value'] = accu[-1] leveled_dict[k]['agg_value'] = accu[-1]
prev_level = curr_level prev_level = curr_level
# root node # root node
rk = keys[-1] rk = keys[-1]
levels_dict[rk]['agg_value'] = sum(accu) + levels_dict[rk]['self_value'] leveled_dict[rk]['agg_value'] = sum(accu) + leveled_dict[rk]['self_value']
def get_row(name:str, value:float, is_bold:int, indent:int): def get_row(name:str, value:float, is_bold:int, indent:int):
@ -153,3 +139,32 @@ def get_row(name:str, value:float, is_bold:int, indent:int):
if is_bold: if is_bold:
item_group = frappe.bold(item_group) item_group = frappe.bold(item_group)
return frappe._dict(item_group=item_group, cogs_debit=value, indent=indent) return frappe._dict(item_group=item_group, cogs_debit=value, indent=indent)
def assign_item_groups_to_svd_list(svd_list):
ig_map = get_item_groups_map(svd_list)
for item in svd_list:
item.item_group = ig_map[item.get("item_code")]
def get_item_groups_map(svd_list):
# for items in svd_list: [{'item_code':'item_group'}]
item_codes = set([i['item_code'] for i in svd_list])
ig_list = frappe.get_list('Item',
fields=['item_code','item_group'],
filters=[('item_code', 'in', item_codes)])
return {i['item_code']:i['item_group'] for i in ig_list}
def append_blank(data):
if len(data) == 0:
data.append(get_row("", 0, 0, 0))
def get_item_groups_dict():
item_groups_list = frappe.get_all("Item Group", fields=("name", "is_group", "lft", "rgt"))
return { (i['lft'],i['rgt']):{'name':i['name'], 'is_group':i['is_group']}
for i in item_groups_list }
def update_leveled_dict(leveled_dict):
for k in leveled_dict: leveled_dict[k].update({'self_value':0, 'agg_value':0})