arrr-erp/app/modules/accounting/tally_connector.py

from __future__ import annotations

from dataclasses import asdict, dataclass
import html
import re
import urllib.error
import urllib.request
import xml.etree.ElementTree as ET

DEFAULT_TALLY_HOST = "127.0.0.1"
DEFAULT_TALLY_PORT = 9000
DEFAULT_TALLY_TIMEOUT_SECONDS = 8


class TallyConnectionError(RuntimeError):
    """Raised when the local TallyPrime XML/HTTP endpoint cannot be reached."""


def _clean_xml_response(xml_text: str) -> str:
    if not xml_text:
        return ""
    xml_text = re.sub(r"&#(0?[0-8]|1[0-9]|2[0-9]|3[01]);", "", xml_text)
    xml_text = re.sub(r"[\x00-\x08\x0B\x0C\x0E-\x1F]", "", xml_text)
    return xml_text


def _child_text(element: ET.Element, tag_name: str) -> str:
    wanted = tag_name.upper()
    for child in list(element):
        if str(child.tag).split("}")[-1].upper() == wanted:
            return (child.text or "").strip()
    return ""


@dataclass(frozen=True)
class TallyCompany:
    name: str
    guid: str = ""
    gstin: str = ""

    def as_dict(self) -> dict[str, str]:
        return asdict(self)


class TallyLiveConnector:
    """Read-only connector for a running local TallyPrime instance.

    Uses Tally XML-over-HTTP.  Phase 1 performs only discovery/status reads and
    never creates, updates or deletes data in Tally.
    """

    def __init__(
        self,
        host: str = DEFAULT_TALLY_HOST,
        port: int = DEFAULT_TALLY_PORT,
        timeout: int = DEFAULT_TALLY_TIMEOUT_SECONDS,
    ) -> None:
        self.host = (host or DEFAULT_TALLY_HOST).strip()
        self.port = int(port or DEFAULT_TALLY_PORT)
        self.timeout = max(1, int(timeout or DEFAULT_TALLY_TIMEOUT_SECONDS))

    @property
    def url(self) -> str:
        return f"http://{self.host}:{self.port}"

    @staticmethod
    def _xml_escape(value: str) -> str:
        return html.escape(str(value or ""), quote=False)

    def _post_xml(self, xml_text: str) -> str:
        request = urllib.request.Request(
            self.url,
            data=xml_text.encode("utf-8"),
            headers={"Content-Type": "application/xml; charset=utf-8"},
            method="POST",
        )
        try:
            with urllib.request.urlopen(request, timeout=self.timeout) as response:
                return response.read().decode("utf-8", errors="replace")
        except (urllib.error.URLError, TimeoutError, OSError) as exc:
            raise TallyConnectionError(
                f"Could not connect to TallyPrime at {self.url}. "
                f"Open TallyPrime and enable its HTTP/XML server on port {self.port}. "
                f"Details: {exc}"
            ) from exc

    def get_loaded_companies(self) -> list[TallyCompany]:
        xml = """<ENVELOPE>
  <HEADER>
    <VERSION>1</VERSION>
    <TALLYREQUEST>Export</TALLYREQUEST>
    <TYPE>Collection</TYPE>
    <ID>ARRRAccountingLoadedCompanies</ID>
  </HEADER>
  <BODY>
    <DESC>
      <STATICVARIABLES>
        <SVEXPORTFORMAT>$$SysName:XML</SVEXPORTFORMAT>
      </STATICVARIABLES>
      <TDL>
        <TDLMESSAGE>
          <COLLECTION NAME="ARRRAccountingLoadedCompanies" ISMODIFY="No">
            <TYPE>Company</TYPE>
            <FETCH>Name,GUID,GSTRegistrationNumber</FETCH>
          </COLLECTION>
        </TDLMESSAGE>
      </TDL>
    </DESC>
  </BODY>
</ENVELOPE>"""
        raw = self._post_xml(xml)
        return self._parse_companies(raw)

    def status(self) -> dict:
        try:
            companies = self.get_loaded_companies()
            return {
                "connected": True,
                "url": self.url,
                "host": self.host,
                "port": self.port,
                "company_count": len(companies),
                "companies": [company.as_dict() for company in companies],
                "error": None,
            }
        except TallyConnectionError as exc:
            return {
                "connected": False,
                "url": self.url,
                "host": self.host,
                "port": self.port,
                "company_count": 0,
                "companies": [],
                "error": str(exc),
            }

    @staticmethod
    def _parse_companies(xml_text: str) -> list[TallyCompany]:
        cleaned = _clean_xml_response(xml_text)
        if not cleaned.strip():
            return []

        try:
            root = ET.fromstring(cleaned.encode("utf-8"))
        except Exception:
            # Keep the same conservative fallback behaviour as the working
            # Office Automation connector: discover names even if Tally emits
            # malformed XML around otherwise useful response data.
            names: list[str] = []
            for match in re.finditer(r"<NAME[^>]*>(.*?)</NAME>", cleaned, flags=re.I | re.S):
                value = re.sub(r"<.*?>", "", match.group(1)).strip()
                if value and value not in names:
                    names.append(value)
            return [TallyCompany(name=name) for name in names]

        companies: list[TallyCompany] = []
        seen: set[str] = set()
        for element in root.iter():
            if str(element.tag).split("}")[-1].upper() != "COMPANY":
                continue
            name = (
                element.attrib.get("NAME")
                or element.attrib.get("name")
                or _child_text(element, "NAME")
                or _child_text(element, "BASICCOMPANYFORMALNAME")
            ).strip()
            key = name.upper()
            if not name or key in seen:
                continue
            seen.add(key)
            companies.append(
                TallyCompany(
                    name=name,
                    guid=_child_text(element, "GUID"),
                    gstin=_child_text(element, "GSTREGISTRATIONNUMBER"),
                )
            )
        return companies