From 806f30fa87c0c8040aaa37d2c7d362e9aa7bcac3 Mon Sep 17 00:00:00 2001 From: Bibin <17405044+bibinqcs@users.noreply.github.com> Date: Sun, 14 Jun 2026 10:20:06 +0000 Subject: [PATCH] refactor: FTA Audit File and UAE VAT Reports --- .../v16_0/add_arabic_company_name_field.py | 19 +- .../fta_audit_file/fta_audit_file.json | 2 +- .../doctype/fta_audit_file/fta_audit_file.py | 145 ++++- .../report/uae_vat_201/uae_vat_201.py | 602 +++++++----------- .../uae_vat_register/uae_vat_register.py | 197 +++--- 5 files changed, 445 insertions(+), 520 deletions(-) diff --git a/erpnext/patches/v16_0/add_arabic_company_name_field.py b/erpnext/patches/v16_0/add_arabic_company_name_field.py index 57b8db59f97..100bf5502a4 100644 --- a/erpnext/patches/v16_0/add_arabic_company_name_field.py +++ b/erpnext/patches/v16_0/add_arabic_company_name_field.py @@ -1,10 +1,21 @@ import frappe - -from erpnext.regional.united_arab_emirates.setup import make_custom_fields +from frappe.custom.doctype.custom_field.custom_field import create_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, + ) diff --git a/erpnext/regional/doctype/fta_audit_file/fta_audit_file.json b/erpnext/regional/doctype/fta_audit_file/fta_audit_file.json index db64126ae59..061b50e219f 100644 --- a/erpnext/regional/doctype/fta_audit_file/fta_audit_file.json +++ b/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 }, diff --git a/erpnext/regional/doctype/fta_audit_file/fta_audit_file.py b/erpnext/regional/doctype/fta_audit_file/fta_audit_file.py index fd8d937735e..1356c76a744 100644 --- a/erpnext/regional/doctype/fta_audit_file/fta_audit_file.py +++ b/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 = { - t: _resolve_tax_code(t) - for t in {item.item_tax_template for items in items_by_invoice.values() for item in items} - if t - } + tax_code_bands = _bulk_tax_code_bands( + { + item.item_tax_template + 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 = { - t: _resolve_tax_code(t) - for t in {item.item_tax_template for items in items_by_invoice.values() for item in items} - if t - } + tax_code_bands = _bulk_tax_code_bands( + { + item.item_tax_template + 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; - otherwise defaults to ``SR`` (Standard Rated). Setting ``tax_category`` - on each Item Tax is the supported way to control this — there is no - heuristic fallback on the template name. + Picks the Item Tax row whose ``valid_from`` is the most recent value + that is still on or before ``posting_date``; rows with no + ``valid_from`` are treated as always-valid and used only as a + 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( - "Item Tax", - {"item_tax_template": item_tax_template}, - "tax_category", - ) - if tax_category and tax_category in _FTA_TAX_CODES: - return tax_category + bands = bands_map.get(item_tax_template) or [] + posting = getdate(posting_date) if posting_date else None + fallback_category = None + for valid_from, tax_category in bands: + if valid_from is None: + fallback_category = fallback_category or 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" diff --git a/erpnext/regional/report/uae_vat_201/uae_vat_201.py b/erpnext/regional/report/uae_vat_201/uae_vat_201.py index 6dc18501af7..ae50b88187c 100644 --- a/erpnext/regional/report/uae_vat_201/uae_vat_201.py +++ b/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 `` 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'{escape(str(text))}' + + 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)): - if data[i].get("legend") == "Standard rated supplies in Abu Dhabi": - legend_link = f""" - {data[i].get("legend")} - """ - 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) + for row in data: + legend = row.get("legend") + new_legend = legend - if address: - if address[0].get("emirate"): - name = "Standard rated supplies in" + " " + address[0].get("emirate") - else: - name = "Standard rated supplies in Dubai" - else: - name = "Standard rated supplies in Dubai" + if legend in emirate_drill_downs: + emirate = emirate_drill_downs[legend] + label = dubai_label_override if legend == dubai_legend and dubai_label_override else legend + new_legend = _drill_down_link( + label, filters, doc_type="Sales Invoice", vat=emirate, category="Standard" + ) + 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""" - {name} - """ - 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 Sharjah": - legend_link = f""" - {data[i].get("legend")} - """ - 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""" - {data[i].get("legend")} - """ - 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""" - {data[i].get("legend")} - """ - 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""" - {data[i].get("legend")} - """ - 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""" - {data[i].get("legend")} - """ - 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""" - {data[i].get("legend")} - """ - 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""" - {data[i].get("legend")} - """ - 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""" - {data[i].get("legend")} - """ - 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""" - {data[i].get("legend")} - """ - 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"), - } - ) + final_data.append( + { + "no": row.get("no"), + "legend": new_legend, + "amount": row.get("amount"), + "vat_amount": row.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") - s = frappe.qb.DocType("Sales Invoice") + si = frappe.qb.DocType("Sales Invoice") + sii = frappe.qb.DocType("Sales Invoice Item") query = ( - frappe.qb.from_(i) - .inner_join(s) - .on(i.parent == s.name) - .select(s.vat_emirate.as_("emirate"), Sum(i.base_net_amount).as_("total"), Sum(i.tax_amount)) - .where((s.docstatus == 1) & (i.is_exempt != 1) & (i.is_zero_rated != 1)) - .groupby(s.vat_emirate) + frappe.qb.from_(sii) + .inner_join(si) + .on(sii.parent == si.name) + .where(si.docstatus == 1) + .where(sii.is_exempt != 1) + .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 = query.where(condition) - try: - return query.run() - except (IndexError, TypeError): - return 0 + query = _apply_period_filters(query, si, filters) + return query.run() 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): - """The conditions to be used to filter data to calculate the total sale.""" - query_filters = [] +def _apply_period_filters(query, table, filters): + """Apply company / posting-date filters from ``filters`` to a frappe.qb query.""" + 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"]]) - return query_filters + query = query.where(table.posting_date <= filters["to_date"]) + 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) - query_filters.append(["reverse_charge", "=", "Y"]) - query_filters.append(["docstatus", "=", 1]) - try: - 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 + return _sum_invoice_field( + "Purchase Invoice", + "base_net_total", + filters, + extra_where=lambda t: [t.reverse_charge == "Y"], + ) @_cached def get_reverse_charge_tax(filters): """Returns the sum of the tax of each Purchase invoice made.""" - p = frappe.qb.DocType("Purchase Invoice") - 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 + return _sum_vat_account_debit(filters) @_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) - query_filters.append(["reverse_charge", "=", "Y"]) - query_filters.append(["recoverable_reverse_charge", ">", "0"]) - query_filters.append(["docstatus", "=", 1]) - try: - 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 + return _sum_invoice_field( + "Purchase Invoice", + "base_net_total", + filters, + extra_where=lambda t: [t.reverse_charge == "Y", t.recoverable_reverse_charge > 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") - 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 + return _sum_vat_account_debit(filters, recoverable=True) @_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) - query_filters.append(["recoverable_standard_rated_expenses", ">", 0]) - query_filters.append(["docstatus", "=", 1]) - try: - 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 + return _sum_invoice_field( + "Purchase Invoice", + "base_net_total", + filters, + extra_where=lambda t: [t.recoverable_standard_rated_expenses > 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) - query_filters.append(["recoverable_standard_rated_expenses", ">", 0]) - query_filters.append(["docstatus", "=", 1]) - try: - return ( - 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 + return _sum_invoice_field( + "Purchase Invoice", + "recoverable_standard_rated_expenses", + filters, + extra_where=lambda t: [t.recoverable_standard_rated_expenses > 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) - query_filters.append(["tourist_tax_return", ">", 0]) - query_filters.append(["docstatus", "=", 1]) - try: - return ( - 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 + return _sum_invoice_field( + "Sales Invoice", + "base_net_total", + filters, + extra_where=lambda t: [t.tourist_tax_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) - query_filters.append(["tourist_tax_return", ">", 0]) - query_filters.append(["docstatus", "=", 1]) - try: - return ( - 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 + return _sum_invoice_field( + "Sales Invoice", + "tourist_tax_return", + filters, + extra_where=lambda t: [t.tourist_tax_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") - s = frappe.qb.DocType("Sales Invoice") - query = ( - frappe.qb.from_(i) - .inner_join(s) - .on(i.parent == s.name) - .select(Sum(i.base_net_amount).as_("total")) - .where((s.docstatus == 1) & (i.is_zero_rated == 1)) + return _sum_item_field( + "Sales Invoice", + "Sales Invoice Item", + "base_net_amount", + filters, + extra_item_where=lambda i: [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") - s = frappe.qb.DocType("Sales Invoice") - query = ( - frappe.qb.from_(i) - .inner_join(s) - .on(i.parent == s.name) - .select(Sum(i.base_net_amount).as_("total")) - .where((s.docstatus == 1) & (i.is_exempt == 1)) + return _sum_item_field( + "Sales Invoice", + "Sales Invoice Item", + "base_net_amount", + filters, + extra_item_where=lambda i: [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 diff --git a/erpnext/regional/report/uae_vat_register/uae_vat_register.py b/erpnext/regional/report/uae_vat_register/uae_vat_register.py index 04427f7359a..534e914936c 100644 --- a/erpnext/regional/report/uae_vat_register/uae_vat_register.py +++ b/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): - conditions, params = build_conditions(filters) - category_clause = sales_category_clause(filters.get("category")) +def _fetch_rows(filters, is_sales): + """Build the VAT register query for either Sales or Purchase Invoices. - emirate_clause = "" - if filters.get("vat"): - emirate_clause = "AND s.vat_emirate = %(vat)s" - params["vat"] = filters["vat"] + Item-wise mode returns one row per Sales/Purchase Invoice Item; the + default mode aggregates back to one row per invoice with summed qty, + net, VAT, and total. ``COALESCE(i.tax_amount, 0)`` is used everywhere + 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"): - return frappe.db.sql( - f""" - SELECT - s.name, s.posting_date, s.customer AS party, - COALESCE(i.cost_center, s.cost_center) AS cost_center, - s.vat_emirate AS emirate, - 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 `tabSales Invoice` s - INNER JOIN `tabSales Invoice Item` i ON i.parent = s.name - WHERE s.docstatus = 1 {conditions} {category_clause} {emirate_clause} - ORDER BY s.posting_date, s.name, i.idx - """, - params, - as_dict=True, + party_field = parent.customer if is_sales else parent.supplier + party_extra = parent.vat_emirate.as_("emirate") if is_sales else parent.reverse_charge + + tax_amount = Coalesce(child.tax_amount, 0) + gross = child.base_net_amount + tax_amount + + if item_wise: + 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"), + Coalesce(child.cost_center, parent.cost_center).as_("cost_center"), + 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( - f""" - SELECT - s.name, s.posting_date, s.customer AS party, s.cost_center, - s.vat_emirate AS emirate, - 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 `tabSales Invoice` s - INNER JOIN `tabSales Invoice Item` i ON i.parent = s.name - WHERE s.docstatus = 1 {conditions} {category_clause} {emirate_clause} - 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, - ) + query = _apply_period_filters(query, parent, filters) + + if is_sales and filters.get("vat"): + query = query.where(parent.vat_emirate == filters["vat"]) + if not is_sales and filters.get("reverse_charge") in ("Y", "N"): + query = query.where(parent.reverse_charge == filters["reverse_charge"]) + if is_sales: + category_criterion = _sales_category_criterion(child, filters.get("category")) + if category_criterion is not None: + query = query.where(category_criterion) + + return query.run(as_dict=True) -def fetch_purchase_rows(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 = {} +def _apply_period_filters(query, parent, filters): if filters.get("company"): - conditions += " AND s.company = %(company)s" - params["company"] = filters["company"] + query = query.where(parent.company == filters["company"]) if filters.get("from_date"): - conditions += " AND s.posting_date >= %(from_date)s" - params["from_date"] = filters["from_date"] + query = query.where(parent.posting_date >= filters["from_date"]) if filters.get("to_date"): - conditions += " AND s.posting_date <= %(to_date)s" - params["to_date"] = filters["to_date"] - return conditions, params + query = query.where(parent.posting_date <= filters["to_date"]) + return query -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 "" + return child.is_exempt == 1 + return None