Files
arrr-erp/app/modules/accounting/act_store.py
T
2026-08-19 10:10:46 +05:30

419 lines
17 KiB
Python

from __future__ import annotations
from contextlib import contextmanager
from datetime import datetime, timezone
from pathlib import Path
import json
import os
import sqlite3
from typing import Iterator, Sequence
ACT_SCHEMA_VERSION = 2
class AccountingActStoreError(RuntimeError):
pass
def _utc_now_iso() -> str:
return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
def _safe_client_id(client_id: int) -> int:
value = int(client_id)
if value <= 0:
raise AccountingActStoreError("client_id must be a positive integer")
return value
class AccountingActStore:
"""Client-scoped SQLite storage for ERP accounting data.
Phase 1 provides accounting storage and read-only Tally discovery.
Phase 2 adds durable client/registration -> Tally company mapping keyed by
Tally GUID. Accounting master/transaction sync and write-back remain out of
scope.
"""
def __init__(self, root: str | Path) -> None:
self.root = Path(root).expanduser().resolve()
def client_dir(self, client_id: int) -> Path:
cid = _safe_client_id(client_id)
return self.root / "Accounting" / f"client_{cid:08d}"
def db_path(self, client_id: int) -> Path:
cid = _safe_client_id(client_id)
return self.client_dir(cid) / f"client_{cid:08d}.act"
@contextmanager
def connect(self, client_id: int) -> Iterator[sqlite3.Connection]:
path = self.db_path(client_id)
path.parent.mkdir(parents=True, exist_ok=True)
connection = sqlite3.connect(path, timeout=30)
connection.row_factory = sqlite3.Row
try:
connection.execute("PRAGMA foreign_keys = ON")
connection.execute("PRAGMA journal_mode = WAL")
connection.execute("PRAGMA synchronous = NORMAL")
yield connection
connection.commit()
except Exception:
connection.rollback()
raise
finally:
connection.close()
def initialize(
self,
client_id: int,
*,
tenant_id: int | None = None,
client_name: str = "",
created_by_user_id: int | None = None,
) -> Path:
path = self.db_path(client_id)
with self.connect(client_id) as db:
db.executescript(
"""
CREATE TABLE IF NOT EXISTS act_meta (
key TEXT PRIMARY KEY,
value TEXT NOT NULL,
updated_at_utc TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS tally_companies (
id INTEGER PRIMARY KEY AUTOINCREMENT,
tally_guid TEXT NOT NULL DEFAULT '',
company_name TEXT NOT NULL,
gstin TEXT NOT NULL DEFAULT '',
first_seen_at_utc TEXT NOT NULL,
last_seen_at_utc TEXT NOT NULL,
is_currently_loaded INTEGER NOT NULL DEFAULT 1,
UNIQUE(tally_guid, company_name)
);
CREATE TABLE IF NOT EXISTS tally_company_mapping (
id INTEGER PRIMARY KEY AUTOINCREMENT,
client_id INTEGER NOT NULL,
registration_id INTEGER,
registration_type_code TEXT NOT NULL DEFAULT '',
registration_number TEXT NOT NULL DEFAULT '',
registration_legal_name TEXT NOT NULL DEFAULT '',
registration_trade_name TEXT NOT NULL DEFAULT '',
business_unit_id INTEGER,
client_branch_id INTEGER,
tally_guid TEXT NOT NULL,
company_name TEXT NOT NULL,
gstin TEXT NOT NULL DEFAULT '',
is_active INTEGER NOT NULL DEFAULT 1,
mapped_at_utc TEXT NOT NULL,
mapped_by_user_id INTEGER,
updated_at_utc TEXT NOT NULL,
unmapped_at_utc TEXT,
unmapped_by_user_id INTEGER
);
CREATE TABLE IF NOT EXISTS tally_connection_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
checked_at_utc TEXT NOT NULL,
connected INTEGER NOT NULL,
tally_url TEXT NOT NULL DEFAULT '',
company_count INTEGER NOT NULL DEFAULT 0,
error_message TEXT,
payload_json TEXT NOT NULL DEFAULT '{}'
);
CREATE TABLE IF NOT EXISTS tally_sync_runs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
sync_type TEXT NOT NULL,
status TEXT NOT NULL,
started_at_utc TEXT NOT NULL,
completed_at_utc TEXT,
requested_by_user_id INTEGER,
records_received INTEGER NOT NULL DEFAULT 0,
error_message TEXT,
details_json TEXT NOT NULL DEFAULT '{}'
);
CREATE INDEX IF NOT EXISTS ix_tally_companies_last_seen
ON tally_companies(last_seen_at_utc);
CREATE INDEX IF NOT EXISTS ix_tally_company_mapping_client_active
ON tally_company_mapping(client_id, is_active);
CREATE INDEX IF NOT EXISTS ix_tally_company_mapping_guid
ON tally_company_mapping(tally_guid, is_active);
CREATE INDEX IF NOT EXISTS ix_tally_connection_history_checked
ON tally_connection_history(checked_at_utc);
CREATE INDEX IF NOT EXISTS ix_tally_sync_runs_started
ON tally_sync_runs(started_at_utc);
"""
)
self._ensure_mapping_columns(db)
now = _utc_now_iso()
metadata = {
"schema_version": str(ACT_SCHEMA_VERSION),
"client_id": str(_safe_client_id(client_id)),
"tenant_id": "" if tenant_id is None else str(int(tenant_id)),
"client_name": str(client_name or ""),
"created_by_user_id": "" if created_by_user_id is None else str(int(created_by_user_id)),
"storage_kind": "ERP_ACCOUNTING_ACT_SQLITE",
}
for key, value in metadata.items():
db.execute(
"""
INSERT INTO act_meta(key, value, updated_at_utc)
VALUES (?, ?, ?)
ON CONFLICT(key) DO UPDATE SET
value=excluded.value,
updated_at_utc=excluded.updated_at_utc
""",
(key, value, now),
)
return path
@staticmethod
def _ensure_mapping_columns(db: sqlite3.Connection) -> None:
columns = {row["name"] for row in db.execute("PRAGMA table_info(tally_company_mapping)").fetchall()}
additions = {
"registration_type_code": "TEXT NOT NULL DEFAULT ''",
"registration_number": "TEXT NOT NULL DEFAULT ''",
"registration_legal_name": "TEXT NOT NULL DEFAULT ''",
"registration_trade_name": "TEXT NOT NULL DEFAULT ''",
"business_unit_id": "INTEGER",
"client_branch_id": "INTEGER",
"updated_at_utc": "TEXT",
"unmapped_at_utc": "TEXT",
"unmapped_by_user_id": "INTEGER",
}
for name, ddl in additions.items():
if name not in columns:
db.execute(f"ALTER TABLE tally_company_mapping ADD COLUMN {name} {ddl}")
def record_tally_status(self, client_id: int, status: dict) -> None:
self.initialize(client_id)
now = _utc_now_iso()
companies: Sequence[dict] = status.get("companies") or []
with self.connect(client_id) as db:
db.execute("UPDATE tally_companies SET is_currently_loaded = 0")
for company in companies:
name = str(company.get("name") or "").strip()
if not name:
continue
guid = str(company.get("guid") or "").strip()
gstin = str(company.get("gstin") or "").strip().upper()
existing = db.execute(
"""
SELECT id FROM tally_companies
WHERE tally_guid = ? AND company_name = ?
LIMIT 1
""",
(guid, name),
).fetchone()
if existing:
db.execute(
"""
UPDATE tally_companies
SET gstin=?, last_seen_at_utc=?, is_currently_loaded=1
WHERE id=?
""",
(gstin, now, int(existing["id"])),
)
else:
db.execute(
"""
INSERT INTO tally_companies(
tally_guid, company_name, gstin,
first_seen_at_utc, last_seen_at_utc, is_currently_loaded
) VALUES (?, ?, ?, ?, ?, 1)
""",
(guid, name, gstin, now, now),
)
db.execute(
"""
INSERT INTO tally_connection_history(
checked_at_utc, connected, tally_url, company_count,
error_message, payload_json
) VALUES (?, ?, ?, ?, ?, ?)
""",
(
now,
1 if status.get("connected") else 0,
str(status.get("url") or ""),
int(status.get("company_count") or 0),
str(status.get("error") or "") or None,
json.dumps(status, ensure_ascii=False, separators=(",", ":")),
),
)
def map_company(
self,
client_id: int,
*,
tally_guid: str,
company_name: str,
gstin: str = "",
registration: dict | None = None,
mapped_by_user_id: int | None = None,
) -> dict:
self.initialize(client_id)
guid = str(tally_guid or "").strip()
name = str(company_name or "").strip()
if not guid:
raise AccountingActStoreError("Tally company GUID is required for permanent mapping.")
if not name:
raise AccountingActStoreError("Tally company name is required.")
registration = registration or {}
registration_id = registration.get("id")
registration_id = int(registration_id) if registration_id not in (None, "") else None
now = _utc_now_iso()
with self.connect(client_id) as db:
if registration_id is None:
db.execute(
"""
UPDATE tally_company_mapping
SET is_active=0, updated_at_utc=?, unmapped_at_utc=?
WHERE client_id=? AND registration_id IS NULL AND is_active=1
""",
(now, now, int(client_id)),
)
else:
db.execute(
"""
UPDATE tally_company_mapping
SET is_active=0, updated_at_utc=?, unmapped_at_utc=?
WHERE client_id=? AND registration_id=? AND is_active=1
""",
(now, now, int(client_id), registration_id),
)
db.execute(
"""
UPDATE tally_company_mapping
SET is_active=0, updated_at_utc=?, unmapped_at_utc=?
WHERE client_id=? AND tally_guid=? AND is_active=1
""",
(now, now, int(client_id), guid),
)
cursor = db.execute(
"""
INSERT INTO tally_company_mapping(
client_id, registration_id, registration_type_code,
registration_number, registration_legal_name,
registration_trade_name, business_unit_id, client_branch_id,
tally_guid, company_name, gstin, is_active,
mapped_at_utc, mapped_by_user_id, updated_at_utc
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?, ?)
""",
(
int(client_id),
registration_id,
str(registration.get("registration_type_code") or ""),
str(registration.get("registration_number") or ""),
str(registration.get("legal_name") or ""),
str(registration.get("trade_name") or ""),
registration.get("business_unit_id"),
registration.get("client_branch_id"),
guid,
name,
str(gstin or "").strip().upper(),
now,
mapped_by_user_id,
now,
),
)
mapping_id = int(cursor.lastrowid)
return self.get_mapping(client_id, mapping_id)
def unmap_company(self, client_id: int, mapping_id: int, *, unmapped_by_user_id: int | None = None) -> dict:
self.initialize(client_id)
now = _utc_now_iso()
with self.connect(client_id) as db:
row = db.execute(
"SELECT id FROM tally_company_mapping WHERE id=? AND client_id=? AND is_active=1",
(int(mapping_id), int(client_id)),
).fetchone()
if not row:
raise AccountingActStoreError("Active Tally company mapping was not found.")
db.execute(
"""
UPDATE tally_company_mapping
SET is_active=0, updated_at_utc=?, unmapped_at_utc=?, unmapped_by_user_id=?
WHERE id=? AND client_id=?
""",
(now, now, unmapped_by_user_id, int(mapping_id), int(client_id)),
)
return {"mapping_id": int(mapping_id), "unmapped": True}
def get_mapping(self, client_id: int, mapping_id: int) -> dict:
with self.connect(client_id) as db:
row = db.execute(
"""
SELECT *
FROM tally_company_mapping
WHERE id=? AND client_id=?
LIMIT 1
""",
(int(mapping_id), int(client_id)),
).fetchone()
if not row:
raise AccountingActStoreError("Tally company mapping was not found.")
return dict(row)
def snapshot(self, client_id: int) -> dict:
self.initialize(client_id)
with self.connect(client_id) as db:
meta_rows = db.execute("SELECT key, value FROM act_meta ORDER BY key").fetchall()
companies = db.execute(
"""
SELECT tally_guid AS guid, company_name AS name, gstin,
first_seen_at_utc, last_seen_at_utc, is_currently_loaded
FROM tally_companies
ORDER BY is_currently_loaded DESC, company_name COLLATE NOCASE
"""
).fetchall()
mappings = db.execute(
"""
SELECT id, client_id, registration_id, registration_type_code,
registration_number, registration_legal_name,
registration_trade_name, business_unit_id, client_branch_id,
tally_guid, company_name, gstin, is_active,
mapped_at_utc, mapped_by_user_id, updated_at_utc,
unmapped_at_utc, unmapped_by_user_id
FROM tally_company_mapping
WHERE is_active=1
ORDER BY CASE WHEN registration_id IS NULL THEN 0 ELSE 1 END,
registration_type_code, registration_number, id
"""
).fetchall()
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()
loaded_guids = {str(row["guid"] or "") for row in companies if int(row["is_currently_loaded"] or 0)}
mapping_rows = []
for row in mappings:
item = dict(row)
item["currently_loaded"] = bool(item.get("tally_guid") and item["tally_guid"] in loaded_guids)
mapping_rows.append(item)
return {
"db_path": str(self.db_path(client_id)),
"metadata": {row["key"]: row["value"] for row in meta_rows},
"companies": [dict(row) for row in companies],
"mappings": mapping_rows,
"latest_connection": dict(latest) if latest else None,
}
def default_accounting_root() -> Path:
env = os.getenv("AUDIT_ACCOUNTING_STORAGE_ROOT", "").strip()
if env:
return Path(env).expanduser().resolve()
return Path.cwd() / "data" / "accounting"