fix(vat audit report): fallback to item name when item code is missing (#54049)

* fix(vat audit report): fallback to item name when item code is missing

* fix: validate south africa company selection

* fix: simplify parent item lookup

* fix: handle missing item mapping

* fix: use list instead of set

erpnext/regional/doctype/south_africa_vat_settings/south_africa_vat_settings.py

@@ -1,9 +1,12 @@
 # Copyright (c) 2021, Frappe Technologies Pvt. Ltd. and contributors
 # For license information, please see license.txt
 
-# import frappe
+import frappe
+from frappe import _
 from frappe.model.document import Document
 
+from erpnext import get_region
+
 
 class SouthAfricaVATSettings(Document):
 	# begin: auto-generated types
@@ -22,4 +25,9 @@ class SouthAfricaVATSettings(Document):
 		vat_accounts: DF.Table[SouthAfricaVATAccount]
 	# end: auto-generated types
 
-	pass
+	def validate(self):
+		self.validate_company_region()
+
+	def validate_company_region(self):
+		if self.company and get_region(self.company) != "South Africa":
+			frappe.throw(_("Company {0} is not in South Africa.").format(frappe.bold(self.company)))

erpnext/regional/report/vat_audit_report/vat_audit_report.js

@@ -10,6 +10,13 @@ frappe.query_reports["VAT Audit Report"] = {
 			options: "Company",
 			reqd: 1,
 			default: frappe.defaults.get_user_default("Company"),
+			get_query: function () {
+				return {
+					filters: {
+						country: "South Africa",
+					},
+				};
+			},
 		},
 		{
 			fieldname: "from_date",

erpnext/regional/report/vat_audit_report/vat_audit_report.py

@@ -6,8 +6,11 @@ import json
 
 import frappe
 from frappe import _
+from frappe.query_builder.functions import Coalesce, NullIf
 from frappe.utils import formatdate, get_link_to_form
 
+from erpnext import get_region
+
 
 def execute(filters=None):
 	return VATAuditReport(filters).run()
@@ -21,19 +24,10 @@ class VATAuditReport:
 		self.doctypes = ["Purchase Invoice", "Sales Invoice"]
 
 	def run(self):
+		self.validate_company_region()
 		self.get_sa_vat_accounts()
 		self.get_columns()
 		for doctype in self.doctypes:
-			self.select_columns = """
-			name as voucher_no,
-			posting_date, remarks"""
-			columns = (
-				", supplier as party, credit_to as account"
-				if doctype == "Purchase Invoice"
-				else ", customer as party, debit_to as account"
-			)
-			self.select_columns += columns
-
 			self.get_invoice_data(doctype)
 
 			if self.invoices:
@@ -43,6 +37,14 @@ class VATAuditReport:
 
 		return self.columns, self.data
 
+	def validate_company_region(self):
+		if self.filters.company and get_region(self.filters.company) != "South Africa":
+			frappe.throw(
+				_(
+					"The company {0} is not in South Africa. VAT Audit Report is only available for companies in South Africa."
+				).format(frappe.bold(self.filters.company))
+			)
+
 	def get_sa_vat_accounts(self):
 		self.sa_vat_accounts = frappe.get_all(
 			"South Africa VAT Account", filters={"parent": self.filters.company}, pluck="account"
@@ -54,47 +56,59 @@ class VATAuditReport:
 			frappe.throw(_("Please set VAT Accounts in {0}").format(link_to_settings))
 
 	def get_invoice_data(self, doctype):
-		conditions = self.get_conditions()
 		self.invoices = frappe._dict()
-
+		invoice_doctype = frappe.qb.DocType(doctype)
-		invoice_data = frappe.db.sql(
+		party_field = invoice_doctype.supplier if doctype == "Purchase Invoice" else invoice_doctype.customer
-			f"""
+		account_field = (
-			SELECT
+			invoice_doctype.credit_to if doctype == "Purchase Invoice" else invoice_doctype.debit_to
-				{self.select_columns}
-			FROM
-				`tab{doctype}`
-			WHERE
-				docstatus = 1 {conditions}
-				and is_opening = 'No'
-			ORDER BY
-				posting_date DESC
-			""",
-			self.filters,
-			as_dict=1,
 		)
 
-		for d in invoice_data:
+		query = (
-			self.invoices.setdefault(d.voucher_no, d)
+			frappe.qb.from_(invoice_doctype)
+			.select(
+				invoice_doctype.name.as_("voucher_no"),
+				invoice_doctype.posting_date,
+				invoice_doctype.remarks,
+				party_field.as_("party"),
+				account_field.as_("account"),
+			)
+			.where(invoice_doctype.docstatus == 1)
+			.where(invoice_doctype.is_opening == "No")
+			.orderby(invoice_doctype.posting_date, order=frappe.qb.desc)
+		)
+
+		if self.filters.get("company"):
+			query = query.where(invoice_doctype.company == self.filters.company)
+		if self.filters.get("from_date"):
+			query = query.where(invoice_doctype.posting_date >= self.filters.from_date)
+		if self.filters.get("to_date"):
+			query = query.where(invoice_doctype.posting_date <= self.filters.to_date)
+
+		invoice_data = query.run(as_dict=True)
+
+		for row in invoice_data:
+			self.invoices.setdefault(row.voucher_no, row)
 
 	def get_invoice_items(self, doctype):
 		self.invoice_items = frappe._dict()
+		item_doctype = frappe.qb.DocType(doctype + " Item")
 
-		items = frappe.db.sql(
+		items = (
-			"""
+			frappe.qb.from_(item_doctype)
-			SELECT
+			.select(
-				item_code, parent, base_net_amount, is_zero_rated
+				Coalesce(NullIf(item_doctype.item_code, ""), item_doctype.item_name).as_("item"),
-			FROM
+				item_doctype.parent,
-				`tab{} Item`
+				item_doctype.base_net_amount,
-			WHERE
+				item_doctype.is_zero_rated,
-				parent in ({})
+			)
-			""".format(doctype, ", ".join(["%s"] * len(self.invoices))),
+			.where(item_doctype.parent.isin(list(self.invoices.keys())))
-			tuple(self.invoices),
+			.run(as_dict=True)
-			as_dict=1,
 		)
-		for d in items:
+
-			self.invoice_items.setdefault(d.parent, {}).setdefault(d.item_code, {"net_amount": 0.0})
+		for row in items:
-			self.invoice_items[d.parent][d.item_code]["net_amount"] += d.get("base_net_amount", 0)
+			self.invoice_items.setdefault(row.parent, {}).setdefault(row.item, {"net_amount": 0.0})
-			self.invoice_items[d.parent][d.item_code]["is_zero_rated"] = d.is_zero_rated
+			self.invoice_items[row.parent][row.item]["net_amount"] += row.get("base_net_amount", 0)
+			self.invoice_items[row.parent][row.item]["is_zero_rated"] = row.is_zero_rated
 
 	def get_items_based_on_tax_rate(self, doctype):
 		self.items_based_on_tax_rate = frappe._dict()
@@ -103,52 +117,54 @@ class VATAuditReport:
 			"Purchase Taxes and Charges" if doctype == "Purchase Invoice" else "Sales Taxes and Charges"
 		)
 
-		self.tax_details = frappe.db.sql(
+		tax_doctype = frappe.qb.DocType(self.tax_doctype)
-			"""
+		self.tax_details = (
-			SELECT
+			frappe.qb.from_(tax_doctype)
-				parent, account_head, item_wise_tax_detail
+			.select(tax_doctype.parent, tax_doctype.account_head, tax_doctype.item_wise_tax_detail)
-			FROM
+			.where(tax_doctype.parenttype == doctype)
-				`tab{}`
+			.where(tax_doctype.docstatus == 1)
-			WHERE
+			.where(tax_doctype.parent.isin(list(self.invoices.keys())))
-				parenttype = {} and docstatus = 1
+			.where(tax_doctype.account_head.isin(self.sa_vat_accounts))
-				and parent in ({})
+			.orderby(tax_doctype.account_head)
-			ORDER BY
+			.run(as_dict=True)
-				account_head
-			""".format(self.tax_doctype, "%s", ", ".join(["%s"] * len(self.invoices.keys()))),
-			tuple([doctype, *list(self.invoices.keys())]),
 		)
 
-		for parent, account, item_wise_tax_detail in self.tax_details:
+		for tax_detail in self.tax_details:
-			if item_wise_tax_detail:
+			if not tax_detail.item_wise_tax_detail:
-				try:
+				continue
-					if account in self.sa_vat_accounts:
-						item_wise_tax_detail = json.loads(item_wise_tax_detail)
-					else:
-						continue
-					for item_code, taxes in item_wise_tax_detail.items():
-						is_zero_rated = self.invoice_items.get(parent).get(item_code).get("is_zero_rated")
-						# to skip items with non-zero tax rate in multiple rows
-						if taxes[0] == 0 and not is_zero_rated:
-							continue
-						tax_rate = self.get_item_amount_map(parent, item_code, taxes)
 
-						if tax_rate is not None:
+			try:
-							rate_based_dict = self.items_based_on_tax_rate.setdefault(parent, {}).setdefault(
+				item_wise_tax_detail = json.loads(tax_detail.item_wise_tax_detail)
-								tax_rate, []
+			except ValueError:
-							)
+				continue
-							if item_code not in rate_based_dict:
+
-								rate_based_dict.append(item_code)
+			parent_items = self.invoice_items.get(tax_detail.parent, {})
-				except ValueError:
+			parent_tax_rates = self.items_based_on_tax_rate.setdefault(tax_detail.parent, {})
+
+			for item, taxes in item_wise_tax_detail.items():
+				is_zero_rated = parent_items.get(item, {}).get("is_zero_rated")
+				# to skip items with non-zero tax rate in multiple rows
+				if taxes[0] == 0 and not is_zero_rated:
 					continue
 
-	def get_item_amount_map(self, parent, item_code, taxes):
+				tax_rate = self.get_item_amount_map(tax_detail.parent, item, taxes)
-		net_amount = self.invoice_items.get(parent).get(item_code).get("net_amount")
+				if tax_rate is not None:
+					rate_based_dict = parent_tax_rates.setdefault(tax_rate, [])
+					if item not in rate_based_dict:
+						rate_based_dict.append(item)
+
+	def get_item_amount_map(self, parent, item, taxes):
+		item_details = self.invoice_items.get(parent, {}).get(item)
+		if not item_details:
+			return None
+
+		net_amount = item_details.get("net_amount", 0)
 		tax_rate = taxes[0]
 		tax_amount = taxes[1]
 		gross_amount = net_amount + tax_amount
 
 		self.item_tax_rate.setdefault(parent, {}).setdefault(
-			item_code,
+			item,
 			{
 				"tax_rate": tax_rate,
 				"gross_amount": 0.0,
@@ -157,24 +173,12 @@ class VATAuditReport:
 			},
 		)
 
-		self.item_tax_rate[parent][item_code]["net_amount"] += net_amount
+		self.item_tax_rate[parent][item]["net_amount"] += net_amount
-		self.item_tax_rate[parent][item_code]["tax_amount"] += tax_amount
+		self.item_tax_rate[parent][item]["tax_amount"] += tax_amount
-		self.item_tax_rate[parent][item_code]["gross_amount"] += gross_amount
+		self.item_tax_rate[parent][item]["gross_amount"] += gross_amount
 
 		return tax_rate
 
-	def get_conditions(self):
-		conditions = ""
-		for opts in (
-			("company", " and company=%(company)s"),
-			("from_date", " and posting_date>=%(from_date)s"),
-			("to_date", " and posting_date<=%(to_date)s"),
-		):
-			if self.filters.get(opts[0]):
-				conditions += opts[1]
-
-		return conditions
-
 	def get_data(self, doctype):
 		consolidated_data = self.get_consolidated_data(doctype)
 		section_name = _("Purchases") if doctype == "Purchase Invoice" else _("Sales")