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()