diff --git a/app/modules/accounting/financial_statements_service.py b/app/modules/accounting/financial_statements_service.py index f8a9843..1ad9667 100644 --- a/app/modules/accounting/financial_statements_service.py +++ b/app/modules/accounting/financial_statements_service.py @@ -47,6 +47,36 @@ SCHEDULES: dict[str, dict[str, Any]] = { 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()) @@ -65,9 +95,10 @@ def _root_group(group_name: str, parents: dict[str, str]) -> str: seen: set[str] = set() last = current while current and _norm(current) not in seen: - seen.add(_norm(current)) + key = _norm(current) + seen.add(key) last = current - parent = parents.get(_norm(current), "") + parent = parents.get(key, "") or DEFAULT_TALLY_GROUP_PARENTS.get(key, "") if not parent or _norm(parent) in {"primary", ""}: break current = parent @@ -79,12 +110,59 @@ def _keyword_any(text: str, words: Iterable[str]) -> bool: return any(_norm(word) in value for word in words) -def classify_default(ledger_name: str, parent_group: str, root_group: str) -> str: +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"}: @@ -138,6 +216,19 @@ def classify_default(ledger_name: str, parent_group: str, root_group: str) -> st 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" @@ -160,14 +251,35 @@ class LedgerLine: mapped_by: str @property - def closing_abs(self) -> float: - return abs(self.closing_balance) + 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: positive balances are debit, negative are credit. + # Tally mirror convention: negative balances are debit, positive are credit. value = _amount(value) - return (value if value > 0 else 0.0, abs(value) if value < 0 else 0.0) + 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]: @@ -202,6 +314,8 @@ def load_mirror_trial_balance( 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] = [] @@ -216,7 +330,7 @@ def load_mirror_trial_balance( code = override.schedule_code mapped_by = "manual" else: - code = classify_default(name, parent, root) + 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")) @@ -256,13 +370,14 @@ def previous_financial_year(financial_year: str) -> str: def _schedule_totals(lines: list[LedgerLine]) -> dict[str, float]: totals: dict[str, float] = defaultdict(float) for line in lines: - totals[line.schedule_code] += line.closing_abs + 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(abs(x.opening_balance) for x in lines if x.schedule_code == "INVENTORIES") - closing = sum(abs(x.closing_balance) for x in lines if x.schedule_code == "INVENTORIES") + 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) @@ -466,6 +581,7 @@ def build_excel_workbook( _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") @@ -495,6 +611,7 @@ def build_excel_workbook( 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") @@ -536,6 +653,7 @@ def build_excel_workbook( 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") @@ -570,8 +688,8 @@ def build_excel_workbook( 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.closing_abs) - old=prev_by_name.get(_norm(line.ledger_name)); _money(ws.cell(r,3),old.closing_abs if old else 0); r+=1 + 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"]: @@ -605,6 +723,138 @@ def build_excel_workbook( _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)