#!/usr/bin/env python3
"""
tally_agent.py — Enterprise RAG · Tally sync agent (Windows / macOS / Linux).
Runs on the machine where Tally is open. Reads Tally's local XML interface
(http://localhost:9000), builds a compact financial snapshot, and pushes it
OUTBOUND (HTTPS) to your Enterprise RAG cloud. No inbound ports, no firewall
changes. Pure Python standard library — no `pip install` needed.
SETUP (one time)
1. Install Python 3 (macOS usually has `python3`; Windows: https://python.org).
2. In Tally, enable the HTTP/XML gateway on port 9000:
TallyPrime : F1 Help → Settings → Connectivity → Client/Server →
"TallyPrime acts as": Both → Port: 9000
Tally.ERP9 : F12 → Advanced Configuration → Tally acts as: Both → Port 9000
3. Run once to create the config: python tally_agent.py
Then edit tally_agent.config.json (cloud_url + token from your app's
Settings → Tally page; set your company name + financial year dates).
4. Run it: python tally_agent.py (syncs now, then every hour)
One-off: python tally_agent.py --once
Keep it running (or schedule it: Windows Task Scheduler / macOS launchd / cron).
"""
import html
import json
import os
import re
import sys
import time
import urllib.request
from datetime import datetime, timezone
HERE = os.path.dirname(os.path.abspath(__file__))
CONFIG_PATH = os.path.join(HERE, "tally_agent.config.json")
CONFIG_TEMPLATE = {
"cloud_url": "https://demo.taskaisystems.com/tally/sync",
"token": "PASTE-YOUR-TALLY-TOKEN-FROM-SETTINGS",
"tally_host": "localhost",
"tally_port": 9000,
"company": "", # leave blank to use the active company in Tally
"from_date": "20240401", # YYYYMMDD — financial year start
"to_date": "20250331", # YYYYMMDD — financial year end
"interval_seconds": 3600,
"max_ledgers": 250,
}
def log(msg: str) -> None:
print(f"[{datetime.now().strftime('%H:%M:%S')}] {msg}", flush=True)
def load_config() -> dict:
if not os.path.exists(CONFIG_PATH):
with open(CONFIG_PATH, "w", encoding="utf-8") as fh:
json.dump(CONFIG_TEMPLATE, fh, indent=2)
log(f"Created {CONFIG_PATH} — edit it (cloud_url + token), then run again.")
sys.exit(0)
with open(CONFIG_PATH, encoding="utf-8") as fh:
cfg = json.load(fh)
if "PASTE-YOUR" in cfg.get("token", ""):
log("Please set your token in tally_agent.config.json (from Settings → Tally).")
sys.exit(1)
return {**CONFIG_TEMPLATE, **cfg}
# ── Tally ─────────────────────────────────────────────────────────────────────
def _trial_balance_request_xml(cfg: dict) -> str:
"""Ask Tally for its built-in Trial Balance report, exploded to sub-groups +
ledgers. This is a LIGHT, native report (no full-object dump) — it won't strain
Tally the way a Ledger collection FETCH does."""
company = (f"{cfg['company']}"
if cfg.get("company") else "")
return (
""
""
""
"Trial Balance"
""
"$$SysName:XML"
f"{company}"
f"{cfg['from_date']}"
f"{cfg['to_date']}"
"Yes"
"Yes"
""
""
)
def _stock_request_xml(cfg: dict) -> str:
"""Tally's native Stock Summary report (item-wise closing quantity + value)."""
company = (f"{cfg['company']}"
if cfg.get("company") else "")
return (
""
""
""
"Stock Summary"
""
"$$SysName:XML"
f"{company}"
f"{cfg['from_date']}"
f"{cfg['to_date']}"
"Yes"
""
""
)
def _post(cfg: dict, xml: str) -> str:
host = str(cfg["tally_host"]).strip()
if host.lower() == "localhost":
host = "127.0.0.1" # force IPv4 — 'localhost' can resolve to IPv6 ::1 and time out
url = f"http://{host}:{cfg['tally_port']}"
req = urllib.request.Request(url, data=xml.encode("utf-8"),
headers={"Content-Type": "text/xml"})
# Bypass any system/corporate proxy for the local Tally call.
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
with opener.open(req, timeout=90) as r: # real company files can be large
raw = r.read()
# Tally usually returns ISO-8859-1; strip invalid XML control chars.
text = raw.decode("iso-8859-1", "replace")
return re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f]", "", text)
def fetch_tally(cfg: dict) -> str:
return _post(cfg, _trial_balance_request_xml(cfg))
def fetch_stock(cfg: dict) -> str:
return _post(cfg, _stock_request_xml(cfg))
# Stock Summary rows: item name + closing quantity + closing value. Tolerant —
# pairs each DSPDISPNAME with its own DSPCLQTY/DSPCLAMTA without crossing into the
# next item. (Tag names verified-tolerant; tune against a real sample if needed.)
_STK_ROW = re.compile(
r"(.*?)"
r"(?:(?!).)*?(.*?)"
r"(?:(?!).)*?(.*?)",
re.S,
)
def parse_stock(xml_text: str) -> list[dict]:
items = []
for m in _STK_ROW.finditer(xml_text):
name = html.unescape(m.group(1)).replace("\x04", "").strip()
qty = html.unescape(m.group(2) or "").strip()
val = abs(_num(m.group(3)))
if not name:
continue
items.append({"name": name, "qty": qty, "value": round(val, 2)})
return items
def _num(s: str) -> float:
try:
return float((s or "").replace(",", "").strip() or 0)
except ValueError:
return 0.0
# Trial Balance XML is a flat sequence of (account name, closing Dr/Cr) pairs:
# NAME
# dr
# cr
# Tally shows debit closing amounts as negative and credit as positive; we expose
# both as positive Dr/Cr columns (exactly like a printed trial balance).
_TB_ROW = re.compile(
r"(.*?).*?"
r"(.*?).*?"
r"(.*?)",
re.S,
)
def parse_trial_balance(xml_text: str) -> list[dict]:
accounts = []
for name_raw, dr_raw, cr_raw in _TB_ROW.findall(xml_text):
name = html.unescape(name_raw).replace("\x04", "").strip()
dr = abs(_num(dr_raw)) # debit closing (shown -ve in Tally)
cr = _num(cr_raw) # credit closing
if not name or (dr == 0 and cr == 0):
continue
accounts.append({"name": name, "dr": round(dr, 2), "cr": round(cr, 2)})
return accounts
def build_summary(accounts: list[dict]) -> dict:
def find(group_name: str):
for a in accounts:
if a["name"].strip().lower() == group_name.lower():
return a
return None
def asset(group_name: str) -> float: # debit-natured: +ve = held / owed to us
a = find(group_name)
return round(a["dr"] - a["cr"], 2) if a else 0.0
def liab(group_name: str) -> float: # credit-natured: +ve = we owe / earned
a = find(group_name)
return round(a["cr"] - a["dr"], 2) if a else 0.0
return {
"bank": asset("Bank Accounts"),
"cash": asset("Cash-in-Hand"),
"receivables": asset("Sundry Debtors"),
"payables": liab("Sundry Creditors"),
"sales": liab("Sales Accounts"),
"purchases": asset("Purchase Accounts"),
"duties_taxes": liab("Duties & Taxes"),
"account_count": len(accounts),
}
# ── Push to cloud ─────────────────────────────────────────────────────────────
def push(cfg: dict, payload: dict) -> None:
data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(cfg["cloud_url"], data=data, method="POST",
headers={"Content-Type": "application/json",
"X-Tally-Token": cfg["token"]})
with urllib.request.urlopen(req, timeout=30) as r:
r.read()
def sync_once(cfg: dict) -> bool:
try:
xml_text = fetch_tally(cfg)
except Exception as exc:
log(f"Could not reach Tally at {cfg['tally_host']}:{cfg['tally_port']} ({exc}). "
"Is Tally open with the gateway enabled (port 9000)?")
return False
accounts = parse_trial_balance(xml_text)
if not accounts:
log("Connected to Tally but parsed 0 accounts — is a company loaded with data, "
"and is Tally idle at Gateway (no pop-up open)?")
accounts.sort(key=lambda a: max(a["dr"], a["cr"]), reverse=True)
summary = build_summary(accounts)
# Inventory (Stock Summary) — best-effort; never fail the sync if it's empty
# (service businesses have no stock) or unparseable.
stock = []
try:
stock = parse_stock(fetch_stock(cfg))
except Exception:
stock = []
stock.sort(key=lambda s: s["value"], reverse=True)
payload = {
"company": cfg.get("company") or "(active company)",
"from_date": cfg["from_date"], "to_date": cfg["to_date"],
"synced_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
"currency": "INR",
"summary": summary,
"accounts": accounts[: int(cfg.get("max_ledgers", 250))],
"stock": {
"item_count": len(stock),
"total_value": round(sum(s["value"] for s in stock), 2),
"items": stock[: int(cfg.get("max_ledgers", 250))],
},
}
try:
push(cfg, payload)
except Exception as exc:
log(f"Sync upload failed ({exc}). Check cloud_url + token.")
return False
log(f"Synced {len(accounts)} accounts + {len(stock)} stock items to the cloud ✓ "
f"(bank ₹{summary['bank']:,.0f}, receivables ₹{summary['receivables']:,.0f}, "
f"payables ₹{summary['payables']:,.0f})")
return True
def main() -> None:
cfg = load_config()
once = "--once" in sys.argv
log(f"Tally agent started (every {cfg['interval_seconds']}s). Ctrl-C to stop.")
while True:
sync_once(cfg)
if once:
break
time.sleep(max(60, int(cfg["interval_seconds"])))
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print("\nStopped.")