A R R R Associates
545 lines
19 KiB
Python
545 lines
19 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from datetime import datetime, timezone
|
|
from difflib import SequenceMatcher
|
|
|
|
from sqlalchemy import delete, func, select
|
|
|
|
from app.modules.accounting.opening_balance_models import (
|
|
AccountingOpeningBalanceRun,
|
|
AccountingOpeningLedgerItem,
|
|
AccountingOpeningMasterMapping,
|
|
AccountingOpeningStockItem,
|
|
)
|
|
|
|
|
|
def _utcnow():
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def _s(value):
|
|
return str(value or "").strip()
|
|
|
|
|
|
def _n(value):
|
|
try:
|
|
return float(value or 0)
|
|
except Exception:
|
|
return 0.0
|
|
|
|
|
|
def _norm(value):
|
|
return re.sub(r"[^a-z0-9]+", " ", _s(value).casefold()).strip()
|
|
|
|
|
|
def _near(a, b, tolerance=0.01):
|
|
return abs(float(a or 0) - float(b or 0)) <= tolerance
|
|
|
|
|
|
|
|
def _is_balance_sheet_ledger(row):
|
|
"""Return True only for ledger masters that belong to the Balance Sheet.
|
|
|
|
Tally exports ISREVENUE for Ledger masters. Revenue ledgers (income,
|
|
expense, purchase and sales accounts) must not carry an opening balance
|
|
into a new financial year. Stock Items are handled separately below.
|
|
"""
|
|
flag = _s((row or {}).get("is_revenue")).casefold()
|
|
if flag in {"yes", "y", "true", "1"}:
|
|
return False
|
|
if flag in {"no", "n", "false", "0"}:
|
|
return True
|
|
|
|
# Defensive fallback for older Tally responses that omitted ISREVENUE.
|
|
# Exclude only unmistakable P&L roots; retain Balance-Sheet masters.
|
|
parent = _norm((row or {}).get("parent"))
|
|
pnl_roots = (
|
|
"direct expenses", "indirect expenses", "direct incomes",
|
|
"indirect incomes", "sales accounts", "purchase accounts",
|
|
)
|
|
return parent not in pnl_roots
|
|
|
|
|
|
def _balance_sheet_ledgers(masters):
|
|
return [row for row in list((masters or {}).get("ledgers") or []) if _is_balance_sheet_ledger(row)]
|
|
|
|
def _mapping_index(db, *, tenant_id, client_id, master_type, previous_company_guid, current_company_guid):
|
|
rows = list(
|
|
db.execute(
|
|
select(AccountingOpeningMasterMapping).where(
|
|
AccountingOpeningMasterMapping.tenant_id == int(tenant_id),
|
|
AccountingOpeningMasterMapping.client_id == int(client_id),
|
|
AccountingOpeningMasterMapping.master_type == master_type,
|
|
AccountingOpeningMasterMapping.previous_company_guid == _s(previous_company_guid),
|
|
AccountingOpeningMasterMapping.current_company_guid == _s(current_company_guid),
|
|
)
|
|
).scalars().all()
|
|
)
|
|
return {row.previous_name_norm: row.current_name for row in rows}
|
|
|
|
|
|
def save_master_mapping(
|
|
db,
|
|
*,
|
|
tenant_id,
|
|
client_id,
|
|
master_type,
|
|
previous_company_guid,
|
|
previous_name,
|
|
current_company_guid,
|
|
current_name,
|
|
user_id,
|
|
note="",
|
|
):
|
|
master_type = _s(master_type).lower()
|
|
if master_type not in {"ledger", "stock_item"}:
|
|
raise ValueError("Master mapping type must be ledger or stock_item.")
|
|
previous_norm = _norm(previous_name)
|
|
if not previous_norm or not _s(current_name):
|
|
raise ValueError("Both previous-year and current-year master names are required.")
|
|
|
|
row = db.execute(
|
|
select(AccountingOpeningMasterMapping).where(
|
|
AccountingOpeningMasterMapping.tenant_id == int(tenant_id),
|
|
AccountingOpeningMasterMapping.client_id == int(client_id),
|
|
AccountingOpeningMasterMapping.master_type == master_type,
|
|
AccountingOpeningMasterMapping.previous_company_guid == _s(previous_company_guid),
|
|
AccountingOpeningMasterMapping.previous_name_norm == previous_norm,
|
|
AccountingOpeningMasterMapping.current_company_guid == _s(current_company_guid),
|
|
)
|
|
).scalar_one_or_none()
|
|
|
|
if row is None:
|
|
row = AccountingOpeningMasterMapping(
|
|
tenant_id=int(tenant_id),
|
|
client_id=int(client_id),
|
|
master_type=master_type,
|
|
previous_company_guid=_s(previous_company_guid),
|
|
previous_name=_s(previous_name),
|
|
previous_name_norm=previous_norm,
|
|
current_company_guid=_s(current_company_guid),
|
|
current_name=_s(current_name),
|
|
created_by_user_id=int(user_id),
|
|
)
|
|
else:
|
|
row.previous_name = _s(previous_name)
|
|
row.current_name = _s(current_name)
|
|
row.note = _s(note)
|
|
|
|
db.add(row)
|
|
db.commit()
|
|
return row
|
|
|
|
|
|
def _match_ledgers(previous, current, learned):
|
|
current_by_norm = {}
|
|
for row in current:
|
|
current_by_norm.setdefault(_norm(row.get("name")), []).append(row)
|
|
|
|
used = set()
|
|
result = []
|
|
|
|
for py in previous:
|
|
py_name = _s(py.get("name"))
|
|
py_norm = _norm(py_name)
|
|
target_name = learned.get(py_norm)
|
|
match = None
|
|
method = ""
|
|
confidence = 0
|
|
|
|
if target_name:
|
|
candidates = [row for row in current if _s(row.get("name")).casefold() == target_name.casefold()]
|
|
if len(candidates) == 1:
|
|
match = candidates[0]
|
|
method = "learned_mapping"
|
|
confidence = 100
|
|
|
|
if match is None:
|
|
exact = current_by_norm.get(py_norm, [])
|
|
if len(exact) == 1:
|
|
match = exact[0]
|
|
method = "exact_normalized_name"
|
|
confidence = 98
|
|
|
|
if match is not None:
|
|
used.add(id(match))
|
|
py_close = _n(py.get("closing_balance"))
|
|
cy_open = _n(match.get("opening_balance"))
|
|
diff = round(py_close - cy_open, 2)
|
|
status = "matched" if _near(py_close, cy_open) else "difference"
|
|
result.append({
|
|
"previous": py, "current": match, "difference": diff,
|
|
"status": status, "method": method, "confidence": confidence,
|
|
})
|
|
else:
|
|
result.append({
|
|
"previous": py, "current": None, "difference": _n(py.get("closing_balance")),
|
|
"status": "missing_in_current_year", "method": "", "confidence": 0,
|
|
})
|
|
|
|
for cy in current:
|
|
if id(cy) in used:
|
|
continue
|
|
result.append({
|
|
"previous": None, "current": cy, "difference": -_n(cy.get("opening_balance")),
|
|
"status": "new_in_current_year", "method": "", "confidence": 0,
|
|
})
|
|
|
|
return result
|
|
|
|
|
|
def _stock_score(py, cy):
|
|
py_name = _norm(py.get("name"))
|
|
cy_name = _norm(cy.get("name"))
|
|
score = int(round(SequenceMatcher(None, py_name, cy_name).ratio() * 60))
|
|
|
|
py_hsn = _s(py.get("hsn_code"))
|
|
cy_hsn = _s(cy.get("hsn_code"))
|
|
if py_hsn and cy_hsn:
|
|
if py_hsn == cy_hsn:
|
|
score += 25
|
|
elif py_hsn[:4] and py_hsn[:4] == cy_hsn[:4]:
|
|
score += 12
|
|
else:
|
|
score -= 20
|
|
|
|
if _norm(py.get("parent")) == _norm(cy.get("parent")) and _norm(py.get("parent")):
|
|
score += 10
|
|
if _norm(py.get("base_units")) == _norm(cy.get("base_units")) and _norm(py.get("base_units")):
|
|
score += 10
|
|
return max(0, min(100, score))
|
|
|
|
|
|
def _match_stock(previous, current, learned):
|
|
current_by_norm = {}
|
|
for row in current:
|
|
current_by_norm.setdefault(_norm(row.get("name")), []).append(row)
|
|
|
|
used = set()
|
|
result = []
|
|
|
|
for py in previous:
|
|
py_name = _s(py.get("name"))
|
|
py_norm = _norm(py_name)
|
|
match = None
|
|
method = ""
|
|
confidence = 0
|
|
|
|
target_name = learned.get(py_norm)
|
|
if target_name:
|
|
candidates = [row for row in current if _s(row.get("name")).casefold() == target_name.casefold()]
|
|
if len(candidates) == 1:
|
|
match = candidates[0]
|
|
method = "learned_mapping"
|
|
confidence = 100
|
|
|
|
if match is None:
|
|
exact = current_by_norm.get(py_norm, [])
|
|
if len(exact) == 1:
|
|
match = exact[0]
|
|
method = "exact_normalized_name"
|
|
confidence = 98
|
|
|
|
if match is None:
|
|
scored = sorted(
|
|
[(_stock_score(py, cy), cy) for cy in current if id(cy) not in used],
|
|
key=lambda row: (-row[0], _s(row[1].get("name"))),
|
|
)
|
|
if scored:
|
|
top = scored[0][0]
|
|
second = scored[1][0] if len(scored) > 1 else 0
|
|
if top >= 90 and top - second >= 10:
|
|
match = scored[0][1]
|
|
method = "unique_hsn_name_match"
|
|
confidence = top
|
|
|
|
if match is None:
|
|
result.append({
|
|
"previous": py, "current": None,
|
|
"qty_difference": _n(py.get("closing_balance")),
|
|
"value_difference": _n(py.get("closing_value")),
|
|
"status": "missing_in_current_year", "method": "", "confidence": 0,
|
|
})
|
|
continue
|
|
|
|
used.add(id(match))
|
|
py_qty = _n(py.get("closing_balance"))
|
|
py_val = _n(py.get("closing_value"))
|
|
cy_qty = _n(match.get("opening_balance"))
|
|
cy_val = _n(match.get("opening_value"))
|
|
qty_diff = round(py_qty - cy_qty, 6)
|
|
val_diff = round(py_val - cy_val, 2)
|
|
|
|
py_unit = _norm(py.get("base_units"))
|
|
cy_unit = _norm(match.get("base_units"))
|
|
if py_unit and cy_unit and py_unit != cy_unit:
|
|
status = "unit_difference"
|
|
elif _near(py_qty, cy_qty, 0.000001) and _near(py_val, cy_val):
|
|
status = "matched"
|
|
elif not _near(py_qty, cy_qty, 0.000001) and not _near(py_val, cy_val):
|
|
status = "quantity_and_value_difference"
|
|
elif not _near(py_qty, cy_qty, 0.000001):
|
|
status = "quantity_difference"
|
|
else:
|
|
status = "value_difference"
|
|
|
|
result.append({
|
|
"previous": py, "current": match,
|
|
"qty_difference": qty_diff, "value_difference": val_diff,
|
|
"status": status, "method": method, "confidence": confidence,
|
|
})
|
|
|
|
for cy in current:
|
|
if id(cy) in used:
|
|
continue
|
|
result.append({
|
|
"previous": None, "current": cy,
|
|
"qty_difference": -_n(cy.get("opening_balance")),
|
|
"value_difference": -_n(cy.get("opening_value")),
|
|
"status": "new_in_current_year", "method": "", "confidence": 0,
|
|
})
|
|
return result
|
|
|
|
|
|
def create_comparison_run(
|
|
db,
|
|
*,
|
|
tenant_id,
|
|
client_id,
|
|
previous_company,
|
|
current_company,
|
|
previous_masters,
|
|
current_masters,
|
|
user_id,
|
|
):
|
|
py_guid = _s(previous_company.get("guid"))
|
|
cy_guid = _s(current_company.get("guid"))
|
|
|
|
ledger_map = _mapping_index(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
client_id=client_id,
|
|
master_type="ledger",
|
|
previous_company_guid=py_guid,
|
|
current_company_guid=cy_guid,
|
|
)
|
|
stock_map = _mapping_index(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
client_id=client_id,
|
|
master_type="stock_item",
|
|
previous_company_guid=py_guid,
|
|
current_company_guid=cy_guid,
|
|
)
|
|
|
|
ledger_matches = _match_ledgers(
|
|
_balance_sheet_ledgers(previous_masters),
|
|
_balance_sheet_ledgers(current_masters),
|
|
ledger_map,
|
|
)
|
|
stock_matches = _match_stock(
|
|
list(previous_masters.get("stock_items") or []),
|
|
list(current_masters.get("stock_items") or []),
|
|
stock_map,
|
|
)
|
|
|
|
run = AccountingOpeningBalanceRun(
|
|
tenant_id=int(tenant_id),
|
|
client_id=int(client_id),
|
|
previous_company_name=_s(previous_company.get("name")),
|
|
previous_company_guid=py_guid,
|
|
current_company_name=_s(current_company.get("name")),
|
|
current_company_guid=cy_guid,
|
|
status="completed",
|
|
ledger_count=len(ledger_matches),
|
|
stock_item_count=len(stock_matches),
|
|
created_by_user_id=int(user_id),
|
|
)
|
|
db.add(run)
|
|
db.flush()
|
|
|
|
for row in ledger_matches:
|
|
py = row["previous"] or {}
|
|
cy = row["current"] or {}
|
|
db.add(
|
|
AccountingOpeningLedgerItem(
|
|
run_id=run.id,
|
|
previous_name=_s(py.get("name")),
|
|
previous_guid=_s(py.get("guid")),
|
|
previous_group=_s(py.get("parent")),
|
|
previous_closing_balance=_n(py.get("closing_balance")),
|
|
current_name=_s(cy.get("name")),
|
|
current_guid=_s(cy.get("guid")),
|
|
current_group=_s(cy.get("parent")),
|
|
current_opening_balance=_n(cy.get("opening_balance")),
|
|
difference=float(row["difference"]),
|
|
match_status=row["status"],
|
|
match_method=row["method"],
|
|
confidence=int(row["confidence"]),
|
|
)
|
|
)
|
|
|
|
for row in stock_matches:
|
|
py = row["previous"] or {}
|
|
cy = row["current"] or {}
|
|
db.add(
|
|
AccountingOpeningStockItem(
|
|
run_id=run.id,
|
|
previous_name=_s(py.get("name")),
|
|
previous_guid=_s(py.get("guid")),
|
|
previous_group=_s(py.get("parent")),
|
|
previous_hsn=_s(py.get("hsn_code")),
|
|
previous_unit=_s(py.get("base_units")),
|
|
previous_closing_qty=_n(py.get("closing_balance")),
|
|
previous_closing_value=_n(py.get("closing_value")),
|
|
previous_closing_rate=_s(py.get("closing_rate")),
|
|
current_name=_s(cy.get("name")),
|
|
current_guid=_s(cy.get("guid")),
|
|
current_group=_s(cy.get("parent")),
|
|
current_hsn=_s(cy.get("hsn_code")),
|
|
current_unit=_s(cy.get("base_units")),
|
|
current_opening_qty=_n(cy.get("opening_balance")),
|
|
current_opening_value=_n(cy.get("opening_value")),
|
|
quantity_difference=float(row["qty_difference"]),
|
|
value_difference=float(row["value_difference"]),
|
|
match_status=row["status"],
|
|
match_method=row["method"],
|
|
confidence=int(row["confidence"]),
|
|
)
|
|
)
|
|
|
|
summary = {
|
|
"ledgers": {},
|
|
"stock_items": {},
|
|
}
|
|
for key in {
|
|
"matched", "difference", "missing_in_current_year", "new_in_current_year"
|
|
}:
|
|
summary["ledgers"][key] = sum(1 for row in ledger_matches if row["status"] == key)
|
|
|
|
for key in {
|
|
"matched", "quantity_difference", "value_difference",
|
|
"quantity_and_value_difference", "unit_difference",
|
|
"missing_in_current_year", "new_in_current_year",
|
|
}:
|
|
summary["stock_items"][key] = sum(1 for row in stock_matches if row["status"] == key)
|
|
|
|
run.summary_json = json.dumps(summary, ensure_ascii=False)
|
|
db.add(run)
|
|
db.commit()
|
|
db.refresh(run)
|
|
return run
|
|
|
|
|
|
def list_runs(db, *, tenant_id, client_id, limit=30):
|
|
return list(
|
|
db.execute(
|
|
select(AccountingOpeningBalanceRun)
|
|
.where(
|
|
AccountingOpeningBalanceRun.tenant_id == int(tenant_id),
|
|
AccountingOpeningBalanceRun.client_id == int(client_id),
|
|
)
|
|
.order_by(AccountingOpeningBalanceRun.id.desc())
|
|
.limit(limit)
|
|
).scalars().all()
|
|
)
|
|
|
|
|
|
def ledger_items(db, *, run_id, status=""):
|
|
stmt = select(AccountingOpeningLedgerItem).where(
|
|
AccountingOpeningLedgerItem.run_id == int(run_id)
|
|
)
|
|
if _s(status):
|
|
stmt = stmt.where(AccountingOpeningLedgerItem.match_status == _s(status))
|
|
return list(db.execute(stmt.order_by(AccountingOpeningLedgerItem.previous_name, AccountingOpeningLedgerItem.current_name)).scalars().all())
|
|
|
|
|
|
def stock_items(db, *, run_id, status=""):
|
|
stmt = select(AccountingOpeningStockItem).where(
|
|
AccountingOpeningStockItem.run_id == int(run_id)
|
|
)
|
|
if _s(status):
|
|
stmt = stmt.where(AccountingOpeningStockItem.match_status == _s(status))
|
|
return list(db.execute(stmt.order_by(AccountingOpeningStockItem.previous_name, AccountingOpeningStockItem.current_name)).scalars().all())
|
|
|
|
|
|
def correction_payload(db, *, run, ledger_ids, stock_ids):
|
|
ledger_rows = []
|
|
stock_rows = []
|
|
|
|
for row in db.execute(
|
|
select(AccountingOpeningLedgerItem).where(
|
|
AccountingOpeningLedgerItem.run_id == run.id,
|
|
AccountingOpeningLedgerItem.id.in_([int(x) for x in ledger_ids] or [-1]),
|
|
)
|
|
).scalars().all():
|
|
if row.match_status != "difference" or not row.current_name:
|
|
continue
|
|
ledger_rows.append({
|
|
"item_id": row.id,
|
|
"name": row.current_name,
|
|
"guid": row.current_guid,
|
|
"expected_opening": row.current_opening_balance,
|
|
"target_opening": row.previous_closing_balance,
|
|
})
|
|
|
|
for row in db.execute(
|
|
select(AccountingOpeningStockItem).where(
|
|
AccountingOpeningStockItem.run_id == run.id,
|
|
AccountingOpeningStockItem.id.in_([int(x) for x in stock_ids] or [-1]),
|
|
)
|
|
).scalars().all():
|
|
if row.match_status not in {
|
|
"quantity_difference", "value_difference", "quantity_and_value_difference"
|
|
}:
|
|
continue
|
|
if not row.current_name or _norm(row.previous_unit) != _norm(row.current_unit):
|
|
continue
|
|
|
|
qty = row.previous_closing_qty
|
|
value = row.previous_closing_value
|
|
rate = abs(value / qty) if abs(qty) > 0.0000001 else 0
|
|
stock_rows.append({
|
|
"item_id": row.id,
|
|
"name": row.current_name,
|
|
"guid": row.current_guid,
|
|
"unit": row.current_unit or row.previous_unit,
|
|
"expected_opening_qty": row.current_opening_qty,
|
|
"expected_opening_value": row.current_opening_value,
|
|
"target_opening_qty": qty,
|
|
"target_opening_value": value,
|
|
"target_opening_rate": rate,
|
|
})
|
|
|
|
return ledger_rows, stock_rows
|
|
|
|
|
|
def apply_correction_results(db, *, run_id, result, user_id):
|
|
now = _utcnow()
|
|
for row in result.get("ledgers") or []:
|
|
item = db.get(AccountingOpeningLedgerItem, int(row.get("item_id") or 0))
|
|
if not item or item.run_id != int(run_id):
|
|
continue
|
|
item.correction_status = "verified" if row.get("verified") else "failed"
|
|
item.correction_note = _s(row.get("message"))
|
|
item.verified_opening_balance = _n(row.get("verified_opening"))
|
|
item.corrected_by_user_id = int(user_id)
|
|
item.corrected_at_utc = now
|
|
db.add(item)
|
|
|
|
for row in result.get("stock_items") or []:
|
|
item = db.get(AccountingOpeningStockItem, int(row.get("item_id") or 0))
|
|
if not item or item.run_id != int(run_id):
|
|
continue
|
|
item.correction_status = "verified" if row.get("verified") else "failed"
|
|
item.correction_note = _s(row.get("message"))
|
|
item.verified_opening_qty = _n(row.get("verified_opening_qty"))
|
|
item.verified_opening_value = _n(row.get("verified_opening_value"))
|
|
item.corrected_by_user_id = int(user_id)
|
|
item.corrected_at_utc = now
|
|
db.add(item)
|
|
|
|
db.commit()
|