Add Phase 6 controlled Tally writeback

2026-08-19 22:33:15 +05:30
parent 66d286fd55
commit 193fd4b71c
11 changed files with 464 additions and 21 deletions
@@ -1,2 +1,2 @@
-__version__ = "1.6.0"
+__version__ = "1.7.0"
 AGENT_NAME = "ERP Local Agent"
@@ -7,7 +7,7 @@ import sqlite3
 from typing import Sequence


-SCHEMA_VERSION = "5"
+SCHEMA_VERSION = "6"

 MASTER_TABLES = {
    "groups": "tally_groups",
@@ -32,8 +32,8 @@ class LocalAccountingStore:
    Phase 3 preserves Phase 1/2 metadata and mappings and adds company-scoped
    read-only Tally master snapshots. Phase 4 adds date-scoped vouchers and their
    ledger, inventory, bill, cost-centre and bank allocations. Phase 5 adds
-    reviewable Income-tax depreciation rate rules and draft calculations only.
-    Existing .act files are upgraded in place and no Tally write-back occurs.
+    reviewable Income-tax depreciation calculations. Phase 6 adds controlled Partner approval and audited Tally write-back for approved depreciation journals.
+    Existing .act files are upgraded in place.
    """

    def __init__(self, storage_root: Path):
@@ -121,6 +121,24 @@ class LocalAccountingStore:
        CREATE INDEX IF NOT EXISTS ix_{table}_master_guid ON {table}(tally_guid, master_guid);
        """

+    @staticmethod
+    def _ensure_phase6_columns(db: sqlite3.Connection) -> None:
+        columns = {row["name"] for row in db.execute("PRAGMA table_info(it_depreciation_runs)").fetchall()}
+        additions = {
+            "approved_by_user_id": "INTEGER",
+            "approved_at_utc": "TEXT",
+            "approval_note": "TEXT NOT NULL DEFAULT ''",
+            "posted_by_user_id": "INTEGER",
+            "posted_at_utc": "TEXT",
+            "posted_voucher_type": "TEXT NOT NULL DEFAULT ''",
+            "posted_voucher_number": "TEXT NOT NULL DEFAULT ''",
+            "tally_posting_status": "TEXT NOT NULL DEFAULT ''",
+            "tally_posting_error": "TEXT NOT NULL DEFAULT ''",
+        }
+        for name, ddl in additions.items():
+            if name not in columns:
+                db.execute(f"ALTER TABLE it_depreciation_runs ADD COLUMN {name} {ddl}")
+
    def initialize(self, client_id: int, client_name: str = "", tenant_id: int | None = None, created_by_user_id: int | None = None) -> Path:
        path = self.db_path(client_id)
        with self.connect(client_id) as db:
@@ -274,8 +292,36 @@ class LocalAccountingStore:
                    FOREIGN KEY(run_id) REFERENCES it_depreciation_runs(id) ON DELETE CASCADE
                );
                CREATE INDEX IF NOT EXISTS ix_it_dep_lines_run ON it_depreciation_run_lines(run_id, line_no);
+
+                CREATE TABLE IF NOT EXISTS tally_writeback_attempts (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    depreciation_run_id INTEGER NOT NULL,
+                    tally_guid TEXT NOT NULL,
+                    company_name TEXT NOT NULL,
+                    voucher_type TEXT NOT NULL DEFAULT 'Journal',
+                    voucher_date TEXT NOT NULL,
+                    reference TEXT NOT NULL DEFAULT '',
+                    debit_ledger TEXT NOT NULL,
+                    credit_ledger TEXT NOT NULL,
+                    amount REAL NOT NULL DEFAULT 0,
+                    requested_by_user_id INTEGER,
+                    started_at_utc TEXT NOT NULL,
+                    completed_at_utc TEXT,
+                    status TEXT NOT NULL,
+                    tally_created INTEGER NOT NULL DEFAULT 0,
+                    tally_altered INTEGER NOT NULL DEFAULT 0,
+                    tally_errors INTEGER NOT NULL DEFAULT 0,
+                    tally_last_voucher_id TEXT NOT NULL DEFAULT '',
+                    request_xml TEXT NOT NULL DEFAULT '',
+                    response_xml TEXT NOT NULL DEFAULT '',
+                    error_message TEXT,
+                    FOREIGN KEY(depreciation_run_id) REFERENCES it_depreciation_runs(id) ON DELETE CASCADE
+                );
+                CREATE INDEX IF NOT EXISTS ix_tally_writeback_run ON tally_writeback_attempts(depreciation_run_id, id);
+                CREATE INDEX IF NOT EXISTS ix_tally_writeback_company ON tally_writeback_attempts(tally_guid, started_at_utc);
                """
            )
+            self._ensure_phase6_columns(db)

            now = _utc_now_iso()
            meta = {
@@ -697,9 +743,121 @@ class LocalAccountingStore:
            run=db.execute("SELECT * FROM it_depreciation_runs WHERE id=?",(int(run_id),)).fetchone()
            if not run: raise ValueError("Depreciation draft was not found.")
            lines=db.execute("SELECT * FROM it_depreciation_run_lines WHERE run_id=? ORDER BY line_no",(int(run_id),)).fetchall()
-        result=dict(run); result["lines"]=[dict(x) for x in lines]; result["no_tally_writeback"]=True
+        result=dict(run); result["lines"]=[dict(x) for x in lines]
+        result["writeback_attempts"] = self.get_writeback_attempts(client_id, run_id) if "tally_writeback_attempts" else []
+        result["no_tally_writeback"] = False
        return result

+    def approve_it_depreciation(self, client_id: int, run_id: int, approved_by_user_id: int, approval_note: str = "") -> dict:
+        self.initialize(client_id)
+        now = _utc_now_iso()
+        with self.connect(client_id) as db:
+            run = db.execute("SELECT * FROM it_depreciation_runs WHERE id=?", (int(run_id),)).fetchone()
+            if not run:
+                raise ValueError("Depreciation draft was not found.")
+            status = str(run["status"] or "").strip().lower()
+            if status == "posted":
+                raise ValueError("This depreciation draft has already been posted to Tally.")
+            if status not in {"draft", "approved"}:
+                raise ValueError(f"Depreciation draft cannot be approved from status '{status}'.")
+            if float(run["total_depreciation"] or 0) <= 0:
+                raise ValueError("Depreciation amount must be greater than zero before approval.")
+            if not str(run["depreciation_expense_ledger"] or "").strip() or not str(run["depreciation_reserve_ledger"] or "").strip():
+                raise ValueError("Select both the depreciation expense ledger and the depreciation reserve/credit ledger before approval.")
+            db.execute(
+                """UPDATE it_depreciation_runs
+                   SET status='approved', approved_by_user_id=?, approved_at_utc=?, approval_note=?,
+                       tally_posting_status=CASE WHEN tally_posting_status='failed' THEN '' ELSE tally_posting_status END,
+                       tally_posting_error=''
+                   WHERE id=?""",
+                (int(approved_by_user_id), now, str(approval_note or "").strip(), int(run_id)),
+            )
+        return self.get_it_depreciation_run(client_id, run_id)
+
+    def begin_writeback_attempt(self, client_id: int, run_id: int, requested_by_user_id: int, *, voucher_date: str, reference: str, request_xml: str) -> dict:
+        self.initialize(client_id)
+        now = _utc_now_iso()
+        with self.connect(client_id) as db:
+            run = db.execute("SELECT * FROM it_depreciation_runs WHERE id=?", (int(run_id),)).fetchone()
+            if not run:
+                raise ValueError("Depreciation draft was not found.")
+            if str(run["status"] or "").strip().lower() != "approved":
+                raise ValueError("The depreciation draft must be approved before posting.")
+            successful = db.execute(
+                "SELECT id FROM tally_writeback_attempts WHERE depreciation_run_id=? AND status='posted' LIMIT 1",
+                (int(run_id),),
+            ).fetchone()
+            if successful:
+                raise ValueError("This depreciation draft has already been posted to Tally.")
+            cur = db.execute(
+                """INSERT INTO tally_writeback_attempts(
+                       depreciation_run_id,tally_guid,company_name,voucher_type,voucher_date,reference,
+                       debit_ledger,credit_ledger,amount,requested_by_user_id,started_at_utc,status,request_xml
+                   ) VALUES(?,?,?,?,?,?,?,?,?,?,?,'posting',?)""",
+                (
+                    int(run_id), str(run["tally_guid"] or ""), str(run["company_name"] or ""), "Journal",
+                    str(voucher_date), str(reference or ""), str(run["depreciation_expense_ledger"] or ""),
+                    str(run["depreciation_reserve_ledger"] or ""), float(run["total_depreciation"] or 0),
+                    int(requested_by_user_id), now, str(request_xml or ""),
+                ),
+            )
+            attempt_id = int(cur.lastrowid)
+            db.execute(
+                "UPDATE it_depreciation_runs SET tally_posting_status='posting', tally_posting_error='' WHERE id=?",
+                (int(run_id),),
+            )
+        return {"attempt_id": attempt_id, "run": self.get_it_depreciation_run(client_id, run_id)}
+
+    def finish_writeback_attempt(self, client_id: int, attempt_id: int, *, posted: bool, result: dict | None = None, response_xml: str = "", error_message: str = "", posted_by_user_id: int | None = None) -> dict:
+        self.initialize(client_id)
+        now = _utc_now_iso()
+        result = result or {}
+        with self.connect(client_id) as db:
+            attempt = db.execute("SELECT * FROM tally_writeback_attempts WHERE id=?", (int(attempt_id),)).fetchone()
+            if not attempt:
+                raise ValueError("Tally write-back attempt was not found.")
+            run_id = int(attempt["depreciation_run_id"])
+            status = "posted" if posted else "failed"
+            db.execute(
+                """UPDATE tally_writeback_attempts
+                   SET completed_at_utc=?, status=?, tally_created=?, tally_altered=?, tally_errors=?,
+                       tally_last_voucher_id=?, response_xml=?, error_message=?
+                   WHERE id=?""",
+                (
+                    now, status, int(result.get("created") or 0), int(result.get("altered") or 0),
+                    int(result.get("errors") or 0), str(result.get("last_voucher_id") or ""),
+                    str(response_xml or result.get("raw_response") or ""), str(error_message or ""), int(attempt_id),
+                ),
+            )
+            if posted:
+                db.execute(
+                    """UPDATE it_depreciation_runs
+                       SET status='posted', posted_by_user_id=?, posted_at_utc=?, posted_voucher_type='Journal',
+                           posted_voucher_number=?, tally_posting_status='posted', tally_posting_error=''
+                       WHERE id=?""",
+                    (int(posted_by_user_id) if posted_by_user_id is not None else None, now, str(result.get("last_voucher_id") or ""), run_id),
+                )
+            else:
+                db.execute(
+                    """UPDATE it_depreciation_runs
+                       SET status='approved', tally_posting_status='failed', tally_posting_error=?
+                       WHERE id=?""",
+                    (str(error_message or "Tally rejected the journal voucher."), run_id),
+                )
+        return self.get_it_depreciation_run(client_id, run_id)
+
+    def get_writeback_attempts(self, client_id: int, run_id: int) -> list[dict]:
+        self.initialize(client_id)
+        with self.connect(client_id) as db:
+            rows = db.execute(
+                """SELECT id,depreciation_run_id,tally_guid,company_name,voucher_type,voucher_date,reference,
+                          debit_ledger,credit_ledger,amount,requested_by_user_id,started_at_utc,completed_at_utc,status,
+                          tally_created,tally_altered,tally_errors,tally_last_voucher_id,error_message
+                   FROM tally_writeback_attempts WHERE depreciation_run_id=? ORDER BY id DESC""",
+                (int(run_id),),
+            ).fetchall()
+        return [dict(row) for row in rows]
+
    def snapshot(self, client_id: int) -> dict:
        path = self.db_path(client_id)
        if not path.is_file():
@@ -719,7 +877,7 @@ class LocalAccountingStore:
            latest = db.execute("SELECT checked_at_utc, connected, tally_url, company_count, error_message FROM tally_connection_history ORDER BY id DESC LIMIT 1").fetchone()
            latest_sync = db.execute("SELECT id, sync_type, tally_guid, company_name, mapping_id, requested_by_user_id, status, started_at_utc, completed_at_utc, rows_processed, error_message, details_json FROM tally_sync_runs WHERE sync_type='masters' ORDER BY id DESC LIMIT 1").fetchone()
            latest_transaction_sync = db.execute("SELECT id, sync_type, tally_guid, company_name, mapping_id, requested_by_user_id, status, started_at_utc, completed_at_utc, rows_processed, error_message, details_json FROM tally_sync_runs WHERE sync_type='transactions' ORDER BY id DESC LIMIT 1").fetchone()
-            latest_depreciation = db.execute("SELECT id, tally_guid, company_name, fy_start, fy_end, total_depreciation, depreciation_expense_ledger, depreciation_reserve_ledger, status, created_at_utc FROM it_depreciation_runs ORDER BY id DESC LIMIT 1").fetchone()
+            latest_depreciation = db.execute("SELECT id, tally_guid, company_name, fy_start, fy_end, total_depreciation, depreciation_expense_ledger, depreciation_reserve_ledger, status, created_at_utc, approved_by_user_id, approved_at_utc, posted_by_user_id, posted_at_utc, tally_posting_status, tally_posting_error FROM it_depreciation_runs ORDER BY id DESC LIMIT 1").fetchone()
            loaded_guids = {str(row["guid"] or "") for row in companies if int(row["is_currently_loaded"] or 0)}
            mapped = []
            for row in mappings:
@@ -41,6 +41,10 @@ class AgentCommandProcessor:
                result = self._calculate_it_depreciation(payload)
            elif action == "accounting_get_it_depreciation_run":
                result = self._get_it_depreciation_run(payload)
+            elif action == "accounting_approve_it_depreciation":
+                result = self._approve_it_depreciation(payload)
+            elif action == "accounting_post_it_depreciation":
+                result = self._post_it_depreciation(payload)
            else:
                raise ValueError(f"Unsupported local-agent command: {action}")
            ok = True
@@ -56,6 +60,7 @@ class AgentCommandProcessor:
            "tally_mapping_capability": True, "tally_master_sync_capability": True,
            "tally_transaction_sync_capability": True,
            "it_depreciation_capability": True,
+            "tally_writeback_capability": True,
        }

    def _status(self, payload: dict[str, Any]) -> dict[str, Any]:
@@ -207,3 +212,71 @@ class AgentCommandProcessor:
        client_id = int(payload.get("client_id")); run_id = int(payload.get("run_id"))
        return {"depreciation": self.store.get_it_depreciation_run(client_id, run_id), "agent": self._agent_info()}

+    def _approve_it_depreciation(self, payload: dict[str, Any]) -> dict[str, Any]:
+        client_id = int(payload.get("client_id"))
+        run_id = int(payload.get("run_id"))
+        user_id = int(payload.get("approved_by_user_id"))
+        result = self.store.approve_it_depreciation(
+            client_id, run_id, user_id, str(payload.get("approval_note") or "").strip()
+        )
+        self.logger.info("IT depreciation approved client_id=%s run_id=%s approved_by=%s", client_id, run_id, user_id)
+        return {"approved": True, "depreciation": result, "agent": self._agent_info()}
+
+    def _post_it_depreciation(self, payload: dict[str, Any]) -> dict[str, Any]:
+        client_id = int(payload.get("client_id"))
+        run_id = int(payload.get("run_id"))
+        posted_by = int(payload.get("posted_by_user_id"))
+        run = self.store.get_it_depreciation_run(client_id, run_id)
+        if str(run.get("status") or "").strip().lower() != "approved":
+            raise ValueError("The depreciation draft must be approved before Tally write-back.")
+        tally_guid = str(run.get("tally_guid") or "").strip()
+        company_name = str(run.get("company_name") or "").strip()
+
+        tally_status = self.tally.status()
+        if not tally_status.get("connected"):
+            raise ValueError(str(tally_status.get("error") or "TallyPrime is not connected."))
+        loaded = next(
+            (row for row in (tally_status.get("companies") or [])
+             if str(row.get("guid") or "").strip() == tally_guid),
+            None,
+        )
+        if not loaded:
+            raise ValueError("The mapped Tally company is not currently open. Open the mapped company in TallyPrime and retry.")
+        if str(loaded.get("name") or "").strip() != company_name:
+            raise ValueError("The currently open Tally company GUID matches the mapping but its company name differs. Refresh mapping/status before posting.")
+
+        voucher_date = str(run.get("fy_end") or "").strip()
+        reference = f"ARRR-ITDEP-{run_id}"
+        narration = f"Income-tax depreciation for FY {run.get('fy_start')} to {run.get('fy_end')} · ERP draft #{run_id}"
+        debit = str(run.get("depreciation_expense_ledger") or "").strip()
+        credit = str(run.get("depreciation_reserve_ledger") or "").strip()
+        amount = float(run.get("total_depreciation") or 0)
+        request_xml = self.tally.build_journal_import_xml(
+            company_name, voucher_date=voucher_date, debit_ledger=debit,
+            credit_ledger=credit, amount=amount, narration=narration, reference=reference,
+        )
+        attempt = self.store.begin_writeback_attempt(
+            client_id, run_id, posted_by, voucher_date=voucher_date,
+            reference=reference, request_xml=request_xml,
+        )
+        attempt_id = int(attempt["attempt_id"])
+        try:
+            result = self.tally.post_journal_voucher(
+                company_name, voucher_date=voucher_date, debit_ledger=debit,
+                credit_ledger=credit, amount=amount, narration=narration, reference=reference,
+            )
+            updated = self.store.finish_writeback_attempt(
+                client_id, attempt_id, posted=True, result=result,
+                response_xml=str(result.get("raw_response") or ""), posted_by_user_id=posted_by,
+            )
+            self.logger.warning(
+                "CONTROLLED TALLY WRITEBACK posted client_id=%s company=%s run_id=%s amount=%.2f by_user=%s tally_voucher=%s",
+                client_id, company_name, run_id, amount, posted_by, result.get("last_voucher_id"),
+            )
+            return {"posted": True, "depreciation": updated, "tally_result": {k:v for k,v in result.items() if k not in {"raw_response","request_xml"}}, "agent": self._agent_info()}
+        except Exception as exc:
+            self.store.finish_writeback_attempt(
+                client_id, attempt_id, posted=False, result={}, error_message=str(exc), posted_by_user_id=posted_by,
+            )
+            raise
+
@@ -99,9 +99,7 @@ class TallyCompany:
 class TallyLiveConnector:
    """Read-only TallyPrime XML/HTTP connector used by the ERP Local Agent.

-    Phase 3 adds master discovery for the currently loaded Tally companies. Phase 4
-    adds date-scoped read-only voucher/transaction export. Every export is scoped
-    with SVCURRENTCOMPANY and never creates, alters or deletes Tally data.
+    Phase 3 adds master discovery; Phase 4 adds transaction export. Phase 6 adds one narrowly-scoped controlled write-back operation for an approved depreciation Journal. All read operations remain read-only.
    """

    MASTER_SPECS = {
@@ -286,6 +284,84 @@ class TallyLiveConnector:
        vouchers = self.export_vouchers(company_name, start, end)
        return {"date_from": start, "date_to": end, "vouchers": vouchers}

+    def build_journal_import_xml(self, company_name: str, *, voucher_date: str, debit_ledger: str, credit_ledger: str, amount: float, narration: str, reference: str = "") -> str:
+        company = str(company_name or "").strip()
+        debit = str(debit_ledger or "").strip()
+        credit = str(credit_ledger or "").strip()
+        if not company:
+            raise ValueError("Tally company name is required for write-back.")
+        if not debit or not credit:
+            raise ValueError("Both debit and credit ledgers are required for write-back.")
+        if debit.casefold() == credit.casefold():
+            raise ValueError("Debit and credit ledgers cannot be the same.")
+        value = round(float(amount or 0), 2)
+        if value <= 0:
+            raise ValueError("Journal amount must be greater than zero.")
+        dt = _tally_date(voucher_date)
+        ref_xml = f"<REFERENCE>{self._xml_escape(reference)}</REFERENCE>" if str(reference or "").strip() 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="Journal" ACTION="Create" OBJVIEW="Accounting Voucher View">
+      <DATE>{dt}</DATE>
+      <VOUCHERTYPENAME>Journal</VOUCHERTYPENAME>
+      <PERSISTEDVIEW>Accounting Voucher View</PERSISTEDVIEW>
+      {ref_xml}
+      <NARRATION>{self._xml_escape(narration)}</NARRATION>
+      <ALLLEDGERENTRIES.LIST>
+        <LEDGERNAME>{self._xml_escape(debit)}</LEDGERNAME>
+        <ISDEEMEDPOSITIVE>Yes</ISDEEMEDPOSITIVE>
+        <AMOUNT>-{value:.2f}</AMOUNT>
+      </ALLLEDGERENTRIES.LIST>
+      <ALLLEDGERENTRIES.LIST>
+        <LEDGERNAME>{self._xml_escape(credit)}</LEDGERNAME>
+        <ISDEEMEDPOSITIVE>No</ISDEEMEDPOSITIVE>
+        <AMOUNT>{value:.2f}</AMOUNT>
+      </ALLLEDGERENTRIES.LIST>
+    </VOUCHER>
+  </TALLYMESSAGE></DATA></BODY>
+</ENVELOPE>"""
+
+    def post_journal_voucher(self, company_name: str, *, voucher_date: str, debit_ledger: str, credit_ledger: str, amount: float, narration: str, reference: str = "") -> dict:
+        request_xml = self.build_journal_import_xml(
+            company_name, voucher_date=voucher_date, debit_ledger=debit_ledger,
+            credit_ledger=credit_ledger, amount=amount, narration=narration, reference=reference,
+        )
+        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 Journal 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 Journal voucher.")
+        if created < 1 and altered < 1:
+            raise ValueError("Tally did not confirm creation of the Journal 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)