src/ExceptionManager.py - master - Flipper/Marginal

src/ExceptionManager.py
126 строк4 KB
Alex
Marginal v1.0.0 RELEASE
11 янв 2026, 18:02
11 янв 2026, 18:025e1eff8
import sys
import threading
import traceback


class ExceptionManager:
    """
    Centralized global exception handler.

    Hooks into sys.excepthook and threading.excepthook to ensure uncaught
    exceptions from Python threads are logged consistently.

    Note:
        Qt (QThread) exceptions are NOT handled here and must be caught
        explicitly inside worker run methods.
    """

    def __init__(self, backbone):
        """
        Initialize the exception manager and register global hooks.

        Args:
            backbone: Application backbone providing access to Logger.
        """
        self.logger = backbone.logger
        self.sessionTimestamp = self.logger.getSessionTimeStamp()

        # Register global exception hooks
        sys.excepthook = self.globalExceptHook

        # Register threading exception hook (Python 3.8+)
        if hasattr(threading, "excepthook"):
            threading.excepthook = self.threadExceptHook
        else:
            self._setupLegacyThreadExceptionHandler()

    def _setupLegacyThreadExceptionHandler(self):
        """
        Install a fallback exception handler for Python versions < 3.8.

        Wraps threading.Thread.run() to intercept uncaught exceptions.
        """
        originalRun = threading.Thread.run

        def wrappedRun(*args, **kwargs):
            try:
                return originalRun(*args, **kwargs)
            except SystemExit:
                raise
            except Exception:
                excType, excValue, excTraceback = sys.exc_info()
                self.logError(
                    excType,
                    excValue,
                    excTraceback,
                    context=f"Thread: {threading.current_thread().name}"
                )
                raise

        threading.Thread.run = wrappedRun

    def logError(self, excType, excValue, excTraceback, context: str = "Main Thread"):
        """
        Log an exception using the application's Logger.

        Args:
            excType: Exception class
            excValue: Exception instance
            excTraceback: Traceback object
            context (str): Logical execution context
        """
        self.logger.logError(excType, excValue, excTraceback, context)

        self.logger.log(
            "ExceptionManager",
            "Exception in {cont}:\n {excType}: {excValue}\n Check error log for more details.",
            cont=context,
            excType=excType.__name__,
            excValue=excValue
        )

    def globalExceptHook(self, excType, excValue, excTraceback):
        """
        Global sys.excepthook handler.

        Logs the exception and forwards it to Python's default handler.
        """
        self.logError(excType, excValue, excTraceback, context="Main Thread")
        sys.__excepthook__(excType, excValue, excTraceback)

    def threadExceptHook(self, args):
        """
        threading.excepthook handler (Python 3.8+).

        Args:
            args: threading.ExceptHookArgs instance
        """
        excType = args.exc_type
        excValue = args.exc_value
        excTraceback = args.exc_traceback
        thread = args.thread

        context = f"Thread: {thread.name} (id: {thread.ident})"
        self.logError(excType, excValue, excTraceback, context)

        if hasattr(threading, "__excepthook__"):
            threading.__excepthook__(args)
        else:
            self.logError(context=thread.name)


    def handleExceptionInThread(self, excInfo, threadName: str | None = None):
        """
        Manually forward a caught thread exception to the ExceptionManager.

        Args:
            excInfo: Tuple (excType, excValue, excTraceback)
            threadName (str | None): Optional thread name override
        """
        excType, excValue, excTraceback = excInfo
        context = (
            f"Thread: {threadName}"
            if threadName
            else f"Thread: {threading.current_thread().name}"
        )
        self.logError(excType, excValue, excTraceback, context)