Add Phase 16 Chart of Accounts and ledger group intelligence
This commit is contained in:
A R R R Associates
@@ -0,0 +1,514 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import delete, func, or_, select
|
||||
|
||||
from app.modules.accounting.chart_models import (
|
||||
AccountingChartGroup,
|
||||
AccountingChartLedger,
|
||||
AccountingChartSyncRun,
|
||||
)
|
||||
|
||||
|
||||
LEDGER_ROLES = [
|
||||
("CUSTOMER", "Customer / Sundry Debtor"),
|
||||
("SUPPLIER", "Supplier / Sundry Creditor"),
|
||||
("SALES", "Sales"),
|
||||
("PURCHASE", "Purchase"),
|
||||
("DIRECT_INCOME", "Direct Income"),
|
||||
("OTHER_INCOME", "Other / Indirect Income"),
|
||||
("DIRECT_EXPENSE", "Direct Expense"),
|
||||
("INDIRECT_EXPENSE", "Indirect Expense"),
|
||||
("GST_INPUT", "GST Input Tax"),
|
||||
("GST_OUTPUT", "GST Output Tax"),
|
||||
("TAX_OTHER", "Other Duties & Taxes"),
|
||||
("BANK", "Bank"),
|
||||
("CASH", "Cash"),
|
||||
("FIXED_ASSET", "Fixed Asset"),
|
||||
("CURRENT_ASSET", "Current Asset"),
|
||||
("CURRENT_LIABILITY", "Current Liability"),
|
||||
("LOAN", "Loan / Borrowing"),
|
||||
("CAPITAL", "Capital"),
|
||||
("RESERVE", "Reserve / Surplus"),
|
||||
("INVENTORY", "Inventory / Stock"),
|
||||
("INVESTMENT", "Investment"),
|
||||
("ROUND_OFF", "Round Off"),
|
||||
("DISCOUNT", "Discount"),
|
||||
("FREIGHT_CHARGE", "Freight / Loading / Ancillary Charge"),
|
||||
("DEPRECIATION", "Depreciation"),
|
||||
("INTER_BRANCH", "Branch / Division"),
|
||||
("SUSPENSE", "Suspense / Review"),
|
||||
("REVIEW", "Review Required"),
|
||||
]
|
||||
|
||||
ROLE_LABELS = dict(LEDGER_ROLES)
|
||||
VALID_ROLE_CODES = set(ROLE_LABELS)
|
||||
|
||||
|
||||
def _utcnow():
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
def _s(value: Any) -> str:
|
||||
return str(value or "").strip()
|
||||
|
||||
|
||||
def _key(value: Any) -> str:
|
||||
return re.sub(r"[^a-z0-9]+", " ", _s(value).casefold()).strip()
|
||||
|
||||
|
||||
PRIMARY_ROLE_MAP = {
|
||||
"sundry debtors": ("CUSTOMER", 100, "Tally root group is Sundry Debtors."),
|
||||
"sundry creditors": ("SUPPLIER", 100, "Tally root group is Sundry Creditors."),
|
||||
"sales accounts": ("SALES", 100, "Tally root group is Sales Accounts."),
|
||||
"purchase accounts": ("PURCHASE", 100, "Tally root group is Purchase Accounts."),
|
||||
"direct incomes": ("DIRECT_INCOME", 98, "Tally root group is Direct Incomes."),
|
||||
"indirect incomes": ("OTHER_INCOME", 98, "Tally root group is Indirect Incomes."),
|
||||
"direct expenses": ("DIRECT_EXPENSE", 98, "Tally root group is Direct Expenses."),
|
||||
"indirect expenses": ("INDIRECT_EXPENSE", 98, "Tally root group is Indirect Expenses."),
|
||||
"bank accounts": ("BANK", 100, "Tally root group is Bank Accounts."),
|
||||
"bank od a c": ("LOAN", 96, "Tally root group is Bank OD A/c."),
|
||||
"cash in hand": ("CASH", 100, "Tally root group is Cash-in-Hand."),
|
||||
"fixed assets": ("FIXED_ASSET", 100, "Tally root group is Fixed Assets."),
|
||||
"current assets": ("CURRENT_ASSET", 92, "Tally root group is Current Assets."),
|
||||
"current liabilities": ("CURRENT_LIABILITY", 92, "Tally root group is Current Liabilities."),
|
||||
"loans liability": ("LOAN", 98, "Tally root group is Loans (Liability)."),
|
||||
"secured loans": ("LOAN", 98, "Tally root group is Secured Loans."),
|
||||
"unsecured loans": ("LOAN", 98, "Tally root group is Unsecured Loans."),
|
||||
"capital account": ("CAPITAL", 100, "Tally root group is Capital Account."),
|
||||
"reserves surplus": ("RESERVE", 100, "Tally root group is Reserves & Surplus."),
|
||||
"stock in hand": ("INVENTORY", 100, "Tally root group is Stock-in-Hand."),
|
||||
"investments": ("INVESTMENT", 100, "Tally root group is Investments."),
|
||||
"duties taxes": ("TAX_OTHER", 94, "Tally root group is Duties & Taxes."),
|
||||
"branch divisions": ("INTER_BRANCH", 98, "Tally root group is Branch/Divisions."),
|
||||
"suspense a c": ("SUSPENSE", 100, "Tally root group is Suspense A/c."),
|
||||
}
|
||||
|
||||
|
||||
def resolve_group_hierarchy(groups: list[dict]) -> dict[str, dict]:
|
||||
by_name = {_key(row.get("name")): dict(row) for row in groups if _s(row.get("name"))}
|
||||
resolved: dict[str, dict] = {}
|
||||
|
||||
def resolve(name: str, trail: set[str] | None = None):
|
||||
k = _key(name)
|
||||
if not k:
|
||||
return {"root_group_name": "", "depth": 0}
|
||||
if k in resolved:
|
||||
return resolved[k]
|
||||
|
||||
trail = set(trail or set())
|
||||
if k in trail:
|
||||
result = {"root_group_name": _s(name), "depth": 0}
|
||||
resolved[k] = result
|
||||
return result
|
||||
trail.add(k)
|
||||
|
||||
row = by_name.get(k)
|
||||
if not row:
|
||||
result = {"root_group_name": _s(name), "depth": 0}
|
||||
resolved[k] = result
|
||||
return result
|
||||
|
||||
parent = _s(row.get("parent"))
|
||||
parent_key = _key(parent)
|
||||
if not parent or parent_key in {"primary", "primary group"} or parent_key == k:
|
||||
result = {"root_group_name": _s(row.get("name")), "depth": 0}
|
||||
else:
|
||||
parent_result = resolve(parent, trail)
|
||||
result = {
|
||||
"root_group_name": parent_result["root_group_name"] or _s(row.get("name")),
|
||||
"depth": int(parent_result["depth"]) + 1,
|
||||
}
|
||||
resolved[k] = result
|
||||
return result
|
||||
|
||||
for row in groups:
|
||||
name = _s(row.get("name"))
|
||||
if name:
|
||||
resolve(name)
|
||||
return resolved
|
||||
|
||||
|
||||
def classify_ledger(ledger: dict, root_group_name: str) -> tuple[str, int, str]:
|
||||
name = _key(ledger.get("name"))
|
||||
parent = _key(ledger.get("parent"))
|
||||
root = _key(root_group_name)
|
||||
|
||||
# Strong ledger-name exceptions first.
|
||||
if "round off" in name or "rounded off" in name:
|
||||
return "ROUND_OFF", 100, "Ledger name identifies a round-off ledger."
|
||||
if "depreciation" in name:
|
||||
return "DEPRECIATION", 96, "Ledger name contains Depreciation."
|
||||
if "discount" in name:
|
||||
return "DISCOUNT", 88, "Ledger name contains Discount."
|
||||
if any(token in name for token in ("freight", "loading", "carriage outward", "delivery charge", "transport charge")):
|
||||
return "FREIGHT_CHARGE", 82, "Ledger name indicates freight/loading/ancillary charges."
|
||||
|
||||
# Duties & Taxes needs input/output distinction.
|
||||
if root == "duties taxes" or parent == "duties taxes":
|
||||
gst_text = " ".join(
|
||||
[
|
||||
name,
|
||||
_key(ledger.get("tax_type")),
|
||||
_key(ledger.get("gst_applicable")),
|
||||
]
|
||||
)
|
||||
if any(token in gst_text for token in ("input", "itc", "input cgst", "input sgst", "input igst")):
|
||||
return "GST_INPUT", 98, "Duties & Taxes ledger is identified as input GST."
|
||||
if any(token in gst_text for token in ("output", "output cgst", "output sgst", "output igst")):
|
||||
return "GST_OUTPUT", 98, "Duties & Taxes ledger is identified as output GST."
|
||||
return "TAX_OTHER", 94, "Ledger belongs to Duties & Taxes."
|
||||
|
||||
mapped = PRIMARY_ROLE_MAP.get(root)
|
||||
if mapped:
|
||||
return mapped
|
||||
|
||||
# Useful parent-level fallback when the root could not be resolved.
|
||||
mapped = PRIMARY_ROLE_MAP.get(parent)
|
||||
if mapped:
|
||||
code, confidence, reason = mapped
|
||||
return code, min(confidence, 90), f"Parent group classification: {reason}"
|
||||
|
||||
# Tally metadata fallback.
|
||||
tax_type = _key(ledger.get("tax_type"))
|
||||
if tax_type and tax_type not in {"not applicable", "none"}:
|
||||
return "TAX_OTHER", 72, f"Tally TaxType is '{_s(ledger.get('tax_type'))}'."
|
||||
|
||||
return "REVIEW", 25, "No sufficiently reliable Chart-of-Accounts rule matched this ledger."
|
||||
|
||||
|
||||
def persist_chart_snapshot(
|
||||
db,
|
||||
*,
|
||||
tenant_id: int,
|
||||
client_id: int,
|
||||
tally_guid: str,
|
||||
company_name: str,
|
||||
groups: list[dict],
|
||||
ledgers: list[dict],
|
||||
user_id: int,
|
||||
):
|
||||
tenant_id = int(tenant_id)
|
||||
client_id = int(client_id)
|
||||
tally_guid = _s(tally_guid)
|
||||
company_name = _s(company_name)
|
||||
if not tally_guid:
|
||||
raise ValueError("Tally company GUID is required.")
|
||||
if not company_name:
|
||||
raise ValueError("Tally company name is required.")
|
||||
|
||||
hierarchy = resolve_group_hierarchy(groups)
|
||||
now = _utcnow()
|
||||
|
||||
# Preserve manual ledger overrides across refresh.
|
||||
existing_ledgers = list(
|
||||
db.execute(
|
||||
select(AccountingChartLedger).where(
|
||||
AccountingChartLedger.tenant_id == tenant_id,
|
||||
AccountingChartLedger.client_id == client_id,
|
||||
AccountingChartLedger.tally_guid == tally_guid,
|
||||
)
|
||||
).scalars().all()
|
||||
)
|
||||
manual = {
|
||||
_key(row.name): {
|
||||
"manual_role_code": row.manual_role_code,
|
||||
"manual_note": row.manual_note,
|
||||
"manual_by_user_id": row.manual_by_user_id,
|
||||
"manual_at_utc": row.manual_at_utc,
|
||||
}
|
||||
for row in existing_ledgers
|
||||
if row.manual_role_code
|
||||
}
|
||||
|
||||
db.execute(
|
||||
delete(AccountingChartGroup).where(
|
||||
AccountingChartGroup.tenant_id == tenant_id,
|
||||
AccountingChartGroup.client_id == client_id,
|
||||
AccountingChartGroup.tally_guid == tally_guid,
|
||||
)
|
||||
)
|
||||
db.execute(
|
||||
delete(AccountingChartLedger).where(
|
||||
AccountingChartLedger.tenant_id == tenant_id,
|
||||
AccountingChartLedger.client_id == client_id,
|
||||
AccountingChartLedger.tally_guid == tally_guid,
|
||||
)
|
||||
)
|
||||
|
||||
for row in groups:
|
||||
name = _s(row.get("name"))
|
||||
if not name:
|
||||
continue
|
||||
info = hierarchy.get(_key(name), {"root_group_name": name, "depth": 0})
|
||||
db.add(
|
||||
AccountingChartGroup(
|
||||
tenant_id=tenant_id,
|
||||
client_id=client_id,
|
||||
tally_guid=tally_guid,
|
||||
company_name=company_name,
|
||||
master_guid=_s(row.get("guid")),
|
||||
name=name,
|
||||
parent_name=_s(row.get("parent")),
|
||||
root_group_name=_s(info.get("root_group_name")),
|
||||
depth=int(info.get("depth") or 0),
|
||||
reserved_name=_s(row.get("reserved_name")),
|
||||
is_revenue=_s(row.get("is_revenue")),
|
||||
is_deemed_positive=_s(row.get("is_deemed_positive")),
|
||||
source_payload_json=json.dumps(row, ensure_ascii=False, separators=(",", ":")),
|
||||
synced_at_utc=now,
|
||||
)
|
||||
)
|
||||
|
||||
classified_count = 0
|
||||
review_count = 0
|
||||
for row in ledgers:
|
||||
name = _s(row.get("name"))
|
||||
if not name:
|
||||
continue
|
||||
parent = _s(row.get("parent"))
|
||||
parent_info = hierarchy.get(
|
||||
_key(parent),
|
||||
{"root_group_name": parent, "depth": 0},
|
||||
)
|
||||
root_name = _s(parent_info.get("root_group_name")) or parent
|
||||
role, confidence, reason = classify_ledger(row, root_name)
|
||||
if role == "REVIEW":
|
||||
review_count += 1
|
||||
else:
|
||||
classified_count += 1
|
||||
|
||||
override = manual.get(_key(name), {})
|
||||
db.add(
|
||||
AccountingChartLedger(
|
||||
tenant_id=tenant_id,
|
||||
client_id=client_id,
|
||||
tally_guid=tally_guid,
|
||||
company_name=company_name,
|
||||
master_guid=_s(row.get("guid")),
|
||||
name=name,
|
||||
parent_group_name=parent,
|
||||
root_group_name=root_name,
|
||||
opening_balance=float(row.get("opening_balance") or 0),
|
||||
closing_balance=float(row.get("closing_balance") or 0),
|
||||
is_billwise_on=_s(row.get("is_billwise_on")),
|
||||
tax_type=_s(row.get("tax_type")),
|
||||
gst_applicable=_s(row.get("gst_applicable")),
|
||||
gst_registration_type=_s(row.get("gst_registration_type")),
|
||||
gst_type_of_supply=_s(row.get("gst_type_of_supply")),
|
||||
hsn_code=_s(row.get("hsn_code")),
|
||||
party_gstin=re.sub(r"[^A-Z0-9]", "", _s(row.get("party_gstin")).upper())[:15],
|
||||
suggested_role_code=role,
|
||||
suggested_confidence=int(confidence),
|
||||
suggestion_reason=reason,
|
||||
manual_role_code=_s(override.get("manual_role_code")),
|
||||
manual_note=_s(override.get("manual_note")),
|
||||
manual_by_user_id=override.get("manual_by_user_id"),
|
||||
manual_at_utc=override.get("manual_at_utc"),
|
||||
source_payload_json=json.dumps(row, ensure_ascii=False, separators=(",", ":")),
|
||||
synced_at_utc=now,
|
||||
)
|
||||
)
|
||||
|
||||
run = AccountingChartSyncRun(
|
||||
tenant_id=tenant_id,
|
||||
client_id=client_id,
|
||||
tally_guid=tally_guid,
|
||||
company_name=company_name,
|
||||
status="completed",
|
||||
group_count=len([x for x in groups if _s(x.get("name"))]),
|
||||
ledger_count=len([x for x in ledgers if _s(x.get("name"))]),
|
||||
classified_count=classified_count,
|
||||
review_count=review_count,
|
||||
requested_by_user_id=user_id,
|
||||
)
|
||||
db.add(run)
|
||||
db.commit()
|
||||
db.refresh(run)
|
||||
return run
|
||||
|
||||
|
||||
def chart_summary(db, *, tenant_id: int, client_id: int, tally_guid: str = ""):
|
||||
where = [
|
||||
AccountingChartLedger.tenant_id == int(tenant_id),
|
||||
AccountingChartLedger.client_id == int(client_id),
|
||||
]
|
||||
group_where = [
|
||||
AccountingChartGroup.tenant_id == int(tenant_id),
|
||||
AccountingChartGroup.client_id == int(client_id),
|
||||
]
|
||||
if tally_guid:
|
||||
where.append(AccountingChartLedger.tally_guid == tally_guid)
|
||||
group_where.append(AccountingChartGroup.tally_guid == tally_guid)
|
||||
|
||||
ledger_count = int(db.scalar(select(func.count(AccountingChartLedger.id)).where(*where)) or 0)
|
||||
group_count = int(db.scalar(select(func.count(AccountingChartGroup.id)).where(*group_where)) or 0)
|
||||
manual_count = int(
|
||||
db.scalar(
|
||||
select(func.count(AccountingChartLedger.id)).where(
|
||||
*where,
|
||||
AccountingChartLedger.manual_role_code != "",
|
||||
)
|
||||
)
|
||||
or 0
|
||||
)
|
||||
review_count = int(
|
||||
db.scalar(
|
||||
select(func.count(AccountingChartLedger.id)).where(
|
||||
*where,
|
||||
AccountingChartLedger.manual_role_code == "",
|
||||
AccountingChartLedger.suggested_role_code == "REVIEW",
|
||||
)
|
||||
)
|
||||
or 0
|
||||
)
|
||||
return {
|
||||
"groups": group_count,
|
||||
"ledgers": ledger_count,
|
||||
"manual": manual_count,
|
||||
"review": review_count,
|
||||
"classified": max(0, ledger_count - review_count),
|
||||
}
|
||||
|
||||
|
||||
def list_ledgers(
|
||||
db,
|
||||
*,
|
||||
tenant_id: int,
|
||||
client_id: int,
|
||||
tally_guid: str = "",
|
||||
q: str = "",
|
||||
role: str = "",
|
||||
root_group: str = "",
|
||||
page: int = 1,
|
||||
per_page: int = 50,
|
||||
):
|
||||
where = [
|
||||
AccountingChartLedger.tenant_id == int(tenant_id),
|
||||
AccountingChartLedger.client_id == int(client_id),
|
||||
]
|
||||
if tally_guid:
|
||||
where.append(AccountingChartLedger.tally_guid == tally_guid)
|
||||
if q.strip():
|
||||
term = f"%{q.strip()}%"
|
||||
where.append(
|
||||
or_(
|
||||
AccountingChartLedger.name.ilike(term),
|
||||
AccountingChartLedger.parent_group_name.ilike(term),
|
||||
AccountingChartLedger.root_group_name.ilike(term),
|
||||
AccountingChartLedger.party_gstin.ilike(term),
|
||||
)
|
||||
)
|
||||
if root_group.strip():
|
||||
where.append(AccountingChartLedger.root_group_name == root_group.strip())
|
||||
if role.strip():
|
||||
role = role.strip().upper()
|
||||
where.append(
|
||||
or_(
|
||||
AccountingChartLedger.manual_role_code == role,
|
||||
(
|
||||
AccountingChartLedger.manual_role_code == ""
|
||||
)
|
||||
& (AccountingChartLedger.suggested_role_code == role),
|
||||
)
|
||||
)
|
||||
|
||||
total = int(db.scalar(select(func.count(AccountingChartLedger.id)).where(*where)) or 0)
|
||||
per_page = max(20, min(200, int(per_page or 50)))
|
||||
pages = max(1, (total + per_page - 1) // per_page)
|
||||
page = max(1, min(int(page or 1), pages))
|
||||
|
||||
rows = list(
|
||||
db.execute(
|
||||
select(AccountingChartLedger)
|
||||
.where(*where)
|
||||
.order_by(
|
||||
AccountingChartLedger.root_group_name,
|
||||
AccountingChartLedger.parent_group_name,
|
||||
AccountingChartLedger.name,
|
||||
)
|
||||
.offset((page - 1) * per_page)
|
||||
.limit(per_page)
|
||||
).scalars().all()
|
||||
)
|
||||
return rows, total, page, pages
|
||||
|
||||
|
||||
def list_groups(db, *, tenant_id: int, client_id: int, tally_guid: str = ""):
|
||||
where = [
|
||||
AccountingChartGroup.tenant_id == int(tenant_id),
|
||||
AccountingChartGroup.client_id == int(client_id),
|
||||
]
|
||||
if tally_guid:
|
||||
where.append(AccountingChartGroup.tally_guid == tally_guid)
|
||||
return list(
|
||||
db.execute(
|
||||
select(AccountingChartGroup)
|
||||
.where(*where)
|
||||
.order_by(
|
||||
AccountingChartGroup.root_group_name,
|
||||
AccountingChartGroup.depth,
|
||||
AccountingChartGroup.name,
|
||||
)
|
||||
).scalars().all()
|
||||
)
|
||||
|
||||
|
||||
def root_group_options(db, *, tenant_id: int, client_id: int, tally_guid: str = ""):
|
||||
where = [
|
||||
AccountingChartLedger.tenant_id == int(tenant_id),
|
||||
AccountingChartLedger.client_id == int(client_id),
|
||||
AccountingChartLedger.root_group_name != "",
|
||||
]
|
||||
if tally_guid:
|
||||
where.append(AccountingChartLedger.tally_guid == tally_guid)
|
||||
return list(
|
||||
db.execute(
|
||||
select(AccountingChartLedger.root_group_name)
|
||||
.where(*where)
|
||||
.distinct()
|
||||
.order_by(AccountingChartLedger.root_group_name)
|
||||
).scalars().all()
|
||||
)
|
||||
|
||||
|
||||
def set_manual_ledger_role(
|
||||
db,
|
||||
*,
|
||||
tenant_id: int,
|
||||
client_id: int,
|
||||
ledger_id: int,
|
||||
role_code: str,
|
||||
note: str,
|
||||
user_id: int,
|
||||
):
|
||||
row = db.execute(
|
||||
select(AccountingChartLedger).where(
|
||||
AccountingChartLedger.id == int(ledger_id),
|
||||
AccountingChartLedger.tenant_id == int(tenant_id),
|
||||
AccountingChartLedger.client_id == int(client_id),
|
||||
)
|
||||
).scalar_one_or_none()
|
||||
if not row:
|
||||
raise ValueError("Chart-of-Accounts ledger was not found.")
|
||||
|
||||
role = _s(role_code).upper()
|
||||
if role and role not in VALID_ROLE_CODES:
|
||||
raise ValueError("Invalid accounting ledger role.")
|
||||
|
||||
row.manual_role_code = role
|
||||
row.manual_note = _s(note)
|
||||
row.manual_by_user_id = int(user_id) if role else None
|
||||
row.manual_at_utc = _utcnow() if role else None
|
||||
db.add(row)
|
||||
db.commit()
|
||||
db.refresh(row)
|
||||
return row
|
||||
|
||||
|
||||
def effective_role(row: AccountingChartLedger) -> str:
|
||||
return row.manual_role_code or row.suggested_role_code or "REVIEW"
|
||||
Reference in New Issue
Block a user