Rakseeibra/arrr-erp
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Restore proven ledger master extraction and add Tally group hierarchy

Browse Source
This commit is contained in:
A R R R Associates
2026-09-08 20:41:47 +05:30
parent 614cab163f
commit f927b50ef1
6 changed files with 15 additions and 289 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/modules/documents/agent_package.py
+1 -1
View File
@@ -4,7 +4,7 @@ import io
from pathlib import Path
import zipfile
ERP_LOCAL_AGENT_VERSION = "1.26.5"
ERP_LOCAL_AGENT_VERSION = "1.26.6"
ERP_LOCAL_AGENT_NAME = "ERP Local Agent"
RUNTIME_ROOT = Path(__file__).resolve().parent / "local_agent_runtime"
_DETERMINISTIC_ZIP_TIMESTAMP = (2026, 1, 1, 0, 0, 0)
app/modules/documents/local_agent_runtime/erp_local_agent/__init__.py
+1 -1
View File
@@ -1,2 +1,2 @@
__version__ = "1.26.5"
__version__ = "1.26.6"
AGENT_NAME = "ERP Local Agent"
app/modules/documents/local_agent_runtime/erp_local_agent/accounting_mirror.py
+1 -211
View File
@@ -249,185 +249,6 @@ class AccountingMirrorEngine:
finally:
db.close()
@staticmethod
def _copy_table_rows(source_path: Path, target_path: Path, table: str) -> int:
if not source_path.is_file() or not target_path.is_file():
return 0
src = sqlite3.connect(source_path, timeout=30)
dst = sqlite3.connect(target_path, timeout=30)
src.row_factory = sqlite3.Row
try:
try:
src_cols = [r[1] for r in src.execute(f"PRAGMA table_info({table})").fetchall()]
dst_cols = [r[1] for r in dst.execute(f"PRAGMA table_info({table})").fetchall()]
common = [c for c in dst_cols if c in src_cols]
if not common:
return 0
rows = src.execute(f"SELECT {','.join(common)} FROM {table}").fetchall()
if not rows:
return 0
dst.execute(f"DELETE FROM {table}")
placeholders = ",".join("?" for _ in common)
dst.executemany(
f"INSERT INTO {table} ({','.join(common)}) VALUES ({placeholders})",
[tuple(row[c] for c in common) for row in rows],
)
dst.commit()
return len(rows)
except sqlite3.Error:
return 0
finally:
dst.close()
src.close()
@staticmethod
def _recover_ledger_master_from_accounting_db(accounting_db: Path, target_path: Path,
company_guid: str, company_name: str) -> int:
if not accounting_db.is_file() or not target_path.is_file():
return 0
src = sqlite3.connect(accounting_db, timeout=30)
src.row_factory = sqlite3.Row
dst = sqlite3.connect(target_path, timeout=30)
try:
try:
rows = src.execute(
"""SELECT master_guid,name,parent,opening_balance,closing_balance,
is_revenue,gst_registration_type
FROM tally_ledgers
WHERE tally_guid=?
ORDER BY name""",
(str(company_guid or "").strip(),),
).fetchall()
if not rows and company_name:
rows = src.execute(
"""SELECT master_guid,name,parent,opening_balance,closing_balance,
is_revenue,gst_registration_type
FROM tally_ledgers
WHERE lower(trim(company_name))=lower(trim(?))
ORDER BY name""",
(str(company_name or "").strip(),),
).fetchall()
if not rows:
return 0
company = dst.execute(
"SELECT company_instance_id,company_guid FROM company_master ORDER BY synced_at DESC LIMIT 1"
).fetchone()
if not company:
return 0
instance = company[0]
guid = company[1] or company_guid
dst.execute("DELETE FROM ledger_master WHERE company_instance_id=?", (instance,))
now = _utc_now()
inserted = 0
for idx, row in enumerate(rows, 1):
master_guid = _text(row["master_guid"]) or f"RECOVERED_LEDGER_{idx:08d}"
dst.execute(
"""INSERT OR REPLACE INTO ledger_master(
company_instance_id,company_guid,ledger_guid,alter_id,ledger_name,
parent_group,opening_balance,closing_balance,is_revenue,party_gstin,
gst_registration_type,synced_at)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?)""",
(
instance, guid, master_guid, 0, _text(row["name"]),
_text(row["parent"]), _number(row["opening_balance"]),
_number(row["closing_balance"]), 1 if str(row["is_revenue"]).strip().lower() in ("1","yes","true") else 0,
"", _text(row["gst_registration_type"]), now,
),
)
inserted += 1
dst.commit()
return inserted
except sqlite3.Error:
return 0
finally:
dst.close()
src.close()
@staticmethod
def _recover_group_master_from_accounting_db(accounting_db: Path, target_path: Path,
company_guid: str, company_name: str) -> int:
if not accounting_db.is_file() or not target_path.is_file():
return 0
src = sqlite3.connect(accounting_db, timeout=30)
src.row_factory = sqlite3.Row
dst = sqlite3.connect(target_path, timeout=30)
try:
try:
rows = src.execute(
"""SELECT master_guid,name,parent,reserved_name,is_revenue,
is_deemed_positive,is_active
FROM tally_groups WHERE tally_guid=? ORDER BY name""",
(str(company_guid or "").strip(),),
).fetchall()
if not rows and company_name:
rows = src.execute(
"""SELECT master_guid,name,parent,reserved_name,is_revenue,
is_deemed_positive,is_active
FROM tally_groups
WHERE lower(trim(company_name))=lower(trim(?))
ORDER BY name""",
(str(company_name or "").strip(),),
).fetchall()
if not rows:
return 0
dst.execute("""CREATE TABLE IF NOT EXISTS group_master(
master_guid TEXT,name TEXT,parent TEXT,reserved_name TEXT,
is_revenue TEXT,is_deemed_positive TEXT,is_active TEXT)""")
dst.execute("DELETE FROM group_master")
for row in rows:
dst.execute(
"INSERT INTO group_master VALUES (?,?,?,?,?,?,?)",
tuple(_text(row[k]) for k in (
"master_guid","name","parent","reserved_name",
"is_revenue","is_deemed_positive","is_active"
)),
)
dst.commit()
return len(rows)
except sqlite3.Error:
return 0
finally:
dst.close()
src.close()
def _recover_missing_masters(self, paths: MirrorPaths, candidate: Path,
expected_company_guid: str, expected_company_name: str) -> dict[str, int]:
recovered = {"ledgers": 0, "groups": 0}
summary = self._summary(candidate)
counts = summary.get("counts") or {}
if int(counts.get("ledger_master") or 0) > 0:
return recovered
# First prefer the currently published mirror/cache because it preserves
# the exact Accounting Mirror schema.
candidates = []
if paths.current_mirror.is_file() and paths.current_mirror.resolve() != candidate.resolve():
candidates.append(paths.current_mirror)
if paths.cache_root.is_dir():
candidates.extend(
p for p in sorted(paths.cache_root.glob("*.act"), key=lambda x: x.stat().st_mtime, reverse=True)
if p.resolve() != candidate.resolve() and p not in candidates
)
for previous in candidates:
previous_summary = self._summary(previous)
previous_counts = previous_summary.get("counts") or {}
if int(previous_counts.get("ledger_master") or 0) > 0:
recovered["ledgers"] = self._copy_table_rows(previous, candidate, "ledger_master")
if int(previous_counts.get("group_master") or 0) > 0:
recovered["groups"] = self._copy_table_rows(previous, candidate, "group_master")
if recovered["ledgers"] > 0:
return recovered
# Final local recovery source: the client Accounting .act snapshot.
recovered["ledgers"] = self._recover_ledger_master_from_accounting_db(
paths.accounting_db, candidate, expected_company_guid, expected_company_name
)
recovered["groups"] = self._recover_group_master_from_accounting_db(
paths.accounting_db, candidate, expected_company_guid, expected_company_name
)
return recovered
def _cache_file(self, paths: MirrorPaths, company_name: str, company_guid: str = "") -> Path:
digest = hashlib.sha256((_text(company_guid) + "|" + _text(company_name).casefold()).encode()).hexdigest()[:12]
return paths.cache_root / f"{_safe_key(company_name)}_{digest}.act"
@@ -496,11 +317,7 @@ class AccountingMirrorEngine:
"COMPANY - ROWS READ:": (14, "Company read"),
"LEDGER MASTER - BEFORE QUERY": (15, "Reading ledgers"),
"LEDGER MASTER - ROWS READ:": (25, "Ledgers read"),
"LEDGER MASTER FALLBACK - BEFORE QUERY": (18, "Retrying ledgers from built-in Tally table"),
"LEDGER MASTER FALLBACK - ROWS READ:": (25, "Ledgers read from built-in Tally table"),
"GROUP MASTER - BEFORE QUERY": (25, "Reading group hierarchy"),
"GROUP MASTER - ROWS READ:": (26, "Group hierarchy read"),
"STOCK MASTER - BEFORE QUERY": (27, "Reading stock items"),
"STOCK MASTER - BEFORE QUERY": (26, "Reading stock items"),
"STOCK MASTER - ROWS READ:": (34, "Stock items read"),
"STOCK HSN MASTER - BEFORE QUERY": (35, "Reading HSN details"),
"STOCK HSN MASTER - ROWS READ:": (43, "HSN details read"),
@@ -568,32 +385,6 @@ class AccountingMirrorEngine:
self._add_auxiliary_masters(candidate, output_dir)
summary = self._summary(candidate)
counts = summary.get("counts") or {}
recovery_counts = {"ledgers": 0, "groups": 0}
if int(counts.get("ledger_master") or 0) == 0:
recovery_counts = self._recover_missing_masters(
paths, candidate, expected_company_guid, expected_company_name
)
summary = self._summary(candidate)
counts = summary.get("counts") or {}
if recovery_counts.get("ledgers"):
validation_message = (
f"Recovered {recovery_counts['ledgers']} ledger masters"
+ (f" and {recovery_counts['groups']} groups" if recovery_counts.get("groups") else "")
+ " from the last valid local Accounting master snapshot."
)
report(94, "Recovered master data", validation_message, rows=dict(rows))
ledger_count = int(counts.get("ledger_master") or 0)
voucher_line_count = int(counts.get("voucher_ledger_entry") or 0)
voucher_count = int(counts.get("voucher") or 0)
if ledger_count == 0 and (voucher_line_count > 0 or voucher_count > 0):
raise RuntimeError(
"Accounting Mirror master extraction is incomplete: Tally returned voucher data "
f"({voucher_count} vouchers / {voucher_line_count} voucher lines) but zero ledger masters. "
"The incomplete mirror was NOT published. Repair/refresh the Accounting Mirror TDL and retry."
)
company = summary.get("company") or {}
actual_name = _text(company.get("company_name"))
actual_guid = _text(company.get("company_guid"))
@@ -652,7 +443,6 @@ class AccountingMirrorEngine:
"odbc_company_guid": actual_guid,
"company_cache_path": str(cache),
"validation_warnings": validation_warnings,
"master_recovery": recovery_counts,
"extractor_log_tail": log[-1500:], "loader_log_tail": llog[-1500:],
**self.status(client_id),
}
app/modules/documents/local_agent_runtime/erp_local_agent/accounting_store.py
+1 -20
View File
@@ -654,23 +654,8 @@ class LocalAccountingStore:
run_id = int(cur.lastrowid)
total = 0
synced_at = _utc_now_iso()
preserved_empty_master_keys: list[str] = []
for key, table in MASTER_TABLES.items():
rows = masters.get(key) or []
existing_count = int(
db.execute(
f"SELECT COUNT(*) FROM {table} WHERE tally_guid=?",
(tally_guid,),
).fetchone()[0]
)
# A zero-row Ledger/Group refresh is never authoritative when
# a valid local master snapshot already exists. Preserve it
# instead of destroying data needed by Depreciation/TDS/Cash.
if key in ("ledgers", "groups") and not rows and existing_count > 0:
counts[key] = existing_count
total += existing_count
preserved_empty_master_keys.append(key)
continue
db.execute(f"DELETE FROM {table} WHERE tally_guid=?", (tally_guid,))
for row in rows:
payload = json.dumps(row, ensure_ascii=False, separators=(",", ":"))
@@ -687,11 +672,7 @@ class LocalAccountingStore:
)
counts[key] = len(rows)
total += len(rows)
details = {
"counts": counts,
"schema_version": SCHEMA_VERSION,
"preserved_empty_master_keys": preserved_empty_master_keys,
}
details = {"counts": counts, "schema_version": SCHEMA_VERSION}
db.execute(
"UPDATE tally_sync_runs SET status='completed', completed_at_utc=?, rows_processed=?, details_json=? WHERE id=?",
(_utc_now_iso(), total, json.dumps(details, ensure_ascii=False, separators=(",", ":")), run_id),
app/modules/documents/local_agent_runtime/erp_local_agent/mirror_runtime/ERP_Accounting_Mirror_v3_0_1.tdl
+5
View File
@@ -19,6 +19,11 @@
Fetch : Name, Parent, GUID, AlterID, OpeningBalance, ClosingBalance, IsRevenue, PartyGSTIN, GSTRegistrationType
Is ODBC Table : Yes
[Collection : ERPAMGroupMaster]
Type : Group
Fetch : Name, Parent, GUID
Is ODBC Table : Yes
[Collection : ERPAMStockMaster]
Type : StockItem
Fetch : Name, Parent, GUID, AlterID, BaseUnits, OpeningBalance, OpeningValue, GSTApplicable, GSTRepUoM, GSTTypeofSupply, GSTItemHSNCodeEx
app/modules/documents/local_agent_runtime/erp_local_agent/mirror_runtime/TallyAccountingMirrorV301.cs
+6 -56
View File
@@ -58,38 +58,6 @@ class TallyAccountingMirrorV301
Console.Out.Flush();
}
static bool TryExtract(OdbcConnection cn, string label, string sql, string path,
out int rows, out double sec)
{
rows = 0;
sec = 0;
try {
Extract(cn, label, sql, path, out rows, out sec);
return true;
}
catch(Exception ex) {
Console.WriteLine(label + " - OPTIONAL QUERY FAILED: " + ex.Message);
Console.Out.Flush();
return false;
}
}
static bool TryExtractFirst(OdbcConnection cn, string label, string[] sqls, string path,
out int rows, out double sec)
{
rows = 0;
sec = 0;
foreach(string sql in sqls) {
int r; double s;
if (TryExtract(cn, label, sql, path, out r, out s) && r > 0) {
rows = r;
sec = s;
return true;
}
}
return false;
}
static int Main(string[] args)
{
string dsn="TallyODBC64_9000", outDir="mirror_output_v3_0_1";
@@ -120,30 +88,12 @@ class TallyAccountingMirrorV301
"SELECT $GUID,$AlterID,$Name,$Parent,$OpeningBalance,$ClosingBalance,$IsRevenue,$PartyGSTIN,$GSTRegistrationType FROM ERPAMLedgerMaster",
Path.Combine(outDir,"ledger_master.csv"),out l,out sl);
// Some TallyPrime/ODBC builds expose vouchers from the custom
// collection but unexpectedly return zero Ledger objects. A
// company with voucher-ledger rows cannot legitimately have no
// ledger masters, so retry through Tally's built-in Ledger table.
if (l == 0) {
Console.WriteLine("LEDGER MASTER - PRIMARY COLLECTION EMPTY; TRYING BUILT-IN LEDGER TABLE");
TryExtractFirst(cn, "LEDGER MASTER FALLBACK",
new string[] {
"SELECT $GUID,$AlterID,$Name,$Parent,$OpeningBalance,$ClosingBalance,$IsRevenue,$PartyGSTIN,$GSTRegistrationType FROM Ledger",
"SELECT $GUID,$AlterID,$Name,$Parent,$OpeningBalance,$ClosingBalance,$IsRevenue,$PartyGSTIN,$GSTRegistrationType FROM Ledgers"
},
Path.Combine(outDir,"ledger_master.csv"), out l, out sl);
}
// Group hierarchy is required for recursive Fixed Assets and
// Capital/Partner-current-account classification. It is optional
// at extraction time so older builds do not break the whole mirror.
int groupRows = 0; double groupSec = 0;
TryExtractFirst(cn, "GROUP MASTER",
new string[] {
"SELECT $GUID,$Name,$Parent,$ReservedName,$IsRevenue,$IsDeemedPositive,$IsActive FROM Group",
"SELECT $GUID,$Name,$Parent,$ReservedName,$IsRevenue,$IsDeemedPositive,$IsActive FROM Groups"
},
Path.Combine(outDir,"group_master.csv"), out groupRows, out groupSec);
// Group hierarchy only. Ledger extraction above is intentionally
// the same proven ERPAMLedgerMaster path used before.
int groupRows; double groupSec;
Extract(cn,"GROUP MASTER",
"SELECT $GUID,$Name,$Parent FROM ERPAMGroupMaster",
Path.Combine(outDir,"group_master.csv"),out groupRows,out groupSec);
Extract(cn,"STOCK MASTER",
"SELECT $GUID,$AlterID,$Name,$Parent,$BaseUnits,$OpeningBalance,$OpeningValue,$GSTApplicable,$GSTRepUoM,$GSTTypeofSupply,$GSTItemHSNCodeEx FROM ERPAMStockMaster",