core/error_handler.py

# core/error_handler.py
"""
Обработчик ошибок с понятными сообщениями.
Адаптировано для Streamlit-приложений.
"""

from __future__ import annotations
import logging
import json
import re
import traceback
from datetime import datetime
from pathlib import Path
from typing import Any, Callable, Awaitable

logger = logging.getLogger(__name__)
class ErrorHandler:
    """
    Обработчик ошибок с понятными сообщениями.
    Адаптировано для Streamlit-приложений.
    """

    _VALID_LOG_LEVELS = {"debug", "info", "error"}

    def __init__(self, log_level: str = "info", log_file: str | None = None):
        """
        Инициализация обработчика ошибок.

        Args:
            log_level: Уровень логирования ("info", "debug", "error")
            log_file: Путь к файлу для записи логов (опционально)
        """
        normalized_level = (log_level or "info").strip().lower()
        self.log_level = normalized_level if normalized_level in self._VALID_LOG_LEVELS else "info"
        self.log_file = log_file
        self.errors: list[dict[str, Any]] = []

    def handle(self, error: Exception, context: dict[str, Any] | None = None) -> dict[str, Any]:
        """
        Обрабатывает ошибку и возвращает понятный ответ для пользователя.
        """
        error_info: dict[str, Any] = {
            "timestamp": datetime.now().isoformat(),
            "type": type(error).__name__,
            "message": str(error),
            "traceback": "".join(traceback.format_exception(type(error), error, error.__traceback__)),
            "context": context or {},
        }

        self.errors.append(error_info)
        self._log_error(error_info)
        return self._format_error(error_info)

    def _log_error(self, error_info: dict[str, Any]) -> None:
        """
        Логирует ошибку в консоль и/или файл.
        """
        message = (
            f"[ERROR] {error_info.get('timestamp', '')} "
            f"{error_info.get('type', 'Unknown')}: {error_info.get('message', '')}"
        )

        trace = str(error_info.get("traceback", "") or "")
        if self.log_level == "debug" and trace.strip() and trace.strip() != "NoneType: None":
            logger.error("%s\n%s", message, trace)
        else:
            logger.error("%s", message)

        if self.log_file:
            try:
                log_path = Path(self.log_file)
                log_path.parent.mkdir(parents=True, exist_ok=True)

                with log_path.open("a", encoding="utf-8") as file:
                    file.write(message + "\n")
                    if self.log_level == "debug":
                        trace = str(error_info.get("traceback", "") or "")
                        if trace.strip():
                            file.write(trace + "\n")
                    file.write("-" * 80 + "\n")
            except OSError:
                logger.exception("Не удалось записать лог ошибок в %s", self.log_file)

    def _format_error(self, error_info: dict[str, Any]) -> dict[str, Any]:
        """
        Форматирует ошибку для пользователя.
        """
        error_type = str(error_info.get("type", "Exception"))
        message = str(error_info.get("message", ""))

        error_messages = {
            "ModuleNotFoundError": (
                f"⚠️ Отсутствует необходимая библиотека. "
                f"Установите: pip install {self._extract_module_name(message)}"
            ),
            "FileNotFoundError": f"📁 Файл не найден: {message}",
            "KeyError": f"🔑 Отсутствует ключ в данных: {message}",
            "ValueError": f"📊 Ошибка в данных: {message}",
            "ConnectionError": "🌐 Ошибка подключения. Проверьте интернет-соединение.",
            "TimeoutError": "⏰ Превышено время ожидания. Попробуйте позже.",
            "PermissionError": f"🔒 Нет доступа к файлу: {message}",
            "JSONDecodeError": "📄 Ошибка парсинга JSON. Проверьте формат данных.",
            "ZeroDivisionError": "⚠️ Деление на ноль. Проверьте исходные данные.",
            "TypeError": f"⚠️ Неверный тип данных: {message}",
            "AttributeError": f"⚠️ Ошибка доступа к атрибуту: {message}",
            "ImportError": f"⚠️ Ошибка импорта: {message}",
            "RuntimeError": f"⚠️ Ошибка выполнения: {message}",
            "MemoryError": "⚠️ Недостаточно памяти. Попробуйте уменьшить объём данных.",
            "IndexError": "⚠️ Выход за границы списка. Проверьте индексы.",
            "NotImplementedError": "⚠️ Функция ещё не реализована.",
            "StopIteration": "⚠️ Итерация завершена.",
            "OverflowError": "⚠️ Число слишком велико для обработки.",
            "RecursionError": "⚠️ Превышена глубина рекурсии.",
            "KeyboardInterrupt": "⏹️ Операция прервана пользователем.",
            "OSError": f"⚠️ Системная ошибка: {message}",
            "IOError": f"⚠️ Ошибка ввода/вывода: {message}",
            "UnicodeDecodeError": "⚠️ Ошибка декодирования. Проверьте кодировку файла.",
            "UnicodeEncodeError": "⚠️ Ошибка кодирования. Проверьте текстовые данные.",
        }

        message_lower = message.lower()

        if "FormulaEngine" in error_type or "calculation" in message_lower:
            user_message = f"📐 Ошибка расчёта: {message}"
        elif "QASystem" in error_type or "search" in message_lower:
            user_message = f"🔍 Ошибка поиска: {message}"
        elif "AgentLoop" in error_type:
            user_message = f"🤖 Ошибка агента: {message}"
        else:
            user_message = error_messages.get(error_type, f"❌ Ошибка: {message or 'Неизвестная ошибка'}")

        return {
            "user_message": user_message,
            "debug_info": error_info if self.log_level == "debug" else None,
            "is_error": True,
            "type": error_type,
            "message": message,
            "timestamp": str(error_info.get("timestamp", "") or ""),
        }

    @staticmethod
    def _extract_module_name(message: str) -> str:
        """Извлекает имя модуля из сообщения об ошибке."""
        match = re.search(r"'([^']+)'", message)
        return match.group(1) if match else "название-библиотеки"

    def get_last_error(self) -> dict[str, Any] | None:
        """Возвращает последнюю ошибку."""
        return self.errors[-1] if self.errors else None

    def get_all_errors(self) -> list[dict[str, Any]]:
        """Возвращает все ошибки."""
        return list(self.errors)

    def clear_errors(self) -> None:
        """Очищает список ошибок."""
        self.errors.clear()

    @staticmethod
    def format_for_ui(error_info: dict[str, Any]) -> str:
        """Форматирует ошибку для отображения в UI."""
        if error_info.get("is_error"):
            return str(error_info.get("user_message", "❌ Неизвестная ошибка"))
        return f"ℹ️ {error_info.get('message', 'Неизвестная ошибка')}"

    def get_error_summary(self) -> dict[str, Any]:
        """Возвращает сводку по ошибкам."""
        if not self.errors:
            return {"total": 0, "types": {}}

        types: dict[str, int] = {}
        for error in self.errors:
            error_type = str(error.get("type", "Unknown"))
            types[error_type] = types.get(error_type, 0) + 1

        last_error = self.errors[-1] if self.errors else None

        return {
            "total": len(self.errors),
            "types": types,
            "last_error": last_error,
            "last_timestamp": last_error.get("timestamp") if last_error else None,
        }

    @staticmethod
    def log_to_file(error_info: dict[str, Any], file_path: str) -> None:
        """Записывает ошибку в отдельный файл лога в формате JSON."""
        try:
            path = Path(file_path)
            path.parent.mkdir(parents=True, exist_ok=True)

            with path.open("a", encoding="utf-8") as file:
                json.dump(error_info, file, ensure_ascii=False)
                file.write("\n" + "-" * 80 + "\n")
        except OSError as exc:
            logger.warning("Не удалось записать лог %s: %s", file_path, exc)

    def get_user_friendly_message(self, error: Exception) -> str:
        """Быстрый метод для получения понятного сообщения об ошибке."""
        result = self.handle(error)
        return str(result["user_message"])

    @staticmethod
    def is_critical(error_info: dict[str, Any]) -> bool:
        """Определяет, является ли ошибка критической."""
        critical_types = {
            "MemoryError",
            "RecursionError",
            "KeyboardInterrupt",
            "SystemError",
        }
        return str(error_info.get("type")) in critical_types


def safe_execute(
    func: Callable[..., Any],
    *args: Any,
    **kwargs: Any,
) -> tuple[Any | None, Exception | None]:
    """Безопасное выполнение функции с обработкой ошибок."""
    try:
        result = func(*args, **kwargs)
        return result, None
    except Exception as exc:
        logger.exception("Ошибка безопасного синхронного вызова")
        return None, exc


async def safe_execute_async(
    async_func: Callable[..., Awaitable[Any]],
    *args: Any,
    **kwargs: Any,
) -> tuple[Any | None, Exception | None]:
    """Безопасное выполнение асинхронной функции с обработкой ошибок."""
    try:
        result = await async_func(*args, **kwargs)
        return result, None
    except Exception as exc:
        logger.exception("Ошибка безопасного асинхронного вызова")
        return None, exc


def _run_self_tests() -> None:
    from unittest.mock import patch
    handler = ErrorHandler(log_level="debug")
    info = handler.handle(ValueError("проверка журнала"), {"stage": "self-test"})
    assert info["is_error"] is True
    assert handler.get_error_summary()["total"] == 1
    assert "ValueError: проверка журнала" in handler.errors[0]["traceback"]
    with patch.object(logger, "exception") as log:
        result, error = safe_execute(lambda: 1 / 0)
        assert result is None and isinstance(error, ZeroDivisionError)
        assert log.called


if __name__ == "__main__":
    _run_self_tests()

# ИСПРАВЛЕНО: traceback сохраняется и вне except; safe_execute/async записывают причину ошибки в журнал.