from __future__ import annotations from collections import defaultdict from dataclasses import dataclass from io import BytesIO from typing import Any, Iterable from openpyxl import Workbook from openpyxl.styles import Alignment, Border, Font, PatternFill, Side from openpyxl.utils import get_column_letter from sqlalchemy import select from app.modules.accounting.accounting_mirror_service import mirror_query, trial_balance from app.modules.accounting.financial_statements_models import AccountingFinancialLedgerMapping # Canonical schedules. The labels intentionally follow the conventional # financial-statement presentation seen in the firm's existing statements. SCHEDULES: dict[str, dict[str, Any]] = { "PARTNER_FUNDS": {"statement": "BS", "label": "Partner's / Proprietor's Funds", "bs_side": "LIABILITY", "order": 10}, "RESERVES": {"statement": "BS", "label": "Reserves & Surplus", "bs_side": "LIABILITY", "order": 20}, "NONCURRENT_BORROWINGS": {"statement": "BS", "label": "Long-Term Borrowings", "bs_side": "LIABILITY", "order": 30}, "SHORT_TERM_BORROWINGS": {"statement": "BS", "label": "Short-Term Borrowings", "bs_side": "LIABILITY", "order": 40}, "TRADE_PAYABLES": {"statement": "BS", "label": "Trade Payables", "bs_side": "LIABILITY", "order": 50}, "STATUTORY_LIABILITIES": {"statement": "BS", "label": "Statutory Liabilities", "bs_side": "LIABILITY", "order": 60}, "PROVISIONS": {"statement": "BS", "label": "Short-Term Provisions", "bs_side": "LIABILITY", "order": 70}, "OTHER_CURRENT_LIABILITIES": {"statement": "BS", "label": "Other Current Liabilities", "bs_side": "LIABILITY", "order": 80}, "PPE": {"statement": "BS", "label": "Property, Plant and Equipment", "bs_side": "ASSET", "order": 110}, "INVESTMENTS": {"statement": "BS", "label": "Investments", "bs_side": "ASSET", "order": 120}, "INVENTORIES": {"statement": "BS", "label": "Inventories", "bs_side": "ASSET", "order": 130}, "LOANS_ADVANCES": {"statement": "BS", "label": "Short-Term Loans and Advances", "bs_side": "ASSET", "order": 140}, "TRADE_RECEIVABLES": {"statement": "BS", "label": "Trade Receivables", "bs_side": "ASSET", "order": 150}, "CASH_BANK": {"statement": "BS", "label": "Cash and Bank Balances", "bs_side": "ASSET", "order": 160}, "OTHER_CURRENT_ASSETS": {"statement": "BS", "label": "Other Current Assets", "bs_side": "ASSET", "order": 170}, "MISC_ASSET": {"statement": "BS", "label": "Miscellaneous Expenditure / Other Assets", "bs_side": "ASSET", "order": 180}, "REVENUE": {"statement": "PL", "label": "Revenue from operations", "pl_side": "INCOME", "order": 210}, "OTHER_INCOME": {"statement": "PL", "label": "Other Income", "pl_side": "INCOME", "order": 220}, "PURCHASES": {"statement": "PL", "label": "Purchases", "pl_side": "EXPENSE", "order": 230}, "DIRECT_EXPENSES": {"statement": "PL", "label": "Direct Expenses", "pl_side": "EXPENSE", "order": 240}, "EMPLOYEE_BENEFITS": {"statement": "PL", "label": "Employee Benefits Expense", "pl_side": "EXPENSE", "order": 250}, "FINANCE_COST": {"statement": "PL", "label": "Finance Cost", "pl_side": "EXPENSE", "order": 260}, "DEPRECIATION": {"statement": "PL", "label": "Depreciation and Amortization Expense", "pl_side": "EXPENSE", "order": 270}, "OTHER_EXPENSES": {"statement": "PL", "label": "Other Expenses", "pl_side": "EXPENSE", "order": 280}, "UNMAPPED": {"statement": "UNMAPPED", "label": "Unmapped / Review Required", "order": 999}, } MAPPING_CHOICES = [(code, meta["label"], meta["statement"]) for code, meta in SCHEDULES.items() if code != "UNMAPPED"] # TallyPrime predefined subgroup hierarchy. Some mirror extractions expose the # subgroup name on the ledger but do not carry parent rows for predefined # groups in group_master. Keep this local fallback so standard Tally groups # never become "Unmapped" merely because the hierarchy master is incomplete. DEFAULT_TALLY_GROUP_PARENTS: dict[str, str] = { "bank accounts": "Current Assets", "cash-in-hand": "Current Assets", "cash in hand": "Current Assets", "deposits (asset)": "Current Assets", "deposits asset": "Current Assets", "loans and advances (asset)": "Current Assets", "loans & advances (asset)": "Current Assets", "stock-in-hand": "Current Assets", "stock in hand": "Current Assets", "sundry debtors": "Current Assets", "sundry debtor": "Current Assets", "duties and taxes": "Current Liabilities", "duties & taxes": "Current Liabilities", "provisions": "Current Liabilities", "sundry creditors": "Current Liabilities", "sundry creditor": "Current Liabilities", "bank od a/c": "Loans (Liability)", "bank od account": "Loans (Liability)", "secured loans": "Loans (Liability)", "unsecured loans": "Loans (Liability)", "reserves & surplus": "Capital Account", "reserves and surplus": "Capital Account", } def _norm(value: Any) -> str: return " ".join(str(value or "").strip().lower().replace("&", "and").split()) def _amount(value: Any) -> float: try: return round(float(value or 0), 2) except (TypeError, ValueError): return 0.0 def _root_group(group_name: str, parents: dict[str, str]) -> str: current = str(group_name or "").strip() if not current: return "" seen: set[str] = set() last = current while current and _norm(current) not in seen: key = _norm(current) seen.add(key) last = current parent = parents.get(key, "") or DEFAULT_TALLY_GROUP_PARENTS.get(key, "") if not parent or _norm(parent) in {"primary", ""}: break current = parent return last def _keyword_any(text: str, words: Iterable[str]) -> bool: value = _norm(text) return any(_norm(word) in value for word in words) def classify_default(ledger_name: str, parent_group: str, root_group: str, closing_balance: float = 0.0) -> str: ledger = _norm(ledger_name) parent = _norm(parent_group) root = _norm(root_group) combined = f"{ledger} {parent} {root}" # First classify the exact Tally default subgroup carried on the ledger. # This works even when group_master does not contain predefined group rows. if parent in {"bank accounts", "bank account"}: # Tally balance convention: positive = credit, negative = debit. A bank # account with a credit balance is presented as an overdraft/borrowing. return "SHORT_TERM_BORROWINGS" if _amount(closing_balance) > 0 else "CASH_BANK" if parent in {"cash-in-hand", "cash in hand"}: return "CASH_BANK" if parent in {"deposits (asset)", "deposits asset", "loans and advances (asset)", "loans & advances (asset)"}: return "LOANS_ADVANCES" if parent in {"stock-in-hand", "stock in hand"}: return "INVENTORIES" if parent in {"sundry debtors", "sundry debtor"}: return "TRADE_RECEIVABLES" if parent in {"sundry creditors", "sundry creditor"}: return "TRADE_PAYABLES" if parent in {"bank od a/c", "bank od account"}: return "SHORT_TERM_BORROWINGS" if parent in {"secured loans", "unsecured loans"}: return "NONCURRENT_BORROWINGS" if parent in {"reserves & surplus", "reserves and surplus"}: return "RESERVES" if parent == "provisions": return "PROVISIONS" # Common custom groups used by firms. These are still deterministic group # mappings; users can override any exceptional ledger from the UI. if parent in {"current account", "partners current account", "partner current account"}: return "PARTNER_FUNDS" if parent in {"creditor for expenses", "expense creditors", "expenses payable"}: return "OTHER_CURRENT_LIABILITIES" # Duties & Taxes can legitimately hold either asset or liability balances. # Prefer the ledger's purpose, then use Tally's balance sign as a fallback. if parent in {"duties and taxes", "duties & taxes"}: if _keyword_any(ledger, [ "input", "receivable", "advance tax", "tcs - pur", "tcs pur", "tds receivable", "tcs receivable", "gst input", "itc" ]): return "OTHER_CURRENT_ASSETS" if _keyword_any(ledger, [ "payable", "output", "tds on", "tcs payable", "rcm payable", "pf payable", "esi payable" ]): return "STATUTORY_LIABILITIES" return "OTHER_CURRENT_ASSETS" if _amount(closing_balance) < 0 else "STATUTORY_LIABILITIES" if root in {"sales accounts", "sales account"}: return "REVENUE" if root in {"purchase accounts", "purchase account"}: return "PURCHASES" if root in {"direct expenses", "direct expense"}: return "DIRECT_EXPENSES" if root in {"direct incomes", "direct income"}: return "REVENUE" if root in {"indirect incomes", "indirect income"}: return "OTHER_INCOME" if root in {"indirect expenses", "indirect expense"}: if _keyword_any(combined, ["salary", "salaries", "wages", "bonus", "staff welfare", "employee", "remuneration"]): return "EMPLOYEE_BENEFITS" if _keyword_any(combined, ["interest", "bank charge", "finance charge", "loan processing", "od interest"]): return "FINANCE_COST" if _keyword_any(combined, ["depreciation", "amortisation", "amortization"]): return "DEPRECIATION" return "OTHER_EXPENSES" if root in {"capital account", "capital accounts"}: if "reserve" in combined or "surplus" in combined: return "RESERVES" return "PARTNER_FUNDS" if root in {"loans (liability)", "loans liability"}: if _keyword_any(combined, ["bank od", "overdraft", "cash credit", "cc a/c", "cc account", "working capital"]): return "SHORT_TERM_BORROWINGS" return "NONCURRENT_BORROWINGS" if root in {"current liabilities", "current liability"}: if parent in {"sundry creditors", "sundry creditor"} or "creditor" in combined or "trade payable" in combined: return "TRADE_PAYABLES" if parent in {"duties and taxes", "duties & taxes"} or _keyword_any(combined, ["gst", "tds", "tcs", "pf payable", "esi payable", "tax payable"]): return "STATUTORY_LIABILITIES" if parent == "provisions" or "provision" in combined or "payable" in ledger: return "PROVISIONS" return "OTHER_CURRENT_LIABILITIES" if root in {"fixed assets", "fixed asset"}: return "PPE" if root in {"investments", "investment"}: return "INVESTMENTS" if root in {"current assets", "current asset"}: if parent in {"stock-in-hand", "stock in hand"} or _keyword_any(combined, ["stock in hand", "inventory", "closing stock"]): return "INVENTORIES" if parent in {"sundry debtors", "sundry debtor"} or "debtor" in combined or "trade receivable" in combined: return "TRADE_RECEIVABLES" if parent in {"cash-in-hand", "cash in hand", "bank accounts", "bank account"} or _keyword_any(combined, ["cash in hand", "bank account"]): return "CASH_BANK" if parent in {"loans and advances (asset)", "loans & advances (asset)", "deposits (asset)", "deposits asset"} or _keyword_any(combined, ["deposit", "advance tax", "tds receivable", "tcs receivable", "gst input", "advance"]): return "LOANS_ADVANCES" return "OTHER_CURRENT_ASSETS" if root in {"misc. expenses (asset)", "misc expenses (asset)", "suspense a/c", "suspense account"}: return "MISC_ASSET" if root in {"branch/divisions", "branch / divisions"}: return "OTHER_CURRENT_ASSETS" # Tally's special Profit & Loss A/c is a balance-sheet carried-forward item # when it appears as a Primary ledger; do not leave it unmapped. if parent == "primary" and _keyword_any(ledger, ["profit & loss", "profit and loss", "p&l"]): return "RESERVES" # Last deterministic name fallbacks for accounts whose custom group master # is not available. These do not override recognised Tally groups. if _keyword_any(ledger, ["capital account"]): return "PARTNER_FUNDS" if _keyword_any(ledger, ["current account"]) and not _keyword_any(ledger, ["bank current account"]): return "PARTNER_FUNDS" return "UNMAPPED" @dataclass class LedgerLine: ledger_name: str parent_group: str root_group: str opening_balance: float debit: float credit: float closing_balance: float opening_debit: float opening_credit: float closing_debit: float closing_credit: float schedule_code: str schedule_label: str statement_code: str mapped_by: str @property def reporting_amount(self) -> float: """Financial-statement amount with the natural schedule sign. Tally balance convention in the mirror is positive=credit and negative=debit. Assets/expenses therefore use -balance while liabilities/income use +balance. Abnormal balances remain negative and print in parentheses rather than being silently converted to absolute. """ meta = SCHEDULES.get(self.schedule_code, SCHEDULES["UNMAPPED"]) if meta.get("statement") == "BS": return round(self.closing_balance if meta.get("bs_side") == "LIABILITY" else -self.closing_balance, 2) if meta.get("statement") == "PL": return round(self.closing_balance if meta.get("pl_side") == "INCOME" else -self.closing_balance, 2) return round(self.closing_balance, 2) @property def opening_reporting_amount(self) -> float: meta = SCHEDULES.get(self.schedule_code, SCHEDULES["UNMAPPED"]) if meta.get("statement") == "BS": return round(self.opening_balance if meta.get("bs_side") == "LIABILITY" else -self.opening_balance, 2) if meta.get("statement") == "PL": return round(self.opening_balance if meta.get("pl_side") == "INCOME" else -self.opening_balance, 2) return round(self.opening_balance, 2) def _balance_split(value: float) -> tuple[float, float]: # Tally mirror convention: negative balances are debit, positive are credit. value = _amount(value) return (abs(value) if value < 0 else 0.0, value if value > 0 else 0.0) def read_overrides(db, *, tenant_id: int, client_id: int) -> dict[str, AccountingFinancialLedgerMapping]: rows = db.execute( select(AccountingFinancialLedgerMapping).where( AccountingFinancialLedgerMapping.tenant_id == int(tenant_id), AccountingFinancialLedgerMapping.client_id == int(client_id), ) ).scalars().all() return {_norm(row.ledger_name): row for row in rows} def load_mirror_trial_balance( *, node_code: str, accounting_payload: dict[str, Any], db=None, tenant_id: int = 0, client_id: int = 0 ) -> dict[str, Any]: tb = trial_balance(node_code=node_code, accounting_payload=accounting_payload, limit=5000) group_rows: list[dict[str, Any]] = [] hierarchy_warning = "" try: groups = mirror_query( node_code=node_code, accounting_payload=accounting_payload, query="group_master", filters={"limit": 5000}, ) group_rows = list(groups.get("rows") or []) except Exception as exc: hierarchy_warning = str(exc) parents = { _norm(row.get("name")): str(row.get("parent") or "").strip() for row in group_rows if str(row.get("name") or "").strip() } for child, parent_name in DEFAULT_TALLY_GROUP_PARENTS.items(): parents.setdefault(child, parent_name) overrides = read_overrides(db, tenant_id=tenant_id, client_id=client_id) if db is not None else {} lines: list[LedgerLine] = [] for row in tb.get("rows") or []: name = str(row.get("ledger_name") or "").strip() if not name: continue parent = str(row.get("parent_group") or "").strip() root = _root_group(parent, parents) if parents else parent override = overrides.get(_norm(name)) if override and override.schedule_code in SCHEDULES: code = override.schedule_code mapped_by = "manual" else: code = classify_default(name, parent, root, _amount(row.get("closing_balance"))) mapped_by = "tally_group" meta = SCHEDULES.get(code, SCHEDULES["UNMAPPED"]) opening = _amount(row.get("opening_balance")) closing = _amount(row.get("closing_balance")) od, oc = _balance_split(opening) cd, cc = _balance_split(closing) lines.append(LedgerLine( ledger_name=name, parent_group=parent, root_group=root, opening_balance=opening, debit=_amount(row.get("debit")), credit=_amount(row.get("credit")), closing_balance=closing, opening_debit=od, opening_credit=oc, closing_debit=cd, closing_credit=cc, schedule_code=code, schedule_label=str(meta.get("label") or code), statement_code=str(meta.get("statement") or "UNMAPPED"), mapped_by=mapped_by, )) return { "lines": lines, "group_count": len(group_rows), "hierarchy_warning": hierarchy_warning, "count": len(lines), } def previous_financial_year(financial_year: str) -> str: start = int(str(financial_year).split("-")[0]) return f"{start - 1}-{str(start)[-2:]}" def _schedule_totals(lines: list[LedgerLine]) -> dict[str, float]: totals: dict[str, float] = defaultdict(float) for line in lines: if line.schedule_code != "UNMAPPED": totals[line.schedule_code] += line.reporting_amount return {k: round(v, 2) for k, v in totals.items()} def _inventory_values(lines: list[LedgerLine]) -> tuple[float, float]: opening = sum(x.opening_reporting_amount for x in lines if x.schedule_code == "INVENTORIES") closing = sum(x.reporting_amount for x in lines if x.schedule_code == "INVENTORIES") return round(opening, 2), round(closing, 2) def build_financial_statements( *, current_lines: list[LedgerLine], previous_lines: list[LedgerLine] | None = None, partnership_shares: list[dict[str, Any]] | None = None, ) -> dict[str, Any]: previous_lines = previous_lines or [] current_totals = _schedule_totals(current_lines) previous_totals = _schedule_totals(previous_lines) def pl_values(lines: list[LedgerLine]) -> dict[str, float]: totals = _schedule_totals(lines) opening_stock, closing_stock = _inventory_values(lines) purchases = totals.get("PURCHASES", 0.0) cost_of_goods = round(opening_stock + purchases - closing_stock, 2) revenue = totals.get("REVENUE", 0.0) other_income = totals.get("OTHER_INCOME", 0.0) direct = totals.get("DIRECT_EXPENSES", 0.0) employee = totals.get("EMPLOYEE_BENEFITS", 0.0) finance = totals.get("FINANCE_COST", 0.0) depreciation = totals.get("DEPRECIATION", 0.0) other_exp = totals.get("OTHER_EXPENSES", 0.0) total_income = round(revenue + other_income, 2) total_expenses = round(cost_of_goods + direct + employee + finance + depreciation + other_exp, 2) profit = round(total_income - total_expenses, 2) return { "REVENUE": revenue, "OTHER_INCOME": other_income, "OPENING_STOCK": opening_stock, "PURCHASES": purchases, "CLOSING_STOCK": closing_stock, "COST_OF_GOODS_SOLD": cost_of_goods, "DIRECT_EXPENSES": direct, "EMPLOYEE_BENEFITS": employee, "FINANCE_COST": finance, "DEPRECIATION": depreciation, "OTHER_EXPENSES": other_exp, "TOTAL_INCOME": total_income, "TOTAL_EXPENSES": total_expenses, "PROFIT": profit, } current_pl = pl_values(current_lines) previous_pl = pl_values(previous_lines) allocations: list[dict[str, Any]] = [] shares = partnership_shares or [] ratio_total = round(sum(float(x.get("profit_share_percent") or 0) for x in shares), 4) if shares and abs(ratio_total - 100.0) < 0.0001: remaining = current_pl["PROFIT"] for idx, share in enumerate(shares): if idx == len(shares) - 1: amount = round(remaining, 2) else: amount = round(current_pl["PROFIT"] * float(share.get("profit_share_percent") or 0) / 100.0, 2) remaining = round(remaining - amount, 2) allocations.append({**share, "amount": amount}) # The mirror may contain either pre-appropriation or already-finalised capital balances. # Do not blindly add profit to capital: first compare the raw Balance Sheet gap with # the P&L result. If Assets - Liabilities equals current-year profit/loss, the books # are pre-appropriation and the ERP adds that result to Partner/Proprietor Funds. # If the Balance Sheet already balances, the profit is already reflected in the # source capital/current accounts and no duplicate adjustment is posted. allocation_total = round(sum(x["amount"] for x in allocations), 2) liability_codes = [k for k,v in SCHEDULES.items() if v.get("statement") == "BS" and v.get("bs_side") == "LIABILITY"] asset_codes = [k for k,v in SCHEDULES.items() if v.get("statement") == "BS" and v.get("bs_side") == "ASSET"] current_bs = dict(current_totals) raw_current_liabilities = round(sum(current_bs.get(k,0.0) for k in liability_codes),2) current_assets = round(sum(current_bs.get(k,0.0) for k in asset_codes),2) current_gap = round(current_assets - raw_current_liabilities, 2) current_profit = round(current_pl["PROFIT"], 2) tolerance = max(1.0, abs(current_profit) * 0.0001) profit_transfer_applied = False profit_transfer_status = "already_reflected" if abs(current_gap) <= tolerance else "review_required" if abs(current_gap - current_profit) <= tolerance: current_bs["PARTNER_FUNDS"] = round(current_bs.get("PARTNER_FUNDS", 0.0) + current_profit, 2) profit_transfer_applied = True profit_transfer_status = "erp_finalisation_adjustment" current_liabilities = round(sum(current_bs.get(k,0.0) for k in liability_codes),2) previous_bs = dict(previous_totals) raw_previous_liabilities = round(sum(previous_bs.get(k,0.0) for k in liability_codes),2) previous_assets = round(sum(previous_bs.get(k,0.0) for k in asset_codes),2) previous_gap = round(previous_assets - raw_previous_liabilities, 2) previous_profit = round(previous_pl["PROFIT"], 2) previous_tolerance = max(1.0, abs(previous_profit) * 0.0001) if abs(previous_gap - previous_profit) <= previous_tolerance: previous_bs["PARTNER_FUNDS"] = round(previous_bs.get("PARTNER_FUNDS", 0.0) + previous_profit, 2) previous_liabilities = round(sum(previous_bs.get(k,0.0) for k in liability_codes),2) return { "current_totals": current_totals, "previous_totals": previous_totals, "previous_bs": previous_bs, "current_pl": current_pl, "previous_pl": previous_pl, "current_bs": current_bs, "current_liabilities": current_liabilities, "current_assets": current_assets, "previous_liabilities": previous_liabilities, "previous_assets": previous_assets, "partnership_allocations": allocations, "profit_share_total": ratio_total, "profit_transfer_applied": profit_transfer_applied, "profit_transfer_status": profit_transfer_status, "balance_sheet_difference": round(current_assets - current_liabilities, 2), "unmapped": [x for x in current_lines if x.schedule_code == "UNMAPPED"], } THIN = Side(style="thin", color="000000") MEDIUM = Side(style="medium", color="000000") DOUBLE = Side(style="double", color="000000") def _set_page(ws, *, landscape: bool = False, repeat_rows: str | None = None) -> None: ws.sheet_view.showGridLines = False ws.freeze_panes = "A5" ws.page_setup.orientation = "landscape" if landscape else "portrait" ws.page_setup.paperSize = ws.PAPERSIZE_A4 ws.page_setup.fitToWidth = 1 ws.page_setup.fitToHeight = 0 ws.sheet_properties.pageSetUpPr.fitToPage = True ws.print_options.horizontalCentered = True ws.page_margins.left = 0.25 ws.page_margins.right = 0.25 ws.page_margins.top = 0.35 ws.page_margins.bottom = 0.35 if repeat_rows: ws.print_title_rows = repeat_rows def _title(ws, client_name: str, address: str, title: str, end_date: str, cols: int = 4) -> int: ws.merge_cells(start_row=1, start_column=1, end_row=1, end_column=cols) ws.cell(1,1,client_name).font = Font(bold=True, size=11) ws.merge_cells(start_row=2, start_column=1, end_row=2, end_column=cols) ws.cell(2,1,address).font = Font(size=9) ws.merge_cells(start_row=3, start_column=1, end_row=3, end_column=cols) ws.cell(3,1,f"{title} for the Year Ended {end_date}").font = Font(bold=True, underline="single", size=10) return 5 def _money(cell, value: float) -> None: cell.value = round(float(value or 0),2) cell.number_format = '#,##0;[Red](#,##0);-' cell.alignment = Alignment(horizontal="right") def _apply_table_border(ws, min_row: int, max_row: int, min_col: int, max_col: int, outer_medium: bool = True) -> None: for row in ws.iter_rows(min_row=min_row, max_row=max_row, min_col=min_col, max_col=max_col): for cell in row: left = MEDIUM if outer_medium and cell.column == min_col else THIN right = MEDIUM if outer_medium and cell.column == max_col else THIN top = MEDIUM if outer_medium and cell.row == min_row else THIN bottom = MEDIUM if outer_medium and cell.row == max_row else THIN cell.border = Border(left=left,right=right,top=top,bottom=bottom) def _header_row(ws, row: int, labels: list[str]) -> None: for col, label in enumerate(labels,1): c=ws.cell(row,col,label) c.font=Font(bold=True,size=9) c.alignment=Alignment(horizontal="center", vertical="center", wrap_text=True) c.fill=PatternFill("solid", fgColor="F2F2F2") c.border=Border(top=MEDIUM,bottom=MEDIUM,left=THIN,right=THIN) def _schedule_sheet_name(note_no: int, label: str) -> str: clean = ''.join(ch for ch in label if ch not in '[]:*?/\\') return f"N{note_no:02d} {clean}"[:31] def build_excel_workbook( *, client_name: str, address: str, financial_year: str, previous_fy: str, current_lines: list[LedgerLine], previous_lines: list[LedgerLine], statements: dict[str, Any], partnership_shares: list[dict[str, Any]] | None = None, ) -> bytes: wb=Workbook() wb.remove(wb.active) start_year=int(financial_year.split('-')[0]) current_end=f"31st March {start_year+1}" previous_end=f"31st March {start_year}" schedule_codes=[] for code,meta in sorted(SCHEDULES.items(), key=lambda kv:int(kv[1].get('order',999))): if code=="UNMAPPED": continue if any(x.schedule_code==code for x in current_lines) or any(x.schedule_code==code for x in previous_lines): schedule_codes.append(code) note_map={code:i+1 for i,code in enumerate(schedule_codes)} # Trial Balance ws=wb.create_sheet("Trial Balance") _set_page(ws, landscape=True, repeat_rows="1:5") _title(ws,client_name,address,"Trial Balance",current_end,8) _header_row(ws,5,["Ledger","Tally Group","Opening Dr","Opening Cr","Current Dr","Current Cr","Closing Dr","Closing Cr"]) r=6 for line in sorted(current_lines,key=lambda x:(x.root_group.lower(),x.parent_group.lower(),x.ledger_name.lower())): ws.cell(r,1,line.ledger_name); ws.cell(r,2,line.parent_group) for col,val in [(3,line.opening_debit),(4,line.opening_credit),(5,line.debit),(6,line.credit),(7,line.closing_debit),(8,line.closing_credit)]: _money(ws.cell(r,col),val) r+=1 ws.cell(r,1,"Total").font=Font(bold=True) for col in range(3,9): _money(ws.cell(r,col),sum(float(ws.cell(rr,col).value or 0) for rr in range(6,r))) ws.cell(r,col).font=Font(bold=True); ws.cell(r,col).border=Border(top=DOUBLE,bottom=DOUBLE) _apply_table_border(ws,5,r,1,8) widths=[34,24,14,14,14,14,14,14] for i,w in enumerate(widths,1): ws.column_dimensions[get_column_letter(i)].width=w ws.print_area = f"A1:H{r}" # Balance Sheet in sample format ws=wb.create_sheet("Balance Sheet") _set_page(ws, repeat_rows="1:5") _title(ws,client_name,address,"Statement of Balance Sheet",current_end,4) _header_row(ws,5,["Particulars","Note",current_end,previous_end]) r=6 ws.cell(r,1,"I EQUITY AND LIABILITIES").font=Font(bold=True); r+=1 liability_codes=[c for c in schedule_codes if SCHEDULES[c].get("bs_side")=="LIABILITY"] for code in liability_codes: meta=SCHEDULES[code]; ws.cell(r,1,meta["label"]); ws.cell(r,2,note_map[code]) ws.cell(r,2).hyperlink=f"#'{_schedule_sheet_name(note_map[code],meta['label'])}'!A1" _money(ws.cell(r,3),statements["current_bs"].get(code,0)); _money(ws.cell(r,4),statements.get("previous_bs", statements["previous_totals"]).get(code,0)); r+=1 ws.cell(r,1,"Total Liabilities").font=Font(bold=True) _money(ws.cell(r,3),statements["current_liabilities"]); _money(ws.cell(r,4),statements["previous_liabilities"]) for c in range(1,5): ws.cell(r,c).border=Border(top=DOUBLE,bottom=DOUBLE) r+=2; ws.cell(r,1,"II ASSETS").font=Font(bold=True); r+=1 asset_codes=[c for c in schedule_codes if SCHEDULES[c].get("bs_side")=="ASSET"] for code in asset_codes: meta=SCHEDULES[code]; ws.cell(r,1,meta["label"]); ws.cell(r,2,note_map[code]) ws.cell(r,2).hyperlink=f"#'{_schedule_sheet_name(note_map[code],meta['label'])}'!A1" _money(ws.cell(r,3),statements["current_bs"].get(code,0)); _money(ws.cell(r,4),statements.get("previous_bs", statements["previous_totals"]).get(code,0)); r+=1 ws.cell(r,1,"Total Assets").font=Font(bold=True) _money(ws.cell(r,3),statements["current_assets"]); _money(ws.cell(r,4),statements["previous_assets"]) for c in range(1,5): ws.cell(r,c).border=Border(top=DOUBLE,bottom=DOUBLE) r+=2; ws.merge_cells(start_row=r,start_column=1,end_row=r,end_column=4) ws.cell(r,1,"The accompanying notes are an integral part of the financial statements").font=Font(italic=True,size=8) _apply_table_border(ws,5,r-2,1,4) for col,w in {1:46,2:9,3:18,4:18}.items(): ws.column_dimensions[get_column_letter(col)].width=w ws.print_area = f"A1:D{r}" # Profit and Loss ws=wb.create_sheet("Profit and Loss") _set_page(ws, repeat_rows="1:5") _title(ws,client_name,address,"Statement of Profit and Loss",current_end,4) _header_row(ws,5,["Particulars","Note",current_end,previous_end]) cp=statements["current_pl"]; pp=statements["previous_pl"] rows=[ ("I", "Revenue from operations", "REVENUE", "REVENUE"), ("II", "Other Income", "OTHER_INCOME", "OTHER_INCOME"), ("III", "Total Income (I+II)", None, "TOTAL_INCOME"), ("IV", "Expenses:", None, None), ("(a)", "Cost of goods sold", None, "COST_OF_GOODS_SOLD"), ("(b)", "Direct Expenses", "DIRECT_EXPENSES", "DIRECT_EXPENSES"), ("(c)", "Employee benefits expense", "EMPLOYEE_BENEFITS", "EMPLOYEE_BENEFITS"), ("(d)", "Finance Cost", "FINANCE_COST", "FINANCE_COST"), ("(e)", "Depreciation and amortization expense", "DEPRECIATION", "DEPRECIATION"), ("(f)", "Other expenses", "OTHER_EXPENSES", "OTHER_EXPENSES"), ("", "Total expenses", None, "TOTAL_EXPENSES"), ("V", "Profit/(loss) before exceptional and extraordinary items (III-IV)", None, "PROFIT"), ("VI", "Exceptional items", None, None), ("VII", "Profit/(loss) before extraordinary items (V-VI)", None, "PROFIT"), ("VIII", "Extraordinary items", None, None), ("IX", "Profit before tax (VII-VIII)", None, "PROFIT"), ("X", "Profit/(Loss) for the year", None, "PROFIT"), ] r=6 for roman,label,note_code,key in rows: ws.cell(r,1,f"{roman} {label}".strip()) if roman in {"III","IV","V","VII","IX","X"} or label=="Total expenses": ws.cell(r,1).font=Font(bold=True) if note_code and note_code in note_map: ws.cell(r,2,note_map[note_code]); ws.cell(r,2).hyperlink=f"#'{_schedule_sheet_name(note_map[note_code],SCHEDULES[note_code]['label'])}'!A1" if key: _money(ws.cell(r,3),cp.get(key,0)); _money(ws.cell(r,4),pp.get(key,0)) if roman in {"III","V","VII","IX","X"} or label=="Total expenses": for c in range(1,5): ws.cell(r,c).border=Border(top=THIN,bottom=DOUBLE) r+=1 r+=1; ws.merge_cells(start_row=r,start_column=1,end_row=r,end_column=4) ws.cell(r,1,"The accompanying notes are an integral part of the financial statements").font=Font(italic=True,size=8) _apply_table_border(ws,5,r-2,1,4) for col,w in {1:56,2:9,3:18,4:18}.items(): ws.column_dimensions[get_column_letter(col)].width=w ws.print_area = f"A1:D{r}" # Cost of goods sold schedule (opening inventory + purchases - closing inventory). ws=wb.create_sheet("Cost of Goods Sold") _set_page(ws) _title(ws,client_name,address,"Note - Cost of Goods Sold",current_end,3) _header_row(ws,5,["Particulars",current_end,previous_end]) cpl=statements["current_pl"]; ppl=statements["previous_pl"] cogs_rows=[ ("Inventory at the beginning of the year","OPENING_STOCK"), ("Add: Purchases during the year","PURCHASES"), ("Less: Inventory at the end of the year","CLOSING_STOCK"), ("Cost of goods sold","COST_OF_GOODS_SOLD"), ] r=6 for label,key in cogs_rows: ws.cell(r,1,label); _money(ws.cell(r,2),cpl.get(key,0)); _money(ws.cell(r,3),ppl.get(key,0)) if key=="COST_OF_GOODS_SOLD": ws.cell(r,1).font=Font(bold=True) for c in range(1,4): ws.cell(r,c).border=Border(top=DOUBLE,bottom=DOUBLE) r+=1 _apply_table_border(ws,5,r-1,1,3) ws.column_dimensions['A'].width=52; ws.column_dimensions['B'].width=18; ws.column_dimensions['C'].width=18 # Notes/schedules prev_by_name={_norm(x.ledger_name):x for x in previous_lines} for code in schedule_codes: meta=SCHEDULES[code]; note=note_map[code] ws=wb.create_sheet(_schedule_sheet_name(note,meta["label"])) _set_page(ws, repeat_rows="1:5") _title(ws,client_name,address,f"Note - {note}: {meta['label']}",current_end,3) _header_row(ws,5,["Particulars",current_end,previous_end]) cur=[x for x in current_lines if x.schedule_code==code] r=6 for line in sorted(cur,key=lambda x:x.ledger_name.lower()): ws.cell(r,1,line.ledger_name); _money(ws.cell(r,2),line.reporting_amount) old=prev_by_name.get(_norm(line.ledger_name)); _money(ws.cell(r,3),old.reporting_amount if old else 0); r+=1 if code=="PARTNER_FUNDS" and statements.get("profit_transfer_applied"): if statements.get("partnership_allocations"): for alloc in statements["partnership_allocations"]: ws.cell(r,1,f"Add: Share of Profit / (Loss) - {alloc.get('partner_name')}") _money(ws.cell(r,2),alloc.get("amount") or 0); _money(ws.cell(r,3),0); r+=1 else: ws.cell(r,1,"Add: Current Year Profit / (Loss)") _money(ws.cell(r,2),statements.get("current_pl",{}).get("PROFIT",0)); _money(ws.cell(r,3),0); r+=1 ws.cell(r,1,"Total").font=Font(bold=True) _money(ws.cell(r,2),statements["current_bs"].get(code,statements["current_totals"].get(code,0))) _money(ws.cell(r,3),statements.get("previous_bs", statements["previous_totals"]).get(code,0)) for c in range(1,4): ws.cell(r,c).border=Border(top=DOUBLE,bottom=DOUBLE) _apply_table_border(ws,5,r,1,3) ws.column_dimensions['A'].width=52; ws.column_dimensions['B'].width=18; ws.column_dimensions['C'].width=18 if partnership_shares: ws=wb.create_sheet("Partner Profit Allocation") _set_page(ws) _title(ws,client_name,address,"Partnership Profit Allocation",current_end,4) _header_row(ws,5,["Partner","Profit Share %","Target Capital / Current Ledger","Share of Profit / (Loss)"]) alloc_by_name={_norm(x.get('partner_name')):x for x in statements.get('partnership_allocations') or []} r=6 for share in partnership_shares: ws.cell(r,1,share.get('partner_name')); ws.cell(r,2,float(share.get('profit_share_percent') or 0)); ws.cell(r,2).number_format='0.00%' # Value stored as percent e.g. 20; Excel percentage needs /100 ws.cell(r,2).value=float(share.get('profit_share_percent') or 0)/100.0 ws.cell(r,3,share.get('target_ledger_name') or '') alloc=alloc_by_name.get(_norm(share.get('partner_name'))); _money(ws.cell(r,4),(alloc or {}).get('amount') or 0); r+=1 ws.cell(r,1,"Total").font=Font(bold=True); ws.cell(r,2,sum(float(x.get('profit_share_percent') or 0) for x in partnership_shares)/100.0); ws.cell(r,2).number_format='0.00%' _money(ws.cell(r,4),sum(float((alloc_by_name.get(_norm(x.get('partner_name'))) or {}).get('amount') or 0) for x in partnership_shares)) _apply_table_border(ws,5,r,1,4) for col,w in {1:32,2:16,3:38,4:20}.items(): ws.column_dimensions[get_column_letter(col)].width=w # Consolidated Notes & Schedules: all notes and annexures in one printable # sheet, aligned one below another. Individual note sheets above are kept # (hidden) for backward compatibility with existing workbook consumers. ws_notes = wb.create_sheet("Notes & Schedules") _set_page(ws_notes, repeat_rows=None) ws_notes.freeze_panes = "A5" ws_notes.page_setup.orientation = "portrait" ws_notes.page_setup.fitToWidth = 1 ws_notes.page_setup.fitToHeight = 0 ws_notes.sheet_properties.pageSetUpPr.fitToPage = True ws_notes.sheet_view.showGridLines = False ws_notes.page_margins.left = 0.28 ws_notes.page_margins.right = 0.28 ws_notes.page_margins.top = 0.35 ws_notes.page_margins.bottom = 0.35 ws_notes.merge_cells("A1:C1"); ws_notes["A1"] = client_name; ws_notes["A1"].font = Font(bold=True,size=11) ws_notes.merge_cells("A2:C2"); ws_notes["A2"] = address; ws_notes["A2"].font = Font(size=9) ws_notes.merge_cells("A3:C3"); ws_notes["A3"] = f"Notes Forming Part of the Financial Statements for the Year Ended {current_end}" ws_notes["A3"].font = Font(bold=True,underline="single",size=10) ws_notes.column_dimensions['A'].width = 58 ws_notes.column_dimensions['B'].width = 18 ws_notes.column_dimensions['C'].width = 18 consolidated_anchor: dict[str, int] = {} r_notes = 5 prev_by_name={_norm(x.ledger_name):x for x in previous_lines} for code in schedule_codes: meta = SCHEDULES[code]; note = note_map[code] consolidated_anchor[code] = r_notes ws_notes.merge_cells(start_row=r_notes,start_column=1,end_row=r_notes,end_column=3) title_cell = ws_notes.cell(r_notes,1,f"Note - {note}: {meta['label']}") title_cell.font = Font(bold=True,size=9) title_cell.fill = PatternFill("solid", fgColor="EDEDED") title_cell.border = Border(top=MEDIUM,bottom=MEDIUM,left=MEDIUM,right=MEDIUM) r_notes += 1 _header_row(ws_notes,r_notes,["Particulars",current_end,previous_end]) block_start = r_notes r_notes += 1 cur=[x for x in current_lines if x.schedule_code==code] for line in sorted(cur,key=lambda x:x.ledger_name.lower()): ws_notes.cell(r_notes,1,line.ledger_name) _money(ws_notes.cell(r_notes,2),line.reporting_amount) old=prev_by_name.get(_norm(line.ledger_name)) _money(ws_notes.cell(r_notes,3),old.reporting_amount if old and old.schedule_code==code else 0) r_notes += 1 if code=="PARTNER_FUNDS" and statements.get("profit_transfer_applied"): if statements.get("partnership_allocations"): for alloc in statements["partnership_allocations"]: ws_notes.cell(r_notes,1,f"Add: Share of Profit / (Loss) - {alloc.get('partner_name')}") _money(ws_notes.cell(r_notes,2),alloc.get("amount") or 0); _money(ws_notes.cell(r_notes,3),0); r_notes += 1 else: ws_notes.cell(r_notes,1,"Add: Current Year Profit / (Loss)") _money(ws_notes.cell(r_notes,2),statements.get("current_pl",{}).get("PROFIT",0)); _money(ws_notes.cell(r_notes,3),0); r_notes += 1 ws_notes.cell(r_notes,1,"Total").font = Font(bold=True) _money(ws_notes.cell(r_notes,2),statements["current_bs"].get(code,statements["current_totals"].get(code,0))) _money(ws_notes.cell(r_notes,3),statements.get("previous_bs",statements["previous_totals"]).get(code,0)) for c in range(1,4): ws_notes.cell(r_notes,c).border=Border(top=DOUBLE,bottom=DOUBLE) _apply_table_border(ws_notes,block_start,r_notes,1,3) r_notes += 3 # Cost of Goods Sold annexure follows the numbered notes. cogs_anchor = r_notes ws_notes.merge_cells(start_row=r_notes,start_column=1,end_row=r_notes,end_column=3) c=ws_notes.cell(r_notes,1,"Annexure - Cost of Goods Sold") c.font=Font(bold=True,size=9); c.fill=PatternFill("solid",fgColor="EDEDED") c.border=Border(top=MEDIUM,bottom=MEDIUM,left=MEDIUM,right=MEDIUM) r_notes += 1 _header_row(ws_notes,r_notes,["Particulars",current_end,previous_end]); block_start=r_notes; r_notes += 1 for label,key in [("Inventory at the beginning of the year","OPENING_STOCK"),("Add: Purchases during the year","PURCHASES"),("Less: Inventory at the end of the year","CLOSING_STOCK"),("Cost of goods sold","COST_OF_GOODS_SOLD")]: ws_notes.cell(r_notes,1,label); _money(ws_notes.cell(r_notes,2),statements["current_pl"].get(key,0)); _money(ws_notes.cell(r_notes,3),statements["previous_pl"].get(key,0)) if key=="COST_OF_GOODS_SOLD": ws_notes.cell(r_notes,1).font=Font(bold=True) for cc in range(1,4): ws_notes.cell(r_notes,cc).border=Border(top=DOUBLE,bottom=DOUBLE) r_notes += 1 _apply_table_border(ws_notes,block_start,r_notes-1,1,3) r_notes += 2 if partnership_shares: ws_notes.merge_cells(start_row=r_notes,start_column=1,end_row=r_notes,end_column=3) c=ws_notes.cell(r_notes,1,"Annexure - Partnership Profit Allocation") c.font=Font(bold=True,size=9); c.fill=PatternFill("solid",fgColor="EDEDED") c.border=Border(top=MEDIUM,bottom=MEDIUM,left=MEDIUM,right=MEDIUM) r_notes += 1 _header_row(ws_notes,r_notes,["Partner / Target Ledger","Profit Share %","Share of Profit / (Loss)"]); block_start=r_notes; r_notes += 1 alloc_by_name={_norm(x.get('partner_name')):x for x in statements.get('partnership_allocations') or []} for share in partnership_shares: label=f"{share.get('partner_name') or ''}" if share.get('target_ledger_name'): label += f" - {share.get('target_ledger_name')}" ws_notes.cell(r_notes,1,label) ws_notes.cell(r_notes,2,float(share.get('profit_share_percent') or 0)/100.0); ws_notes.cell(r_notes,2).number_format='0.00%' alloc=alloc_by_name.get(_norm(share.get('partner_name'))); _money(ws_notes.cell(r_notes,3),(alloc or {}).get('amount') or 0); r_notes += 1 ws_notes.cell(r_notes,1,"Total").font=Font(bold=True) ws_notes.cell(r_notes,2,sum(float(x.get('profit_share_percent') or 0) for x in partnership_shares)/100.0); ws_notes.cell(r_notes,2).number_format='0.00%' _money(ws_notes.cell(r_notes,3),sum(float((alloc_by_name.get(_norm(x.get('partner_name'))) or {}).get('amount') or 0) for x in partnership_shares)) for cc in range(1,4): ws_notes.cell(r_notes,cc).border=Border(top=DOUBLE,bottom=DOUBLE) _apply_table_border(ws_notes,block_start,r_notes,1,3) r_notes += 2 ws_notes.print_area = f"A1:C{max(1,r_notes)}" # Point Balance Sheet and P&L note links to the consolidated schedule sheet. for sheet_name in ("Balance Sheet","Profit and Loss"): ws_link = wb[sheet_name] for rr in range(6, ws_link.max_row + 1): note_val = ws_link.cell(rr,2).value try: note_no = int(note_val) except (TypeError,ValueError): continue code = next((k for k,v in note_map.items() if int(v)==note_no),None) if code and code in consolidated_anchor: ws_link.cell(rr,2).hyperlink = f"#'Notes & Schedules'!A{consolidated_anchor[code]}" # Cost-of-goods-sold line links to its annexure without disturbing note numbering. ws_pl = wb["Profit and Loss"] for rr in range(6, ws_pl.max_row + 1): if "cost of goods sold" in _norm(ws_pl.cell(rr,1).value): ws_pl.cell(rr,2).value = "Sch." ws_pl.cell(rr,2).hyperlink = f"#'Notes & Schedules'!A{cogs_anchor}" break # Keep the existing individual note/annexure sheets for compatibility, but # hide them from the normal workbook view now that the consolidated sheet is # the printable schedule pack. for sh in wb.worksheets: if sh.title.startswith("N") and len(sh.title) > 3 and sh.title[1:3].isdigit(): sh.sheet_state = "hidden" if sh.title in {"Cost of Goods Sold", "Partner Profit Allocation"}: sh.sheet_state = "hidden" # Place consolidated schedules immediately after the two statements. wb._sheets.remove(ws_notes) wb._sheets.insert(3, ws_notes) if statements.get("unmapped"): ws=wb.create_sheet("Unmapped Ledgers") _set_page(ws, landscape=True) _title(ws,client_name,address,"Unmapped Ledgers - Review Required",current_end,5) _header_row(ws,5,["Ledger","Immediate Tally Group","Resolved Root Group","Closing Debit","Closing Credit"]) r=6 for line in statements["unmapped"]: ws.cell(r,1,line.ledger_name); ws.cell(r,2,line.parent_group); ws.cell(r,3,line.root_group); _money(ws.cell(r,4),line.closing_debit); _money(ws.cell(r,5),line.closing_credit); r+=1 _apply_table_border(ws,5,r-1,1,5) for col,w in {1:35,2:28,3:28,4:18,5:18}.items(): ws.column_dimensions[get_column_letter(col)].width=w bio=BytesIO(); wb.save(bio); return bio.getvalue()