nsinnovations/erpnext
2
0
Fork 0
mirror of https://github.com/frappe/erpnext.git synced 2026-08-16 16:08:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #56261 from bibinqcs/qcs-uae-regional

Browse Source 
feat: Enhance UAE VAT Reports, UAE FTA Audit File and Add VAT Register
This commit is contained in:
Bibin
2026-06-22 10:32:38 +05:30
committed by GitHub
parent bae0263990 d0988dc32c
commit 7b84478c70
15 changed files with 2207 additions and 286 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
erpnext/patches.txt
View File

@@ -489,3 +489,4 @@ erpnext.patches.v16_0.submit_existing_product_bundles #1
erpnext.patches.v16_0.migrate_subscription_generate_invoice_at
erpnext.patches.v16_0.rename_subscription_billing_period_fields
erpnext.patches.v16_0.drop_redundant_serial_no_index_from_sabb
erpnext.patches.v16_0.add_arabic_company_name_field

21
erpnext/patches/v16_0/add_arabic_company_name_field.py Normal file
View File

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

0
erpnext/regional/doctype/fta_audit_file/__init__.py Normal file
View File

126
erpnext/regional/doctype/fta_audit_file/fta_audit_file.js Normal file
View File

@@ -0,0 +1,126 @@
// Copyright (c) 2025, Frappe Technologies Pvt. Ltd. and contributors
// For license information, please see license.txt
frappe.ui.form.on("FTA Audit File", {
refresh: function (frm) {
// Generate FAF — available from Draft (first generation) and Error
// (retry after a previous attempt failed). Queued/Generating are
// blocked by the server guard; Generated/Submitted are intentionally
// not re-generable.
if (!frm.is_new() && ["Draft", "Error"].includes(frm.doc.status)) {
frm.add_custom_button(
__(frm.doc.status === "Error" ? "Retry FAF Generation" : "Generate FAF"),
function () {
frm.trigger("generate_faf");
},
__("Actions")
);
}
// Add Mark as Submitted button for Generated status
if (frm.doc.status === "Generated") {
frm.add_custom_button(
__("Mark as Submitted"),
function () {
frm.trigger("mark_submitted");
},
__("Actions")
);
}
// Add Download button if file exists
if (frm.doc.faf_file) {
frm.add_custom_button(
__("Download FAF"),
function () {
window.open(frm.doc.faf_file);
},
__("Actions")
);
}
// Show status indicator
frm.trigger("set_status_indicator");
},
generate_faf: function (frm) {
frappe.confirm(
__("This will generate the FTA Audit File for the selected period. Continue?"),
function () {
frm.call({
doc: frm.doc,
method: "generate_faf",
freeze: true,
freeze_message: __("Queuing FTA Audit File generation..."),
}).then((r) => {
if (!r.message) return;
if (r.message.success) {
frappe.show_alert({
message: r.message.message,
indicator: "green",
});
} else {
frappe.msgprint({
title: __("Generation Failed"),
message: r.message.message,
indicator: "red",
});
}
frm.reload_doc();
});
}
);
},
mark_submitted: function (frm) {
frappe.confirm(
__("Mark this FAF as submitted to FTA? This action is for record-keeping only."),
function () {
frm.call({
doc: frm.doc,
method: "mark_as_submitted",
}).then((r) => {
if (r.message && r.message.success) {
frappe.show_alert({
message: r.message.message,
indicator: "green",
});
frm.reload_doc();
}
});
}
);
},
set_status_indicator: function (frm) {
const status_colors = {
Draft: "orange",
Queued: "yellow",
Generating: "blue",
Generated: "green",
Submitted: "blue",
Error: "red",
};
if (frm.doc.status) {
frm.page.set_indicator(__(frm.doc.status), status_colors[frm.doc.status] || "gray");
}
},
from_date: function (frm) {
frm.trigger("validate_dates");
},
to_date: function (frm) {
frm.trigger("validate_dates");
},
validate_dates: function (frm) {
if (frm.doc.from_date && frm.doc.to_date) {
if (frm.doc.from_date > frm.doc.to_date) {
frappe.msgprint(__("From Date cannot be after To Date"));
frm.set_value("to_date", null);
}
}
},
});

221
erpnext/regional/doctype/fta_audit_file/fta_audit_file.json Normal file
View File

@@ -0,0 +1,221 @@
{
"actions": [],
"allow_rename": 0,
"autoname": "naming_series:",
"beta": 1,
"creation": "2025-12-11 00:29:21.891860",
"doctype": "DocType",
"editable_grid": 1,
"engine": "InnoDB",
"field_order": [
"naming_series",
"company",
"column_break_1",
"from_date",
"to_date",
"section_break_agent",
"tax_agency_name",
"tan",
"column_break_agent",
"tax_agent_name",
"taan",
"section_break_options",
"file_type",
"include_opening_balance",
"column_break_options_2",
"status",
"section_break_output",
"faf_file",
"section_break_logs",
"generation_log",
"error_message"
],
"fields": [
{
"fieldname": "naming_series",
"fieldtype": "Select",
"label": "Series",
"options": "FAF-.YYYY.-",
"reqd": 1,
"hidden": 1,
"default": "FAF-.YYYY.-"
},
{
"fieldname": "company",
"fieldtype": "Link",
"in_list_view": 1,
"in_standard_filter": 1,
"label": "Company",
"options": "Company",
"reqd": 1
},
{
"fieldname": "column_break_1",
"fieldtype": "Column Break"
},
{
"fieldname": "from_date",
"fieldtype": "Date",
"in_list_view": 1,
"label": "From Date",
"reqd": 1
},
{
"fieldname": "to_date",
"fieldtype": "Date",
"in_list_view": 1,
"label": "To Date",
"reqd": 1
},
{
"fieldname": "section_break_agent",
"fieldtype": "Section Break",
"label": "Tax Agent (optional)",
"description": "Fill in only if the FAF is being filed through a registered Tax Agency or Tax Agent.",
"collapsible": 1,
"collapsible_depends_on": "eval:!(doc.tax_agency_name || doc.tan || doc.tax_agent_name || doc.taan)"
},
{
"fieldname": "tax_agency_name",
"fieldtype": "Data",
"label": "Tax Agency Name",
"length": 100
},
{
"fieldname": "tan",
"fieldtype": "Data",
"label": "TAN (Tax Agency Number)",
"length": 20
},
{
"fieldname": "column_break_agent",
"fieldtype": "Column Break"
},
{
"fieldname": "tax_agent_name",
"fieldtype": "Data",
"label": "Tax Agent Name",
"length": 100
},
{
"fieldname": "taan",
"fieldtype": "Data",
"label": "TAAN (Tax Agent Approval Number)",
"length": 20
},
{
"fieldname": "section_break_options",
"fieldtype": "Section Break",
"label": "Options"
},
{
"fieldname": "file_type",
"fieldtype": "Select",
"label": "File Type",
"options": "VAT",
"default": "VAT",
"reqd": 1
},
{
"fieldname": "include_opening_balance",
"fieldtype": "Check",
"label": "Include Opening Balance in GL",
"description": "Carry each account's pre-period balance forward into the General Ledger Balance column. Off by default (period-only running balance), matching the lighter interpretation used by Microsoft Dynamics 365 Finance.",
"default": "0"
},
{
"fieldname": "column_break_options_2",
"fieldtype": "Column Break"
},
{
"fieldname": "status",
"fieldtype": "Select",
"label": "Status",
"options": "Draft\nQueued\nGenerating\nGenerated\nSubmitted\nError",
"default": "Draft",
"in_list_view": 1,
"read_only": 1
},
{
"fieldname": "section_break_output",
"fieldtype": "Section Break",
"label": "Generated File",
"depends_on": "eval:doc.status === 'Generated' || doc.status === 'Submitted'"
},
{
"fieldname": "faf_file",
"fieldtype": "Attach",
"label": "FAF File",
"read_only": 1
},
{
"fieldname": "section_break_logs",
"fieldtype": "Section Break",
"label": "Generation Logs",
"collapsible": 1
},
{
"fieldname": "generation_log",
"fieldtype": "Long Text",
"label": "Generation Log",
"read_only": 1
},
{
"fieldname": "error_message",
"fieldtype": "Long Text",
"label": "Error Message",
"read_only": 1,
"depends_on": "eval:doc.status === 'Error'"
}
],
"index_web_pages_for_search": 1,
"links": [],
"modified": "2025-12-11 00:29:21.891860",
"modified_by": "Administrator",
"module": "Regional",
"name": "FTA Audit File",
"naming_rule": "By \"Naming Series\" field",
"owner": "Administrator",
"permissions": [
{
"create": 1,
"delete": 1,
"email": 1,
"export": 1,
"print": 1,
"read": 1,
"report": 1,
"role": "System Manager",
"share": 1,
"write": 1
},
{
"create": 1,
"delete": 1,
"email": 1,
"export": 1,
"print": 1,
"read": 1,
"report": 1,
"role": "Accounts Manager",
"share": 1,
"write": 1
},
{
"create": 1,
"email": 1,
"export": 1,
"print": 1,
"read": 1,
"report": 1,
"role": "Accounts User",
"share": 1,
"write": 1
}
],
"sort_field": "modified",
"sort_order": "DESC",
"states": [],
"title_field": "company",
"track_changes": 1
}

807
erpnext/regional/doctype/fta_audit_file/fta_audit_file.py Normal file
View File

@@ -0,0 +1,807 @@
# Copyright (c) 2025, Frappe Technologies Pvt. Ltd. and contributors
# For license information, please see license.txt
"""
FTA Audit File DocType Controller
Owns generation of FTA Audit Files (FAF) for UAE VAT compliance per the
FTA "Requirements Document for Tax Accounting Software" (October 2017),
Appendix 5.
A conformant VAT FAF contains four CSV tables in this order, each
delimited by an explicit start/end marker row:
1. Company Information (CompInfoStart .. CompInfoEnd)
2. Purchase Listing (PurcDataStart .. PurcDataEnd)
3. Supply Listing (SuppDataStart .. SuppDataEnd)
4. General Ledger (GLDataStart .. GLDataEnd)
The footer of each transactional table carries running totals plus a
transaction count. Primary amount columns are in the company's accounting
currency (typically AED for a UAE-registered entity); foreign-currency
mirrors are emitted alongside when the source invoice is in a different
currency.
"""
import csv
import io
import frappe
from frappe import _
from frappe.model.document import Document
from frappe.utils import flt, getdate, today
from frappe.utils.file_manager import save_file
FAF_VERSION = "FAFv1.0.0"
DEFAULT_COUNTRY = "United Arab Emirates"
DEFAULT_DATE = "31-12-9999"
PRODUCT_VERSION = "ERPNext"
# Number of GL Entry rows read into memory per batch when streaming the
# General Ledger section. Large UAE companies routinely produce hundreds
# of thousands of GL entries per year; reading them all into a single list
# would push the background worker over its memory limit.
GL_PAGE_SIZE = 5000
# 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):
if getdate(self.from_date) > getdate(self.to_date):
frappe.throw(_("From Date cannot be after To Date"))
if not frappe.db.get_value("Company", self.company, "tax_id"):
frappe.throw(
_("Company {0} does not have a Tax ID (TRN). Please set the Tax ID in Company.").format(
self.company
)
)
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.
Returns immediately with status ``Queued``. The actual generation
runs in ``_run_generation`` on the ``long`` queue (Frappe's
``enqueue_doc`` re-fetches a fresh doc inside the worker) and
updates ``status``, ``faf_file``, ``generation_log``, and
``error_message`` when complete.
Under ``frappe.flags.in_test`` the job runs synchronously so tests
can assert on the post-generation state without polling.
"""
# `@frappe.whitelist()` gates network access but not document write
# permission; without this explicit check, a user with read-only
# access could still trigger generation via REST.
self.check_permission("write")
# 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 in ("Generated", "Submitted"):
# UI hides the Generate/Retry button for Generated and Submitted;
# enforce the same lifecycle on the REST endpoint so a direct
# call cannot silently overwrite the attached CSV.
frappe.throw(_("FAF is already {0}; create a new document to regenerate.").format(current_status))
self.status = "Queued"
self.generation_log = ""
self.error_message = None
self.save()
frappe.enqueue_doc(
self.doctype,
self.name,
"_run_generation",
queue="long",
timeout=1500,
enqueue_after_commit=True,
now=bool(frappe.flags.in_test),
)
return {
"success": True,
"message": _("FAF generation has been queued. The status will update when complete."),
"docname": self.name,
"status": self.status,
}
@frappe.whitelist()
def mark_as_submitted(self):
"""Mark the FAF as submitted to the FTA portal (manual record)."""
self.check_permission("write")
if self.status != "Generated":
frappe.throw(_("Only Generated files can be marked as Submitted"))
self.status = "Submitted"
self.save()
return {"success": True, "message": _("FAF marked as submitted")}
def _run_generation(self):
"""Background entry point. Invoked via ``frappe.enqueue_doc`` from
``generate_faf`` (or synchronously under ``frappe.flags.in_test``).
Broad ``except`` is intentional: a long-running batch op records
the failure on the doc itself (status/error_message/log) so the
user sees what went wrong without the request 500'ing. The
exception is re-raised so the queue marks the job as failed and
the traceback is written to the Error Log.
"""
try:
self.status = "Generating"
self.save()
result = self._build_faf()
self.faf_file = result["file_url"]
self.generation_log = result["log"]
self.status = "Generated"
self.save()
except Exception as e:
try:
err_doc = frappe.get_doc(self.doctype, self.name)
err_doc.status = "Error"
err_doc.error_message = str(e)
err_doc.generation_log = (err_doc.generation_log or "") + f"\n\nError: {e}"
err_doc.save()
except Exception:
# Don't lose the original failure if persisting the Error
# state itself fails (e.g. row lock, validation regression);
# log the secondary failure with context, then re-raise the
# original ``e`` below so the job is still marked failed.
frappe.log_error(
title=_("FAF Error-state persistence failed"),
message=f"{self.doctype} {self.name}\n\n{frappe.get_traceback()}",
)
frappe.log_error(
title=_("FAF Generation Error"),
message=frappe.get_traceback(),
)
raise
def _build_faf(self):
"""Build the FAF CSV per Appendix 5 and attach it to this document."""
log_entries = []
def log(msg):
log_entries.append(msg)
log(f"Starting FAF generation for {self.company}")
log(f"Period: {self.from_date} to {self.to_date}")
log(f"File Type: {self.file_type}")
output = io.StringIO()
writer = csv.writer(output)
self._write_company_info(writer)
log("Company Information written")
purchase_count = self._write_purchase_listing(writer)
log(f"Purchase Listing written: {purchase_count} line items")
supply_count = self._write_supply_listing(writer)
log(f"Supply Listing written: {supply_count} line items")
gl_count = self._write_gl_listing(writer)
log(f"General Ledger written: {gl_count} entries")
csv_content = output.getvalue()
output.close()
file_name = f"FAF_{self.company}_{self.from_date}_to_{self.to_date}.csv".replace(" ", "_")
file_doc = save_file(
fname=file_name,
content=csv_content.encode("utf-8"),
dt="FTA Audit File",
dn=self.name,
is_private=1,
)
log(f"FAF file generated: {file_name}")
log("Generation completed successfully")
return {"file_url": file_doc.file_url, "log": "\n".join(log_entries)}
def _write_company_info(self, writer):
"""Emit ``CompInfoStart`` + body row + ``CompInfoEnd`` per Appendix 5."""
writer.writerow(["CompInfoStart"])
info = (
frappe.db.get_value(
"Company",
self.company,
["company_name", "company_name_in_arabic", "tax_id"],
as_dict=True,
)
or {}
)
writer.writerow(
[
_clean(info.get("company_name") or self.company),
_clean(info.get("company_name_in_arabic") or ""),
info.get("tax_id") or "",
_clean(self.tax_agency_name or ""),
_clean(self.tan or ""),
_clean(self.tax_agent_name or ""),
_clean(self.taan or ""),
_format_date(self.from_date),
_format_date(self.to_date),
_format_date(today()),
PRODUCT_VERSION,
FAF_VERSION,
]
)
writer.writerow(["CompInfoEnd"])
def _write_purchase_listing(self, writer):
"""Emit Purchase Listing per Appendix 5 with end-of-table totals row."""
writer.writerow(["PurcDataStart"])
invoices = frappe.get_all(
"Purchase Invoice",
filters={
"company": self.company,
"posting_date": ["between", [self.from_date, self.to_date]],
"docstatus": 1,
},
fields=[
"name",
"supplier",
"supplier_name",
"posting_date",
"permit_no",
"currency",
"conversion_rate",
],
order_by="posting_date asc, name asc",
)
if not invoices:
writer.writerow(["PurcDataEnd", _money(0), _money(0), 0])
return 0
invoice_names = [inv.name for inv in invoices]
supplier_names = list({inv.supplier for inv in invoices if inv.supplier})
supplier_trn_map = _bulk_party_field("Supplier", supplier_names, "tax_id")
items_by_invoice = _bulk_invoice_items(
"Purchase Invoice Item",
invoice_names,
[
"parent",
"idx",
"item_name",
"description",
"base_net_amount",
"net_amount",
"tax_amount",
"item_tax_template",
],
)
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)
total_purchase_company = 0.0
total_vat_company = 0.0
line_count = 0
for inv in invoices:
supplier_trn = supplier_trn_map.get(inv.supplier, "")
fcy_code, fcy_factor = _fcy_for_invoice(inv.currency, inv.conversion_rate, company_currency)
for item in items_by_invoice.get(inv.name, []):
# base_net_amount is company-currency; tax_amount is a UAE
# custom field with options="currency" and therefore stored
# in the document's invoice currency. Multiply by the
# conversion rate to land in company currency.
net_company = flt(item.base_net_amount, 2)
vat_invoice = flt(item.tax_amount or 0, 2)
vat_company = flt(vat_invoice * fcy_factor, 2)
net_fcy = flt(item.net_amount or 0, 2) if fcy_code != "XXX" else 0.0
vat_fcy = vat_invoice if fcy_code != "XXX" else 0.0
writer.writerow(
[
_clean(inv.supplier_name),
supplier_trn,
_format_date(inv.posting_date),
inv.name,
inv.permit_no or "",
item.idx,
_clean(item.description or item.item_name or ""),
_money(net_company),
_money(vat_company),
_resolve_tax_code(item.item_tax_template, inv.posting_date, tax_code_bands),
fcy_code,
_money(net_fcy),
_money(vat_fcy),
]
)
total_purchase_company += net_company
total_vat_company += vat_company
line_count += 1
writer.writerow(
[
"PurcDataEnd",
_money(total_purchase_company),
_money(total_vat_company),
line_count,
]
)
return line_count
def _write_supply_listing(self, writer):
"""Emit Supply Listing per Appendix 5 with end-of-table totals row."""
writer.writerow(["SuppDataStart"])
invoices = frappe.get_all(
"Sales Invoice",
filters={
"company": self.company,
"posting_date": ["between", [self.from_date, self.to_date]],
"docstatus": 1,
},
fields=[
"name",
"customer",
"customer_name",
"posting_date",
"currency",
"conversion_rate",
],
order_by="posting_date asc, name asc",
)
if not invoices:
writer.writerow(["SuppDataEnd", _money(0), _money(0), 0])
return 0
invoice_names = [inv.name for inv in invoices]
customer_names = list({inv.customer for inv in invoices if inv.customer})
customer_trn_map = _bulk_party_field("Customer", customer_names, "tax_id")
customer_country_map = _bulk_party_country("Customer", customer_names)
items_by_invoice = _bulk_invoice_items(
"Sales Invoice Item",
invoice_names,
[
"parent",
"idx",
"item_name",
"description",
"base_net_amount",
"net_amount",
"tax_amount",
"item_tax_template",
"is_zero_rated",
"is_exempt",
],
)
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)
total_supply_company = 0.0
total_vat_company = 0.0
line_count = 0
for inv in invoices:
customer_trn = customer_trn_map.get(inv.customer, "")
customer_country = customer_country_map.get(inv.customer) or DEFAULT_COUNTRY
fcy_code, fcy_factor = _fcy_for_invoice(inv.currency, inv.conversion_rate, company_currency)
for item in items_by_invoice.get(inv.name, []):
# See _write_purchase_listing for the currency convention:
# tax_amount is invoice-currency, base_net_amount is
# company-currency, and fcy_factor converts invoice → company.
net_company = flt(item.base_net_amount, 2)
vat_invoice = flt(item.tax_amount or 0, 2)
vat_company = flt(vat_invoice * fcy_factor, 2)
net_fcy = flt(item.net_amount or 0, 2) if fcy_code != "XXX" else 0.0
vat_fcy = vat_invoice if fcy_code != "XXX" else 0.0
if item.is_zero_rated:
tax_code = "ZR"
elif item.is_exempt:
tax_code = "EX"
else:
tax_code = _resolve_tax_code(item.item_tax_template, inv.posting_date, tax_code_bands)
writer.writerow(
[
_clean(inv.customer_name),
customer_trn,
_format_date(inv.posting_date),
inv.name,
item.idx,
_clean(item.description or item.item_name or ""),
_money(net_company),
_money(vat_company),
tax_code,
_clean(customer_country),
fcy_code,
_money(net_fcy),
_money(vat_fcy),
]
)
total_supply_company += net_company
total_vat_company += vat_company
line_count += 1
writer.writerow(
[
"SuppDataEnd",
_money(total_supply_company),
_money(total_vat_company),
line_count,
]
)
return line_count
def _write_gl_listing(self, writer):
"""Emit General Ledger per Appendix 5 with end-of-table totals row.
GL Entry rows are streamed in pages of ``GL_PAGE_SIZE`` to keep
memory bounded for multi-year exports on large companies; the
running-balance, account-name, and totals state survives across
pages so the output is identical to a single-fetch implementation.
"""
writer.writerow(["GLDataStart"])
company_currency = _company_currency(self.company)
base_filters = {
"company": self.company,
"posting_date": ["between", [self.from_date, self.to_date]],
"is_cancelled": 0,
}
# Opening balances need every account that posts in the period up
# front; without that flag we cache account names lazily as we
# encounter them in each batch.
if self.include_opening_balance:
accounts_in_period = frappe.get_all(
"GL Entry", filters=base_filters, pluck="account", distinct=True
)
running_balance = _opening_balances_by_account(self.company, self.from_date, accounts_in_period)
account_name_map = _bulk_party_field("Account", accounts_in_period, "account_name")
else:
running_balance = {}
account_name_map = {}
source_type_map = {
"Sales Invoice": "AR",
"Purchase Invoice": "AP",
"Journal Entry": "General Journal",
"Payment Entry": "Cash Receipt",
"Stock Entry": "Inventory",
"Delivery Note": "Inventory Sale",
"Purchase Receipt": "Purchases",
}
total_debit = 0.0
total_credit = 0.0
count = 0
start = 0
while True:
batch = frappe.get_all(
"GL Entry",
filters=base_filters,
fields=[
"name",
"posting_date",
"account",
"remarks",
"against",
"voucher_no",
"voucher_type",
"debit",
"credit",
],
order_by="posting_date asc, creation asc",
limit_start=start,
limit_page_length=GL_PAGE_SIZE,
)
if not batch:
break
# Backfill the account-name cache for accounts new to this batch.
new_accounts = [e.account for e in batch if e.account and e.account not in account_name_map]
if new_accounts:
account_name_map.update(_bulk_party_field("Account", new_accounts, "account_name"))
for entry in batch:
account_name = account_name_map.get(entry.account) or entry.account
source_type = source_type_map.get(entry.voucher_type, entry.voucher_type or "")
debit = flt(entry.debit, 2)
credit = flt(entry.credit, 2)
running_balance[entry.account] = running_balance.get(entry.account, 0.0) + debit - credit
balance = flt(running_balance[entry.account], 2)
writer.writerow(
[
_format_date(entry.posting_date),
entry.account,
_clean(account_name),
_clean(entry.remarks or ""),
_clean(entry.against or ""),
entry.voucher_no,
entry.voucher_no,
source_type,
_money(debit),
_money(credit),
_money(balance),
]
)
total_debit += debit
total_credit += credit
count += 1
if len(batch) < GL_PAGE_SIZE:
break
start += GL_PAGE_SIZE
writer.writerow(
[
"GLDataEnd",
_money(total_debit),
_money(total_credit),
count,
company_currency,
]
)
return count
def _clean(value):
"""Sanitize a string for FAF CSV.
The spec mandates that the delimiter (``,``) must not appear inside any
field. We follow Microsoft Dynamics 365's UAE FAF convention and
substitute ``;`` so the original separator stays visible in the data.
Embedded newlines are stripped because they would otherwise break the
CSV row structure.
"""
if value is None:
return ""
return str(value).replace(",", ";").replace("\n", " ").replace("\r", " ").strip()
def _format_date(d):
"""Format a date as DD-MM-YYYY per FTA spec; missing values become 31-12-9999."""
if not d:
return DEFAULT_DATE
return getdate(d).strftime("%d-%m-%Y")
def _money(value):
"""Format a numeric field as ``Decimal[14,2]`` per FTA spec.
Python's ``csv.writer`` calls ``str()`` on numeric values, which
strips trailing zeros (``0.00`` → ``"0.0"``). The spec mandates two
decimal places everywhere a Decimal[14,2] field is emitted, so we
pre-format to a string here.
"""
return f"{flt(value):.2f}"
def _company_currency(company):
return frappe.db.get_value("Company", company, "default_currency") or "AED"
def _fcy_for_invoice(invoice_currency, conversion_rate, company_currency):
"""Resolve foreign-currency code + conversion factor for an invoice.
Returns ``("XXX", 1.0)`` when the invoice is in the company's home
currency (no FCY columns to populate); otherwise the ISO 4217 code
plus the conversion factor (rate to company currency).
"""
if not invoice_currency or invoice_currency == company_currency:
return ("XXX", 1.0)
return (invoice_currency, flt(conversion_rate) or 1.0)
def _bulk_party_field(doctype, names, field):
"""Return ``{name: field_value}`` for the given names. Empty input → empty dict."""
if not names:
return {}
rows = frappe.get_all(
doctype,
filters={"name": ["in", names]},
fields=["name", field],
)
return {r["name"]: (r.get(field) or "") for r in rows}
def _bulk_party_country(party_doctype, party_names):
"""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 {}
dl = frappe.qb.DocType("Dynamic Link")
addr = frappe.qb.DocType("Address")
rows = (
frappe.qb.from_(dl)
.inner_join(addr)
.on(addr.name == dl.parent)
.where(dl.link_doctype == party_doctype)
.where(dl.parenttype == "Address")
.where(dl.link_name.isin(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 = {}
for r in rows:
out.setdefault(r["link_name"], r["country"])
return out
def _opening_balances_by_account(company, period_start, accounts):
"""Net pre-period balance per account: sum(debit) - sum(credit) before ``period_start``.
Returns ``{account: net}`` where net is debit-positive (positive for
asset/expense accounts that carry a debit balance, negative for
liability/equity/revenue accounts that carry a credit balance).
Single aggregated SQL via ``frappe.qb`` — one query regardless of the
number of accounts. Cancelled GL entries are excluded.
"""
if not accounts:
return {}
from frappe.query_builder.functions import Sum
gle = frappe.qb.DocType("GL Entry")
rows = (
frappe.qb.from_(gle)
.where(gle.company == company)
.where(gle.posting_date < period_start)
.where(gle.is_cancelled == 0)
.where(gle.account.isin(accounts))
.groupby(gle.account)
.select(
gle.account,
Sum(gle.debit).as_("debit"),
Sum(gle.credit).as_("credit"),
)
.run(as_dict=True)
)
return {r["account"]: flt(r.get("debit") or 0) - flt(r.get("credit") or 0) for r in rows}
def _bulk_invoice_items(child_doctype, invoice_names, fields):
"""Return ``{parent_invoice: [items...]}`` for the given invoice names."""
if not invoice_names:
return {}
items = frappe.get_all(
child_doctype,
filters={"parent": ["in", invoice_names]},
fields=fields,
order_by="parent asc, idx asc",
)
out = {}
for item in items:
out.setdefault(item["parent"], []).append(item)
return out
_FTA_TAX_CODES = ("SR", "ZR", "EX", "RC", "IG", "OA", "IA")
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.
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"
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"

238
erpnext/regional/doctype/fta_audit_file/test_fta_audit_file.py Normal file
View File

@@ -0,0 +1,238 @@
# Copyright (c) 2025, Frappe Technologies Pvt. Ltd. and contributors
# For license information, please see license.txt
import frappe
from frappe.tests.utils import FrappeTestCase
class TestFTAAuditFile(FrappeTestCase):
def setUp(self):
"""Create a UAE test company with TRN before each test.
Per-test creation (not setUpClass) because FrappeTestCase rolls
back the database after each test, including class-level fixtures.
"""
self.company = self._get_or_create_test_company()
def _get_or_create_test_company(self):
company_name = "_Test Company UAE"
if not frappe.db.exists("Company", company_name):
frappe.get_doc(
{
"doctype": "Company",
"company_name": company_name,
"abbr": "_TCU",
"country": "United Arab Emirates",
"default_currency": "AED",
"tax_id": "100123456789012",
}
).insert(ignore_permissions=True)
else:
company = frappe.get_doc("Company", company_name)
if not company.tax_id:
company.tax_id = "100123456789012"
company.save(ignore_permissions=True)
return company_name
def test_fta_audit_file_creation(self):
"""Test that FTA Audit File can be created."""
doc = frappe.get_doc(
{
"doctype": "FTA Audit File",
"company": self.company,
"from_date": "2024-01-01",
"to_date": "2024-03-31",
"file_type": "VAT",
}
)
doc.insert()
self.assertTrue(doc.name)
self.assertEqual(doc.status, "Draft")
self.assertEqual(doc.file_type, "VAT")
def test_date_validation(self):
"""Test that from_date cannot be after to_date."""
doc = frappe.get_doc(
{
"doctype": "FTA Audit File",
"company": self.company,
"from_date": "2024-03-31",
"to_date": "2024-01-01",
"file_type": "VAT",
}
)
self.assertRaises(frappe.ValidationError, doc.insert)
def test_company_trn_validation(self):
"""Test that company must have a TRN."""
company_no_trn = "_Test Company No TRN"
if not frappe.db.exists("Company", company_no_trn):
company = frappe.get_doc(
{
"doctype": "Company",
"company_name": company_no_trn,
"abbr": "_TCNT",
"country": "United Arab Emirates",
"default_currency": "AED",
}
)
company.insert(ignore_permissions=True)
doc = frappe.get_doc(
{
"doctype": "FTA Audit File",
"company": company_no_trn,
"from_date": "2024-01-01",
"to_date": "2024-03-31",
"file_type": "VAT",
}
)
self.assertRaises(frappe.ValidationError, doc.insert)
def test_generate_faf_empty_period(self):
"""End-to-end smoke test: generate against an empty period.
With ``frappe.flags.in_test`` set by FrappeTestCase, the enqueued
job runs synchronously, so by the time generate_faf() returns the
doc has reached its terminal status.
"""
doc = frappe.get_doc(
{
"doctype": "FTA Audit File",
"company": self.company,
"from_date": "2099-01-01",
"to_date": "2099-01-31",
"file_type": "VAT",
}
)
doc.insert()
result = doc.generate_faf()
self.assertTrue(result["success"])
doc.reload()
self.assertEqual(doc.status, "Generated")
self.assertTrue(doc.faf_file)
self.assertIn("Company Information written", doc.generation_log)
self.assertIn("Purchase Listing written", doc.generation_log)
self.assertIn("Supply Listing written", doc.generation_log)
self.assertIn("General Ledger written", doc.generation_log)
def test_generate_faf_csv_structure(self):
"""The generated CSV must contain the four spec section markers."""
doc = frappe.get_doc(
{
"doctype": "FTA Audit File",
"company": self.company,
"from_date": "2099-02-01",
"to_date": "2099-02-28",
"file_type": "VAT",
}
)
doc.insert()
doc.generate_faf()
doc.reload()
self.assertEqual(doc.status, "Generated")
file_doc = frappe.get_doc("File", {"file_url": doc.faf_file})
csv_content = file_doc.get_content()
if isinstance(csv_content, bytes):
csv_content = csv_content.decode("utf-8")
for marker in (
"CompInfoStart",
"CompInfoEnd",
"PurcDataStart",
"PurcDataEnd",
"SuppDataStart",
"SuppDataEnd",
"GLDataStart",
"GLDataEnd",
):
self.assertIn(marker, csv_content, f"Missing FAF section marker {marker!r}")
self.assertIn("FAFv1.0.0", csv_content)
def test_tax_agent_fields_appear_in_company_info(self):
"""Tax Agency / Tax Agent details must round-trip into the CSV."""
doc = frappe.get_doc(
{
"doctype": "FTA Audit File",
"company": self.company,
"from_date": "2099-04-01",
"to_date": "2099-04-30",
"file_type": "VAT",
"tax_agency_name": "Acme Tax Agency",
"tan": "TAN-555-001",
"tax_agent_name": "Jane Auditor",
"taan": "TAAN-777",
}
)
doc.insert()
doc.generate_faf()
doc.reload()
self.assertEqual(doc.status, "Generated")
file_doc = frappe.get_doc("File", {"file_url": doc.faf_file})
csv_content = file_doc.get_content()
if isinstance(csv_content, bytes):
csv_content = csv_content.decode("utf-8")
for value in ("Acme Tax Agency", "TAN-555-001", "Jane Auditor", "TAAN-777"):
self.assertIn(value, csv_content, f"Missing tax-agent value {value!r} in FAF")
def test_decimal_fields_use_two_decimal_places(self):
"""Decimal[14,2] cells must always emit two decimal places per spec."""
doc = frappe.get_doc(
{
"doctype": "FTA Audit File",
"company": self.company,
"from_date": "2099-05-01",
"to_date": "2099-05-31",
"file_type": "VAT",
}
)
doc.insert()
doc.generate_faf()
doc.reload()
file_doc = frappe.get_doc("File", {"file_url": doc.faf_file})
csv_content = file_doc.get_content()
if isinstance(csv_content, bytes):
csv_content = csv_content.decode("utf-8")
self.assertIn("PurcDataEnd,0.00,0.00,0", csv_content)
self.assertIn("SuppDataEnd,0.00,0.00,0", csv_content)
self.assertIn("GLDataEnd,0.00,0.00,0,AED", csv_content)
self.assertNotIn("PurcDataEnd,0.0,", csv_content)
self.assertNotIn("SuppDataEnd,0.0,", csv_content)
self.assertNotIn("GLDataEnd,0.0,", csv_content)
def test_mark_as_submitted_workflow(self):
"""Generated docs can be marked submitted; non-Generated cannot."""
doc = frappe.get_doc(
{
"doctype": "FTA Audit File",
"company": self.company,
"from_date": "2099-05-01",
"to_date": "2099-05-31",
"file_type": "VAT",
}
)
doc.insert()
self.assertRaises(frappe.ValidationError, doc.mark_as_submitted)
doc.generate_faf()
doc.reload()
self.assertEqual(doc.status, "Generated")
result = doc.mark_as_submitted()
self.assertTrue(result["success"])
doc.reload()
self.assertEqual(doc.status, "Submitted")

141
erpnext/regional/report/uae_vat_201/uae_vat_201.html
View File

@@ -1,77 +1,104 @@
{%
var report_columns = report.get_columns_for_print();
report_columns = report_columns.filter(col => !col.hidden);
var report_columns = report.get_columns_for_print();
report_columns = report_columns.filter(col => !col.hidden);
%}
<style>
.print-format {
padding: 10mm;
font-size: 8.0pt !important;
font-family: Tahoma, sans-serif;
}
.print-format {
padding: 10mm;
font-size: 8pt !important;
font-family: Tahoma, sans-serif;
}
.print-format th.col-no { width: 8%; }
.print-format th.col-legend { width: 58%; }
.print-format th.col-amount,
.print-format th.col-vat { width: 17%; }
.print-format th.nvd-legend { width: 75%; }
.print-format td.num,
.print-format th.num { text-align: right; }
</style>
<h1 style="margin-top:0; text-align: center;">{%= __(report.report_name) %}</h1>
<h1 style="margin-top:0; text-align: center;">{%= __(report.report_name) %}</h1>
<h3 style="margin-top:0; font-weight:500">{%= __("VAT on Sales and All Other Outputs") %}</h2>
<h3 style="margin-top:0; font-weight:500">{%= __("VAT on Sales and All Other Outputs") %}</h3>
<table class="table table-bordered">
<thead>
<th style="width: 13">{%= report_columns[0].label %}</th>
<th style="width: {%= 100 - (report_columns.length - 1) * 13%}%">{%= report_columns[1].label %}</th>
{% for (let i=2; i<report_columns.length; i++) { %}
<th style="width: 13">{%= report_columns[i].label %}</th>
{% } %}
</thead>
<tbody>
{% for (let j=1; j<12; j++) { %}
{%
var row = data[j];
%}
<tr >
{% for (let i=0; i<report_columns.length; i++) { %}
<td >
{% const fieldname = report_columns[i].fieldname; %}
{% if (!is_null(row[fieldname])) { %}
{%= frappe.format(row[fieldname], report_columns[i], {}, row) %}
{% } %}
</td>
{% } %}
<tr>
<th class="col-no">{%= report_columns[0].label %}</th>
<th class="col-legend">{%= report_columns[1].label %}</th>
<th class="col-amount num">{%= report_columns[2].label %}</th>
<th class="col-vat num">{%= report_columns[3].label %}</th>
</tr>
</thead>
<tbody>
{% for (let j=1; j<13; j++) { %}
{% var row = data[j]; %}
{% if (row) { %}
<tr>
{% for (let i=0; i<report_columns.length; i++) { %}
<td class="{%= report_columns[i].fieldtype == 'Currency' ? 'num' : '' %}">
{% const fieldname = report_columns[i].fieldname; %}
{% if (!is_null(row[fieldname])) { %}
{%= frappe.format(row[fieldname], report_columns[i], {}, row) %}
{% } %}
</td>
{% } %}
</tr>
{% } %}
{% } %}
</tbody>
</table>
<h3 style="margin-top:0; font-weight:500">{%= __("VAT on Expenses and All Other Inputs") %}</h2>
<h3 style="margin-top:0; font-weight:500">{%= __("VAT on Expenses and All Other Inputs") %}</h3>
<table class="table table-bordered">
<table class="table table-bordered">
<thead>
<th style="width: 13">{%= report_columns[0].label %}</th>
<th style="width: {%= 100 - (report_columns.length - 1) * 13%}%">{%= report_columns[1].label %}</th>
{% for (let i=2; i<report_columns.length; i++) { %}
<th style="width: 13">{%= report_columns[i].label %}</th>
{% } %}
</thead>
<tbody>
{% for (let j=14; j<data.length; j++) { %}
{%
var row = data[j];
%}
<tr >
{% for (let i=0; i<report_columns.length; i++) { %}
<td >
{% const fieldname = report_columns[i].fieldname; %}
{% if (!is_null(row[fieldname])) { %}
{%= frappe.format(row[fieldname], report_columns[i], {}, row) %}
{% } %}
</td>
{% } %}
<tr>
<th class="col-no">{%= report_columns[0].label %}</th>
<th class="col-legend">{%= report_columns[1].label %}</th>
<th class="col-amount num">{%= report_columns[2].label %}</th>
<th class="col-vat num">{%= report_columns[3].label %}</th>
</tr>
</thead>
<tbody>
{% for (let j=15; j<18; j++) { %}
{% var row = data[j]; %}
{% if (row) { %}
<tr>
{% for (let i=0; i<report_columns.length; i++) { %}
<td class="{%= report_columns[i].fieldtype == 'Currency' ? 'num' : '' %}">
{% const fieldname = report_columns[i].fieldname; %}
{% if (!is_null(row[fieldname])) { %}
{%= frappe.format(row[fieldname], report_columns[i], {}, row) %}
{% } %}
</td>
{% } %}
</tr>
{% } %}
{% } %}
</tbody>
</table>
<h3 style="margin-top:0; font-weight:500">{%= __("Net VAT Due") %}</h3>
<table class="table table-bordered">
<thead>
<tr>
<th class="col-no">{%= report_columns[0].label %}</th>
<th class="nvd-legend">{%= report_columns[1].label %}</th>
<th class="col-vat num">{%= report_columns[3].label %}</th>
</tr>
</thead>
<tbody>
{% for (let j=20; j<23; j++) { %}
{% var row = data[j]; %}
{% if (row) { %}
<tr>
<td>{%= row.no %}</td>
<td>{%= row.legend %}</td>
<td class="num">{%= row.vat_amount %}</td>
</tr>
{% } %}
{% } %}
</tbody>
</table>

5
erpnext/regional/report/uae_vat_201/uae_vat_201.js
View File

@@ -33,11 +33,14 @@ frappe.query_reports["UAE VAT 201"] = {
default: frappe.datetime.get_today(),
},
],
formatter: function (value, row, column, data, default_formatter) {
if (
data &&
(data.legend == "VAT on Sales and All Other Outputs" ||
data.legend == "VAT on Expenses and All Other Inputs") &&
data.legend == "VAT on Expenses and All Other Inputs" ||
data.legend == "Net VAT Due" ||
data.legend == "Total") &&
data.legend == value
) {
value = $(`<span>${value}</span>`);

631
erpnext/regional/report/uae_vat_201/uae_vat_201.py
View File

@@ -2,17 +2,75 @@
# 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
# Per-request memoization cache for the helper functions below. Stored on
# ``frappe.local`` so concurrent requests under gevent/threaded workers
# never share or race on this state; cleared at the start of every
# ``execute()`` so each report run gets fresh data.
_CACHE_ATTR = "_uae_vat_201_cache"
def _get_cache():
cache = getattr(frappe.local, _CACHE_ATTR, None)
if cache is None:
cache = {}
setattr(frappe.local, _CACHE_ATTR, cache)
return 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):
# ``frappe.local`` survives across unit-test methods (it is request
# scoped, not test scoped). Two tests that call the same helper with
# equivalent filter dicts would otherwise share a cached value from
# the first test's data set. Bypass the cache in tests so each
# call hits the DB; production callers (one execute() per HTTP
# request, cache cleared at its start) still see the optimisation.
if frappe.flags.in_test:
return fn(filters, *args, **kwargs)
cache = _get_cache()
key = (fn.__name__, tuple(sorted((filters or {}).items())))
if key not in cache:
cache[key] = fn(filters, *args, **kwargs)
return cache[key]
return wrapper
def execute(filters=None):
filters = filters or {}
validate_company_region(filters)
_get_cache().clear()
columns = get_columns()
data, emirates, amounts_by_emirate = get_data(filters)
data = get_data(filters)
return columns, data
@@ -48,23 +106,86 @@ def get_columns():
def get_data(filters=None):
"""Returns the list of dictionaries. Each dictionary is a row in the datatable and chart data."""
data = []
emirates, amounts_by_emirate = append_vat_on_sales(data, filters)
amounts_by_emirate = append_vat_on_sales(data, filters)
append_vat_on_expenses(data, filters)
return data, emirates, amounts_by_emirate
net_vat_due(data, filters, amounts_by_emirate)
dubai_label_override = _company_emirate_label(filters)
final_data = []
for row in data:
key = row.get("_key")
legend = row.get("legend")
new_legend = legend
if key and key.startswith("emirate:"):
emirate = key.split(":", 1)[1]
label = dubai_label_override if emirate == "Dubai" and dubai_label_override else legend
new_legend = _drill_down_link(
label, filters, doc_type="Sales Invoice", vat=emirate, category="Standard"
)
elif key == "reverse_charge_supplies":
new_legend = _drill_down_link(legend, filters, doc_type="Purchase Invoice", reverse_charge="Y")
elif key == "zero_rated":
new_legend = _drill_down_link(legend, filters, doc_type="Sales Invoice", category="Zero Rated")
elif key == "exempt_supplies":
new_legend = _drill_down_link(legend, filters, doc_type="Sales Invoice", category="Exempt Rated")
elif key == "standard_rated_expenses":
new_legend = _drill_down_link(legend, filters, doc_type="Purchase Invoice")
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"), "", "")
emirates, amounts_by_emirate = standard_rated_expenses_emiratewise(data, filters)
amounts_by_emirate = standard_rated_expenses_emiratewise(data, filters)
si_amount = amounts_by_emirate[1]
si_vat = amounts_by_emirate[2]
append_data(
data,
"2",
_("Tax Refunds provided to Tourists under the Tax Refunds for Tourists Scheme"),
frappe.format((-1) * get_tourist_tax_return_total(filters), "Currency"),
frappe.format((-1) * get_tourist_tax_return_tax(filters), "Currency"),
format_currency_signed((-1) * get_tourist_tax_return_total(filters)),
format_currency_signed((-1) * get_tourist_tax_return_tax(filters)),
)
append_data(
@@ -73,15 +194,48 @@ def append_vat_on_sales(data, filters):
_("Supplies subject to the reverse charge provision"),
frappe.format(get_reverse_charge_total(filters), "Currency"),
frappe.format(get_reverse_charge_tax(filters), "Currency"),
key="reverse_charge_supplies",
)
append_data(data, "4", _("Zero Rated"), frappe.format(get_zero_rated_total(filters), "Currency"), "-")
append_data(
data,
"4",
_("Zero Rated"),
frappe.format(get_zero_rated_total(filters), "Currency"),
"-",
key="zero_rated",
)
append_data(data, "5", _("Exempt Supplies"), frappe.format(get_exempt_total(filters), "Currency"), "-")
append_data(
data,
"5",
_("Exempt Supplies"),
frappe.format(get_exempt_total(filters), "Currency"),
"-",
key="exempt_supplies",
)
append_data(
data,
"8",
_("Total"),
frappe.format(
(-1) * get_tourist_tax_return_total(filters)
+ get_reverse_charge_total(filters)
+ get_zero_rated_total(filters)
+ get_exempt_total(filters)
+ sum(si_amount),
"Currency",
),
frappe.format(
(-1) * get_tourist_tax_return_tax(filters) + get_reverse_charge_tax(filters) + sum(si_vat),
"Currency",
),
)
append_data(data, "", "", "", "")
return emirates, amounts_by_emirate
return amounts_by_emirate
def standard_rated_expenses_emiratewise(data, filters):
@@ -98,16 +252,22 @@ def standard_rated_expenses_emiratewise(data, filters):
"vat_amount": frappe.format(vat, "Currency"),
}
amounts_by_emirate = append_emiratewise_expenses(data, emirates, amounts_by_emirate)
return emirates, amounts_by_emirate
return amounts_by_emirate
def append_emiratewise_expenses(data, emirates, amounts_by_emirate):
"""Append emiratewise standard rated expenses and vat."""
s_amount = []
v_amount = []
for no, emirate in enumerate(emirates, 97):
if emirate in amounts_by_emirate:
amounts_by_emirate[emirate]["no"] = _("1{0}").format(chr(no))
amounts_by_emirate[emirate]["legend"] = _("Standard rated supplies in {0}").format(emirate)
amounts_by_emirate[emirate]["_key"] = f"emirate:{emirate}"
data.append(amounts_by_emirate[emirate])
s_amount.append(amounts_by_emirate[emirate].get("raw_amount") or 0)
v_amount.append(amounts_by_emirate[emirate].get("raw_vat_amount") or 0)
else:
append_data(
data,
@@ -115,8 +275,9 @@ def append_emiratewise_expenses(data, emirates, amounts_by_emirate):
_("Standard rated supplies in {0}").format(emirate),
frappe.format(0, "Currency"),
frappe.format(0, "Currency"),
key=f"emirate:{emirate}",
)
return amounts_by_emirate
return amounts_by_emirate, s_amount, v_amount
def append_vat_on_expenses(data, filters):
@@ -128,6 +289,7 @@ def append_vat_on_expenses(data, filters):
_("Standard Rated Expenses"),
frappe.format(get_standard_rated_expenses_total(filters), "Currency"),
frappe.format(get_standard_rated_expenses_tax(filters), "Currency"),
key="standard_rated_expenses",
)
append_data(
data,
@@ -137,30 +299,103 @@ def append_vat_on_expenses(data, filters):
frappe.format(get_reverse_charge_recoverable_tax(filters), "Currency"),
)
def append_data(data, no, legend, amount, vat_amount):
"""Returns data with appended value."""
data.append({"no": no, "legend": legend, "amount": amount, "vat_amount": vat_amount})
append_data(
data,
"11",
_("Total"),
frappe.format(
get_standard_rated_expenses_total(filters) + get_reverse_charge_recoverable_total(filters),
"Currency",
),
frappe.format(
get_standard_rated_expenses_tax(filters) + get_reverse_charge_recoverable_tax(filters),
"Currency",
),
)
def net_vat_due(data, filters, amounts_by_emirate):
si_vat = amounts_by_emirate[2]
append_data(data, "", "", "", "")
append_data(data, "", _("Net VAT Due"), "", "")
append_data(
data,
"12",
_("Total value of due tax for the period"),
frappe.format(0.00, "Currency"),
frappe.format(
sum(si_vat) + (-1) * get_tourist_tax_return_tax(filters) + get_reverse_charge_tax(filters),
"Currency",
),
)
append_data(
data,
"13",
_("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),
"Currency",
),
)
# Calculate payable tax: Due Tax - Recoverable Tax
due_tax = sum(si_vat) + (-1) * get_tourist_tax_return_tax(filters) + get_reverse_charge_tax(filters)
recoverable_tax = get_standard_rated_expenses_tax(filters) + get_reverse_charge_recoverable_tax(filters)
payable_tax = due_tax - recoverable_tax
append_data(
data,
"14",
_("Payable tax for the period"),
frappe.format(0.00, "Currency"),
frappe.format(payable_tax, "Currency"),
)
def append_data(data, no, legend, amount, vat_amount, key=None):
"""Append one row to ``data``.
``key`` (when provided) is a language-independent identifier used by
``get_data`` to decide which rows get drill-down links. Without it,
dispatch would have to match the localized ``legend`` text and would
silently break under any non-English language.
"""
data.append({"no": no, "legend": legend, "amount": amount, "vat_amount": vat_amount, "_key": key})
def format_currency_signed(value):
"""Format a number as currency, placing the minus sign *before* the currency symbol
when negative (e.g. "-د.إ 5,000.00" rather than "د.إ -5,000.00")."""
if value is None:
value = 0
if value < 0:
return "-" + frappe.format(abs(value), "Currency")
return frappe.format(value, "Currency")
@_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():
@@ -168,245 +403,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"]])
if filters.get("from_date"):
query_filters.append(["posting_date", "<=", filters["to_date"]])
return query_filters
query = query.where(table.posting_date >= filters["from_date"])
if filters.get("to_date"):
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_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_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_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_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

0
erpnext/regional/report/uae_vat_register/__init__.py Normal file
View File

73
erpnext/regional/report/uae_vat_register/uae_vat_register.js Normal file
View File

@@ -0,0 +1,73 @@
// Copyright (c) 2025, Frappe Technologies Pvt. Ltd. and contributors
// For license information, please see license.txt
frappe.query_reports["UAE VAT Register"] = {
filters: [
{
fieldname: "company",
label: __("Company"),
fieldtype: "Link",
options: "Company",
reqd: 1,
default: frappe.defaults.get_user_default("Company"),
},
{
fieldname: "from_date",
label: __("From Date"),
fieldtype: "Date",
reqd: 1,
default: frappe.datetime.add_months(frappe.datetime.get_today(), -3),
},
{
fieldname: "to_date",
label: __("To Date"),
fieldtype: "Date",
reqd: 1,
default: frappe.datetime.get_today(),
},
{
fieldname: "doc_type",
label: __("Document Type"),
fieldtype: "Select",
options: ["Sales Invoice", "Purchase Invoice"],
default: "Sales Invoice",
reqd: 1,
},
{
fieldname: "category",
label: __("Category"),
fieldtype: "Select",
options: ["", "Standard", "Zero Rated", "Exempt Rated"],
depends_on: "eval: doc.doc_type == 'Sales Invoice'",
},
{
fieldname: "vat",
label: __("Emirate"),
fieldtype: "Select",
options: [
"",
"Abu Dhabi",
"Dubai",
"Sharjah",
"Ajman",
"Umm Al Quwain",
"Ras Al Khaimah",
"Fujairah",
],
depends_on: "eval: doc.doc_type == 'Sales Invoice'",
},
{
fieldname: "reverse_charge",
label: __("Reverse Charge"),
fieldtype: "Select",
options: ["", "Y", "N"],
depends_on: "eval: doc.doc_type == 'Purchase Invoice'",
},
{
fieldname: "item_wise",
label: __("Item-wise"),
fieldtype: "Check",
default: 0,
},
],
};

24
erpnext/regional/report/uae_vat_register/uae_vat_register.json Normal file
View File

@@ -0,0 +1,24 @@
{
"add_total_row": 1,
"add_translate_data": 0,
"columns": [],
"creation": "2025-12-11 00:29:21.891860",
"disabled": 0,
"docstatus": 0,
"doctype": "Report",
"filters": [],
"idx": 0,
"is_standard": "Yes",
"letterhead": null,
"modified": "2025-12-11 00:29:21.891860",
"modified_by": "Administrator",
"module": "Regional",
"name": "UAE VAT Register",
"owner": "Administrator",
"prepared_report": 0,
"ref_doctype": "GL Entry",
"report_name": "UAE VAT Register",
"report_type": "Script Report",
"roles": [],
"timeout": 0
}

197
erpnext/regional/report/uae_vat_register/uae_vat_register.py Normal file
View File

@@ -0,0 +1,197 @@
# Copyright (c) 2025, Frappe Technologies Pvt. Ltd. and contributors
# For license information, please see license.txt
import frappe
from frappe import _
from frappe.query_builder.functions import Coalesce, Sum
def execute(filters=None):
if not filters:
filters = {}
columns = get_columns(filters)
data = get_data(filters)
return columns, data
def get_columns(filters):
doc_type = filters.get("doc_type") or "Sales Invoice"
is_sales = doc_type == "Sales Invoice"
item_wise = bool(filters.get("item_wise"))
columns = [
{
"label": _("Invoice"),
"fieldname": "name",
"fieldtype": "Link",
"options": doc_type,
"width": 180,
},
{"label": _("Posting Date"), "fieldname": "posting_date", "fieldtype": "Date", "width": 100},
{
"label": _("Customer") if is_sales else _("Supplier"),
"fieldname": "party",
"fieldtype": "Link",
"options": "Customer" if is_sales else "Supplier",
"width": 150,
},
{
"label": _("Cost Center"),
"fieldname": "cost_center",
"fieldtype": "Link",
"options": "Cost Center",
"width": 130,
},
]
if is_sales:
columns.append({"label": _("Emirate"), "fieldname": "emirate", "fieldtype": "Data", "width": 110})
else:
columns.append(
{
"label": _("Reverse Charge"),
"fieldname": "reverse_charge",
"fieldtype": "Data",
"width": 110,
}
)
if item_wise:
columns.extend(
[
{
"label": _("Item Code"),
"fieldname": "item_code",
"fieldtype": "Link",
"options": "Item",
"width": 150,
},
{"label": _("Qty"), "fieldname": "qty", "fieldtype": "Float", "width": 80},
{"label": _("Rate"), "fieldname": "rate", "fieldtype": "Currency", "width": 100},
]
)
else:
columns.append({"label": _("Qty"), "fieldname": "qty", "fieldtype": "Float", "width": 80})
columns.extend(
[
{"label": _("Net Amount"), "fieldname": "net_amount", "fieldtype": "Currency", "width": 120},
{"label": _("VAT Amount"), "fieldname": "vat_amount", "fieldtype": "Currency", "width": 120},
{
"label": _("Total Amount"),
"fieldname": "total_amount",
"fieldtype": "Currency",
"width": 120,
},
]
)
return columns
def get_data(filters):
doc_type = filters.get("doc_type") or "Sales Invoice"
if doc_type == "Sales Invoice":
return _fetch_rows(filters, is_sales=True)
if doc_type == "Purchase Invoice":
return _fetch_rows(filters, is_sales=False)
return []
def _fetch_rows(filters, is_sales):
"""Build the VAT register query for either Sales or Purchase Invoices.
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"))
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)
)
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 _apply_period_filters(query, parent, filters):
if filters.get("company"):
query = query.where(parent.company == filters["company"])
if filters.get("from_date"):
query = query.where(parent.posting_date >= filters["from_date"])
if filters.get("to_date"):
query = query.where(parent.posting_date <= filters["to_date"])
return query
def _sales_category_criterion(child, category):
"""Translate the ``category`` filter into a Sales Invoice Item criterion."""
if category == "Standard":
return (child.is_zero_rated != 1) & (child.is_exempt != 1)
if category == "Zero Rated":
return child.is_zero_rated == 1
if category == "Exempt Rated":
return child.is_exempt == 1
return None

8
erpnext/regional/united_arab_emirates/setup.py
View File

@@ -200,6 +200,14 @@ def make_custom_fields():
print_hide=1,
),
],
"Company": [
dict(
fieldname="company_name_in_arabic",
label="Company Name in Arabic",
fieldtype="Data",
insert_after="company_name",
),
],
"Customer": [
dict(
fieldname="customer_name_in_arabic",