backend/rag_pipeline/excel_parser.py

import logging
import io
import pandas as pd
from typing import Dict, Any

logger = logging.getLogger(__name__)

def parse_excel_financials(file_path: str) -> Dict[str, Any]:
    """
    Parsuje arkusze Excel (.xlsx, .csv) wyciągając ustrukturyzowane tabele finansowe.
    Używane dla Financial Agent do budowania kontekstu płynności i EBITDA.
    """
    try:
        # Read all sheets
        if file_path.endswith('.csv'):
            df = pd.read_csv(file_path)
            sheets = {"Arkusz1": df}
        else:
            sheets = pd.read_excel(file_path, sheet_name=None)
            
        markdown_output = ""
        structured_data = {}
        
        for sheet_name, df in sheets.items():
            # Clean empty columns/rows
            df.dropna(how="all", inplace=True)
            df.dropna(axis=1, how="all", inplace=True)
            
            markdown_output += f"### Arkusz: {sheet_name}\n"
            markdown_output += df.to_markdown(index=False) + "\n\n"
            
            # Simple heuristic extraction for structured
            text_dump = df.to_string().lower()
            if "ebitda" in text_dump:
                structured_data["has_ebitda"] = True
            if "przychody" in text_dump or "sales" in text_dump:
                structured_data["has_revenue"] = True
                
        return {
            "text": markdown_output,
            "parser": "pandas_excel",
            "metadata": structured_data
        }
    except Exception as e:
        logger.error(f"[ExcelParser] Błąd parsowania pliku {file_path}: {e}")
        return {
            "text": "",
            "parser": "error",
            "metadata": {}
        }