tools/excel_tool.py

import os
import pandas as pd
from langchain.tools import BaseTool

class ExcelTool(BaseTool):
    """
    Инструмент для анализа Excel-файлов.
    Формат input: "<file_url>[,col=<ColumnName>[,agg=<sum|mean>]]".
    Если задан только file_url, возвращает сумму всех числовых значений.
    """
    name: str = "excel_tool"
    description: str = ("Load and analyze Excel (.xls/.xlsx) files. Use for computing totals, filtering rows, summarizing sales data, or any spreadsheet-based calculation.")

    def _run(self, input: str) -> str:
        parts = [p.strip() for p in input.split(',') if p.strip()]
        file_url = parts[0]
        opts = {}
        # Парсим key=value параметры
        for param in parts[1:]:
            if '=' in param:
                k, v = param.split('=', 1)
                opts[k.strip()] = v.strip()
        # Читаем файл
        try:
            df = pd.read_excel(file_url)
        except Exception as e:
            return f"ERROR: Unable to read Excel file: {e}"
        # Если нет опций: сумма всех числовых
        # Применяем фильтры, если указаны
        filt = opts.get('filter')
        if filt:
            try:
                df = df.query(filt)
            except Exception as e:
                return f"ERROR: Filter failed: {e}"
        if len(opts) == (1 if filt else 0):
            try:
                total = df.select_dtypes(include='number').sum().sum()
                return f"{total:.2f}"
            except Exception as e:
                return f"ERROR: Summation failed: {e}"
        if not opts:
            try:
                total = df.select_dtypes(include='number').sum().sum()
                return str(total)
            except Exception as e:
                return f"ERROR: Summation failed: {e}"
        # Обработка опций
        col = opts.get('col')
        agg = opts.get('agg', 'sum')
        if col and col not in df.columns:
            return f"ERROR: Column '{col}' not found."
        data = df[col] if col else df.select_dtypes(include='number')
        try:
            if agg == 'sum':
                result = data.sum()
            elif agg == 'mean':
                result = data.mean()
            else:
                return f"ERROR: Unsupported aggregation '{agg}'"
            # Если результат — серия, переводим в число
            if hasattr(result, 'sum') and not isinstance(result, (int, float)):
                result = result.sum()
            return str(result)
        except Exception as e:
            return f"ERROR: Aggregation failed: {e}"

    async def _arun(self, input: str) -> str:
        raise NotImplementedError("Async not supported.")