refactor: FTA Audit File and UAE VAT Reports

erpnext/patches/v16_0/add_arabic_company_name_field.py

@@ -1,10 +1,21 @@
 import frappe
-
+from frappe.custom.doctype.custom_field.custom_field import create_custom_fields
-from erpnext.regional.united_arab_emirates.setup import make_custom_fields
 
 
 def execute():
-	if not frappe.db.get_value("Company", {"country": "United Arab Emirates"}):
+	if not frappe.db.exists("Company", {"country": "United Arab Emirates"}):
 		return
 
-	make_custom_fields()
+	create_custom_fields(
+		{
+			"Company": [
+				{
+					"fieldname": "company_name_in_arabic",
+					"label": "Company Name in Arabic",
+					"fieldtype": "Data",
+					"insert_after": "company_name",
+				}
+			]
+		},
+		ignore_validate=True,
+	)

erpnext/regional/doctype/fta_audit_file/fta_audit_file.json

@@ -112,7 +112,7 @@
    "fieldname": "file_type",
    "fieldtype": "Select",
    "label": "File Type",
-   "options": "VAT\nExcise",
+   "options": "VAT",
    "default": "VAT",
    "reqd": 1
   },

erpnext/regional/doctype/fta_audit_file/fta_audit_file.py

@@ -35,6 +35,23 @@ DEFAULT_COUNTRY = "United Arab Emirates"
 DEFAULT_DATE = "31-12-9999"
 PRODUCT_VERSION = "ERPNext"
 
+# Inputs that define what the FAF file represents. Once the file has been
+# Generated or Submitted, changing any of these would silently desync the
+# attached CSV from the form, so we lock them.
+LOCKED_INPUT_FIELDS = (
+	"company",
+	"from_date",
+	"to_date",
+	"file_type",
+	"include_opening_balance",
+	"tax_agency_name",
+	"tan",
+	"tax_agent_name",
+	"taan",
+)
+LOCKED_STATUSES = ("Generated", "Submitted")
+IN_FLIGHT_STATUSES = ("Queued", "Generating")
+
 
 class FTAAuditFile(Document):
 	def validate(self):
@@ -48,9 +65,31 @@ class FTAAuditFile(Document):
 				)
 			)
 
+		self._guard_locked_fields()
+
 		if self.status != "Error":
 			self.error_message = None
 
+	def _guard_locked_fields(self):
+		"""Block edits to FAF inputs once the file is Generated or Submitted.
+
+		The status field alone is read-only in the UI, but a user with write
+		permission can still patch fields via REST or scripts; this enforces
+		immutability server-side so the attached CSV always matches the form.
+		"""
+		if self.is_new():
+			return
+
+		previous_status = self.get_db_value("status")
+		if previous_status not in LOCKED_STATUSES:
+			return
+
+		changed = [f for f in LOCKED_INPUT_FIELDS if self.has_value_changed(f)]
+		if changed:
+			frappe.throw(
+				_("Cannot modify {0} after the FAF has been {1}.").format(", ".join(changed), previous_status)
+			)
+
 	@frappe.whitelist()
 	def generate_faf(self):
 		"""Queue FAF generation as a background job.
@@ -64,6 +103,14 @@ class FTAAuditFile(Document):
 		Under ``frappe.flags.in_test`` the job runs synchronously so tests
 		can assert on the post-generation state without polling.
 		"""
+		# Re-read status from DB so two concurrent button clicks can't both
+		# enqueue a job — the second one sees Queued/Generating and bails.
+		current_status = frappe.db.get_value(self.doctype, self.name, "status", for_update=True)
+		if current_status in IN_FLIGHT_STATUSES:
+			frappe.throw(_("FAF generation is already {0} for this document.").format(current_status))
+		if current_status == "Submitted":
+			frappe.throw(_("Cannot regenerate a Submitted FAF."))
+
 		self.status = "Queued"
 		self.generation_log = ""
 		self.error_message = None
@@ -140,9 +187,6 @@ class FTAAuditFile(Document):
 		log(f"Period: {self.from_date} to {self.to_date}")
 		log(f"File Type: {self.file_type}")
 
-		if self.file_type != "VAT":
-			frappe.throw(_("FAF generation for {0} is not yet implemented").format(self.file_type))
-
 		output = io.StringIO()
 		writer = csv.writer(output)
 
@@ -252,11 +296,14 @@ class FTAAuditFile(Document):
 				"item_tax_template",
 			],
 		)
-		tax_code_map = {
+		tax_code_bands = _bulk_tax_code_bands(
-			t: _resolve_tax_code(t)
+			{
-			for t in {item.item_tax_template for items in items_by_invoice.values() for item in items}
+				item.item_tax_template
-			if t
+				for items in items_by_invoice.values()
-		}
+				for item in items
+				if item.item_tax_template
+			}
+		)
 
 		company_currency = _company_currency(self.company)
 
@@ -285,7 +332,7 @@ class FTAAuditFile(Document):
 						_clean(item.description or item.item_name or ""),
 						_money(net_aed),
 						_money(vat_aed),
-						tax_code_map.get(item.item_tax_template, "SR"),
+						_resolve_tax_code(item.item_tax_template, inv.posting_date, tax_code_bands),
 						fcy_code,
 						_money(net_fcy),
 						_money(vat_fcy),
@@ -351,11 +398,14 @@ class FTAAuditFile(Document):
 				"is_exempt",
 			],
 		)
-		tax_code_map = {
+		tax_code_bands = _bulk_tax_code_bands(
-			t: _resolve_tax_code(t)
+			{
-			for t in {item.item_tax_template for items in items_by_invoice.values() for item in items}
+				item.item_tax_template
-			if t
+				for items in items_by_invoice.values()
-		}
+				for item in items
+				if item.item_tax_template
+			}
+		)
 
 		company_currency = _company_currency(self.company)
 
@@ -379,7 +429,7 @@ class FTAAuditFile(Document):
 				elif item.is_exempt:
 					tax_code = "EX"
 				else:
-					tax_code = tax_code_map.get(item.item_tax_template, "SR")
+					tax_code = _resolve_tax_code(item.item_tax_template, inv.posting_date, tax_code_bands)
 
 				writer.writerow(
 					[
@@ -563,7 +613,14 @@ def _bulk_party_field(doctype, names, field):
 
 
 def _bulk_party_country(party_doctype, party_names):
-	"""Return ``{party_name: country}`` taken from each party's first address with a country."""
+	"""Return ``{party_name: country}`` for each party.
+
+	Picks deterministically when a party has multiple addresses: prefer the
+	one flagged ``is_primary_address``, then ``is_shipping_address``, then
+	the lowest address name. Without this ordering, MariaDB would return
+	rows in storage-engine order and the FAF would be non-reproducible
+	across runs.
+	"""
 	if not party_names:
 		return {}
 
@@ -579,6 +636,9 @@ def _bulk_party_country(party_doctype, party_names):
 		.where(addr.country.isnotnull())
 		.where(addr.country != "")
 		.select(dl.link_name, addr.country)
+		.orderby(addr.is_primary_address, order=frappe.qb.desc)
+		.orderby(addr.is_shipping_address, order=frappe.qb.desc)
+		.orderby(addr.name)
 		.run(as_dict=True)
 	)
 	out = {}
@@ -638,22 +698,51 @@ def _bulk_invoice_items(child_doctype, invoice_names, fields):
 _FTA_TAX_CODES = ("SR", "ZR", "EX", "RC", "IG", "OA", "IA")
 
 
-def _resolve_tax_code(item_tax_template):
+def _bulk_tax_code_bands(item_tax_templates):
+	"""Return ``{template: [(valid_from, tax_category), ...]}`` sorted desc by valid_from.
+
+	One ``Item Tax Template`` can have multiple ``Item Tax`` rows with
+	different ``valid_from`` dates (e.g. tax code changing on a regulator
+	cutover). Fetching them all up-front lets ``_resolve_tax_code`` pick
+	the row that was in force on each invoice's posting date without an
+	extra DB hit per line item.
+	"""
+	if not item_tax_templates:
+		return {}
+	rows = frappe.get_all(
+		"Item Tax",
+		filters={"item_tax_template": ["in", list(item_tax_templates)]},
+		fields=["item_tax_template", "tax_category", "valid_from"],
+		order_by="valid_from desc",
+	)
+	out = {}
+	for r in rows:
+		out.setdefault(r["item_tax_template"], []).append((r.get("valid_from"), r.get("tax_category")))
+	return out
+
+
+def _resolve_tax_code(item_tax_template, posting_date, bands_map):
 	"""Derive FTA tax code (SR/ZR/EX/RC/IG/OA/IA) from one Item Tax Template.
 
-	Uses the Item Tax row's ``tax_category`` if it matches an FTA code;
+	Picks the Item Tax row whose ``valid_from`` is the most recent value
-	otherwise defaults to ``SR`` (Standard Rated). Setting ``tax_category``
+	that is still on or before ``posting_date``; rows with no
-	on each Item Tax is the supported way to control this — there is no
+	``valid_from`` are treated as always-valid and used only as a
-	heuristic fallback on the template name.
+	fallback. Defaults to ``SR`` (Standard Rated) when nothing matches or
+	the chosen ``tax_category`` isn't one of the FTA codes.
 	"""
 	if not item_tax_template:
 		return "SR"
 
-	tax_category = frappe.db.get_value(
+	bands = bands_map.get(item_tax_template) or []
-		"Item Tax",
+	posting = getdate(posting_date) if posting_date else None
-		{"item_tax_template": item_tax_template},
+	fallback_category = None
-		"tax_category",
+	for valid_from, tax_category in bands:
-	)
+		if valid_from is None:
-	if tax_category and tax_category in _FTA_TAX_CODES:
+			fallback_category = fallback_category or tax_category
-		return tax_category
+			continue
+		if posting is None or getdate(valid_from) <= posting:
+			return tax_category if tax_category in _FTA_TAX_CODES else "SR"
+
+	if fallback_category and fallback_category in _FTA_TAX_CODES:
+		return fallback_category
 	return "SR"

erpnext/regional/report/uae_vat_201/uae_vat_201.py

@@ -2,9 +2,13 @@
 # For license information, please see license.txt
 
 
+from html import escape
+from urllib.parse import urlencode
+
 import frappe
 from frappe import _
-from frappe.query_builder.functions import Sum
+from frappe.query_builder.functions import Coalesce, Sum
+from frappe.utils import flt
 
 from erpnext import get_region
 
@@ -14,6 +18,25 @@ from erpnext import get_region
 _cache = {}
 
 
+def _drill_down_link(text, filters, **extra):
+	"""Return an `<a>` tag pointing at the UAE VAT Register report.
+
+	Filter values are URL-encoded so company names with ``&`` or other
+	reserved characters don't break the query string, and the link text
+	is HTML-escaped to prevent injection from user-controlled fields.
+	"""
+	params = {}
+	for key in ("company", "from_date", "to_date"):
+		value = (filters or {}).get(key)
+		if value:
+			params[key] = value
+	for key, value in extra.items():
+		if value is not None:
+			params[key] = value
+	query = urlencode(params)
+	return f'<a href="/app/query-report/UAE VAT Register?{query}">{escape(str(text))}</a>'
+
+
 def _cached(fn):
 	def wrapper(filters, *args, **kwargs):
 		key = (fn.__name__, tuple(sorted((filters or {}).items())))
@@ -68,179 +91,68 @@ def get_data(filters=None):
 	append_vat_on_expenses(data, filters)
 	net_vat_due(data, filters, amounts_by_emirate)
 
+	emirate_drill_downs = {f"Standard rated supplies in {emirate}": emirate for emirate in get_emirates()}
+	dubai_legend = "Standard rated supplies in Dubai"
+	dubai_label_override = _company_emirate_label(filters)
+
 	final_data = []
-	for i in range(0, len(data)):
+	for row in data:
-		if data[i].get("legend") == "Standard rated supplies in Abu Dhabi":
+		legend = row.get("legend")
-			legend_link = f"""
+		new_legend = legend
-			<a href= "/app/query-report/UAE%20VAT%20Register?company={filters.get("company")}&from_date={filters.get("from_date")}&to_date={filters.get("to_date")}&doc_type=Sales%20Invoice&vat=Abu%20Dhabi&category=Standard">{data[i].get("legend")}</a>
-			"""
-			final_data.append(
-				{
-					"no": data[i].get("no"),
-					"legend": legend_link,
-					"amount": data[i].get("amount"),
-					"vat_amount": data[i].get("vat_amount"),
-				}
-			)
-		elif data[i].get("legend") == "Standard rated supplies in Dubai":
-			company = frappe.defaults.get_user_default("Company")
-			company_filters = [
-				["Dynamic Link", "link_doctype", "=", "Company"],
-				["Dynamic Link", "link_name", "=", company],
-				["Address", "is_your_company_address", "=", 1],
-			]
-			company_fields = [
-				"name",
-				"address_line1",
-				"address_line2",
-				"city",
-				"state",
-				"country",
-				"emirate",
-			]
-			address = frappe.get_all("Address", filters=company_filters, fields=company_fields)
 
-			if address:
+		if legend in emirate_drill_downs:
-				if address[0].get("emirate"):
+			emirate = emirate_drill_downs[legend]
-					name = "Standard rated supplies in" + " " + address[0].get("emirate")
+			label = dubai_label_override if legend == dubai_legend and dubai_label_override else legend
-				else:
+			new_legend = _drill_down_link(
-					name = "Standard rated supplies in Dubai"
+				label, filters, doc_type="Sales Invoice", vat=emirate, category="Standard"
-			else:
+			)
-				name = "Standard rated supplies in Dubai"
+		elif legend == "Supplies subject to the reverse charge provision":
+			new_legend = _drill_down_link(legend, filters, doc_type="Purchase Invoice", reverse_charge="Y")
+		elif legend == "Zero Rated":
+			new_legend = _drill_down_link(legend, filters, doc_type="Sales Invoice", category="Zero Rated")
+		elif legend == "Exempt Supplies":
+			new_legend = _drill_down_link(legend, filters, doc_type="Sales Invoice", category="Exempt Rated")
+		elif legend == "Standard Rated Expenses":
+			new_legend = _drill_down_link(legend, filters, doc_type="Purchase Invoice")
 
-			legend_link = f"""
+		final_data.append(
-			<a href= "/app/query-report/UAE%20VAT%20Register?company={filters.get("company")}&from_date={filters.get("from_date")}&to_date={filters.get("to_date")}&doc_type=Sales%20Invoice&vat=Dubai&category=Standard">{name}</a>
+			{
-			"""
+				"no": row.get("no"),
-			final_data.append(
+				"legend": new_legend,
-				{
+				"amount": row.get("amount"),
-					"no": data[i].get("no"),
+				"vat_amount": row.get("vat_amount"),
-					"legend": legend_link,
+			}
-					"amount": data[i].get("amount"),
+		)
-					"vat_amount": data[i].get("vat_amount"),
-				}
-			)
-
-		elif data[i].get("legend") == "Standard rated supplies in Sharjah":
-			legend_link = f"""
-			<a href= "/app/query-report/UAE%20VAT%20Register?company={filters.get("company")}&from_date={filters.get("from_date")}&to_date={filters.get("to_date")}&doc_type=Sales%20Invoice&vat=Sharjah&category=Standard">{data[i].get("legend")}</a>
-			"""
-			final_data.append(
-				{
-					"no": data[i].get("no"),
-					"legend": legend_link,
-					"amount": data[i].get("amount"),
-					"vat_amount": data[i].get("vat_amount"),
-				}
-			)
-		elif data[i].get("legend") == "Standard rated supplies in Ajman":
-			legend_link = f"""
-			<a href= "/app/query-report/UAE%20VAT%20Register?company={filters.get("company")}&from_date={filters.get("from_date")}&to_date={filters.get("to_date")}&doc_type=Sales%20Invoice&vat=Ajman&category=Standard">{data[i].get("legend")}</a>
-			"""
-			final_data.append(
-				{
-					"no": data[i].get("no"),
-					"legend": legend_link,
-					"amount": data[i].get("amount"),
-					"vat_amount": data[i].get("vat_amount"),
-				}
-			)
-		elif data[i].get("legend") == "Standard rated supplies in Umm Al Quwain":
-			legend_link = f"""
-			<a href= "/app/query-report/UAE%20VAT%20Register?company={filters.get("company")}&from_date={filters.get("from_date")}&to_date={filters.get("to_date")}&doc_type=Sales%20Invoice&vat=Umm%20Al%20Quwain&category=Standard">{data[i].get("legend")}</a>
-			"""
-			final_data.append(
-				{
-					"no": data[i].get("no"),
-					"legend": legend_link,
-					"amount": data[i].get("amount"),
-					"vat_amount": data[i].get("vat_amount"),
-				}
-			)
-		elif data[i].get("legend") == "Standard rated supplies in Ras Al Khaimah":
-			legend_link = f"""
-			<a href= "/app/query-report/UAE%20VAT%20Register?company={filters.get("company")}&from_date={filters.get("from_date")}&to_date={filters.get("to_date")}&doc_type=Sales%20Invoice&vat=Ras%20Al%20Khaimah&category=Standard">{data[i].get("legend")}</a>
-			"""
-			final_data.append(
-				{
-					"no": data[i].get("no"),
-					"legend": legend_link,
-					"amount": data[i].get("amount"),
-					"vat_amount": data[i].get("vat_amount"),
-				}
-			)
-		elif data[i].get("legend") == "Standard rated supplies in Fujairah":
-			legend_link = f"""
-			<a href= "/app/query-report/UAE%20VAT%20Register?company={filters.get("company")}&from_date={filters.get("from_date")}&to_date={filters.get("to_date")}&doc_type=Sales%20Invoice&vat=Fujairah&category=Standard">{data[i].get("legend")}</a>
-			"""
-			final_data.append(
-				{
-					"no": data[i].get("no"),
-					"legend": legend_link,
-					"amount": data[i].get("amount"),
-					"vat_amount": data[i].get("vat_amount"),
-				}
-			)
-		elif data[i].get("legend") == "Supplies subject to the reverse charge provision":
-			legend_link = f"""
-			<a href= "/app/query-report/UAE%20VAT%20Register?company={filters.get("company")}&from_date={filters.get("from_date")}&to_date={filters.get("to_date")}&doc_type=Purchase%20Invoice&reverse_charge=Y">{data[i].get("legend")}</a>
-			"""
-			final_data.append(
-				{
-					"no": data[i].get("no"),
-					"legend": legend_link,
-					"amount": data[i].get("amount"),
-					"vat_amount": data[i].get("vat_amount"),
-				}
-			)
-		elif data[i].get("legend") == "Zero Rated":
-			legend_link = f"""
-			<a href= "/app/query-report/UAE%20VAT%20Register?company={filters.get("company")}&from_date={filters.get("from_date")}&to_date={filters.get("to_date")}&doc_type=Sales%20Invoice&category=Zero%20Rated">{data[i].get("legend")}</a>
-			"""
-			final_data.append(
-				{
-					"no": data[i].get("no"),
-					"legend": legend_link,
-					"amount": data[i].get("amount"),
-					"vat_amount": data[i].get("vat_amount"),
-				}
-			)
-		elif data[i].get("legend") == "Exempt Supplies":
-			legend_link = f"""
-			<a href= "/app/query-report/UAE%20VAT%20Register?company={filters.get("company")}&from_date={filters.get("from_date")}&to_date={filters.get("to_date")}&doc_type=Sales%20Invoice&category=Exempt%20Rated">{data[i].get("legend")}</a>
-			"""
-			final_data.append(
-				{
-					"no": data[i].get("no"),
-					"legend": legend_link,
-					"amount": data[i].get("amount"),
-					"vat_amount": data[i].get("vat_amount"),
-				}
-			)
-		elif data[i].get("legend") == "Standard Rated Expenses":
-			legend_link = f"""
-			<a href= "/app/query-report/UAE%20VAT%20Register?company={filters.get("company")}&from_date={filters.get("from_date")}&to_date={filters.get("to_date")}&doc_type=Purchase%20Invoice">{data[i].get("legend")}</a>
-			"""
-			final_data.append(
-				{
-					"no": data[i].get("no"),
-					"legend": legend_link,
-					"amount": data[i].get("amount"),
-					"vat_amount": data[i].get("vat_amount"),
-				}
-			)
-		else:
-			final_data.append(
-				{
-					"no": data[i].get("no"),
-					"legend": data[i].get("legend"),
-					"amount": data[i].get("amount"),
-					"vat_amount": data[i].get("vat_amount"),
-				}
-			)
 
 	return final_data
 
 
+def _company_emirate_label(filters):
+	"""Return the home-emirate label for the company in ``filters`` if any.
+
+	The Dubai row is conventionally relabeled with the actual emirate of
+	the filtered company's primary address. Falls back to ``None`` when
+	no company filter is set or the address has no emirate, in which case
+	callers keep the original "Standard rated supplies in Dubai" wording.
+	"""
+	company = (filters or {}).get("company")
+	if not company:
+		return None
+	address = frappe.get_all(
+		"Address",
+		filters=[
+			["Dynamic Link", "link_doctype", "=", "Company"],
+			["Dynamic Link", "link_name", "=", company],
+			["Address", "is_your_company_address", "=", 1],
+		],
+		fields=["emirate"],
+		limit=1,
+	)
+	if address and address[0].get("emirate"):
+		return _("Standard rated supplies in {0}").format(address[0]["emirate"])
+	return None
+
+
 def append_vat_on_sales(data, filters):
 	"""Appends Sales and All Other Outputs."""
 	append_data(data, "", _("VAT on Sales and All Other Outputs"), "", "")
@@ -384,7 +296,7 @@ def net_vat_due(data, filters, amounts_by_emirate):
 	append_data(
 		data,
 		"13",
-		_("Total value of recoverable tax for the period "),
+		_("Total value of recoverable tax for the period"),
 		frappe.format(0.00, "Currency"),
 		frappe.format(
 			get_standard_rated_expenses_tax(filters) + get_reverse_charge_recoverable_tax(filters),
@@ -424,22 +336,24 @@ def format_currency_signed(value):
 @_cached
 def get_total_emiratewise(filters):
 	"""Returns Emiratewise Amount and Taxes."""
-	i = frappe.qb.DocType("Sales Invoice Item")
+	si = frappe.qb.DocType("Sales Invoice")
-	s = frappe.qb.DocType("Sales Invoice")
+	sii = frappe.qb.DocType("Sales Invoice Item")
 	query = (
-		frappe.qb.from_(i)
+		frappe.qb.from_(sii)
-		.inner_join(s)
+		.inner_join(si)
-		.on(i.parent == s.name)
+		.on(sii.parent == si.name)
-		.select(s.vat_emirate.as_("emirate"), Sum(i.base_net_amount).as_("total"), Sum(i.tax_amount))
+		.where(si.docstatus == 1)
-		.where((s.docstatus == 1) & (i.is_exempt != 1) & (i.is_zero_rated != 1))
+		.where(sii.is_exempt != 1)
-		.groupby(s.vat_emirate)
+		.where(sii.is_zero_rated != 1)
+		.groupby(si.vat_emirate)
+		.select(
+			si.vat_emirate.as_("emirate"),
+			Coalesce(Sum(sii.base_net_amount), 0).as_("total"),
+			Coalesce(Sum(sii.tax_amount), 0),
+		)
 	)
-	for condition in get_conditions(filters, s):
+	query = _apply_period_filters(query, si, filters)
-		query = query.where(condition)
+	return query.run()
-	try:
-		return query.run()
-	except (IndexError, TypeError):
-		return 0
 
 
 def get_emirates():
@@ -447,255 +361,185 @@ def get_emirates():
 	return ["Abu Dhabi", "Dubai", "Sharjah", "Ajman", "Umm Al Quwain", "Ras Al Khaimah", "Fujairah"]
 
 
-def get_filters(filters):
+def _apply_period_filters(query, table, filters):
-	"""The conditions to be used to filter data to calculate the total sale."""
+	"""Apply company / posting-date filters from ``filters`` to a frappe.qb query."""
-	query_filters = []
+	filters = filters or {}
 	if filters.get("company"):
-		query_filters.append(["company", "=", filters["company"]])
+		query = query.where(table.company == filters["company"])
 	if filters.get("from_date"):
-		query_filters.append(["posting_date", ">=", filters["from_date"]])
+		query = query.where(table.posting_date >= filters["from_date"])
 	if filters.get("to_date"):
-		query_filters.append(["posting_date", "<=", filters["to_date"]])
+		query = query.where(table.posting_date <= filters["to_date"])
-	return query_filters
+	return query
+
+
+def _sum_invoice_field(doctype, field, filters, extra_where=None):
+	"""Return ``sum(field)`` on a submitted invoice doctype with the standard
+	period filters. ``extra_where(table)`` may yield additional ``Criterion``s."""
+	table = frappe.qb.DocType(doctype)
+	query = frappe.qb.from_(table).where(table.docstatus == 1).select(Coalesce(Sum(table[field]), 0))
+	query = _apply_period_filters(query, table, filters)
+	if extra_where is not None:
+		for criterion in extra_where(table):
+			query = query.where(criterion)
+	result = query.run()
+	return flt(result[0][0]) if result else 0
+
+
+def _sum_item_field(parent_doctype, child_doctype, field, filters, extra_item_where=None):
+	"""Return ``sum(child.field)`` for child rows of submitted parents in the period."""
+	parent = frappe.qb.DocType(parent_doctype)
+	child = frappe.qb.DocType(child_doctype)
+	query = (
+		frappe.qb.from_(child)
+		.inner_join(parent)
+		.on(child.parent == parent.name)
+		.where(parent.docstatus == 1)
+		.select(Coalesce(Sum(child[field]), 0))
+	)
+	query = _apply_period_filters(query, parent, filters)
+	if extra_item_where is not None:
+		for criterion in extra_item_where(child):
+			query = query.where(criterion)
+	result = query.run()
+	return flt(result[0][0]) if result else 0
+
+
+def _sum_vat_account_debit(filters, recoverable=False):
+	"""Sum of GL debit for reverse-charge purchases booked to UAE VAT Accounts.
+
+	With ``recoverable=True``, multiplies the debit by the invoice's
+	``recoverable_reverse_charge`` percentage (and only sums rows with a
+	non-zero recoverable rate). Returns 0 when no company filter is set,
+	since UAE VAT Accounts are scoped per company.
+	"""
+	if not (filters or {}).get("company"):
+		return 0
+
+	pi = frappe.qb.DocType("Purchase Invoice")
+	gl = frappe.qb.DocType("GL Entry")
+	uva = frappe.qb.DocType("UAE VAT Account")
+
+	vat_accounts = frappe.qb.from_(uva).where(uva.parent == filters["company"]).select(uva.account)
+
+	amount = gl.debit
+	if recoverable:
+		amount = amount * pi.recoverable_reverse_charge / 100
+
+	query = (
+		frappe.qb.from_(pi)
+		.inner_join(gl)
+		.on(gl.voucher_no == pi.name)
+		.where(pi.reverse_charge == "Y")
+		.where(pi.docstatus == 1)
+		.where(gl.docstatus == 1)
+		.where(gl.account.isin(vat_accounts))
+		.select(Coalesce(Sum(amount), 0))
+	)
+	if recoverable:
+		query = query.where(pi.recoverable_reverse_charge > 0)
+	query = _apply_period_filters(query, pi, filters)
+	result = query.run()
+	return flt(result[0][0]) if result else 0
 
 
 @_cached
 def get_reverse_charge_total(filters):
 	"""Returns the sum of the total of each Purchase invoice made."""
-	query_filters = get_filters(filters)
+	return _sum_invoice_field(
-	query_filters.append(["reverse_charge", "=", "Y"])
+		"Purchase Invoice",
-	query_filters.append(["docstatus", "=", 1])
+		"base_net_total",
-	try:
+		filters,
-		return (
+		extra_where=lambda t: [t.reverse_charge == "Y"],
-			frappe.db.get_all(
+	)
-				"Purchase Invoice",
-				filters=query_filters,
-				fields=["sum(base_net_total)"],
-				as_list=True,
-				limit=1,
-			)[0][0]
-			or 0
-		)
-	except (IndexError, TypeError):
-		return 0
 
 
 @_cached
 def get_reverse_charge_tax(filters):
 	"""Returns the sum of the tax of each Purchase invoice made."""
-	p = frappe.qb.DocType("Purchase Invoice")
+	return _sum_vat_account_debit(filters)
-	gl = frappe.qb.DocType("GL Entry")
-	uae_vat = frappe.qb.DocType("UAE VAT Account")
-	query = (
-		frappe.qb.from_(p)
-		.inner_join(gl)
-		.on(gl.voucher_no == p.name)
-		.select(Sum(gl.debit))
-		.where(
-			(p.reverse_charge == "Y")
-			& (p.docstatus == 1)
-			& (gl.docstatus == 1)
-			& gl.account.isin(
-				frappe.qb.from_(uae_vat)
-				.select(uae_vat.account)
-				.where(uae_vat.parent == filters.get("company"))
-			)
-		)
-	)
-	for condition in get_conditions_join(filters, p):
-		query = query.where(condition)
-	return query.run()[0][0] or 0
 
 
 @_cached
 def get_reverse_charge_recoverable_total(filters):
 	"""Returns the sum of the total of each Purchase invoice made with recoverable reverse charge."""
-	query_filters = get_filters(filters)
+	return _sum_invoice_field(
-	query_filters.append(["reverse_charge", "=", "Y"])
+		"Purchase Invoice",
-	query_filters.append(["recoverable_reverse_charge", ">", "0"])
+		"base_net_total",
-	query_filters.append(["docstatus", "=", 1])
+		filters,
-	try:
+		extra_where=lambda t: [t.reverse_charge == "Y", t.recoverable_reverse_charge > 0],
-		return (
+	)
-			frappe.db.get_all(
-				"Purchase Invoice",
-				filters=query_filters,
-				fields=["sum(base_net_total)"],
-				as_list=True,
-				limit=1,
-			)[0][0]
-			or 0
-		)
-	except (IndexError, TypeError):
-		return 0
 
 
 @_cached
 def get_reverse_charge_recoverable_tax(filters):
 	"""Returns the sum of the tax of each Purchase invoice made."""
-	p = frappe.qb.DocType("Purchase Invoice")
+	return _sum_vat_account_debit(filters, recoverable=True)
-	gl = frappe.qb.DocType("GL Entry")
-	uae_vat = frappe.qb.DocType("UAE VAT Account")
-	query = (
-		frappe.qb.from_(p)
-		.inner_join(gl)
-		.on(gl.voucher_no == p.name)
-		.select(Sum(gl.debit * p.recoverable_reverse_charge / 100))
-		.where(
-			(p.reverse_charge == "Y")
-			& (p.docstatus == 1)
-			& (p.recoverable_reverse_charge > 0)
-			& (gl.docstatus == 1)
-			& gl.account.isin(
-				frappe.qb.from_(uae_vat)
-				.select(uae_vat.account)
-				.where(uae_vat.parent == filters.get("company"))
-			)
-		)
-	)
-	for condition in get_conditions_join(filters, p):
-		query = query.where(condition)
-	return query.run()[0][0] or 0
-
-
-def get_conditions_join(filters, p):
-	"""The conditions to be used to filter data to calculate the total vat."""
-	conditions = []
-	if filters.get("company"):
-		conditions.append(p.company == filters.get("company"))
-	if filters.get("from_date"):
-		conditions.append(p.posting_date >= filters.get("from_date"))
-	if filters.get("to_date"):
-		conditions.append(p.posting_date <= filters.get("to_date"))
-	return conditions
 
 
 @_cached
 def get_standard_rated_expenses_total(filters):
 	"""Returns the sum of the total of each Purchase invoice made with recoverable reverse charge."""
-	query_filters = get_filters(filters)
+	return _sum_invoice_field(
-	query_filters.append(["recoverable_standard_rated_expenses", ">", 0])
+		"Purchase Invoice",
-	query_filters.append(["docstatus", "=", 1])
+		"base_net_total",
-	try:
+		filters,
-		return (
+		extra_where=lambda t: [t.recoverable_standard_rated_expenses > 0],
-			frappe.db.get_all(
+	)
-				"Purchase Invoice",
-				filters=query_filters,
-				fields=["sum(base_net_total)"],
-				as_list=True,
-				limit=1,
-			)[0][0]
-			or 0
-		)
-	except (IndexError, TypeError):
-		return 0
 
 
 @_cached
 def get_standard_rated_expenses_tax(filters):
 	"""Returns the sum of the tax of each Purchase invoice made."""
-	query_filters = get_filters(filters)
+	return _sum_invoice_field(
-	query_filters.append(["recoverable_standard_rated_expenses", ">", 0])
+		"Purchase Invoice",
-	query_filters.append(["docstatus", "=", 1])
+		"recoverable_standard_rated_expenses",
-	try:
+		filters,
-		return (
+		extra_where=lambda t: [t.recoverable_standard_rated_expenses > 0],
-			frappe.db.get_all(
+	)
-				"Purchase Invoice",
-				filters=query_filters,
-				fields=["sum(recoverable_standard_rated_expenses)"],
-				as_list=True,
-				limit=1,
-			)[0][0]
-			or 0
-		)
-	except (IndexError, TypeError):
-		return 0
 
 
 @_cached
 def get_tourist_tax_return_total(filters):
 	"""Returns the sum of the total of each Sales invoice with non zero tourist_tax_return."""
-	query_filters = get_filters(filters)
+	return _sum_invoice_field(
-	query_filters.append(["tourist_tax_return", ">", 0])
+		"Sales Invoice",
-	query_filters.append(["docstatus", "=", 1])
+		"base_net_total",
-	try:
+		filters,
-		return (
+		extra_where=lambda t: [t.tourist_tax_return > 0],
-			frappe.db.get_all(
+	)
-				"Sales Invoice", filters=query_filters, fields=["sum(base_net_total)"], as_list=True, limit=1
-			)[0][0]
-			or 0
-		)
-	except (IndexError, TypeError):
-		return 0
 
 
 @_cached
 def get_tourist_tax_return_tax(filters):
 	"""Returns the sum of the tax of each Sales invoice with non zero tourist_tax_return."""
-	query_filters = get_filters(filters)
+	return _sum_invoice_field(
-	query_filters.append(["tourist_tax_return", ">", 0])
+		"Sales Invoice",
-	query_filters.append(["docstatus", "=", 1])
+		"tourist_tax_return",
-	try:
+		filters,
-		return (
+		extra_where=lambda t: [t.tourist_tax_return > 0],
-			frappe.db.get_all(
+	)
-				"Sales Invoice",
-				filters=query_filters,
-				fields=["sum(tourist_tax_return)"],
-				as_list=True,
-				limit=1,
-			)[0][0]
-			or 0
-		)
-	except (IndexError, TypeError):
-		return 0
 
 
 @_cached
 def get_zero_rated_total(filters):
 	"""Returns the sum of each Sales Invoice Item Amount which is zero rated."""
-	i = frappe.qb.DocType("Sales Invoice Item")
+	return _sum_item_field(
-	s = frappe.qb.DocType("Sales Invoice")
+		"Sales Invoice",
-	query = (
+		"Sales Invoice Item",
-		frappe.qb.from_(i)
+		"base_net_amount",
-		.inner_join(s)
+		filters,
-		.on(i.parent == s.name)
+		extra_item_where=lambda i: [i.is_zero_rated == 1],
-		.select(Sum(i.base_net_amount).as_("total"))
-		.where((s.docstatus == 1) & (i.is_zero_rated == 1))
 	)
-	for condition in get_conditions(filters, s):
-		query = query.where(condition)
-	try:
-		return query.run()[0][0] or 0
-	except (IndexError, TypeError):
-		return 0
 
 
 @_cached
 def get_exempt_total(filters):
 	"""Returns the sum of each Sales Invoice Item Amount which is Vat Exempt."""
-	i = frappe.qb.DocType("Sales Invoice Item")
+	return _sum_item_field(
-	s = frappe.qb.DocType("Sales Invoice")
+		"Sales Invoice",
-	query = (
+		"Sales Invoice Item",
-		frappe.qb.from_(i)
+		"base_net_amount",
-		.inner_join(s)
+		filters,
-		.on(i.parent == s.name)
+		extra_item_where=lambda i: [i.is_exempt == 1],
-		.select(Sum(i.base_net_amount).as_("total"))
-		.where((s.docstatus == 1) & (i.is_exempt == 1))
 	)
-	for condition in get_conditions(filters, s):
-		query = query.where(condition)
-	try:
-		return query.run()[0][0] or 0
-	except (IndexError, TypeError):
-		return 0
-
-
-def get_conditions(filters, s):
-	"""The conditions to be used to filter data to calculate the total sale."""
-	conditions = []
-	if filters.get("company"):
-		conditions.append(s.company == filters.get("company"))
-	if filters.get("from_date"):
-		conditions.append(s.posting_date >= filters.get("from_date"))
-	if filters.get("to_date"):
-		conditions.append(s.posting_date <= filters.get("to_date"))
-	return conditions

erpnext/regional/report/uae_vat_register/uae_vat_register.py

@@ -4,6 +4,7 @@
 
 import frappe
 from frappe import _
+from frappe.query_builder.functions import Coalesce, Sum
 
 
 def execute(filters=None):
@@ -88,129 +89,109 @@ def get_columns(filters):
 def get_data(filters):
 	doc_type = filters.get("doc_type") or "Sales Invoice"
 	if doc_type == "Sales Invoice":
-		return fetch_sales_rows(filters)
+		return _fetch_rows(filters, is_sales=True)
 	if doc_type == "Purchase Invoice":
-		return fetch_purchase_rows(filters)
+		return _fetch_rows(filters, is_sales=False)
 	return []
 
 
-def fetch_sales_rows(filters):
+def _fetch_rows(filters, is_sales):
-	conditions, params = build_conditions(filters)
+	"""Build the VAT register query for either Sales or Purchase Invoices.
-	category_clause = sales_category_clause(filters.get("category"))
 
-	emirate_clause = ""
+	Item-wise mode returns one row per Sales/Purchase Invoice Item; the
-	if filters.get("vat"):
+	default mode aggregates back to one row per invoice with summed qty,
-		emirate_clause = "AND s.vat_emirate = %(vat)s"
+	net, VAT, and total. ``COALESCE(i.tax_amount, 0)`` is used everywhere
-		params["vat"] = filters["vat"]
+	so a missing VAT amount surfaces as 0 instead of NULL — matching the
+	currency display and avoiding NULLs in client-side totals.
+	"""
+	parent_doctype = "Sales Invoice" if is_sales else "Purchase Invoice"
+	child_doctype = "Sales Invoice Item" if is_sales else "Purchase Invoice Item"
+	parent = frappe.qb.DocType(parent_doctype)
+	child = frappe.qb.DocType(child_doctype)
+	item_wise = bool(filters.get("item_wise"))
 
-	if filters.get("item_wise"):
+	party_field = parent.customer if is_sales else parent.supplier
-		return frappe.db.sql(
+	party_extra = parent.vat_emirate.as_("emirate") if is_sales else parent.reverse_charge
-			f"""
+
-			SELECT
+	tax_amount = Coalesce(child.tax_amount, 0)
-				s.name, s.posting_date, s.customer AS party,
+	gross = child.base_net_amount + tax_amount
-				COALESCE(i.cost_center, s.cost_center) AS cost_center,
+
-				s.vat_emirate AS emirate,
+	if item_wise:
-				i.item_code, i.qty, i.rate,
+		query = (
-				i.base_net_amount AS net_amount,
+			frappe.qb.from_(parent)
-				i.tax_amount AS vat_amount,
+			.inner_join(child)
-				(i.base_net_amount + COALESCE(i.tax_amount, 0)) AS total_amount
+			.on(child.parent == parent.name)
-			FROM `tabSales Invoice` s
+			.where(parent.docstatus == 1)
-			INNER JOIN `tabSales Invoice Item` i ON i.parent = s.name
+			.select(
-			WHERE s.docstatus = 1 {conditions} {category_clause} {emirate_clause}
+				parent.name,
-			ORDER BY s.posting_date, s.name, i.idx
+				parent.posting_date,
-			""",
+				party_field.as_("party"),
-			params,
+				Coalesce(child.cost_center, parent.cost_center).as_("cost_center"),
-			as_dict=True,
+				party_extra,
+				child.item_code,
+				child.qty,
+				child.rate,
+				child.base_net_amount.as_("net_amount"),
+				tax_amount.as_("vat_amount"),
+				gross.as_("total_amount"),
+			)
+			.orderby(parent.posting_date)
+			.orderby(parent.name)
+			.orderby(child.idx)
+		)
+	else:
+		cost_center = parent.cost_center
+		query = (
+			frappe.qb.from_(parent)
+			.inner_join(child)
+			.on(child.parent == parent.name)
+			.where(parent.docstatus == 1)
+			.select(
+				parent.name,
+				parent.posting_date,
+				party_field.as_("party"),
+				cost_center,
+				party_extra,
+				Sum(child.qty).as_("qty"),
+				Coalesce(Sum(child.base_net_amount), 0).as_("net_amount"),
+				Coalesce(Sum(tax_amount), 0).as_("vat_amount"),
+				Coalesce(Sum(gross), 0).as_("total_amount"),
+			)
+			.groupby(parent.name, parent.posting_date, party_field, cost_center, party_extra)
+			.orderby(parent.posting_date)
+			.orderby(parent.name)
 		)
 
-	return frappe.db.sql(
+	query = _apply_period_filters(query, parent, filters)
-		f"""
+
-		SELECT
+	if is_sales and filters.get("vat"):
-			s.name, s.posting_date, s.customer AS party, s.cost_center,
+		query = query.where(parent.vat_emirate == filters["vat"])
-			s.vat_emirate AS emirate,
+	if not is_sales and filters.get("reverse_charge") in ("Y", "N"):
-			SUM(i.qty) AS qty,
+		query = query.where(parent.reverse_charge == filters["reverse_charge"])
-			SUM(i.base_net_amount) AS net_amount,
+	if is_sales:
-			SUM(i.tax_amount) AS vat_amount,
+		category_criterion = _sales_category_criterion(child, filters.get("category"))
-			SUM(i.base_net_amount + COALESCE(i.tax_amount, 0)) AS total_amount
+		if category_criterion is not None:
-		FROM `tabSales Invoice` s
+			query = query.where(category_criterion)
-		INNER JOIN `tabSales Invoice Item` i ON i.parent = s.name
+
-		WHERE s.docstatus = 1 {conditions} {category_clause} {emirate_clause}
+	return query.run(as_dict=True)
-		GROUP BY s.name, s.posting_date, s.customer, s.cost_center, s.vat_emirate
-		ORDER BY s.posting_date, s.name
-		""",
-		params,
-		as_dict=True,
-	)
 
 
-def fetch_purchase_rows(filters):
+def _apply_period_filters(query, parent, filters):
-	conditions, params = build_conditions(filters)
-
-	rc_clause = ""
-	if filters.get("reverse_charge") in ("Y", "N"):
-		rc_clause = "AND s.reverse_charge = %(reverse_charge)s"
-		params["reverse_charge"] = filters["reverse_charge"]
-
-	if filters.get("item_wise"):
-		return frappe.db.sql(
-			f"""
-			SELECT
-				s.name, s.posting_date, s.supplier AS party,
-				COALESCE(i.cost_center, s.cost_center) AS cost_center,
-				s.reverse_charge,
-				i.item_code, i.qty, i.rate,
-				i.base_net_amount AS net_amount,
-				i.tax_amount AS vat_amount,
-				(i.base_net_amount + COALESCE(i.tax_amount, 0)) AS total_amount
-			FROM `tabPurchase Invoice` s
-			INNER JOIN `tabPurchase Invoice Item` i ON i.parent = s.name
-			WHERE s.docstatus = 1 {conditions} {rc_clause}
-			ORDER BY s.posting_date, s.name, i.idx
-			""",
-			params,
-			as_dict=True,
-		)
-
-	return frappe.db.sql(
-		f"""
-		SELECT
-			s.name, s.posting_date, s.supplier AS party, s.cost_center,
-			s.reverse_charge,
-			SUM(i.qty) AS qty,
-			SUM(i.base_net_amount) AS net_amount,
-			SUM(i.tax_amount) AS vat_amount,
-			SUM(i.base_net_amount + COALESCE(i.tax_amount, 0)) AS total_amount
-		FROM `tabPurchase Invoice` s
-		INNER JOIN `tabPurchase Invoice Item` i ON i.parent = s.name
-		WHERE s.docstatus = 1 {conditions} {rc_clause}
-		GROUP BY s.name, s.posting_date, s.supplier, s.cost_center, s.reverse_charge
-		ORDER BY s.posting_date, s.name
-		""",
-		params,
-		as_dict=True,
-	)
-
-
-def build_conditions(filters):
-	conditions = ""
-	params = {}
 	if filters.get("company"):
-		conditions += " AND s.company = %(company)s"
+		query = query.where(parent.company == filters["company"])
-		params["company"] = filters["company"]
 	if filters.get("from_date"):
-		conditions += " AND s.posting_date >= %(from_date)s"
+		query = query.where(parent.posting_date >= filters["from_date"])
-		params["from_date"] = filters["from_date"]
 	if filters.get("to_date"):
-		conditions += " AND s.posting_date <= %(to_date)s"
+		query = query.where(parent.posting_date <= filters["to_date"])
-		params["to_date"] = filters["to_date"]
+	return query
-	return conditions, params
 
 
-def sales_category_clause(category):
+def _sales_category_criterion(child, category):
+	"""Translate the ``category`` filter into a Sales Invoice Item criterion."""
 	if category == "Standard":
-		return "AND i.is_zero_rated != 1 AND i.is_exempt != 1"
+		return (child.is_zero_rated != 1) & (child.is_exempt != 1)
 	if category == "Zero Rated":
-		return "AND i.is_zero_rated = 1"
+		return child.is_zero_rated == 1
 	if category == "Exempt Rated":
-		return "AND i.is_exempt = 1"
+		return child.is_exempt == 1
-	return ""
+	return None