default_thai_company/default_thai_company/tax_withholding.py

import json
import os

import erpnext
import frappe
from erpnext.accounts.doctype.payment_entry.payment_entry import PaymentEntry
from erpnext.accounts.doctype.payment_entry.payment_entry import get_payment_entry as _get_payment_entry
from erpnext.accounts.doctype.sales_invoice.sales_invoice import SalesInvoice
from erpnext.accounts.doctype.tax_withholding_category.tax_withholding_category import (
	get_tax_withholding_details,
	normal_round,
)
from frappe import _
from frappe.contacts.doctype.address.address import get_default_address
from frappe.modules.utils import sync_customizations_for_doctype
from frappe.utils import cint, flt, fmt_money, getdate

from default_thai_company.utils import money_in_words

# Tax accounts every Thai company gets (Company.on_update, fixture import, and the
# create_vat_accounts patch for companies that predate the VAT pair).
# Withholding: liability = tax we withhold from suppliers, remitted on P.N.D.3/53
# (linked on every category); asset = tax customers withhold from us, creditable
# against corporate income tax.
# VAT: output (ภาษีขาย) collected on sales is payable, input (ภาษีซื้อ) paid on
# purchases is recoverable; the P.P.30 return nets the two, so they are kept apart.
LIABILITY_ACCOUNT = "Withholding Tax Payable"
ASSET_ACCOUNT = "Withholding Tax Receivable"
OUTPUT_VAT_ACCOUNT = "Output VAT"
INPUT_VAT_ACCOUNT = "Input VAT"
COMPANY_ACCOUNTS = (
	(LIABILITY_ACCOUNT, "Liability"),
	(ASSET_ACCOUNT, "Asset"),
	(OUTPUT_VAT_ACCOUNT, "Liability"),
	(INPUT_VAT_ACCOUNT, "Asset"),
)

# Preferred parent groups by root type; first match wins, else the root account.
PARENT_GROUPS = {
	"Liability": ("Duties and Taxes", "Current Liabilities"),
	"Asset": ("Tax Assets", "Current Assets"),
}

CATEGORY_FILTERS = {"name": ("like", "WHT %")}


def thai_companies():
	return frappe.get_all("Company", filters={"country": "Thailand"}, pluck="name")


def get_or_create_account(company, account_name, root_type, account_type="Tax", parent_groups=None):
	"""Leaf account `account_name` for `company`, created under the first existing
	group in `parent_groups` (default: PARENT_GROUPS[root_type]), else the root."""
	name = frappe.db.get_value(
		"Account", {"company": company, "account_name": account_name, "is_group": 0, "root_type": root_type}
	)
	if name:
		return name

	account = frappe.get_doc(
		{
			"doctype": "Account",
			"company": company,
			"account_name": account_name,
			"parent_account": find_parent_group(
				company, root_type, parent_groups or PARENT_GROUPS[root_type]
			),
			"root_type": root_type,
			"account_type": account_type,
			"is_group": 0,
		}
	).insert(ignore_permissions=True)
	return account.name


def find_parent_group(company, root_type, group_names):
	for account_name in group_names:
		parent = frappe.db.get_value(
			"Account",
			{"company": company, "account_name": account_name, "is_group": 1, "root_type": root_type},
		)
		if parent:
			return parent

	return frappe.db.get_value(
		"Account",
		{"company": company, "root_type": root_type, "is_group": 1, "parent_account": ("is", "not set")},
	)


def ensure_company_accounts(company):
	"""Create every account in COMPANY_ACCOUNTS for `company`; return the
	withholding liability account name (the one Tax Withholding Categories link)."""
	accounts = {name: get_or_create_account(company, name, root_type) for name, root_type in COMPANY_ACCOUNTS}
	return accounts[LIABILITY_ACCOUNT]


def link_company(category, company, account):
	"""Append `account` for `company` to the category's accounts table if missing."""
	if any(row.company == company for row in category.accounts):
		return False
	category.append("accounts", {"company": company, "account": account})
	return True


def prepare_fixture_accounts(doc, method=None):
	"""Tax Withholding Category fixtures ship without `accounts` (company-specific).

	Fixture import re-inserts the doc on every migrate, so: carry over the rows
	already configured on this site, then link every Thai company, creating the
	withholding accounts on first use. Mandatory is relaxed for sites without a
	Thai company yet (setup wizard not run); `setup_company` links them later.
	"""
	doc.flags.ignore_mandatory = True

	if not doc.accounts and frappe.db.exists(doc.doctype, doc.name):
		for row in frappe.get_all(
			"Tax Withholding Account",
			filters={"parent": doc.name, "parenttype": doc.doctype},
			fields=["company", "account"],
			order_by="idx",
		):
			doc.append("accounts", row)

	for company in thai_companies():
		if not any(row.company == company for row in doc.accounts):
			link_company(doc, company, ensure_company_accounts(company))


def sync_category_customizations():
	"""before_install / before_migrate: create the Tax Withholding Category Custom
	Fields (custom/tax_withholding_category.json) ahead of the fixture import.
	Frappe syncs fixtures before customizations, and a fixture value for a field
	that does not exist yet is dropped."""
	folder = frappe.get_app_path("default_thai_company", "default_thai_company", "custom")
	filename = "tax_withholding_category.json"
	with open(os.path.join(folder, filename)) as f:
		sync_customizations_for_doctype(json.load(f), folder, filename)


def company_ready(doc):
	"""A Thai company whose chart of accounts exists."""
	return doc.country == "Thailand" and frappe.db.exists("Account", {"company": doc.name})


def setup_company(doc, method=None):
	"""Company.on_update: create the tax accounts and link every WHT category."""
	if not company_ready(doc):
		return

	account = ensure_company_accounts(doc.name)
	for name in frappe.get_all("Tax Withholding Category", filters=CATEGORY_FILTERS, pluck="name"):
		category = frappe.get_doc("Tax Withholding Category", name)
		if link_company(category, doc.name, account):
			category.save(ignore_permissions=True)


def is_thai_company(company):
	return frappe.get_cached_value("Company", company, "country") == "Thailand"


def get_withholding_details(category, posting_date, company):
	details = get_tax_withholding_details(category, posting_date, company)
	if not details:
		frappe.throw(
			_("Tax Withholding Category {0} has no account for Company {1}").format(category, company)
		)
	return details


def withholding_on(details, taxable, base_taxable, precision):
	"""Withholding for a taxable amount; the single threshold is checked in company currency."""
	if details.threshold and flt(base_taxable) < flt(details.threshold):
		return 0.0
	amount = flt(taxable) * flt(details.rate) / 100
	return normal_round(amount) if cint(details.round_off_tax_amount) else flt(amount, precision)


class ThaiSalesInvoice(SalesInvoice):
	def set_tax_withholding(self):
		"""Thai customers withhold at payment (see `set_customer_withholding`);
		ERPNext's customer-side handling is Indian TCS, which adds tax on top of
		the invoice. For Thai companies only show the expected withholding and
		the net payable; totals and GL are untouched."""
		if not is_thai_company(self.company):
			return super().set_tax_withholding()

		self.withholding_tax_amount = self.amount_after_withholding = 0
		if not self.tax_withholding_category:
			return

		details = get_withholding_details(self.tax_withholding_category, self.posting_date, self.company)
		self.withholding_tax_amount = withholding_on(
			details, self.net_total, self.base_net_total, self.precision("withholding_tax_amount")
		)
		if self.withholding_tax_amount:
			self.amount_after_withholding = flt(
				(self.rounded_total or self.grand_total) - self.withholding_tax_amount,
				self.precision("amount_after_withholding"),
			)

	def set_total_in_words(self):
		"""SellingController's version, with Thai wording when the saving user's
		language is Thai. Print formats re-derive the text per print language
		(`get_in_words`)."""
		base_amount = abs(
			self.base_grand_total if self.is_rounded_total_disabled() else self.base_rounded_total
		)
		self.base_in_words = money_in_words(base_amount, self.company_currency)
		amount = abs(self.grand_total if self.is_rounded_total_disabled() else self.rounded_total)
		self.in_words = money_in_words(amount, self.currency)


class ThaiPaymentEntry(PaymentEntry):
	def build_gl_map(self):
		"""A "Deduct" tax row on a receipt posts bank Dr gross and bank Cr
		withholding; ERPNext merges them into one entry but only nets it when
		the bank entry itself carries `post_net_value`. Set it so the bank
		ledger shows the amount that actually arrived."""
		gl_entries = super().build_gl_map()
		if self.payment_type == "Receive" and self.party_type == "Customer" and self.get("taxes"):
			for entry in gl_entries:
				if entry.account == self.paid_to:
					entry.post_net_value = True
		return gl_entries

	@frappe.whitelist()
	def allocate_amount_to_references(self, *args, **kwargs):
		"""Called by the form after Get Outstanding Invoices and on Paid Amount
		changes; keep the withholding row in step with the allocation."""
		super().allocate_amount_to_references(*args, **kwargs)
		if self.party_type == "Customer" and self.payment_type == "Receive" and self.source_exchange_rate:
			auto_apply_customer_withholding(self)
			set_customer_withholding(self)


@frappe.whitelist()
def get_payment_entry(
	dt,
	dn,
	party_amount=None,
	bank_account=None,
	bank_amount=None,
	party_type=None,
	payment_type=None,
	reference_date=None,
	created_from_payment_request=False,
):
	"""Create > Payment from a Sales Invoice: arrive with withholding applied.

	Signature mirrors ERPNext's so `frappe.call` drops request-only args
	(`cmd`, ...) instead of forwarding them.
	"""
	pe = _get_payment_entry(
		dt,
		dn,
		party_amount=party_amount,
		bank_account=bank_account,
		bank_amount=bank_amount,
		party_type=party_type,
		payment_type=payment_type,
		reference_date=reference_date,
		created_from_payment_request=created_from_payment_request,
	)
	if auto_apply_customer_withholding(pe) and pe.source_exchange_rate:
		set_customer_withholding(pe)
	return pe


def auto_apply_customer_withholding(doc):
	"""Tick Apply Tax Withholding Amount on a new customer receipt whose
	references include a Sales Invoice that carries withholding. Only for
	unsaved entries: after the first save the checkbox is the user's."""
	if (
		doc.party_type != "Customer"
		or doc.payment_type != "Receive"
		or doc.apply_tax_withholding_amount
		or not doc.is_new()
	):
		return False

	invoices = [r.reference_name for r in doc.references if r.reference_doctype == "Sales Invoice"]
	if not invoices or not frappe.db.exists(
		"Sales Invoice", {"name": ("in", invoices), "withholding_tax_amount": (">", 0)}
	):
		return False

	doc.apply_tax_withholding_amount = 1
	return True


def set_customer_withholding(doc, method=None):
	"""Payment Entry.validate: tax withheld by the customer on a Receive entry
	is posted like ERPNext's supplier TDS, as a "Deduct" row in the taxes
	table against the withholding receivable account.

	`paid_amount` is the gross amount settled against the invoices (what
	"Get Outstanding Invoices" fills in); the bank receives paid minus the
	withholding (`received_amount_after_tax`). Withholding is rate x pre-VAT
	amount of each allocated reference, prorated by the allocation. Runs after
	the controller's validate, then re-runs the tax computation so the row's
	base amounts and totals are final within this save.
	"""
	if doc.party_type != "Customer" or doc.payment_type != "Receive":
		return

	if method == "validate":
		auto_apply_customer_withholding(doc)

	account = frappe.db.get_value(
		"Account",
		{"company": doc.company, "account_name": ASSET_ACCOUNT, "root_type": "Asset", "is_group": 0},
	)
	row = next((d for d in doc.taxes if d.account_head == account), None) if account else None

	if not doc.apply_tax_withholding_amount:
		if row:
			doc.remove(row)
			doc.apply_taxes()
			doc.set_amounts_after_tax()
		return

	if not account:
		frappe.throw(_("Account {0} not found for Company {1}").format(ASSET_ACCOUNT, doc.company))

	if not doc.tax_withholding_category:
		doc.tax_withholding_category = frappe.db.get_value("Customer", doc.party, "tax_withholding_category")

	amount, description = get_customer_withholding(doc)

	if not amount:
		if row:
			doc.remove(row)
	else:
		if not row:
			row = doc.append(
				"taxes", {"account_head": account, "charge_type": "Actual", "add_deduct_tax": "Deduct"}
			)
		row.tax_amount = amount
		row.description = description
		row.cost_center = doc.cost_center or erpnext.get_default_cost_center(doc.company)

	doc.apply_taxes()
	doc.set_amounts_after_tax()


def get_customer_withholding(doc):
	"""Total withheld across allocated references, in company currency.

	A Sales Invoice's own category wins over the Payment Entry's; Sales Orders
	use the Payment Entry's. Withholding is prorated by allocated / grand total
	on the pre-VAT amount, per reference.
	"""
	total = 0.0
	details_by_category = {}
	precision = doc.precision("difference_amount")

	for ref in doc.references:
		if ref.reference_doctype not in ("Sales Invoice", "Sales Order") or not ref.allocated_amount:
			continue

		fields = ["net_total", "grand_total"]
		if ref.reference_doctype == "Sales Invoice":
			fields.append("tax_withholding_category")
		values = frappe.db.get_value(ref.reference_doctype, ref.reference_name, fields, as_dict=True)
		if not values.grand_total:
			continue

		category = values.get("tax_withholding_category") or doc.tax_withholding_category
		if not category:
			frappe.throw(
				_("Please set Tax Withholding Category on {0} {1} or on this Payment Entry").format(
					_(ref.reference_doctype), ref.reference_name
				)
			)
		if category not in details_by_category:
			details_by_category[category] = get_withholding_details(category, doc.posting_date, doc.company)

		base_taxable = (
			flt(ref.allocated_amount)
			* flt(values.net_total)
			/ flt(values.grand_total)
			* flt(doc.source_exchange_rate)
		)
		total += withholding_on(details_by_category[category], base_taxable, base_taxable, precision)

	if len(details_by_category) == 1:
		# category_name; translations/th.csv carries the fixture categories
		description = _(next(iter(details_by_category.values())).description)
	else:
		description = _("Withholding tax deducted by customer")
	return flt(total, precision), description


# Withholding tax certificate (หนังสือรับรองการหักภาษี ณ ที่จ่าย, Sec. 50 bis): the income
# row the payment is reported on, from Tax Withholding Category.income_type (custom field).
# Rows: 1 = Sec. 40(1), 2 = 40(2), 3 = 40(3), 4a = 40(4)(a), 4b = 40(4)(b) dividends
# paid from profits taxed at the standard 20% rate (form row 4(b)(1)(1.3)), 5 = payments
# withheld under Revenue Department orders issued under Sec. 3 tera, 6 = other (specified).
CERTIFICATE_ROWS = {
	"Salary and Wages - Sec. 40(1)": "1",
	"Fees and Commissions - Sec. 40(2)": "2",
	"Royalties - Sec. 40(3)": "3",
	"Interest - Sec. 40(4)(a)": "4a",
	"Dividends - Sec. 40(4)(b)": "4b",
	"Sec. 3 Tera (Services, Rent, Contract Work etc.)": "5",
}

THAI_MONTHS = (
	"มกราคม",
	"กุมภาพันธ์",
	"มีนาคม",
	"เมษายน",
	"พฤษภาคม",
	"มิถุนายน",
	"กรกฎาคม",
	"สิงหาคม",
	"กันยายน",
	"ตุลาคม",
	"พฤศจิกายน",
	"ธันวาคม",
)


def pnd_form(row, supplier_type):
	"""P.N.D. return the certificate row is filed on, as numbered on the form:
	1 = ภ.ง.ด.1ก, 2 = ภ.ง.ด.2, 3 = ภ.ง.ด.3, 53 = ภ.ง.ด.53.

	Juristic payees (Company; Partnership, taken as registered) file on
	ภ.ง.ด.53 whatever the income; individuals by income type: salary on
	ภ.ง.ด.1ก, Sec. 40(3)/(4) on ภ.ง.ด.2, the rest on ภ.ง.ด.3.
	"""
	if supplier_type != "Individual":
		return "53"
	if row == "1":
		return "1"
	if row in ("3", "4a", "4b"):
		return "2"
	return "3"


def one_line_address(address_name):
	"""Address in the order of the Thailand Address Template, on one line."""
	if not address_name:
		return None
	address = frappe.get_cached_doc("Address", address_name)
	parts = [
		address.address_line1,
		address.address_line2,
		address.county,
		address.city,
		address.state,
		address.pincode,
	]
	if address.country and address.country != "Thailand":
		parts.append(address.country)
	return " ".join(part.strip() for part in parts if part and part.strip())


def tax_id_digits(tax_id):
	"""The 13 digits of a Thai tax ID for the form's boxes, or None when it is not one."""
	digits = "".join(ch for ch in (tax_id or "") if ch.isdigit())
	return digits if len(digits) == 13 else None


def get_withholding_certificate(doc):
	"""Certificate data for a Purchase Invoice with Apply Tax Withholding Amount.

	Amounts are in company currency; the form is Thai, so dates are Buddhist Era
	and the tax in words is Thai whatever the print language.
	"""
	company = frappe.get_cached_doc("Company", doc.company)
	currency = erpnext.get_company_currency(doc.company)
	precision = doc.precision("base_grand_total")

	category = (
		frappe.get_cached_doc("Tax Withholding Category", doc.tax_withholding_category)
		if doc.tax_withholding_category
		else None
	)
	row = CERTIFICATE_ROWS.get(category.income_type if category else None, "6")
	supplier_type = frappe.get_cached_value("Supplier", doc.supplier, "supplier_type")

	amount = flt(doc.base_tax_withholding_net_total, precision)
	tax = flt(
		sum(abs(flt(t.base_tax_amount)) for t in doc.get("taxes") if cint(t.is_tax_withholding_account)),
		precision,
	)
	posting_date = getdate(doc.posting_date)

	return frappe._dict(
		payer=frappe._dict(
			name=company.company_name,
			tax_id=company.tax_id,
			tax_id_digits=tax_id_digits(company.tax_id),
			address=one_line_address(get_default_address("Company", doc.company)),
		),
		payee=frappe._dict(
			name=doc.supplier_name,
			tax_id=doc.tax_id,
			tax_id_digits=tax_id_digits(doc.tax_id),
			address=one_line_address(doc.supplier_address or get_default_address("Supplier", doc.supplier)),
		),
		row=row,
		# row 6 prints what was paid for; translations/th.csv carries the fixture categories
		row_note=_(category.category_name) if row == "6" and category else None,
		pnd=pnd_form(row, supplier_type),
		date=f"{posting_date.day:02d}/{posting_date.month:02d}/{posting_date.year + 543}",
		amount=fmt_money(amount, precision),
		tax=fmt_money(tax, precision),
		tax_in_words=money_in_words(tax, currency, lang="th"),
		issued=frappe._dict(
			day=posting_date.day,
			month=THAI_MONTHS[posting_date.month - 1],
			year=posting_date.year + 543,
		),
	)