import json import os import logging from typing import Optional, Dict, Any from openpyxl import Workbook from openpyxl.styles import Font, Alignment, Border, Side, PatternFill logging.basicConfig( level=logging.INFO, format='%(asctime)s %(levelname)s %(name)s %(message)s', ) logger = logging.getLogger("cf_generation") class CashFlowStatementGenerator: """ Generates a Cash Flow Statement Excel file from extracted financial data. """ def __init__(self, extracted_data_file: Optional[str] = None, extracted_data: Optional[Dict[str, Any]] = None): """ Initialize with extracted financial data. Args: extracted_data_file: Path to JSON file with extracted data. extracted_data: Data dict (if already loaded). Raises: ValueError: If neither data file nor dict is provided. """ if extracted_data_file: try: with open(extracted_data_file, 'r') as f: self.data = json.load(f) logger.info(f"Loaded data from {extracted_data_file}") except Exception as e: logger.error(f"Failed to load data from {extracted_data_file}: {e}") raise elif extracted_data: self.data = extracted_data logger.info("Loaded data from provided dictionary.") else: logger.error("Either extracted_data_file or extracted_data must be provided.") raise ValueError("Either extracted_data_file or extracted_data must be provided") @staticmethod def format_amount(amount: Any) -> float: """ Format amount for display - return numeric value, formatting handled by Excel. Args: amount: Value to format. Returns: float: Numeric value (0 if invalid). """ if amount is None or amount == '' or amount == '-': return 0.0 try: return float(amount) except (ValueError, TypeError): return 0.0 def generate_cash_flow_statement_xlsx(self, output_filename: Optional[str] = None) -> Dict[str, Any]: """ Generate the complete Cash Flow Statement in Excel format with openpyxl formatting. Args: output_filename: Output Excel file name (from env or default). Returns: dict: Summary and verification of generated statement. """ output_filename = output_filename or os.getenv("CFS_OUTPUT_FILE", "data/cash_flow_statements.xlsx") try: pl_data = self.data['profit_and_loss'] wc_data = self.data['working_capital'] inv_data = self.data['investing_activities'] fin_data = self.data['financing_activities'] cash_data = self.data['cash_and_equivalents'] except KeyError as e: logger.error(f"Missing key in input data: {e}") raise # Calculations pbt_current = self.format_amount(pl_data['profit_before_tax']['current']) pbt_previous = self.format_amount(pl_data['profit_before_tax']['previous']) dep_current = self.format_amount(pl_data['depreciation']['current']) dep_previous = self.format_amount(pl_data['depreciation']['previous']) int_inc_current = self.format_amount(pl_data['interest_income']['current']) int_inc_previous = self.format_amount(pl_data['interest_income']['previous']) op_profit_current = pbt_current + dep_current - int_inc_current op_profit_previous = pbt_previous + dep_previous - int_inc_previous tr_change = self.format_amount(wc_data['trade_receivables']['change']) inv_change = self.format_amount(wc_data['inventories']['change']) oca_change = self.format_amount(wc_data['other_current_assets']['change']) stla_change = self.format_amount(wc_data['short_term_loans_advances']['change']) cwip_change = 0.0 ltla_change = self.format_amount(wc_data['long_term_loans_advances']['change']) stp_change = self.format_amount(wc_data['short_term_provisions']['change']) tp_change = self.format_amount(wc_data['trade_payables']['change']) ocl_change = self.format_amount(wc_data['other_current_liabilities']['change']) total_wc_change = ( tr_change + inv_change + oca_change + stla_change + cwip_change + ltla_change + stp_change + tp_change + ocl_change ) cash_from_operations = op_profit_current + total_wc_change tax_paid = float(os.getenv("CFS_TAX_PAID", 179.27)) net_operating_cash_flow = cash_from_operations - tax_paid asset_purchases = self.format_amount(inv_data['asset_purchases']['total']) asset_sales = self.format_amount(inv_data['asset_sales']['total']) interest_income = self.format_amount(inv_data['interest_income']['current']) net_investing_cash_flow = -asset_purchases + asset_sales + interest_income dividend_paid = self.format_amount(fin_data['dividend_paid']['current']) borrowing_change = self.format_amount(fin_data['long_term_borrowings']['change']) cmltd_repayment = abs(self.format_amount(fin_data['current_maturities']['change'])) net_financing_cash_flow = -dividend_paid + borrowing_change - cmltd_repayment net_change = net_operating_cash_flow + net_investing_cash_flow + net_financing_cash_flow cash_beginning = self.format_amount(cash_data['total']['previous']) cash_ending = self.format_amount(cash_data['total']['current']) cfs_data = [ ['Particulars', 'March 31, 2024', 'March 31, 2023'], ['', '', ''], ['Cash flow from operating activities', '', ''], ['Profit before taxation', pbt_current, pbt_previous], ['', '', ''], ['Adjustment for:', '', ''], ['Add: Depreciation and Amortisation Expense', dep_current, dep_previous], ['Less: Interest income', -int_inc_current, -int_inc_previous], ['Operating profit before working capital changes', op_profit_current, op_profit_previous], ['', '', ''], ['Movements in working capital:', '', ''], ['(Increase)/Decrease in Trade Receivables', tr_change, ''], ['(Increase)/Decrease in Inventories', inv_change, ''], ['(Increase)/Decrease in Other Current Assets', oca_change, ''], ['(Increase)/Decrease in Short Term Loans & Advances', stla_change, ''], ['(Increase)/Decrease in Capital Work in Progress', cwip_change, ''], ['(Increase)/Decrease in Long Term Loans & Advances', ltla_change, ''], ['Increase/(Decrease) in Short Term Provisions', stp_change, ''], ['Increase/(Decrease) in Trade Payables', tp_change, ''], ['Increase/(Decrease) in Other Current Liabilities', ocl_change, ''], ['Cash used in operations', cash_from_operations, ''], ['Less: Direct taxes paid (net of refunds)', -tax_paid, ''], ['Net cash flow from operating activities (A)', net_operating_cash_flow, ''], ['', '', ''], ['Cash flows from investing activities', '', ''], ['Purchase of Assets', -asset_purchases if asset_purchases > 0 else '', ''], ['Sale of Assets', asset_sales if asset_sales > 0 else '', ''], ['Interest income', interest_income, ''], ['Net cash flow from investing activities (B)', net_investing_cash_flow, ''], ['', '', ''], ['Cash flows from financing activities', '', ''], ['Dividend paid', -dividend_paid if dividend_paid > 0 else '', ''], ['Long Term Borrowings', borrowing_change if borrowing_change > 0 else '', ''], ['Repayment of borrowings', -abs(borrowing_change) if borrowing_change < 0 else '', ''], ['Net cash flow from financing activities (C)', net_financing_cash_flow, ''], ['', '', ''], ['Net increase/(decrease) in cash and cash equivalents (A+B+C)', net_change, ''], ['Cash and cash equivalents at the beginning of the year', cash_beginning, ''], ['Cash and cash equivalents at the end of the year', cash_ending, cash_beginning], ['', '', ''], ['Components of cash and cash equivalents', '', ''], ['Cash on hand', self.format_amount(cash_data['cash_on_hand']['current']), self.format_amount(cash_data['cash_on_hand']['previous'])], ['With banks in Current Accounts', self.format_amount(cash_data['bank_balances']['current']), self.format_amount(cash_data['bank_balances']['previous'])], ['With banks in Fixed Deposits', self.format_amount(cash_data['fixed_deposits']['current']), self.format_amount(cash_data['fixed_deposits']['previous'])], ['Total cash and cash equivalents (Refer note 13)', cash_ending, cash_beginning] ] wb = Workbook() ws = wb.active ws.title = "Cash Flow Statement" # Styles title_font = Font(bold=True, size=14, color="FFFFFF") subtitle_font = Font(bold=True, size=12) header_font = Font(bold=True, size=11) section_font = Font(bold=True, size=11) normal_font = Font(size=10) bold_font = Font(bold=True, size=10) thin_border = Border(left=Side(style="thin"), right=Side(style="thin"), top=Side(style="thin"), bottom=Side(style="thin")) center_align = Alignment(horizontal="center", vertical="center") left_align = Alignment(horizontal="left", vertical="center") right_align = Alignment(horizontal="right", vertical="center") title_fill = PatternFill(start_color="366092", end_color="366092", fill_type="solid") subtitle_fill = PatternFill(start_color="D7E4BC", end_color="D7E4BC", fill_type="solid") header_fill = PatternFill(start_color="F2F2F2", end_color="F2F2F2", fill_type="solid") section_fill = PatternFill(start_color="E7E6E6", end_color="E7E6E6", fill_type="solid") total_fill = PatternFill(start_color="F0F0F0", end_color="F0F0F0", fill_type="solid") # Set column widths ws.column_dimensions["A"].width = 55 ws.column_dimensions["B"].width = 18 ws.column_dimensions["C"].width = 18 # Title row ws.merge_cells("A1:C1") ws["A1"] = "CASH FLOW STATEMENT" ws["A1"].font = title_font ws["A1"].alignment = center_align ws["A1"].fill = title_fill ws.merge_cells("A2:C2") ws["A2"] = "For the year ended March 31, 2024" ws["A2"].font = subtitle_font ws["A2"].alignment = center_align ws["A2"].fill = subtitle_fill ws.merge_cells("A3:C3") ws["A3"] = "(All amounts in Lakhs)" ws["A3"].font = normal_font ws["A3"].alignment = center_align ws["A3"].fill = subtitle_fill # Header row ws["A5"] = "Particulars" ws["B5"] = "March 31, 2024" ws["C5"] = "March 31, 2023" for col in ["A", "B", "C"]: ws[f"{col}5"].font = header_font ws[f"{col}5"].alignment = center_align ws[f"{col}5"].fill = header_fill ws[f"{col}5"].border = thin_border # Write data rows excel_row = 6 for row_data in cfs_data[1:]: particulars, current_val, previous_val = row_data cell_a = ws.cell(row=excel_row, column=1, value=particulars) cell_b = ws.cell(row=excel_row, column=2, value=current_val) cell_c = ws.cell(row=excel_row, column=3, value=previous_val) is_section = any(section in str(particulars).lower() for section in [ 'cash flow from operating', 'cash flows from investing', 'cash flows from financing', 'adjustment for:', 'movements in working capital:', 'components of cash' ]) is_total = any(keyword in str(particulars).lower() for keyword in [ 'net cash flow', 'operating profit before working', 'cash used in operations', 'net increase', 'total cash' ]) if is_section and str(particulars).strip(): cell_a.font = section_font cell_a.fill = section_fill elif str(particulars).strip(): cell_a.font = normal_font else: cell_a.font = normal_font cell_a.alignment = left_align cell_a.border = thin_border for cell, value in zip([cell_b, cell_c], [current_val, previous_val]): if value == '' or value is None: cell.value = '' elif isinstance(value, (int, float)) and value != 0: cell.number_format = '#,##0.00' if is_total: cell.font = bold_font cell.fill = total_fill else: cell.font = normal_font else: cell.value = '' cell.alignment = right_align cell.border = thin_border excel_row += 1 try: wb.save(output_filename) logger.info(f"Cash Flow Statement Excel file saved to {output_filename}") except Exception as e: logger.error(f"Failed to save Excel file: {e}") raise return { 'operating_cash_flow': net_operating_cash_flow, 'investing_cash_flow': net_investing_cash_flow, 'financing_cash_flow': net_financing_cash_flow, 'net_change_in_cash': net_change, 'cash_beginning': cash_beginning, 'cash_ending': cash_ending, 'verification': { 'calculated_net_change': net_change, 'actual_cash_change': cash_ending - cash_beginning, 'difference': net_change - (cash_ending - cash_beginning) }, 'output_file': output_filename, 'detailed_calculations': { 'profit_before_tax': {'current': pbt_current, 'previous': pbt_previous}, 'depreciation': {'current': dep_current, 'previous': dep_previous}, 'interest_income': {'current': int_inc_current, 'previous': int_inc_previous}, 'operating_profit_before_wc': {'current': op_profit_current, 'previous': op_profit_previous}, 'working_capital_changes': { 'trade_receivables': tr_change, 'inventories': inv_change, 'other_current_assets': oca_change, 'short_term_loans_advances': stla_change, 'long_term_loans_advances': ltla_change, 'short_term_provisions': stp_change, 'trade_payables': tp_change, 'other_current_liabilities': ocl_change, 'total': total_wc_change }, 'cash_from_operations': cash_from_operations, 'tax_paid': tax_paid } } def main(): """ Main entry point for generating the Cash Flow Statement. """ extracted_file = os.getenv("CFS_EXTRACTED_FILE", "data/extracted_cfs_data.json") output_file = os.getenv("CFS_OUTPUT_FILE", "data/cash_flow_statements.xlsx") if not os.path.exists(extracted_file): logger.error(f"Extracted data file '{extracted_file}' not found. Please run the Financial Data Extractor first.") return try: cfs_generator = CashFlowStatementGenerator(extracted_data_file=extracted_file) cfs_summary = cfs_generator.generate_cash_flow_statement_xlsx(output_file) logger.info("Cash Flow Statement generation completed successfully.") logger.info(f"File created: {cfs_summary['output_file']}") logger.info(f"Operating Cash Flow: ₹{cfs_summary['operating_cash_flow']:,.2f} Lakhs") logger.info(f"Investing Cash Flow: ₹{cfs_summary['investing_cash_flow']:,.2f} Lakhs") logger.info(f"Financing Cash Flow: ₹{cfs_summary['financing_cash_flow']:,.2f} Lakhs") logger.info(f"Net Change in Cash: ₹{cfs_summary['net_change_in_cash']:,.2f} Lakhs") verification = cfs_summary['verification'] logger.info(f"Verification - Calculated Net Change: ₹{verification['calculated_net_change']:,.2f} Lakhs, " f"Actual Change: ₹{verification['actual_cash_change']:,.2f} Lakhs, " f"Difference: ₹{verification['difference']:,.2f} Lakhs") if abs(verification['difference']) < 1: logger.info("Cash Flow Statement balances correctly!") else: logger.warning("Cash Flow Statement has balancing difference - review calculations.") except Exception as e: logger.error(f"Error during Cash Flow Statement generation: {e}") if __name__ == "__main__": main()