refactor: add type hints, remove comment, sort imports
This commit is contained in:
parent
98c9b0e9ed
commit
865900fd2d
@ -1,14 +1,28 @@
|
||||
# Copyright (c) 2013, Frappe Technologies Pvt. Ltd. and contributors
|
||||
# For license information, please see license.txt
|
||||
|
||||
from collections import OrderedDict
|
||||
import datetime
|
||||
from typing import Dict, List, Tuple, Union
|
||||
|
||||
import frappe
|
||||
from frappe import _
|
||||
from frappe.utils import date_diff
|
||||
from collections import OrderedDict
|
||||
|
||||
from erpnext.accounts.report.general_ledger.general_ledger import get_gl_entries
|
||||
|
||||
|
||||
def execute(filters=None):
|
||||
Filters = frappe._dict
|
||||
Row = frappe._dict
|
||||
Data = List[Row]
|
||||
Columns = List[Dict[str, str]]
|
||||
DateTime = Union[datetime.date, datetime.datetime]
|
||||
FilteredEntries = List[Dict[str, Union[str, float, DateTime, None]]]
|
||||
ItemGroupsDict = Dict[Tuple[int, int], Dict[str, Union[str, int]]]
|
||||
SVDList = List[frappe._dict]
|
||||
|
||||
|
||||
def execute(filters: Filters) -> Tuple[Columns, Data]:
|
||||
update_filters_with_account(filters)
|
||||
validate_filters(filters)
|
||||
columns = get_columns()
|
||||
@ -16,17 +30,17 @@ def execute(filters=None):
|
||||
return columns, data
|
||||
|
||||
|
||||
def update_filters_with_account(filters):
|
||||
def update_filters_with_account(filters: Filters) -> None:
|
||||
account = frappe.get_value("Company", filters.get("company"), "default_expense_account")
|
||||
filters.update(dict(account=account))
|
||||
|
||||
|
||||
def validate_filters(filters):
|
||||
def validate_filters(filters: Filters) -> None:
|
||||
if filters.from_date > filters.to_date:
|
||||
frappe.throw(_("From Date must be before To Date"))
|
||||
|
||||
|
||||
def get_columns():
|
||||
def get_columns() -> Columns:
|
||||
return [
|
||||
{
|
||||
'label': 'Item Group',
|
||||
@ -43,7 +57,7 @@ def get_columns():
|
||||
]
|
||||
|
||||
|
||||
def get_data(filters):
|
||||
def get_data(filters: Filters) -> Data:
|
||||
filtered_entries = get_filtered_entries(filters)
|
||||
svd_list = get_stock_value_difference_list(filtered_entries)
|
||||
leveled_dict = get_leveled_dict()
|
||||
@ -62,7 +76,7 @@ def get_data(filters):
|
||||
return data
|
||||
|
||||
|
||||
def get_filtered_entries(filters):
|
||||
def get_filtered_entries(filters: Filters) -> FilteredEntries:
|
||||
gl_entries = get_gl_entries(filters, [])
|
||||
filtered_entries = []
|
||||
for entry in gl_entries:
|
||||
@ -74,7 +88,7 @@ def get_filtered_entries(filters):
|
||||
return filtered_entries
|
||||
|
||||
|
||||
def get_stock_value_difference_list(filtered_entries):
|
||||
def get_stock_value_difference_list(filtered_entries: FilteredEntries) -> SVDList:
|
||||
voucher_nos = [fe.get('voucher_no') for fe in filtered_entries]
|
||||
svd_list = frappe.get_list(
|
||||
'Stock Ledger Entry', fields=['item_code','stock_value_difference'],
|
||||
@ -84,7 +98,7 @@ def get_stock_value_difference_list(filtered_entries):
|
||||
return svd_list
|
||||
|
||||
|
||||
def get_leveled_dict():
|
||||
def get_leveled_dict() -> OrderedDict:
|
||||
item_groups_dict = get_item_groups_dict()
|
||||
lr_list = sorted(item_groups_dict, key=lambda x : int(x[0]))
|
||||
leveled_dict = OrderedDict()
|
||||
@ -109,14 +123,14 @@ def get_leveled_dict():
|
||||
return leveled_dict
|
||||
|
||||
|
||||
def assign_self_values(leveled_dict, svd_list):
|
||||
def assign_self_values(leveled_dict: OrderedDict, svd_list: SVDList) -> None:
|
||||
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_agg_values(leveled_dict):
|
||||
def assign_agg_values(leveled_dict: OrderedDict) -> None:
|
||||
keys = list(leveled_dict.keys())[::-1]
|
||||
prev_level = leveled_dict[keys[-1]]['level']
|
||||
accu = [0]
|
||||
@ -141,21 +155,21 @@ def assign_agg_values(leveled_dict):
|
||||
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) -> Row:
|
||||
item_group = name
|
||||
if is_bold:
|
||||
item_group = frappe.bold(item_group)
|
||||
return frappe._dict(item_group=item_group, cogs_debit=value, indent=indent)
|
||||
|
||||
|
||||
def assign_item_groups_to_svd_list(svd_list):
|
||||
def assign_item_groups_to_svd_list(svd_list: SVDList) -> None:
|
||||
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])
|
||||
|
||||
def get_item_groups_map(svd_list: SVDList) -> Dict[str, str]:
|
||||
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)]
|
||||
@ -163,12 +177,12 @@ def get_item_groups_map(svd_list):
|
||||
return {i['item_code']:i['item_group'] for i in ig_list}
|
||||
|
||||
|
||||
def get_item_groups_dict():
|
||||
def get_item_groups_dict() -> ItemGroupsDict:
|
||||
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):
|
||||
def update_leveled_dict(leveled_dict: OrderedDict) -> None:
|
||||
for k in leveled_dict:
|
||||
leveled_dict[k].update({'self_value':0, 'agg_value':0})
|
||||
|
Loading…
x
Reference in New Issue
Block a user