Fix Tally GST registration discovery via Tax Unit masters

This commit is contained in:
A R R R Associates
2026-08-20 22:58:06 +05:30
parent fa0f681106
commit 92f06f333a
6 changed files with 120 additions and 15 deletions
+11 -6
View File
@@ -144,15 +144,20 @@ def _enrich_gstin_display(live_result: dict | None, registrations) -> None:
erp_gstins = [] erp_gstins = []
live_company = company_by_guid.get(str(mapping.get("tally_guid") or "").strip()) or {} live_company = company_by_guid.get(str(mapping.get("tally_guid") or "").strip()) or {}
tally_gstin = _normalise_gstin(live_company.get("gstin") or mapping.get("gstin")) tally_gstins: list[str] = []
for value in list(live_company.get("gstins") or []) + [live_company.get("gstin"), mapping.get("gstin")]:
gstin = _normalise_gstin(value)
if gstin and gstin not in tally_gstins:
tally_gstins.append(gstin)
mapping["erp_gstins"] = erp_gstins mapping["erp_gstins"] = erp_gstins
mapping["tally_gstin_display"] = tally_gstin mapping["tally_gstins"] = tally_gstins
mapping["tally_gstin_display"] = ", ".join(tally_gstins)
if erp_gstins and tally_gstin: if erp_gstins and tally_gstins:
mapping["gstin_match_status"] = "matched" if tally_gstin in erp_gstins else "mismatch" mapping["gstin_match_status"] = "matched" if set(erp_gstins).intersection(tally_gstins) else "mismatch"
elif erp_gstins and not tally_gstin: elif erp_gstins and not tally_gstins:
mapping["gstin_match_status"] = "tally_missing" mapping["gstin_match_status"] = "tally_missing"
elif tally_gstin and not erp_gstins: elif tally_gstins and not erp_gstins:
mapping["gstin_match_status"] = "erp_missing" mapping["gstin_match_status"] = "erp_missing"
else: else:
mapping["gstin_match_status"] = "unavailable" mapping["gstin_match_status"] = "unavailable"
+1 -1
View File
@@ -4,7 +4,7 @@ import io
from pathlib import Path from pathlib import Path
import zipfile import zipfile
ERP_LOCAL_AGENT_VERSION = "1.9.4" ERP_LOCAL_AGENT_VERSION = "1.9.5"
ERP_LOCAL_AGENT_NAME = "ERP Local Agent" ERP_LOCAL_AGENT_NAME = "ERP Local Agent"
RUNTIME_ROOT = Path(__file__).resolve().parent / "local_agent_runtime" RUNTIME_ROOT = Path(__file__).resolve().parent / "local_agent_runtime"
_DETERMINISTIC_ZIP_TIMESTAMP = (2026, 1, 1, 0, 0, 0) _DETERMINISTIC_ZIP_TIMESTAMP = (2026, 1, 1, 0, 0, 0)
@@ -1,4 +1,4 @@
ERP Local Agent 1.9.4 ERP Local Agent 1.9.5
Existing storage, WebSocket tunnel, Tally and client .act functionality are preserved. Existing storage, WebSocket tunnel, Tally and client .act functionality are preserved.
@@ -90,3 +90,10 @@ Tally GSTIN enrichment 1.9.4:
- Loaded-company discovery requests GSTRegistrationNumber, GSTRegistrationDetails, GSTIN and PartyGSTIN. - Loaded-company discovery requests GSTRegistrationNumber, GSTRegistrationDetails, GSTIN and PartyGSTIN.
- Company parser searches GSTIN candidates recursively, including nested GST registration details. - Company parser searches GSTIN candidates recursively, including nested GST registration details.
- Existing company GUID/name discovery remains unchanged. - Existing company GUID/name discovery remains unchanged.
Tally GST Registration master enrichment 1.9.5:
- Keeps existing loaded-company Name/GUID discovery unchanged.
- Reads GST registrations from TallyPrime Tax Unit masters using a filtered COMPUTE collection, avoiding the Tax Unit enumeration timeout seen with direct NATIVEMETHOD/FETCH probes.
- Supports multiple GST registrations for one Tally company while preserving the existing primary gstin field for backward compatibility.
- GST enrichment failure is non-fatal: company discovery, GUID mapping and all existing accounting functions remain available.
- Mapping validation accepts the ERP GSTIN when it matches any GST registration reported by the selected Tally company.
@@ -1,2 +1,2 @@
__version__ = "1.9.4" __version__ = "1.9.5"
AGENT_NAME = "ERP Local Agent" AGENT_NAME = "ERP Local Agent"
@@ -111,14 +111,22 @@ class AgentCommandProcessor:
if not company: if not company:
raise ValueError("The selected Tally company is no longer loaded. Refresh Tally companies and try again.") raise ValueError("The selected Tally company is no longer loaded. Refresh Tally companies and try again.")
company_name = str(company.get("name") or "").strip() company_name = str(company.get("name") or "").strip()
company_gstin = str(company.get("gstin") or "").strip().upper() company_gstins: list[str] = []
for value in list(company.get("gstins") or []) + [company.get("gstin")]:
gstin = str(value or "").strip().upper()
if gstin and gstin not in company_gstins:
company_gstins.append(gstin)
company_gstin = company_gstins[0] if company_gstins else ""
if not company_name: if not company_name:
raise ValueError("Tally returned an invalid company name.") raise ValueError("Tally returned an invalid company name.")
if registration: if registration:
reg_type = str(registration.get("registration_type_code") or "").strip().upper() reg_type = str(registration.get("registration_type_code") or "").strip().upper()
reg_number = str(registration.get("registration_number") or "").strip().upper() reg_number = str(registration.get("registration_number") or "").strip().upper()
if reg_type == "GSTIN" and reg_number and company_gstin and reg_number != company_gstin and not allow_gstin_mismatch: if reg_type == "GSTIN" and reg_number and company_gstins and reg_number not in company_gstins and not allow_gstin_mismatch:
raise ValueError(f"GSTIN mismatch: ERP registration is {reg_number}, but Tally company reports {company_gstin}. Verify the company or explicitly allow the mismatch.") reported = ", ".join(company_gstins)
raise ValueError(f"GSTIN mismatch: ERP registration is {reg_number}, but Tally company reports {reported}. Verify the company or explicitly allow the mismatch.")
if reg_type == "GSTIN" and reg_number in company_gstins:
company_gstin = reg_number
self.store.record_tally_status(client_id, tally_status) self.store.record_tally_status(client_id, tally_status)
mapping = self.store.map_company(client_id, tally_guid=requested_guid, company_name=company_name, gstin=company_gstin, registration=registration, mapped_by_user_id=int(mapped_by_user_id) if mapped_by_user_id not in (None, "") else None) mapping = self.store.map_company(client_id, tally_guid=requested_guid, company_name=company_name, gstin=company_gstin, registration=registration, mapped_by_user_id=int(mapped_by_user_id) if mapped_by_user_id not in (None, "") else None)
return {"mapped": True, "mapping": mapping, "accounting": self.store.snapshot(client_id), "tally": tally_status, "agent": self._agent_info()} return {"mapped": True, "mapping": mapping, "accounting": self.store.snapshot(client_id), "tally": tally_status, "agent": self._agent_info()}
@@ -91,8 +91,9 @@ class TallyCompany:
name: str name: str
guid: str = "" guid: str = ""
gstin: str = "" gstin: str = ""
gstins: tuple[str, ...] = ()
def as_dict(self) -> dict[str, str]: def as_dict(self) -> dict:
return asdict(self) return asdict(self)
@@ -205,7 +206,83 @@ class TallyLiveConnector:
</TDLMESSAGE></TDL> </TDLMESSAGE></TDL>
</DESC></BODY> </DESC></BODY>
</ENVELOPE>""" </ENVELOPE>"""
return self._parse_companies(self._post_xml(xml)) companies = self._parse_companies(self._post_xml(xml))
enriched: list[TallyCompany] = []
for company in companies:
gstins = self._get_company_gstins(company.name)
if not gstins and company.gstin:
gstins = (company.gstin.strip().upper(),)
primary_gstin = gstins[0] if gstins else company.gstin
enriched.append(
TallyCompany(
name=company.name,
guid=company.guid,
gstin=primary_gstin,
gstins=gstins,
)
)
return enriched
def _get_company_gstins(self, company_name: str) -> tuple[str, ...]:
"""Return GST registrations configured in the selected Tally company.
TallyPrime 3+ stores GST registrations as Tax Unit masters. A direct
NATIVEMETHOD/FETCH enumeration can stall on some Tally builds, while the
filtered COMPUTE collection below is stable and has been verified against
the GST Registration master shown in TallyPrime. Failure to enrich GSTINs
must never make otherwise-working company discovery unavailable.
"""
company = str(company_name or "").strip()
if not company:
return ()
xml = f"""<ENVELOPE>
<HEADER>
<VERSION>1</VERSION><TALLYREQUEST>Export</TALLYREQUEST><TYPE>Collection</TYPE><ID>ARRRAccountingGSTRegistrations</ID>
</HEADER>
<BODY><DESC>
<STATICVARIABLES>{self._static_variables(company)}</STATICVARIABLES>
<TDL><TDLMESSAGE>
<COLLECTION NAME="ARRRAccountingGSTRegistrations">
<TYPE>Tax Unit</TYPE>
<COMPUTE>MasterName : $Name</COMPUTE>
<COMPUTE>GSTNumber : $GSTRegNumber</COMPUTE>
<COMPUTE>GSTState : $StateName</COMPUTE>
<COMPUTE>GSTUsedFor : $UsedFor</COMPUTE>
<FILTER>ARRRAccountingHasGSTNumber</FILTER>
</COLLECTION>
<SYSTEM TYPE="Formulae" NAME="ARRRAccountingHasGSTNumber">NOT $$IsEmpty:$GSTRegNumber</SYSTEM>
</TDLMESSAGE></TDL>
</DESC></BODY>
</ENVELOPE>"""
try:
return self._parse_gst_registration_numbers(self._post_xml(xml))
except TallyConnectionError:
return ()
@staticmethod
def _parse_gst_registration_numbers(xml_text: str) -> tuple[str, ...]:
cleaned = _clean_xml_response(xml_text)
if not cleaned:
return ()
values: list[str] = []
try:
root = ET.fromstring(cleaned.encode("utf-8"))
for element in root.iter():
if _tag(element) != "TAXUNIT":
continue
candidates = [
_child_text(element, "GSTREGNUMBER"),
str(element.attrib.get("TAXREGISTRATION") or ""),
]
for candidate in candidates:
value = re.sub(r"\s+", "", candidate.strip().upper())
if re.fullmatch(r"[0-9]{2}[A-Z]{5}[0-9]{4}[A-Z][0-9A-Z]Z[0-9A-Z]", value) and value not in values:
values.append(value)
except Exception:
for match in re.findall(r"\b[0-9]{2}[A-Z]{5}[0-9]{4}[A-Z][0-9A-Z]Z[0-9A-Z]\b", cleaned.upper()):
if match not in values:
values.append(match)
return tuple(values)
def status(self) -> dict: def status(self) -> dict:
try: try:
@@ -573,5 +650,13 @@ class TallyLiveConnector:
"GSTNUMBER", "GSTNUMBER",
), ),
) )
companies.append(TallyCompany(name=name, guid=_child_text(element, "GUID"), gstin=gstin)) normalised_gstin = re.sub(r"\s+", "", gstin.strip().upper())
companies.append(
TallyCompany(
name=name,
guid=_child_text(element, "GUID"),
gstin=normalised_gstin,
gstins=(normalised_gstin,) if normalised_gstin else (),
)
)
return companies return companies