Add TDS compliance liability reconciliation and rule master tools
This commit is contained in:
A R R R Associates
@@ -1,2 +1,2 @@
|
||||
__version__ = "1.22.4"
|
||||
__version__ = "1.22.5"
|
||||
AGENT_NAME = "ERP Local Agent"
|
||||
|
||||
@@ -58,6 +58,10 @@ class AgentCommandProcessor:
|
||||
result = self._bank_reconciliation_extract(payload)
|
||||
elif action == "accounting_cash_payment_compliance":
|
||||
result = self._cash_payment_compliance(payload)
|
||||
elif action == "accounting_tds_compliance":
|
||||
result = self._tds_compliance(payload)
|
||||
elif action == "accounting_post_tds_liability":
|
||||
result = self._post_tds_liability(payload)
|
||||
elif action == "accounting_post_bank_voucher":
|
||||
result = self._post_bank_voucher(payload)
|
||||
elif action == "accounting_native_voucher_preflight":
|
||||
@@ -799,6 +803,84 @@ class AgentCommandProcessor:
|
||||
"agent": self._agent_info(),
|
||||
}
|
||||
|
||||
def _tds_compliance(self, payload: dict[str, Any]) -> dict[str, Any]:
|
||||
from collections import defaultdict
|
||||
from datetime import date as _date
|
||||
|
||||
company, company_name = self._resolve_open_company(payload)
|
||||
date_from = str(payload.get("date_from") or "").strip()
|
||||
date_to = str(payload.get("date_to") or "").strip()
|
||||
rules = list(payload.get("rules") or [])
|
||||
if not rules:
|
||||
raise ValueError("No active TDS rules were supplied by ERP.")
|
||||
vouchers = self.tally.export_vouchers(company_name, date_from, date_to)
|
||||
# TDS ledgers are identified by both ledger name and parent/group text; this is deliberately broader than one fixed ledger name.
|
||||
masters = self.tally.export_master_collection(company_name, "ledgers")
|
||||
tds_ledger_names = set()
|
||||
for led in masters:
|
||||
name = str(led.get("name") or "").strip()
|
||||
parent = str(led.get("parent") or "").strip().casefold()
|
||||
text = (name + " " + parent).casefold()
|
||||
if name and ("tds" in text or "tax deducted" in text or "duties & taxes" in parent and "income tax" in text):
|
||||
tds_ledger_names.add(name.casefold())
|
||||
|
||||
candidates = []
|
||||
for v in vouchers:
|
||||
if str(v.get("is_cancelled") or "").lower() in {"yes","true","1"} or str(v.get("is_optional") or "").lower() in {"yes","true","1"}:
|
||||
continue
|
||||
entries = list(v.get("ledger_entries") or [])
|
||||
actual_tds = round(sum(abs(float(e.get("amount") or 0)) for e in entries if str(e.get("ledger_name") or "").strip().casefold() in tds_ledger_names), 2)
|
||||
party = str(v.get("party_ledger_name") or "").strip()
|
||||
non_tds = [e for e in entries if str(e.get("ledger_name") or "").strip().casefold() not in tds_ledger_names]
|
||||
for e in non_tds:
|
||||
lname = str(e.get("ledger_name") or "").strip()
|
||||
amount = abs(float(e.get("amount") or 0))
|
||||
if not lname or amount <= 0 or (party and lname.casefold() == party.casefold()):
|
||||
continue
|
||||
text = lname.casefold()
|
||||
matches=[]
|
||||
for r in rules:
|
||||
kws=[x.strip().casefold() for x in str(r.get("nature_keywords") or "").split(',') if x.strip()]
|
||||
if any(k in text for k in kws): matches.append(r)
|
||||
if not matches: continue
|
||||
# More specific/longer keyword match first; user can refine keywords in Rule Master.
|
||||
rule=max(matches,key=lambda r:max([len(k.strip()) for k in str(r.get("nature_keywords") or "").split(',') if k.strip()] or [0]))
|
||||
candidates.append({"date":str(v.get("date") or v.get("effective_date") or ""),"party":party or "Unidentified party","nature_ledger":lname,"taxable_base":round(amount,2),"actual_tds":actual_tds,"voucher_number":str(v.get("voucher_number") or ""),"voucher_type":str(v.get("voucher_type_name") or ""),"reference":str(v.get("reference") or ""),"rule":rule})
|
||||
|
||||
by_party_rule=defaultdict(list)
|
||||
for x in candidates: by_party_rule[(x["party"].casefold(),str(x["rule"].get("rule_code")))].append(x)
|
||||
transactions=[]
|
||||
for (_party, _code), rows in by_party_rule.items():
|
||||
rows.sort(key=lambda x:x["date"])
|
||||
cumulative=0.0; monthly=defaultdict(float)
|
||||
for x in rows:
|
||||
r=x["rule"]; base=x["taxable_base"]; cumulative=round(cumulative+base,2); month=str(x["date"] or "")[:7]; monthly[month]=round(monthly[month]+base,2)
|
||||
mode=str(r.get("threshold_mode") or "ANNUAL").upper(); single=float(r.get("single_threshold") or 0); agg=float(r.get("aggregate_threshold") or 0)
|
||||
applicable=False; taxable_for_tds=base
|
||||
if mode=="SINGLE": applicable=base>single
|
||||
elif mode=="MONTHLY": applicable=monthly[month]>single
|
||||
elif mode=="BOTH": applicable=(single>0 and base>single) or (agg>0 and cumulative>agg)
|
||||
else: applicable=(agg<=0 or cumulative>agg)
|
||||
if applicable and bool(r.get("deduct_on_excess_only")) and agg>0:
|
||||
prior=max(0.0,cumulative-base); taxable_for_tds=max(0.0,cumulative-max(prior,agg)); taxable_for_tds=min(base,taxable_for_tds)
|
||||
rate=float(r.get("rate_other") or r.get("rate_individual_huf") or 0)
|
||||
expected=round(taxable_for_tds*rate/100.0,2) if applicable else 0.0
|
||||
actual=min(round(float(x["actual_tds"] or 0),2), expected if expected>0 else float(x["actual_tds"] or 0))
|
||||
liability=round(max(0.0,expected-actual),2)
|
||||
status="OK" if expected>0 and liability<=0.009 else ("TDS NOT DEDUCTED" if expected>0 and actual<=0.009 else ("SHORT DEDUCTION" if liability>0 else "BELOW / OUTSIDE THRESHOLD"))
|
||||
transactions.append({**{k:v for k,v in x.items() if k!='rule'},"rule_id":r.get("id"),"rule_code":r.get("rule_code"),"rule_name":r.get("name"),"legacy_section":r.get("legacy_section"),"statutory_reference":r.get("statutory_reference"),"cumulative_amount":cumulative,"expected_tds":expected,"actual_tds":actual,"liability":liability,"status":status})
|
||||
return {"tds_review":{"company_name":company_name,"company_guid":str(company.get("guid") or ""),"date_from":date_from,"date_to":date_to,"summary":{"vouchers_reviewed":len(vouchers),"candidate_transactions":len(transactions),"tds_not_deducted":sum(1 for x in transactions if x["status"]=="TDS NOT DEDUCTED"),"short_deduction":sum(1 for x in transactions if x["status"]=="SHORT DEDUCTION"),"purchase_transactions":sum(1 for x in transactions if x["rule_code"]=="PURCHASE_GOODS")},"transactions":transactions,"tds_ledgers":sorted(tds_ledger_names),"read_only":True},"agent":self._agent_info()}
|
||||
|
||||
def _post_tds_liability(self, payload: dict[str, Any]) -> dict[str, Any]:
|
||||
company, company_name = self._resolve_open_company(payload)
|
||||
voucher_date=str(payload.get("voucher_date") or "").strip(); party=str(payload.get("party_ledger") or "").strip(); tds=str(payload.get("tds_ledger") or "").strip(); amount=round(float(payload.get("amount") or 0),2); ref=str(payload.get("reference") or "").strip()
|
||||
if not voucher_date or not party or not tds or amount<=0 or not ref: raise ValueError("Voucher date, party ledger, TDS ledger, positive amount and ERP reference are required.")
|
||||
for v in self.tally.export_vouchers(company_name,voucher_date,voucher_date):
|
||||
if ref and ref in {str(v.get("reference") or "").strip(), str(v.get("narration") or "").strip()}:
|
||||
raise ValueError(f"TDS entry {ref} already exists in Tally; duplicate posting blocked.")
|
||||
result=self.tally.post_journal_voucher(company_name,voucher_date=voucher_date,debit_ledger=party,credit_ledger=tds,amount=amount,narration=str(payload.get("narration") or f"TDS liability {ref}"),reference=ref)
|
||||
return {"posted":True,"company_name":company_name,"reference":ref,"last_voucher_id":result.get("last_voucher_id"),"result":result,"agent":self._agent_info()}
|
||||
|
||||
def _bank_posting_preflight(self, payload: dict[str, Any]) -> dict[str, Any]:
|
||||
company, company_name = self._resolve_open_company(payload)
|
||||
ledgers = self.tally.bank_posting_ledgers(company_name)
|
||||
|
||||
Reference in New Issue
Block a user