Add Phase 10 controlled Tally purchase posting

2026-08-22 16:12:22 +05:30
parent 82771844fd
commit 4b0ac5ce2b
12 changed files with 1590 additions and 3 deletions
@@ -1 +1 @@
-__version__ = "1.12.0"
+__version__ = "1.13.0"
@@ -1,6 +1,7 @@
 from __future__ import annotations

 from datetime import datetime, timezone
+import re
 from typing import Any

 from . import __version__
@@ -38,6 +39,10 @@ class AgentCommandProcessor:
                result = self._sync_transactions(payload)
            elif action == "accounting_historical_evidence":
                result = self._historical_evidence(payload)
+            elif action == "accounting_purchase_posting_preflight":
+                result = self._purchase_posting_preflight(payload)
+            elif action == "accounting_post_purchase_voucher":
+                result = self._post_purchase_voucher(payload)
            elif action == "accounting_depreciation_preview":
                result = self._depreciation_preview(payload)
            elif action == "accounting_calculate_it_depreciation":
@@ -70,6 +75,8 @@ class AgentCommandProcessor:
            "tally_mapping_capability": True, "tally_master_sync_capability": True,
            "tally_transaction_sync_capability": True,
            "historical_learning_read_capability": True,
+            "purchase_posting_preflight_capability": True,
+            "purchase_voucher_write_capability": True,
            "it_depreciation_capability": True,
            "tally_writeback_capability": True,
        }
@@ -214,6 +221,154 @@ class AgentCommandProcessor:
            voucher_scope=str(payload.get("voucher_scope") or "purchase"),
        )

+
+    def _resolve_open_company(self, payload: dict[str, Any]) -> tuple[dict[str, Any], str]:
+        requested_guid = str(payload.get("tally_guid") or "").strip()
+        if not requested_guid:
+            raise ValueError("Mapped Tally company GUID is required.")
+        tally_status = self.tally.status()
+        if not tally_status.get("connected"):
+            raise ValueError(str(tally_status.get("error") or "TallyPrime is not connected."))
+        company = next(
+            (
+                row for row in (tally_status.get("companies") or [])
+                if str(row.get("guid") or "").strip() == requested_guid
+            ),
+            None,
+        )
+        if not company:
+            raise ValueError(
+                "The mapped Tally company is not currently open on this workstation. "
+                "Open the correct company and run preflight again."
+            )
+        company_name = str(company.get("name") or "").strip()
+        if not company_name:
+            raise ValueError("Tally returned an invalid company name.")
+        expected_name = str(payload.get("company_name") or "").strip()
+        if expected_name and expected_name.casefold() != company_name.casefold():
+            raise ValueError(
+                f"Open-company mismatch: ERP expected '{expected_name}' but Tally reports '{company_name}'."
+            )
+        return company, company_name
+
+    def _purchase_posting_preflight(self, payload: dict[str, Any]) -> dict[str, Any]:
+        client_id = int(payload.get("client_id"))
+        company, company_name = self._resolve_open_company(payload)
+        result = self.tally.purchase_posting_preflight(
+            company_name,
+            supplier_gstin=str(payload.get("supplier_gstin") or ""),
+            supplier_name=str(payload.get("supplier_name") or ""),
+            purchase_ledger_name=str(payload.get("purchase_ledger_name") or ""),
+            invoice_number=str(payload.get("invoice_number") or ""),
+            invoice_date=str(payload.get("invoice_date") or ""),
+            invoice_value=float(payload.get("invoice_value") or 0),
+        )
+        return {
+            **result,
+            "company_name": company_name,
+            "company_guid": str(company.get("guid") or ""),
+            "company_gstin": str(company.get("gstin") or ""),
+            "client_id": client_id,
+            "agent": self._agent_info(),
+        }
+
+    def _post_purchase_voucher(self, payload: dict[str, Any]) -> dict[str, Any]:
+        client_id = int(payload.get("client_id"))
+        company, company_name = self._resolve_open_company(payload)
+
+        invoice_number = str(payload.get("invoice_number") or "").strip()
+        invoice_date = str(payload.get("invoice_date") or "").strip()
+        party = str(payload.get("party_ledger_name") or "").strip()
+        purchase_ledger = str(payload.get("purchase_ledger_name") or "").strip()
+
+        rows = self.tally.purchase_posting_ledgers(company_name)
+        for label, ledger in (
+            ("creditor", party),
+            ("purchase/expense", purchase_ledger),
+            ("Input IGST", str(payload.get("input_igst_ledger_name") or "").strip()),
+            ("Input CGST", str(payload.get("input_cgst_ledger_name") or "").strip()),
+            ("Input SGST", str(payload.get("input_sgst_ledger_name") or "").strip()),
+            ("Input Cess", str(payload.get("input_cess_ledger_name") or "").strip()),
+        ):
+            if ledger and not self.tally._ledger_exists(rows, ledger):
+                raise ValueError(
+                    f"{label} ledger '{ledger}' is not present in the currently opened Tally company."
+                )
+
+        expected_gstin = re.sub(
+            r"[^A-Z0-9]", "", str(payload.get("supplier_gstin") or "").upper()
+        )[:15]
+        party_row = next(
+            (r for r in rows if str(r.get("name") or "").strip().casefold() == party.casefold()),
+            None,
+        )
+        if not party_row:
+            raise ValueError("Selected creditor ledger was not found during final validation.")
+        party_gstin = str(party_row.get("gstin") or "").strip().upper()
+        if expected_gstin and party_gstin and expected_gstin != party_gstin:
+            raise ValueError(
+                f"Supplier GSTIN mismatch: ERP has {expected_gstin}, "
+                f"but Tally ledger '{party}' reports {party_gstin}."
+            )
+
+        duplicates = self.tally.find_purchase_duplicates(
+            company_name,
+            invoice_date=invoice_date,
+            reference=invoice_number,
+            party_ledger_names=[party],
+            invoice_value=float(payload.get("invoice_value") or 0),
+        )
+        if duplicates:
+            raise ValueError(
+                "Possible duplicate Purchase voucher already exists in Tally for this "
+                "supplier/invoice/date. Posting has been blocked."
+            )
+
+        tally_result = self.tally.post_purchase_voucher(
+            company_name,
+            voucher_date=invoice_date,
+            party_ledger_name=party,
+            purchase_ledger_name=purchase_ledger,
+            taxable_value=float(payload.get("taxable_value") or 0),
+            input_igst_ledger_name=str(payload.get("input_igst_ledger_name") or "").strip(),
+            igst=float(payload.get("igst") or 0),
+            input_cgst_ledger_name=str(payload.get("input_cgst_ledger_name") or "").strip(),
+            cgst=float(payload.get("cgst") or 0),
+            input_sgst_ledger_name=str(payload.get("input_sgst_ledger_name") or "").strip(),
+            sgst=float(payload.get("sgst") or 0),
+            input_cess_ledger_name=str(payload.get("input_cess_ledger_name") or "").strip(),
+            cess=float(payload.get("cess") or 0),
+            reference=str(payload.get("reference") or invoice_number),
+            narration=str(payload.get("narration") or ""),
+        )
+
+        safe_result = {
+            k: v for k, v in tally_result.items()
+            if k not in {"raw_response", "request_xml"}
+        }
+        self.logger.info(
+            "CONTROLLED PURCHASE WRITEBACK posted client_id=%s company=%s "
+            "erp_purchase_id=%s invoice=%s party=%s tally_voucher=%s",
+            client_id,
+            company_name,
+            payload.get("erp_purchase_id"),
+            invoice_number,
+            party,
+            safe_result.get("last_voucher_id"),
+        )
+        return {
+            "posted": True,
+            "client_id": client_id,
+            "erp_purchase_id": payload.get("erp_purchase_id"),
+            "company_name": company_name,
+            "company_guid": str(company.get("guid") or ""),
+            "reference": str(payload.get("reference") or invoice_number),
+            "party_ledger_name": party,
+            "purchase_ledger_name": purchase_ledger,
+            "tally_result": safe_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()
@@ -439,6 +439,405 @@ class TallyLiveConnector:
            "request_xml": request_xml,
        }

+
+    def purchase_posting_ledgers(self, company_name: str) -> list[dict]:
+        company = str(company_name or "").strip()
+        if not company:
+            raise ValueError("Tally company name is required.")
+        xml = f"""<ENVELOPE>
+  <HEADER><VERSION>1</VERSION><TALLYREQUEST>Export</TALLYREQUEST><TYPE>Collection</TYPE><ID>ARRRPurchasePostingLedgers</ID></HEADER>
+  <BODY><DESC>
+    <STATICVARIABLES>{self._static_variables(company)}</STATICVARIABLES>
+    <TDL><TDLMESSAGE>
+      <COLLECTION NAME="ARRRPurchasePostingLedgers" ISMODIFY="No">
+        <TYPE>Ledger</TYPE>
+        <FETCH>Name,GUID,Parent,PartyGSTIN,GSTIN,GSTRegistrationNumber,TaxType,GSTApplicable,GSTRegistrationType,IsBillWiseOn</FETCH>
+      </COLLECTION>
+    </TDLMESSAGE></TDL>
+  </DESC></BODY>
+</ENVELOPE>"""
+        raw = self._post_xml(xml)
+        cleaned = _clean_xml_response(raw)
+        try:
+            root = ET.fromstring(cleaned.encode("utf-8"))
+        except Exception as exc:
+            raise ValueError(
+                f"Tally returned invalid XML while validating purchase ledgers: {exc}"
+            ) from exc
+
+        rows = []
+        seen = set()
+        for element in root.iter():
+            if _tag(element) != "LEDGER":
+                continue
+            name = (
+                element.attrib.get("NAME")
+                or element.attrib.get("name")
+                or _child_text(element, "NAME")
+            ).strip()
+            if not name or name.casefold() in seen:
+                continue
+            seen.add(name.casefold())
+            gstin = _first_text(
+                element,
+                ("PARTYGSTIN", "GSTIN", "GSTREGISTRATIONNUMBER", "GSTNO", "GSTNUMBER"),
+            )
+            rows.append({
+                "name": name,
+                "guid": _child_text(element, "GUID"),
+                "parent": _child_text(element, "PARENT"),
+                "gstin": re.sub(r"[^A-Z0-9]", "", str(gstin or "").upper())[:15],
+                "tax_type": _child_text(element, "TAXTYPE"),
+                "gst_applicable": _child_text(element, "GSTAPPLICABLE"),
+                "gst_registration_type": _child_text(element, "GSTREGISTRATIONTYPE"),
+                "is_billwise_on": _child_text(element, "ISBILLWISEON"),
+            })
+        return rows
+
+    @staticmethod
+    def _ledger_exists(rows: list[dict], name: str) -> bool:
+        key = str(name or "").strip().casefold()
+        return bool(
+            key
+            and any(
+                str(row.get("name") or "").strip().casefold() == key
+                for row in rows
+            )
+        )
+
+    @staticmethod
+    def _tax_ledger_candidates(rows: list[dict], tax: str) -> list[dict]:
+        target = str(tax or "").strip().upper()
+        patterns = {
+            "IGST": ("IGST", "INTEGRATED"),
+            "CGST": ("CGST", "CENTRAL"),
+            "SGST": ("SGST", "STATE GST", "STATE TAX", "UTGST", "UT GST"),
+            "CESS": ("CESS",),
+        }.get(target, (target,))
+        matches = []
+        for row in rows:
+            text = " ".join([
+                str(row.get("name") or ""),
+                str(row.get("parent") or ""),
+                str(row.get("tax_type") or ""),
+            ]).upper()
+            if any(pattern in text for pattern in patterns):
+                matches.append(row)
+        matches.sort(
+            key=lambda row: (
+                0
+                if any(
+                    word in str(row.get("name") or "").upper()
+                    for word in ("INPUT", "PURCHASE")
+                )
+                else 1,
+                str(row.get("name") or "").casefold(),
+            )
+        )
+        return matches[:20]
+
+    def purchase_posting_preflight(
+        self,
+        company_name: str,
+        *,
+        supplier_gstin: str,
+        supplier_name: str,
+        purchase_ledger_name: str,
+        invoice_number: str,
+        invoice_date: str,
+        invoice_value: float,
+    ) -> dict:
+        rows = self.purchase_posting_ledgers(company_name)
+        gstin = re.sub(
+            r"[^A-Z0-9]", "", str(supplier_gstin or "").upper()
+        )[:15]
+        supplier_key = re.sub(
+            r"[^A-Z0-9]", "", str(supplier_name or "").upper()
+        )
+
+        gstin_matches = [
+            row for row in rows if gstin and row.get("gstin") == gstin
+        ]
+        name_matches = []
+        if supplier_key:
+            for row in rows:
+                key = re.sub(
+                    r"[^A-Z0-9]", "", str(row.get("name") or "").upper()
+                )
+                if key == supplier_key:
+                    name_matches.append(row)
+
+        party = gstin_matches or name_matches
+        if not party and supplier_key:
+            party = [
+                row
+                for row in rows
+                if supplier_key
+                in re.sub(r"[^A-Z0-9]", "", str(row.get("name") or "").upper())
+                or re.sub(
+                    r"[^A-Z0-9]", "", str(row.get("name") or "").upper()
+                )
+                in supplier_key
+            ][:20]
+
+        duplicates = self.find_purchase_duplicates(
+            company_name,
+            invoice_date=invoice_date,
+            reference=invoice_number,
+            party_ledger_names=[
+                str(row.get("name") or "") for row in party
+            ],
+            invoice_value=invoice_value,
+        )
+
+        return {
+            "party_candidates": party[:20],
+            "purchase_ledger_verified": self._ledger_exists(
+                rows, purchase_ledger_name
+            ),
+            "igst_candidates": self._tax_ledger_candidates(rows, "IGST"),
+            "cgst_candidates": self._tax_ledger_candidates(rows, "CGST"),
+            "sgst_candidates": self._tax_ledger_candidates(rows, "SGST"),
+            "cess_candidates": self._tax_ledger_candidates(rows, "CESS"),
+            "duplicate_candidates": duplicates,
+            "ledger_count": len(rows),
+        }
+
+    def find_purchase_duplicates(
+        self,
+        company_name: str,
+        *,
+        invoice_date: str,
+        reference: str,
+        party_ledger_names: list[str] | None = None,
+        invoice_value: float = 0,
+    ) -> list[dict]:
+        target_ref = re.sub(
+            r"[^A-Z0-9]", "", str(reference or "").upper()
+        )
+        parties = {
+            str(value or "").strip().casefold()
+            for value in (party_ledger_names or [])
+            if str(value or "").strip()
+        }
+        expected = round(abs(float(invoice_value or 0)), 2)
+        if not invoice_date:
+            return []
+
+        vouchers = self.export_vouchers(
+            company_name, invoice_date, invoice_date
+        )
+        result = []
+        for voucher in vouchers:
+            if "PURCHASE" not in str(
+                voucher.get("voucher_type_name") or ""
+            ).upper():
+                continue
+            ref = re.sub(
+                r"[^A-Z0-9]",
+                "",
+                str(voucher.get("reference") or "").upper(),
+            )
+            party = str(voucher.get("party_ledger_name") or "").strip()
+            if target_ref and ref != target_ref:
+                continue
+            if parties and party.casefold() not in parties:
+                continue
+            party_amount = 0.0
+            for entry in voucher.get("ledger_entries") or []:
+                if (
+                    str(entry.get("ledger_name") or "")
+                    .strip()
+                    .casefold()
+                    == party.casefold()
+                ):
+                    party_amount = max(
+                        party_amount,
+                        abs(float(entry.get("amount") or 0)),
+                    )
+            if (
+                expected
+                and party_amount
+                and abs(party_amount - expected)
+                > max(2.0, expected * 0.005)
+            ):
+                continue
+            result.append({
+                "guid": voucher.get("guid") or "",
+                "voucher_number": voucher.get("voucher_number") or "",
+                "reference": voucher.get("reference") or "",
+                "party_ledger_name": party,
+                "date": voucher.get("date") or "",
+                "amount": party_amount,
+            })
+        return result[:20]
+
+    def build_purchase_import_xml(
+        self,
+        company_name: str,
+        *,
+        voucher_date: str,
+        party_ledger_name: str,
+        purchase_ledger_name: str,
+        taxable_value: float,
+        input_igst_ledger_name: str = "",
+        igst: float = 0,
+        input_cgst_ledger_name: str = "",
+        cgst: float = 0,
+        input_sgst_ledger_name: str = "",
+        sgst: float = 0,
+        input_cess_ledger_name: str = "",
+        cess: float = 0,
+        reference: str = "",
+        narration: str = "",
+    ) -> str:
+        company = str(company_name or "").strip()
+        party = str(party_ledger_name or "").strip()
+        purchase = str(purchase_ledger_name or "").strip()
+
+        if not company or not party or not purchase:
+            raise ValueError(
+                "Company, creditor ledger and purchase/expense ledger are required."
+            )
+
+        taxable = round(abs(float(taxable_value or 0)), 2)
+        taxes = [
+            (
+                str(input_igst_ledger_name or "").strip(),
+                round(abs(float(igst or 0)), 2),
+                "IGST",
+            ),
+            (
+                str(input_cgst_ledger_name or "").strip(),
+                round(abs(float(cgst or 0)), 2),
+                "CGST",
+            ),
+            (
+                str(input_sgst_ledger_name or "").strip(),
+                round(abs(float(sgst or 0)), 2),
+                "SGST",
+            ),
+            (
+                str(input_cess_ledger_name or "").strip(),
+                round(abs(float(cess or 0)), 2),
+                "Cess",
+            ),
+        ]
+
+        if taxable <= 0:
+            raise ValueError("Taxable value must be greater than zero.")
+        for ledger, amount, label in taxes:
+            if amount > 0 and not ledger:
+                raise ValueError(
+                    f"{label} amount exists but no input tax ledger was selected."
+                )
+
+        total = round(
+            taxable + sum(amount for _, amount, _ in taxes), 2
+        )
+        dt = _tally_date(voucher_date)
+        ref_xml = (
+            f"<REFERENCE>{self._xml_escape(reference)}</REFERENCE>"
+            if str(reference or "").strip()
+            else ""
+        )
+
+        debit_entries = [
+            f"""<ALLLEDGERENTRIES.LIST>
+        <LEDGERNAME>{self._xml_escape(purchase)}</LEDGERNAME>
+        <ISDEEMEDPOSITIVE>Yes</ISDEEMEDPOSITIVE>
+        <AMOUNT>-{taxable:.2f}</AMOUNT>
+      </ALLLEDGERENTRIES.LIST>"""
+        ]
+        for ledger, amount, _ in taxes:
+            if amount > 0:
+                debit_entries.append(
+                    f"""<ALLLEDGERENTRIES.LIST>
+        <LEDGERNAME>{self._xml_escape(ledger)}</LEDGERNAME>
+        <ISDEEMEDPOSITIVE>Yes</ISDEEMEDPOSITIVE>
+        <AMOUNT>-{amount:.2f}</AMOUNT>
+      </ALLLEDGERENTRIES.LIST>"""
+                )
+
+        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="Purchase" ACTION="Create" OBJVIEW="Accounting Voucher View">
+      <DATE>{dt}</DATE>
+      <VOUCHERTYPENAME>Purchase</VOUCHERTYPENAME>
+      <PERSISTEDVIEW>Accounting Voucher View</PERSISTEDVIEW>
+      <PARTYLEDGERNAME>{self._xml_escape(party)}</PARTYLEDGERNAME>
+      {ref_xml}
+      <NARRATION>{self._xml_escape(narration)}</NARRATION>
+      {''.join(debit_entries)}
+      <ALLLEDGERENTRIES.LIST>
+        <LEDGERNAME>{self._xml_escape(party)}</LEDGERNAME>
+        <ISDEEMEDPOSITIVE>No</ISDEEMEDPOSITIVE>
+        <AMOUNT>{total:.2f}</AMOUNT>
+        <BILLALLOCATIONS.LIST>
+          <NAME>{self._xml_escape(reference or 'New Ref')}</NAME>
+          <BILLTYPE>New Ref</BILLTYPE>
+          <AMOUNT>{total:.2f}</AMOUNT>
+        </BILLALLOCATIONS.LIST>
+      </ALLLEDGERENTRIES.LIST>
+    </VOUCHER>
+  </TALLYMESSAGE></DATA></BODY>
+</ENVELOPE>"""
+
+    def post_purchase_voucher(
+        self, company_name: str, **kwargs
+    ) -> dict:
+        request_xml = self.build_purchase_import_xml(
+            company_name, **kwargs
+        )
+        response_xml = self._post_xml(request_xml)
+        cleaned = _clean_xml_response(response_xml)
+
+        try:
+            root = ET.fromstring(cleaned.encode("utf-8"))
+        except Exception as exc:
+            raise ValueError(
+                f"Tally returned invalid XML after Purchase import: {exc}"
+            ) from exc
+
+        def first_int(tag_name: str) -> int:
+            text = _first_text(root, [tag_name])
+            try:
+                return int(float(str(text or "0")))
+            except Exception:
+                return 0
+
+        created = first_int("CREATED")
+        altered = first_int("ALTERED")
+        errors = first_int("ERRORS")
+        last_voucher_id = _first_text(
+            root,
+            ["LASTVCHID", "LASTMID", "LASTVOUCHERID"],
+        )
+        line_error = _first_text(root, ["LINEERROR"])
+
+        if errors > 0 or line_error:
+            raise ValueError(
+                line_error
+                or f"Tally reported {errors} error(s) while creating the Purchase voucher."
+            )
+        if created < 1 and altered < 1:
+            raise ValueError(
+                "Tally did not confirm creation of the Purchase voucher."
+            )
+
+        return {
+            "created": created,
+            "altered": altered,
+            "errors": errors,
+            "last_voucher_id": str(last_voucher_id or ""),
+            "raw_response": response_xml,
+            "request_xml": request_xml,
+        }
+
    @staticmethod
    def _parse_vouchers(xml_text: str) -> list[dict]:
        cleaned = _clean_xml_response(xml_text)
@@ -190,7 +190,7 @@ class StorageAgentTunnel:

    def _capabilities(self) -> list[str]:
        return [
-            "storage", "local_dashboard", "manual_updates", "historical_learning_read", "accounting_act", "agent.jobs.v2",
+            "storage", "local_dashboard", "manual_updates", "historical_learning_read", "accounting_act", "agent.jobs.v2", "purchase_posting.preflight", "tally.purchase.write",
            "tally.status", "tally.company_identity", "tally.read_groups",
            "tally.read_ledgers", "tally.read_stock_items", "tally.read_vouchers",
            "tally.read_trial_balance", "tally.master_sync", "tally.transaction_sync",