Add Phase 11 bank to Tally with multi-bank contra intelligence

2026-08-22 19:59:40 +05:30
parent 215ad2c6c6
commit 8748ee9539
16 changed files with 1140 additions and 28 deletions
@@ -1,2 +1,2 @@
-__version__ = "1.13.1"
+__version__ = "1.14.0"
 AGENT_NAME = "ERP Local Agent"
@@ -43,6 +43,10 @@ class AgentCommandProcessor:
                result = self._purchase_posting_preflight(payload)
            elif action == "accounting_post_purchase_voucher":
                result = self._post_purchase_voucher(payload)
+            elif action == "accounting_bank_posting_preflight":
+                result = self._bank_posting_preflight(payload)
+            elif action == "accounting_post_bank_voucher":
+                result = self._post_bank_voucher(payload)
            elif action == "accounting_depreciation_preview":
                result = self._depreciation_preview(payload)
            elif action == "accounting_calculate_it_depreciation":
@@ -369,6 +373,75 @@ class AgentCommandProcessor:
            "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)
+        duplicates = self.tally.find_bank_voucher_duplicates(
+            company_name,
+            voucher_date=str(payload.get("transaction_date") or ""),
+            voucher_type=str(payload.get("voucher_type") or ""),
+            reference=str(payload.get("reference") or ""),
+            amount=float(payload.get("amount") or 0),
+        )
+        return {
+            **ledgers,
+            "duplicate_candidates": duplicates,
+            "company_name": company_name,
+            "company_guid": str(company.get("guid") or ""),
+            "agent": self._agent_info(),
+        }
+
+    def _post_bank_voucher(self, payload: dict[str, Any]) -> dict[str, Any]:
+        company, company_name = self._resolve_open_company(payload)
+        ledgers = self.tally.bank_posting_ledgers(company_name)
+        all_names = {str(x.get("name") or "").strip().casefold() for x in ledgers.get("all_ledgers") or []}
+        bank_names = {str(x.get("name") or "").strip().casefold() for x in ledgers.get("bank_ledgers") or []}
+
+        bank = str(payload.get("bank_ledger_name") or "").strip()
+        counter = str(payload.get("counter_ledger_name") or "").strip()
+        other = str(payload.get("other_bank_ledger_name") or "").strip()
+        vtype = str(payload.get("voucher_type") or "").title()
+
+        if bank.casefold() not in bank_names:
+            raise ValueError("Selected bank ledger is not present under Tally bank accounts.")
+        if vtype == "Contra":
+            if other.casefold() not in bank_names or other.casefold() == bank.casefold():
+                raise ValueError("The other bank ledger is invalid for Contra.")
+        elif counter.casefold() not in all_names:
+            raise ValueError("Selected counter ledger is not present in Tally.")
+
+        duplicates = self.tally.find_bank_voucher_duplicates(
+            company_name,
+            voucher_date=str(payload.get("transaction_date") or ""),
+            voucher_type=vtype,
+            reference=str(payload.get("reference") or ""),
+            amount=float(payload.get("amount") or 0),
+        )
+        if duplicates:
+            raise ValueError("Possible duplicate bank voucher already exists in Tally. Posting blocked.")
+
+        result = self.tally.post_bank_voucher(
+            company_name,
+            voucher_type=vtype,
+            voucher_date=str(payload.get("transaction_date") or ""),
+            bank_ledger_name=bank,
+            counter_ledger_name=counter,
+            other_bank_ledger_name=other,
+            direction=str(payload.get("direction") or ""),
+            amount=float(payload.get("amount") or 0),
+            narration=str(payload.get("narration") or ""),
+            reference=str(payload.get("reference") or ""),
+        )
+        return {
+            "posted": True,
+            "company_name": company_name,
+            "company_guid": str(company.get("guid") or ""),
+            "reference": str(payload.get("reference") or ""),
+            "tally_result": result,
+            "agent": self._agent_info(),
+        }
+
    def _depreciation_preview(self, payload: dict[str, Any]) -> dict[str, Any]:
        client_id = int(payload.get("client_id"))
        tally_guid = str(payload.get("tally_guid") or "").strip()
@@ -838,6 +838,101 @@ class TallyLiveConnector:
            "request_xml": request_xml,
        }

+
+    def bank_posting_ledgers(self, company_name: str) -> dict:
+        rows = self.purchase_posting_ledgers(company_name)
+        bank_rows = [
+            row for row in rows
+            if "BANK" in str(row.get("parent") or "").upper()
+            or "BANK ACCOUNT" in str(row.get("parent") or "").upper()
+        ]
+        return {"bank_ledgers": bank_rows, "all_ledgers": rows}
+
+    def find_bank_voucher_duplicates(self, company_name: str, *, voucher_date: str, voucher_type: str, reference: str, amount: float) -> list[dict]:
+        target_ref = re.sub(r"[^A-Z0-9]", "", str(reference or "").upper())
+        expected = round(abs(float(amount or 0)), 2)
+        result = []
+        for voucher in self.export_vouchers(company_name, voucher_date, voucher_date):
+            if str(voucher.get("voucher_type_name") or "").strip().casefold() != str(voucher_type or "").strip().casefold():
+                continue
+            ref = re.sub(r"[^A-Z0-9]", "", str(voucher.get("reference") or "").upper())
+            if target_ref and ref and ref != target_ref:
+                continue
+            values = [abs(float(x.get("amount") or 0)) for x in voucher.get("ledger_entries") or []]
+            if expected and values and all(abs(v-expected) > max(1.0, expected*0.002) for v in values):
+                continue
+            result.append({
+                "guid": voucher.get("guid") or "",
+                "voucher_number": voucher.get("voucher_number") or "",
+                "reference": voucher.get("reference") or "",
+                "date": voucher.get("date") or "",
+            })
+        return result[:20]
+
+    def build_bank_voucher_xml(self, company_name: str, *, voucher_type: str, voucher_date: str, bank_ledger_name: str, counter_ledger_name: str = "", other_bank_ledger_name: str = "", direction: str = "", amount: float, narration: str = "", reference: str = "") -> str:
+        vtype = str(voucher_type or "").title()
+        if vtype not in {"Payment", "Receipt", "Contra"}:
+            raise ValueError("Unsupported bank voucher type.")
+        company = str(company_name or "").strip()
+        bank = str(bank_ledger_name or "").strip()
+        counter = str(counter_ledger_name or "").strip()
+        other_bank = str(other_bank_ledger_name or "").strip()
+        value = round(abs(float(amount or 0)), 2)
+        if not company or not bank or value <= 0:
+            raise ValueError("Company, bank ledger and positive amount are required.")
+        if vtype == "Contra" and not other_bank:
+            raise ValueError("Other bank ledger is required for Contra.")
+        if vtype != "Contra" and not counter:
+            raise ValueError("Counter ledger is required for Payment/Receipt.")
+
+        if vtype == "Payment":
+            debit_ledger, credit_ledger = counter, bank
+        elif vtype == "Receipt":
+            debit_ledger, credit_ledger = bank, counter
+        else:
+            # Row direction describes movement in the uploaded bank account.
+            if str(direction or "").lower() == "debit":
+                debit_ledger, credit_ledger = other_bank, bank
+            else:
+                debit_ledger, credit_ledger = bank, other_bank
+
+        ref_xml = f"<REFERENCE>{self._xml_escape(reference)}</REFERENCE>" if reference else ""
+        return f"""<ENVELOPE>
+  <HEADER><VERSION>1</VERSION><TALLYREQUEST>Import</TALLYREQUEST><TYPE>Data</TYPE><ID>Vouchers</ID></HEADER>
+  <BODY><DESC><STATICVARIABLES><SVCURRENTCOMPANY>{self._xml_escape(company)}</SVCURRENTCOMPANY></STATICVARIABLES></DESC>
+  <DATA><TALLYMESSAGE xmlns:UDF="TallyUDF">
+    <VOUCHER VCHTYPE="{vtype}" ACTION="Create" OBJVIEW="Accounting Voucher View">
+      <DATE>{_tally_date(voucher_date)}</DATE>
+      <VOUCHERTYPENAME>{vtype}</VOUCHERTYPENAME>
+      <PERSISTEDVIEW>Accounting Voucher View</PERSISTEDVIEW>
+      {ref_xml}
+      <NARRATION>{self._xml_escape(narration)}</NARRATION>
+      <ALLLEDGERENTRIES.LIST><LEDGERNAME>{self._xml_escape(debit_ledger)}</LEDGERNAME><ISDEEMEDPOSITIVE>Yes</ISDEEMEDPOSITIVE><AMOUNT>-{value:.2f}</AMOUNT></ALLLEDGERENTRIES.LIST>
+      <ALLLEDGERENTRIES.LIST><LEDGERNAME>{self._xml_escape(credit_ledger)}</LEDGERNAME><ISDEEMEDPOSITIVE>No</ISDEEMEDPOSITIVE><AMOUNT>{value:.2f}</AMOUNT></ALLLEDGERENTRIES.LIST>
+    </VOUCHER>
+  </TALLYMESSAGE></DATA></BODY>
+</ENVELOPE>"""
+
+    def post_bank_voucher(self, company_name: str, **kwargs) -> dict:
+        request_xml = self.build_bank_voucher_xml(company_name, **kwargs)
+        response_xml = self._post_xml(request_xml)
+        root = ET.fromstring(_clean_xml_response(response_xml).encode("utf-8"))
+        def _i(tag):
+            try: return int(float(_first_text(root, [tag]) or 0))
+            except Exception: return 0
+        errors = _i("ERRORS")
+        line_error = _first_text(root, ["LINEERROR"])
+        created = _i("CREATED")
+        if errors or line_error:
+            raise ValueError(line_error or f"Tally reported {errors} bank voucher error(s).")
+        if created < 1 and _i("ALTERED") < 1:
+            raise ValueError("Tally did not confirm creation of the bank voucher.")
+        return {
+            "created": created,
+            "errors": errors,
+            "last_voucher_id": str(_first_text(root, ["LASTVCHID","LASTMID","LASTVOUCHERID"]) or ""),
+        }
+
    @staticmethod
    def _parse_vouchers(xml_text: str) -> list[dict]:
        cleaned = _clean_xml_response(xml_text)