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

Add Phase 5 Income-tax depreciation tool

Browse Source
This commit is contained in:
A R R R Associates
2026-08-19 21:56:26 +05:30
parent 102520bb25
commit 66d286fd55
11 changed files with 407 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/modules/documents/local_agent_runtime/erp_local_agent/__init__.py
+1 -1
View File
@@ -1,2 +1,2 @@
__version__ = "1.5.0"
__version__ = "1.6.0"
AGENT_NAME = "ERP Local Agent"
app/modules/documents/local_agent_runtime/erp_local_agent/accounting_store.py
+189 -4
View File
@@ -1,13 +1,13 @@
from __future__ import annotations
from datetime import datetime, timezone
from datetime import date, datetime, timedelta, timezone
import json
from pathlib import Path
import sqlite3
from typing import Sequence
SCHEMA_VERSION = "4"
SCHEMA_VERSION = "5"
MASTER_TABLES = {
"groups": "tally_groups",
@@ -31,8 +31,9 @@ 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. Existing .act files
are upgraded in place.
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.
"""
def __init__(self, storage_root: Path):
@@ -244,6 +245,35 @@ class LocalAccountingStore:
FOREIGN KEY(ledger_entry_id) REFERENCES tally_voucher_ledger_entries(id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS ix_tally_bank_allocations_ledger_entry ON tally_bank_allocations(ledger_entry_id, line_no);
CREATE TABLE IF NOT EXISTS it_depreciation_rate_rules (
id INTEGER PRIMARY KEY AUTOINCREMENT,
tally_guid TEXT NOT NULL, ledger_key TEXT NOT NULL, ledger_guid TEXT NOT NULL DEFAULT '',
ledger_name TEXT NOT NULL, rate_percent REAL NOT NULL DEFAULT 0,
updated_by_user_id INTEGER, updated_at_utc TEXT NOT NULL,
UNIQUE(tally_guid, ledger_key)
);
CREATE INDEX IF NOT EXISTS ix_it_dep_rate_rules_company ON it_depreciation_rate_rules(tally_guid, ledger_name);
CREATE TABLE IF NOT EXISTS it_depreciation_runs (
id INTEGER PRIMARY KEY AUTOINCREMENT, tally_guid TEXT NOT NULL, company_name TEXT NOT NULL,
mapping_id INTEGER, fy_start TEXT NOT NULL, fy_end TEXT NOT NULL,
depreciation_expense_ledger TEXT NOT NULL DEFAULT '', depreciation_reserve_ledger TEXT NOT NULL DEFAULT '',
requested_by_user_id INTEGER, status TEXT NOT NULL DEFAULT 'draft', created_at_utc TEXT NOT NULL,
total_depreciation REAL NOT NULL DEFAULT 0, details_json TEXT NOT NULL DEFAULT '{}'
);
CREATE INDEX IF NOT EXISTS ix_it_dep_runs_company_fy ON it_depreciation_runs(tally_guid, fy_start, fy_end, id);
CREATE TABLE IF NOT EXISTS it_depreciation_run_lines (
id INTEGER PRIMARY KEY AUTOINCREMENT, run_id INTEGER NOT NULL, line_no INTEGER NOT NULL,
ledger_key TEXT NOT NULL, ledger_guid TEXT NOT NULL DEFAULT '', ledger_name TEXT NOT NULL, parent_group TEXT NOT NULL DEFAULT '',
opening_wdv REAL NOT NULL DEFAULT 0, additions_full_rate REAL NOT NULL DEFAULT 0,
additions_half_rate REAL NOT NULL DEFAULT 0, reductions REAL NOT NULL DEFAULT 0,
rate_percent REAL NOT NULL DEFAULT 0, depreciation_amount REAL NOT NULL DEFAULT 0,
closing_wdv REAL NOT NULL DEFAULT 0, payload_json TEXT NOT NULL DEFAULT '{}',
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);
"""
)
@@ -517,6 +547,159 @@ class LocalAccountingStore:
result[key] = int(db.execute(f"SELECT COUNT(*) FROM {table} WHERE tally_guid=?", (tally_guid,)).fetchone()[0])
return result
@staticmethod
def _ledger_key(row: dict) -> str:
guid = str(row.get("master_guid") or row.get("guid") or "").strip()
name = str(row.get("name") or row.get("ledger_name") or "").strip()
return ("G:" + guid) if guid else ("N:" + name.casefold())
@staticmethod
def _is_debit_entry(row: sqlite3.Row | dict) -> bool:
flag = str(row["is_deemed_positive"] if isinstance(row, sqlite3.Row) else row.get("is_deemed_positive") or "").strip().lower()
if flag in {"yes", "true", "1"}:
return True
if flag in {"no", "false", "0"}:
return False
amount = float(row["amount"] if isinstance(row, sqlite3.Row) else row.get("amount") or 0)
return amount < 0
@staticmethod
def _half_rate_addition(voucher_date: str, fy_end: date) -> bool:
try:
acquired = date.fromisoformat(str(voucher_date or ""))
except Exception:
return False
# Transaction date is a review proxy for put-to-use date. Half-rate is
# flagged only when fewer than 180 calendar days remain including that date.
return ((fy_end - acquired).days + 1) < 180
def _fixed_asset_group_names(self, db: sqlite3.Connection, tally_guid: str) -> set[str]:
rows = db.execute("SELECT name, parent FROM tally_groups WHERE tally_guid=?", (tally_guid,)).fetchall()
all_groups = {str(r["name"] or "").strip(): str(r["parent"] or "").strip() for r in rows}
fixed = {name for name in all_groups if name.casefold() == "fixed assets"}
changed = True
while changed:
changed = False
fixed_fold = {x.casefold() for x in fixed}
for name, parent in all_groups.items():
if name not in fixed and parent.casefold() in fixed_fold:
fixed.add(name); changed = True
return fixed
def _transaction_coverage(self, db: sqlite3.Connection, tally_guid: str, fy_start: date, fy_end: date) -> dict:
rows = db.execute(
"SELECT details_json FROM tally_sync_runs WHERE sync_type='transactions' AND tally_guid=? AND status='completed' ORDER BY id",
(tally_guid,),
).fetchall()
ranges = []
for row in rows:
try:
d = json.loads(row["details_json"] or "{}")
a = date.fromisoformat(str(d.get("date_from") or "")); b = date.fromisoformat(str(d.get("date_to") or ""))
if b >= fy_start and a <= fy_end:
ranges.append((max(a, fy_start), min(b, fy_end)))
except Exception:
continue
ranges.sort()
merged = []
for a,b in ranges:
if not merged or a > merged[-1][1] + timedelta(days=1): merged.append([a,b])
elif b > merged[-1][1]: merged[-1][1] = b
complete = bool(merged and merged[0][0] <= fy_start and merged[-1][1] >= fy_end and all(merged[i][1] + timedelta(days=1) >= merged[i+1][0] for i in range(len(merged)-1)))
return {"complete": complete, "ranges": [{"from":a.isoformat(),"to":b.isoformat()} for a,b in merged]}
def depreciation_preview(self, client_id: int, *, tally_guid: str, fy_start: str, fy_end: str) -> dict:
if not self.exists(client_id):
raise ValueError("Accounting storage is not initialized for this client.")
mapping = self.get_active_mapping_by_guid(client_id, tally_guid)
start = date.fromisoformat(str(fy_start)); end = date.fromisoformat(str(fy_end))
if start > end: raise ValueError("Financial-year start cannot be after end date.")
with self.connect(client_id) as db:
fixed_groups = self._fixed_asset_group_names(db, tally_guid)
if not fixed_groups:
raise ValueError("Fixed Assets group was not found in synchronized Tally masters. Synchronize Phase 3 masters first.")
ledgers = db.execute(
"SELECT master_guid, name, parent, opening_balance, closing_balance, payload_json FROM tally_ledgers WHERE tally_guid=? ORDER BY name COLLATE NOCASE",
(tally_guid,),
).fetchall()
rule_rows = db.execute("SELECT ledger_key, rate_percent FROM it_depreciation_rate_rules WHERE tally_guid=?", (tally_guid,)).fetchall()
rules = {str(r["ledger_key"]): float(r["rate_percent"] or 0) for r in rule_rows}
coverage = self._transaction_coverage(db, tally_guid, start, end)
fixed_fold = {x.casefold() for x in fixed_groups}
result=[]
for row in ledgers:
if str(row["parent"] or "").strip().casefold() not in fixed_fold: continue
item=dict(row); key=self._ledger_key(item); name=str(row["name"] or "").strip()
entries=db.execute(
"""SELECT v.voucher_date, e.amount, e.is_deemed_positive
FROM tally_voucher_ledger_entries e JOIN tally_vouchers v ON v.id=e.voucher_id
WHERE v.tally_guid=? AND e.ledger_name=? AND v.voucher_date>=? AND v.voucher_date<=?
AND lower(COALESCE(v.is_cancelled,'')) NOT IN ('yes','true','1')
AND lower(COALESCE(v.is_optional,'')) NOT IN ('yes','true','1')""",
(tally_guid,name,start.isoformat(),end.isoformat()),
).fetchall()
full=half=reductions=0.0
for e in entries:
amt=abs(float(e["amount"] or 0))
if self._is_debit_entry(e):
if self._half_rate_addition(str(e["voucher_date"] or ""), end): half += amt
else: full += amt
else: reductions += amt
result.append({
"ledger_key":key,"ledger_guid":str(row["master_guid"] or ""),"ledger_name":name,"parent_group":str(row["parent"] or ""),
"opening_wdv":abs(float(row["opening_balance"] or 0)),"tally_closing_balance":abs(float(row["closing_balance"] or 0)),
"additions_full_rate":round(full,2),"additions_half_rate":round(half,2),"reductions":round(reductions,2),
"rate_percent":rules.get(key,0.0),
})
all_ledgers=[str(r["name"] or "").strip() for r in ledgers if str(r["name"] or "").strip()]
latest=db.execute("SELECT id, fy_start, fy_end, total_depreciation, depreciation_expense_ledger, depreciation_reserve_ledger, created_at_utc FROM it_depreciation_runs WHERE tally_guid=? ORDER BY id DESC LIMIT 1",(tally_guid,)).fetchone()
return {"mapping":mapping,"fy_start":start.isoformat(),"fy_end":end.isoformat(),"transaction_coverage":coverage,"fixed_asset_groups":sorted(fixed_groups,key=str.casefold),"fixed_asset_ledgers":result,"all_ledgers":all_ledgers,"latest_run":dict(latest) if latest else None}
def calculate_it_depreciation(self, client_id: int, *, tally_guid: str, fy_start: str, fy_end: str, assignments: list[dict], depreciation_expense_ledger: str = "", depreciation_reserve_ledger: str = "", requested_by_user_id: int | None = None) -> dict:
preview=self.depreciation_preview(client_id,tally_guid=tally_guid,fy_start=fy_start,fy_end=fy_end)
if not preview["transaction_coverage"].get("complete"):
raise ValueError("Phase 4 transaction data does not completely cover the selected financial year. Synchronize the full financial year before calculating depreciation.")
by_key={str(x.get("ledger_key")):x for x in preview["fixed_asset_ledgers"]}
assignment_map={str(x.get("ledger_key") or ""): float(x.get("rate_percent") or 0) for x in assignments}
unknown=[k for k in assignment_map if k not in by_key]
if unknown: raise ValueError("One or more submitted fixed-asset ledgers are no longer available. Refresh the depreciation page.")
for rate in assignment_map.values():
if rate < 0 or rate > 100: raise ValueError("Depreciation rate must be between 0% and 100%.")
now=_utc_now_iso(); lines=[]; total=0.0
for index,row in enumerate(preview["fixed_asset_ledgers"],1):
rate=float(assignment_map.get(row["ledger_key"], row.get("rate_percent") or 0))
opening=float(row["opening_wdv"] or 0); full=float(row["additions_full_rate"] or 0); half=float(row["additions_half_rate"] or 0); reductions=float(row["reductions"] or 0)
closing_before=max(0.0, opening+full+half-reductions)
full_base=max(0.0, opening+full-reductions)
dep=min(closing_before, (full_base*rate/100.0)+(half*rate/200.0))
dep=round(dep,2); closing=round(max(0.0,closing_before-dep),2); total+=dep
line={**row,"rate_percent":rate,"depreciation_amount":dep,"closing_wdv":closing,"closing_before_depreciation":round(closing_before,2)}
lines.append(line)
mapping=preview["mapping"]
with self.connect(client_id) as db:
for row in lines:
db.execute("""INSERT INTO it_depreciation_rate_rules(tally_guid,ledger_key,ledger_guid,ledger_name,rate_percent,updated_by_user_id,updated_at_utc)
VALUES(?,?,?,?,?,?,?) ON CONFLICT(tally_guid,ledger_key) DO UPDATE SET ledger_guid=excluded.ledger_guid,ledger_name=excluded.ledger_name,rate_percent=excluded.rate_percent,updated_by_user_id=excluded.updated_by_user_id,updated_at_utc=excluded.updated_at_utc""",
(tally_guid,row["ledger_key"],row["ledger_guid"],row["ledger_name"],row["rate_percent"],requested_by_user_id,now))
details={"calculation_basis":"Ledger-level draft using synchronized opening balance and voucher movements; transaction date is used as the review proxy for put-to-use date for the 180-day test.","transaction_coverage":preview["transaction_coverage"],"no_tally_writeback":True}
cur=db.execute("""INSERT INTO it_depreciation_runs(tally_guid,company_name,mapping_id,fy_start,fy_end,depreciation_expense_ledger,depreciation_reserve_ledger,requested_by_user_id,status,created_at_utc,total_depreciation,details_json)
VALUES(?,?,?,?,?,?,?,?, 'draft',?,?,?)""",
(tally_guid,str(mapping.get("company_name") or ""),mapping.get("id"),str(fy_start),str(fy_end),str(depreciation_expense_ledger or ""),str(depreciation_reserve_ledger or ""),requested_by_user_id,now,round(total,2),json.dumps(details,ensure_ascii=False,separators=(",",":"))))
run_id=int(cur.lastrowid)
for i,row in enumerate(lines,1):
db.execute("""INSERT INTO it_depreciation_run_lines(run_id,line_no,ledger_key,ledger_guid,ledger_name,parent_group,opening_wdv,additions_full_rate,additions_half_rate,reductions,rate_percent,depreciation_amount,closing_wdv,payload_json)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
(run_id,i,row["ledger_key"],row["ledger_guid"],row["ledger_name"],row["parent_group"],row["opening_wdv"],row["additions_full_rate"],row["additions_half_rate"],row["reductions"],row["rate_percent"],row["depreciation_amount"],row["closing_wdv"],json.dumps(row,ensure_ascii=False,separators=(",",":"))))
return {"run_id":run_id,"status":"draft","company_name":str(mapping.get("company_name") or ""),"fy_start":str(fy_start),"fy_end":str(fy_end),"total_depreciation":round(total,2),"depreciation_expense_ledger":str(depreciation_expense_ledger or ""),"depreciation_reserve_ledger":str(depreciation_reserve_ledger or ""),"lines":lines,"proposed_entry":{"debit_ledger":str(depreciation_expense_ledger or "Depreciation"),"credit_ledger":str(depreciation_reserve_ledger or "Depreciation Reserve"),"amount":round(total,2)},"no_tally_writeback":True}
def get_it_depreciation_run(self, client_id: int, run_id: int) -> dict:
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.")
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
return result
def snapshot(self, client_id: int) -> dict:
path = self.db_path(client_id)
if not path.is_file():
@@ -536,6 +719,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()
loaded_guids = {str(row["guid"] or "") for row in companies if int(row["is_currently_loaded"] or 0)}
mapped = []
for row in mappings:
@@ -565,4 +749,5 @@ class LocalAccountingStore:
"mappings": mapped,
"latest_master_sync": latest_sync_dict,
"latest_transaction_sync": latest_transaction_sync_dict,
"latest_it_depreciation": dict(latest_depreciation) if latest_depreciation else None,
}
app/modules/documents/local_agent_runtime/erp_local_agent/commands.py
+36 -1
View File
@@ -23,7 +23,7 @@ class AgentCommandProcessor:
error: str | None = None
ok = False
try:
if action in {"tally_status", "phase1_status", "phase2_status", "phase3_status", "phase4_status"}:
if action in {"tally_status", "phase1_status", "phase2_status", "phase3_status", "phase4_status", "phase5_status"}:
result = self._status(payload)
elif action == "accounting_initialize":
result = self._initialize(payload)
@@ -35,6 +35,12 @@ class AgentCommandProcessor:
result = self._sync_masters(payload)
elif action == "accounting_sync_transactions":
result = self._sync_transactions(payload)
elif action == "accounting_depreciation_preview":
result = self._depreciation_preview(payload)
elif action == "accounting_calculate_it_depreciation":
result = self._calculate_it_depreciation(payload)
elif action == "accounting_get_it_depreciation_run":
result = self._get_it_depreciation_run(payload)
else:
raise ValueError(f"Unsupported local-agent command: {action}")
ok = True
@@ -49,6 +55,7 @@ class AgentCommandProcessor:
"tally_capability": True, "accounting_act_capability": True,
"tally_mapping_capability": True, "tally_master_sync_capability": True,
"tally_transaction_sync_capability": True,
"it_depreciation_capability": True,
}
def _status(self, payload: dict[str, Any]) -> dict[str, Any]:
@@ -172,3 +179,31 @@ class AgentCommandProcessor:
)
self.logger.info("Tally transaction sync completed client_id=%s company=%s period=%s..%s rows=%s", client_id, company_name, date_from, date_to, sync.get("rows_processed"))
return {"synced": True, "sync": sync, "accounting": self.store.snapshot(client_id), "tally": tally_status, "agent": self._agent_info()}
def _depreciation_preview(self, payload: dict[str, Any]) -> dict[str, Any]:
client_id = int(payload.get("client_id"))
tally_guid = str(payload.get("tally_guid") or "").strip()
fy_start = str(payload.get("fy_start") or "").strip()
fy_end = str(payload.get("fy_end") or "").strip()
if not tally_guid: raise ValueError("Select a mapped Tally company for depreciation.")
return {"preview": self.store.depreciation_preview(client_id, tally_guid=tally_guid, fy_start=fy_start, fy_end=fy_end), "agent": self._agent_info()}
def _calculate_it_depreciation(self, payload: dict[str, Any]) -> dict[str, Any]:
client_id = int(payload.get("client_id"))
result = self.store.calculate_it_depreciation(
client_id,
tally_guid=str(payload.get("tally_guid") or "").strip(),
fy_start=str(payload.get("fy_start") or "").strip(),
fy_end=str(payload.get("fy_end") or "").strip(),
assignments=list(payload.get("assignments") or []),
depreciation_expense_ledger=str(payload.get("depreciation_expense_ledger") or "").strip(),
depreciation_reserve_ledger=str(payload.get("depreciation_reserve_ledger") or "").strip(),
requested_by_user_id=int(payload.get("requested_by_user_id")) if payload.get("requested_by_user_id") not in (None, "") else None,
)
self.logger.info("IT depreciation draft calculated client_id=%s company=%s run_id=%s total=%s", client_id, result.get("company_name"), result.get("run_id"), result.get("total_depreciation"))
return {"calculated": True, "depreciation": result, "accounting": self.store.snapshot(client_id), "agent": self._agent_info()}
def _get_it_depreciation_run(self, payload: dict[str, Any]) -> dict[str, Any]:
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()}