# -*- coding: utf-8 -*- # error_logger.py """Централизованный модуль логирования ошибок. Записывает ошибки в logs/error.log с ротацией и идентификацией источника. Использование ------------- :: from error_logger import setup_error_logging, log_exception # Один раз при старте приложения: setup_error_logging() # В except-блоках: except Exception as e: log_exception(__name__, "func_name", e) Формат записи лога ------------------ :: 2026-03-27 14:30:00 | ERROR | module | function | ExcType | message Traceback (most recent call last): ... """ from __future__ import annotations import faulthandler import logging import sys import threading import traceback from logging.handlers import RotatingFileHandler from pathlib import Path _LOG_DIR = Path(__file__).resolve().parent / "logs" _LOG_FILE = _LOG_DIR / "error.log" _INTERPRETER_LOG_FILE = _LOG_DIR / "interpreter.log" _MAX_BYTES = 5 * 1024 * 1024 _BACKUP_COUNT = 3 _error_logger: logging.Logger = logging.getLogger("usms.errors") _configured = False _interpreter_hooks_installed = False _interpreter_stream = None class _SafeRotatingFileHandler(RotatingFileHandler): """Rotating handler, устойчивый к lock-ошибкам Windows при rollover.""" def handleError(self, record: logging.LogRecord) -> None: exc = sys.exc_info()[1] if not _is_locked_rollover_error(exc): super().handleError(record) return self._reopen_stream() try: logging.FileHandler.emit(self, record) except Exception as exc: try: sys.__stderr__.write(f"error_logger emit failure: {type(exc).__name__}: {exc}\n") sys.__stderr__.flush() except Exception as _fallback_exc: return return def _reopen_stream(self) -> None: try: if self.stream is not None: self.stream.close() except Exception as exc: try: sys.__stderr__.write(f"error_logger stream close failure: {type(exc).__name__}: {exc}\n") sys.__stderr__.flush() except Exception as _fallback_exc: return self.stream = self._open() def _is_locked_rollover_error(exc: BaseException | None) -> bool: """Определить отказ rollover из-за блокировки log-файла на Windows.""" if not isinstance(exc, PermissionError): return False return getattr(exc, "winerror", None) == 32 def setup_error_logging() -> None: """Настроить файловый handler для логирования ошибок. Создаёт директорию ``logs/`` и добавляет ``RotatingFileHandler`` к логгеру ``usms``, чтобы все дочерние логгеры (``usms.warehouse``, ``usms.errors`` и т.д.) автоматически записывали ошибки уровня ERROR и выше в файл. """ global _configured if _configured: return _LOG_DIR.mkdir(parents=True, exist_ok=True) usms_root = logging.getLogger("usms") if any( isinstance(handler, _SafeRotatingFileHandler) and Path(getattr(handler, "baseFilename", "")) == _LOG_FILE for handler in usms_root.handlers ): _configured = True return file_handler = _SafeRotatingFileHandler( str(_LOG_FILE), maxBytes=_MAX_BYTES, backupCount=_BACKUP_COUNT, encoding="utf-8", ) file_handler.setLevel(logging.ERROR) formatter = logging.Formatter( "%(asctime)s | %(levelname)s | %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) file_handler.setFormatter(formatter) usms_root.addHandler(file_handler) console_handler = logging.StreamHandler(sys.stderr) console_handler.setLevel(logging.ERROR) console_handler.setFormatter(formatter) usms_root.addHandler(console_handler) if usms_root.level == logging.NOTSET or usms_root.level > logging.DEBUG: usms_root.setLevel(logging.DEBUG) _configured = True def install_interpreter_hooks() -> None: """Подключить глобальные hooks для traceback и фатальных ошибок.""" global _interpreter_hooks_installed, _interpreter_stream if _interpreter_hooks_installed: return _LOG_DIR.mkdir(parents=True, exist_ok=True) _interpreter_stream = _INTERPRETER_LOG_FILE.open("a", encoding="utf-8") def _handle_unhandled_exception( exc_type: type[BaseException], exc_value: BaseException, exc_tb, ) -> None: if issubclass(exc_type, KeyboardInterrupt): sys.__excepthook__(exc_type, exc_value, exc_tb) return formatted = "".join(traceback.format_exception(exc_type, exc_value, exc_tb)) _write_interpreter_diagnostics("UNHANDLED EXCEPTION", formatted) _error_logger.error( "%s | %s | %s\n%s", "__main__", "sys.excepthook", exc_type.__name__, formatted.rstrip(), ) def _handle_thread_exception(args: threading.ExceptHookArgs) -> None: formatted = "".join( traceback.format_exception(args.exc_type, args.exc_value, args.exc_traceback) ) _write_interpreter_diagnostics( f"THREAD EXCEPTION: {getattr(args.thread, 'name', 'unknown')}", formatted, ) _error_logger.error( "%s | %s | %s\n%s", "__main__", "threading.excepthook", args.exc_type.__name__, formatted.rstrip(), ) def _handle_unraisable(unraisable) -> None: formatted = "".join( traceback.format_exception( type(unraisable.exc_value), unraisable.exc_value, unraisable.exc_traceback, ) ) object_repr = repr(getattr(unraisable, "object", None)) _write_interpreter_diagnostics( f"UNRAISABLE EXCEPTION: {object_repr}", formatted, ) _error_logger.error( "%s | %s | %s\n%s", "__main__", "sys.unraisablehook", type(unraisable.exc_value).__name__, formatted.rstrip(), ) sys.excepthook = _handle_unhandled_exception threading.excepthook = _handle_thread_exception if hasattr(sys, "unraisablehook"): sys.unraisablehook = _handle_unraisable try: faulthandler.enable(file=_interpreter_stream, all_threads=True) except Exception as exc: try: sys.__stderr__.write(f"faulthandler enable failure: {type(exc).__name__}: {exc}\n") sys.__stderr__.flush() except Exception as _fallback_exc: return _interpreter_hooks_installed = True def _write_interpreter_diagnostics(title: str, payload: str) -> None: """Вывести интерпретаторную диагностику и в консоль, и в файл.""" message = f"\n=== {title} ===\n{payload}" try: sys.__stderr__.write(message) sys.__stderr__.flush() except Exception as exc: try: sys.stderr.write(f"interpreter diagnostics stderr failure: {type(exc).__name__}: {exc}\n") sys.stderr.flush() except Exception as _fallback_exc: return try: if _interpreter_stream is not None: _interpreter_stream.write(message) _interpreter_stream.flush() except Exception as exc: try: sys.__stderr__.write(f"interpreter log file write failure: {type(exc).__name__}: {exc}\n") sys.__stderr__.flush() except Exception as _fallback_exc: return def log_exception(module: str, func: str, exc: BaseException) -> None: """Залогировать перехваченное исключение с идентификацией источника. Parameters ---------- module : str Имя модуля (обычно ``__name__``). func : str Имя функции / метода, в котором произошла ошибка. exc : BaseException Перехваченное исключение. """ exc_type = type(exc).__name__ if exc.__traceback__ is not None: formatted_traceback = "".join( traceback.format_exception(type(exc), exc, exc.__traceback__) ) else: formatted_traceback = f"{exc_type}: {exc}" _error_logger.error( "%s | %s | %s | %s\n%s", module, func, exc_type, exc, formatted_traceback, )