default_thai_company/default_thai_company/tax_withholding.py

import erpnext
import frappe
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.utils import cint, flt

# Liability: tax we withhold from suppliers and remit on P.N.D.3/53 (linked on every category).
# Asset: tax customers withhold from us, creditable against corporate income tax.
LIABILITY_ACCOUNT = "Withholding Tax Payable"
ASSET_ACCOUNT = "Withholding Tax Receivable"

# 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):
	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),
			"root_type": root_type,
			"account_type": "Tax",
			"is_group": 0,
		}
	).insert(ignore_permissions=True)
	return account.name


def find_parent_group(company, root_type):
	for account_name in PARENT_GROUPS[root_type]:
		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 both withholding accounts for `company`; return the liability account name."""
	get_or_create_account(company, ASSET_ACCOUNT, "Asset")
	return get_or_create_account(company, LIABILITY_ACCOUNT, "Liability")


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 setup_company(doc, method=None):
	"""Company.on_update: create withholding accounts and link every WHT category."""
	if doc.country != "Thailand" or not frappe.db.exists("Account", {"company": doc.name}):
		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_customer_withholding(doc, method=None):
	"""Payment Entry.validate: tax withheld by the customer on a Receive entry
	becomes a deduction to the company's withholding receivable account.

	Withholding is rate x pre-VAT amount of each allocated reference, prorated
	by the allocation. `paid_amount` is the cash actually received, so the
	deduction closes the difference against the gross allocation. Runs after
	the controller's validate (exchange rates and allocations are final), then
	re-derives the two amounts that depend on deductions.
	"""
	if doc.party_type != "Customer" or doc.payment_type != "Receive" or not doc.apply_tax_withholding_amount:
		return

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

	account = frappe.db.get_value(
		"Account",
		{"company": doc.company, "account_name": ASSET_ACCOUNT, "root_type": "Asset", "is_group": 0},
	)
	if not account:
		frappe.throw(_("Account {0} not found for Company {1}").format(ASSET_ACCOUNT, doc.company))

	amount, description = get_customer_withholding(doc)
	row = next((d for d in doc.deductions if d.account == account), None)

	if not amount:
		if row:
			doc.remove(row)
	else:
		if not row:
			row = doc.append("deductions", {"account": account})
		row.amount = amount
		row.description = description
		row.cost_center = doc.cost_center or erpnext.get_default_cost_center(doc.company)

	doc.set_unallocated_amount()
	doc.set_difference_amount()


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:
		description = next(iter(details_by_category.values())).description
	else:
		description = _("Withholding tax deducted by customer")
	return flt(total, precision), description