codekingpro/portable-devtools

1import asyncio2import collections3import logging4from collections.abc import Callable5 6from mitmproxy import command7from mitmproxy import log8from mitmproxy.log import LogEntry9from mitmproxy.utils import signals10 11 12class EventStore:13    def __init__(self, size: int = 10000) -> None:14        self.data: collections.deque[LogEntry] = collections.deque(maxlen=size)15        self.sig_add = signals.SyncSignal(lambda entry: None)16        self.sig_refresh = signals.SyncSignal(lambda: None)17 18        self.logger = CallbackLogger(self._add_log)19        self.logger.install()20 21    def done(self):22        self.logger.uninstall()23 24    def _add_log(self, entry: LogEntry) -> None:25        self.data.append(entry)26        self.sig_add.send(entry)27 28    @property29    def size(self) -> int | None:30        return self.data.maxlen31 32    @command.command("eventstore.clear")33    def clear(self) -> None:34        """35        Clear the event log.36        """37        self.data.clear()38        self.sig_refresh.send()39 40 41class CallbackLogger(log.MitmLogHandler):42    def __init__(43        self,44        callback: Callable[[LogEntry], None],45    ):46        super().__init__()47        self.callback = callback48        self.event_loop = asyncio.get_running_loop()49        self.formatter = log.MitmFormatter(colorize=False)50 51    def emit(self, record: logging.LogRecord) -> None:52        entry = LogEntry(53            msg=self.format(record),54            level=log.LOGGING_LEVELS_TO_LOGENTRY.get(record.levelno, "error"),55        )56        self.event_loop.call_soon_threadsafe(self.callback, entry)57