codekingpro/portable-devtools

1import asyncio2import logging3import sys4 5from mitmproxy import log6from mitmproxy.contrib import click as miniclick7from mitmproxy.utils import vt_codes8 9 10class ErrorCheck:11    """Monitor startup for error log entries, and terminate immediately if there are some."""12 13    repeat_errors_on_stderr: bool14    """15    Repeat all errors on stderr before exiting.16    This is useful for the console UI, which otherwise swallows all output.17    """18 19    def __init__(self, repeat_errors_on_stderr: bool = False) -> None:20        self.repeat_errors_on_stderr = repeat_errors_on_stderr21 22        self.logger = ErrorCheckHandler()23        self.logger.install()24 25    def finish(self):26        self.logger.uninstall()27 28    async def shutdown_if_errored(self):29        # don't run immediately, wait for all logging tasks to finish.30        await asyncio.sleep(0)31        if self.logger.has_errored:32            plural = "s" if len(self.logger.has_errored) > 1 else ""33            if self.repeat_errors_on_stderr:34                message = f"Error{plural} logged during startup:"35                if vt_codes.ensure_supported(sys.stderr):  # pragma: no cover36                    message = miniclick.style(message, fg="red")37                details = "\n".join(38                    self.logger.format(r) for r in self.logger.has_errored39                )40                print(f"{message}\n{details}", file=sys.stderr)41            else:42                print(43                    f"Error{plural} logged during startup, exiting...", file=sys.stderr44                )45 46            sys.exit(1)47 48 49class ErrorCheckHandler(log.MitmLogHandler):50    def __init__(self) -> None:51        super().__init__(logging.ERROR)52        self.has_errored: list[logging.LogRecord] = []53 54    def emit(self, record: logging.LogRecord) -> None:55        self.has_errored.append(record)56