diff --git a/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.js b/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.js index 27a38912a86..edfde8f52e1 100644 --- a/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.js +++ b/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.js @@ -5,7 +5,7 @@ frappe.ui.form.on("Period Closing Voucher", { onload: function (frm) { if (!frm.doc.transaction_date) frm.doc.transaction_date = frappe.datetime.obj_to_str(new Date()); - frm.ignore_doctypes_on_cancel_all = ["Process Period Closing Voucher"]; + frm.ignore_doctypes_on_cancel_all = ["Process Period Closing Voucher", "MapReduce Job"]; }, setup: function (frm) { diff --git a/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.py b/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.py index 1f4a60e6f14..83c1db7e7d2 100644 --- a/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.py +++ b/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.py @@ -5,9 +5,20 @@ import copy import frappe -from frappe import _ -from frappe.query_builder.functions import Max, Sum -from frappe.utils import add_days, flt, fmt_money, formatdate, get_link_to_form, getdate +from frappe import _, qb +from frappe.query_builder.custom import ConstantColumn +from frappe.query_builder.functions import Count, Max, Min, Sum +from frappe.utils import ( + add_days, + ceil, + cint, + flt, + fmt_money, + formatdate, + get_datetime, + get_link_to_form, + getdate, +) from erpnext import is_perpetual_inventory_enabled from erpnext.accounts.doctype.account_closing_balance.account_closing_balance import ( @@ -265,8 +276,17 @@ class PeriodClosingVoucher(AccountsController): if frappe.get_single_value("Accounts Settings", "use_legacy_controller_for_pcv"): self.make_gl_entries() else: - ppcv = frappe.get_doc({"doctype": "Process Period Closing Voucher", "parent_pcv": self.name}) - ppcv.save().submit() + from frappe.utils.background_jobs import mapreduce + + data = self.get_data_for_mapreduce() + mapreduce( + "erpnext.accounts.doctype.period_closing_voucher.period_closing_voucher.mapper", + "erpnext.accounts.doctype.period_closing_voucher.period_closing_voucher.reducer", + "erpnext.accounts.doctype.period_closing_voucher.period_closing_voucher.summarize_and_post_ledger", + data, + self.doctype, + self.name, + ) def on_cancel(self): self.ignore_linked_doctypes = ( @@ -275,11 +295,16 @@ class PeriodClosingVoucher(AccountsController): "Payment Ledger Entry", "Account Closing Balance", "Process Period Closing Voucher", + "MapReduce Job", ) + self.block_if_future_closing_voucher_exists() self.validate_accounts_not_frozen(for_cancellation=True) if not frappe.get_single_value("Accounts Settings", "use_legacy_controller_for_pcv"): + from frappe.utils.background_jobs import cancel_mapreduce_job + + cancel_mapreduce_job(self.doctype, self.name) self.cancel_process_pcv_docs() self.db_set("gle_processing_status", "In Progress") @@ -292,6 +317,11 @@ class PeriodClosingVoucher(AccountsController): def on_trash(self): super().on_trash() + if not frappe.get_single_value("Accounts Settings", "use_legacy_controller_for_pcv"): + from frappe.utils.background_jobs import remove_mapreduce_job + + remove_mapreduce_job(self.doctype, self.name) + ppcvs = frappe.db.get_all( "Process Period Closing Voucher", {"parent_pcv": self.name, "docstatus": ["in", [1, 2]]} ) @@ -594,6 +624,135 @@ class PeriodClosingVoucher(AccountsController): {"voucher_type": "Period Closing Voucher", "voucher_no": self.name, "is_cancelled": 0}, ) + def get_data_for_mapreduce(self): + return self.generate_tasks_for_normal_balance() + self.generate_tasks_for_opening_balance() + + def get_period_range_for_tasks(self, start_date, end_date, step_size, report_type, balance_type): + start_date = getdate(start_date) + end_date = getdate(end_date) + + # split period into date ranges + curr_date = getdate(start_date) + date_splits = [] + while True: + next_date = getdate(add_days(curr_date, step_size)) + if next_date < end_date: + date_splits.append( + { + "from_date": str(curr_date), + "to_date": str(next_date), + "pcv": self.name, + "report_type": report_type, + "balance_type": balance_type, + } + ) + curr_date = getdate(add_days(next_date, 1)) + else: + date_splits.append( + { + "from_date": str(curr_date), + "to_date": str(end_date), + "pcv": self.name, + "report_type": report_type, + "balance_type": balance_type, + } + ) + break + + return date_splits + + def generate_tasks_for_normal_balance(self): + # estimation can be wrong by a factor of 2 + gl = qb.DocType("GL Entry") + raw_query = ( + qb.from_(gl) + .select(Count(gl.star)) + .where( + gl.is_cancelled.eq(0) & gl.posting_date.between(self.period_start_date, self.period_end_date) + ) + .get_sql() + ) + + # estimation can be wrong by a factor of 2 + correction_factor = 2 + if frappe.db.db_type == "postgres": + analyzer = frappe.json.loads( + ( + frappe.db.sql( + f"explain (format json) {raw_query}", + ) + )[0][0] + ) + + estimated_count = analyzer[0].get("Plan").get("Plans")[0].get("Plan Rows") * correction_factor + else: + estimated_count = ( + cint( + frappe.db.sql( + f"explain {raw_query}", + as_dict=True, + )[0].rows + ) + * correction_factor + ) + + job_count = ( + 1 if estimated_count / 2000000 < 1 else ceil(estimated_count / 2000000) + ) # conservative chunk size + days = (getdate(self.period_end_date) - getdate(self.period_start_date)).days + step_size = 1 if days / job_count < 1 else ceil(days / job_count) + return self.get_period_range_for_tasks( + self.period_start_date, self.period_end_date, step_size, "Balance Sheet", "Normal Balance" + ) + self.get_period_range_for_tasks( + self.period_start_date, self.period_end_date, step_size, "Profit and Loss", "Normal Balance" + ) + + def generate_tasks_for_opening_balance(self): + tasks = [] + if self.is_first_period_closing_voucher(): + gl = qb.DocType("GL Entry") + min = qb.from_(gl).select(Min(gl.posting_date)).run()[0][0] + max = qb.from_(gl).select(Max(gl.posting_date)).run()[0][0] + + raw_query = ( + qb.from_(gl) + .select(Count(gl.star)) + .where(gl.is_cancelled.eq(0) & gl.is_opening.eq("Yes") & gl.posting_date.between(min, max)) + .get_sql() + ) + + # estimation can be wrong by a factor of 2 + correction_factor = 2 + if frappe.db.db_type == "postgres": + analyzer = frappe.json.loads( + ( + frappe.db.sql( + f"explain (format json) {raw_query}", + ) + )[0][0] + ) + + estimated_count = analyzer[0].get("Plan").get("Plans")[0].get("Plan Rows") * correction_factor + else: + estimated_count = ( + cint( + frappe.db.sql( + f"explain {raw_query};", + as_dict=True, + )[0].rows + ) + * correction_factor + ) + + job_count = ( + 1 if estimated_count / 2000000 < 1 else ceil(estimated_count / 2000000) + ) # conservative chunk size + days = (getdate(self.period_end_date) - getdate(self.period_start_date)).days + step_size = 1 if days / job_count < 1 else ceil(days / job_count) + tasks = self.get_period_range_for_tasks(min, max, step_size, "Balance Sheet", "Opening Balance") + + return tasks + def process_gl_and_closing_entries(doc): from erpnext.accounts.general_ledger import make_gl_entries @@ -673,3 +832,119 @@ def get_previous_closed_period_in_current_year(fiscal_year, company): order_by="period_end_date desc", ) return prev_closed_period_end_date + + +def mapper(val): + start_date = val.from_date + end_date = val.to_date + pcv = val.pcv + report_type = val.report_type + balance_type = val.balance_type + company = frappe.db.get_value("Period Closing Voucher", pcv, "company") + dimensions = get_dimensions() + + accounts = frappe.db.get_all( + "Account", filters={"company": company, "report_type": report_type}, pluck="name" + ) + + gle = qb.DocType("GL Entry") + query = qb.from_(gle).select(gle.account) + for dim in dimensions: + query = query.select(gle[dim]) + query = query.select( + Sum(gle.debit).as_("debit"), + Sum(gle.credit).as_("credit"), + Sum(gle.debit_in_account_currency).as_("debit_in_account_currency"), + Sum(gle.credit_in_account_currency).as_("credit_in_account_currency"), + # account_currency is constant per grouped account -> Max() keeps the GROUP BY postgres-valid + Max(gle.account_currency).as_("account_currency"), + ConstantColumn(balance_type).as_("balance_type"), + ConstantColumn(report_type).as_("report_type"), + ).where( + (gle.company.eq(company)) + & (gle.is_cancelled.eq(0)) + & (gle.posting_date.between(start_date, end_date)) + & (gle.account.isin(accounts)) + ) + + if balance_type == "Opening Balance": + query = query.where(gle.is_opening.eq("Yes")) + else: + # Keep balances aligned with legacy PCV logic (non-opening transactions only) + query = query.where(gle.is_opening.eq("No")) + + query = query.groupby(gle.account) + for dim in dimensions: + query = query.groupby(gle[dim]) + + res = query.run(as_dict=True) + return res + + +def reducer(final, partial_res): + if final is None: + final = [] + + if partial_res: + final.extend([frappe._dict(x) for x in partial_res]) + + return final + + +def get_dimensions(): + from erpnext.accounts.doctype.accounting_dimension.accounting_dimension import ( + get_accounting_dimensions, + ) + + default_dimensions = ["cost_center", "finance_book", "project"] + dimensions = default_dimensions + get_accounting_dimensions() + return dimensions + + +def summarize_and_post_ledger(result, ref_dt, ref_dn): + pcv = frappe.get_doc(ref_dt, ref_dn) + + from erpnext.accounts.doctype.process_period_closing_voucher.process_period_closing_voucher import ( + build_dimension_wise_balance_dict, + get_bs_closing_entries, + get_closing_account_closing_entry, + get_gle_for_closing_account, + get_gle_for_pl_account, + get_p_l_closing_entries, + ) + + result = [frappe._dict(x) for x in result] + + # generate and post closing entries for P&L accounts + pl_entries = [x for x in result if x.report_type == "Profit and Loss"] + pl_dimension_wise_acc_balance = build_dimension_wise_balance_dict(pl_entries) + + # build gl map + pl_accounts_reverse_gle = [] + closing_account_gle = [] + + for dimensions, account_balances in pl_dimension_wise_acc_balance.items(): + for acc, balances in account_balances.items(): + balance_in_company_currency = flt(balances.debit) - flt(balances.credit) + if balance_in_company_currency: + pl_accounts_reverse_gle.append(get_gle_for_pl_account(pcv, acc, balances, dimensions)) + + closing_account_gle.append(get_gle_for_closing_account(pcv, account_balances["balances"], dimensions)) + + gl_entries = pl_accounts_reverse_gle + closing_account_gle + if gl_entries: + from erpnext.accounts.general_ledger import make_gl_entries + + make_gl_entries(gl_entries, merge_entries=False) + + # generate and post account closing balance for balance sheet accounts + bs_entries = [x for x in result if x.report_type == "Balance Sheet"] + bs_dimension_wise_acc_balance = build_dimension_wise_balance_dict(bs_entries) + pl_closing_entries = get_p_l_closing_entries(pl_accounts_reverse_gle, pcv) + bs_closing_entries = get_bs_closing_entries(bs_dimension_wise_acc_balance, pcv) + closing_entries_for_closing_account = get_closing_account_closing_entry(closing_account_gle, pcv) + closing_entries = pl_closing_entries + bs_closing_entries + closing_entries_for_closing_account + + make_closing_entries(closing_entries, pcv.name, pcv.company, pcv.period_end_date) + + frappe.db.set_value("Period Closing Voucher", pcv.name, "gle_processing_status", "Completed") diff --git a/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher_dashboard.py b/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher_dashboard.py new file mode 100644 index 00000000000..fcbeb8350ba --- /dev/null +++ b/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher_dashboard.py @@ -0,0 +1,8 @@ +from frappe import _ + + +def get_data(): + return { + "non_standard_fieldnames": {"MapReduce Job": "document_name"}, + "transactions": [{"label": _("Job"), "items": ["MapReduce Job"]}], + }