Spaces:
Running
Running
| # 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 ""), | |
| } | |
| 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() | |
| 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, | |
| } | |
| 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"]) | |
| 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 записывают причину ошибки в журнал. | |