my-botty/src/logger.py

import logging
import io
import os
import sys
import shutil
import threading
import traceback
import warnings
import zipfile
from logging.handlers import RotatingFileHandler
from version import __version__
from colorama import Fore, Back, Style, init
import time

init()

# Hard cap on the live log file so a long/spammy session can't balloon log.txt
# to many GB (it has happened — a single session produced a 22 GB log). Each
# rotated file is zipped into log/archive/; backups + archives are both bounded,
# so total disk use is capped. Override the per-file cap with BOTTY_LOG_MAX_MB.
try:
    LOG_FILE_MAX_BYTES = int(float(os.environ.get("BOTTY_LOG_MAX_MB", "50")) * 1024 * 1024)
except (TypeError, ValueError):
    LOG_FILE_MAX_BYTES = 50 * 1024 * 1024
LOG_FILE_BACKUPS = 5          # rotated log.txt.1..5 before zipping
LOG_ARCHIVE_MAX = 30          # keep at most this many zipped archives


class ArchiveRotatingFileHandler(RotatingFileHandler):
    """Size-capped RotatingFileHandler that zips rotated files into log/archive/.

    Size-based (not time-based) so the live log.txt can never exceed maxBytes —
    rotation also fires mid-session, not only at midnight. The .1/.2 backup
    naming matches what RotatingFileHandler produces (the old TimedRotating
    base produced date-suffixed names this archiver never found)."""

    def doRollover(self):
        # Let the parent rotate the file (creates log.txt.1, etc.)
        super().doRollover()

        # Archive directory
        archive_dir = os.path.join("log", "archive")
        os.makedirs(archive_dir, exist_ok=True)

        # Zip all rotated backup files into log/archive/
        base_file = self.baseFilename  # e.g. "log/log.txt"
        for i in range(1, self.backupCount + 1):
            rotated = f"{base_file}.{i}"
            if os.path.exists(rotated):
                # Build zip name from the rotated file suffix
                # log.txt.1 -> log_20260528.zip (use modification time)
                mtime = os.path.getmtime(rotated)
                zip_name = f"log_{time.strftime('%Y%m%d_%H%M%S', time.localtime(mtime))}.zip"
                zip_path = os.path.join(archive_dir, zip_name)
                try:
                    with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zf:
                        zf.write(rotated, os.path.basename(rotated))
                    os.remove(rotated)
                except Exception as e:
                    # If zipping fails, leave the rotated file in place
                    pass

        # Prune old archives so log/archive/ can't grow without bound either.
        try:
            zips = sorted(
                (os.path.join(archive_dir, f) for f in os.listdir(archive_dir)
                 if f.startswith("log_") and f.endswith(".zip")),
                key=os.path.getmtime,
            )
            for old in zips[:-LOG_ARCHIVE_MAX]:
                os.remove(old)
        except Exception:
            pass


class CustomFormatter(logging.Formatter):
    _format = f'[{__version__} %(asctime)s] %(levelname)-10s %(message)s'

    FORMATS = {
        logging.DEBUG:    Fore.WHITE          + _format + Fore.WHITE,
        logging.INFO:     Fore.LIGHTBLUE_EX   + _format + Fore.WHITE,
        logging.WARNING:  Fore.LIGHTYELLOW_EX + _format + Fore.WHITE,
        logging.ERROR:    Fore.LIGHTRED_EX    + _format + Fore.WHITE,
        logging.CRITICAL: Fore.RED            + _format + Fore.WHITE
    }


    def format(self, record):
        log_fmt = self.FORMATS.get(record.levelno)
        formatter = logging.Formatter(log_fmt)
        return formatter.format(record)

class Logger:
    """Manage logging"""
    os.makedirs("log", exist_ok=True)
    _logger_level = None
    _log_contents = io.StringIO()
    _current_log_file_path = "log/log.txt"
    _output = ""  # intercepted output from stdout and stderr
    string_handler = None
    file_handler = None
    console_handler = None
    logger = None

    @staticmethod
    def debug(data: str):
        if Logger.logger is None:
            Logger.init()
        Logger.logger.debug(data)

    @staticmethod
    def info(data: str):
        if Logger.logger is None:
            Logger.init()
        Logger.logger.info(data)

    @staticmethod
    def warning(data: str):
        if Logger.logger is None:
            Logger.init()
        Logger.logger.warning(data)

    @staticmethod
    def error(data: str):
        if Logger.logger is None:
            Logger.init()
        Logger.logger.error(data)

    @staticmethod
    def exception(data: str):
        if Logger.logger is None:
            Logger.init()
        Logger.logger.exception(data)

    @staticmethod
    def install_exception_hooks():
        def log_uncaught_exception(exc_type, exc_value, exc_traceback):
            if issubclass(exc_type, (KeyboardInterrupt, SystemExit)):
                sys.__excepthook__(exc_type, exc_value, exc_traceback)
                return
            Logger.error(
                "Uncaught exception:\n"
                + "".join(traceback.format_exception(exc_type, exc_value, exc_traceback))
            )

        def log_thread_exception(args):
            if issubclass(args.exc_type, SystemExit):
                return
            Logger.error(
                f"Uncaught exception in thread {args.thread.name}:\n"
                + "".join(traceback.format_exception(args.exc_type, args.exc_value, args.exc_traceback))
            )

        sys.excepthook = log_uncaught_exception
        threading.excepthook = log_thread_exception

    @staticmethod
    def init(lvl = logging.DEBUG):
        """
        Setup logger for StringIO, console and file handler
        """
        Logger._logger_level = lvl

        if Logger.logger is not None:
            Logger.logger.warning("WARNING: logger was setup already, deleting all previously existing handlers")
            for hdlr in Logger.logger.handlers[:]:  # remove all old handlers
                Logger.logger.removeHandler(hdlr)

        # Create the logger
        Logger.logger = logging.getLogger("botty")
        for hdlr in Logger.logger.handlers:
            Logger.logger.removeHandler(hdlr)
        Logger.logger.setLevel(Logger._logger_level)
        Logger.logger.propagate = False

        # Setup the StringIO handler
        Logger._log_contents = io.StringIO()
        Logger.string_handler = logging.StreamHandler(Logger._log_contents)
        Logger.string_handler.setLevel(Logger._logger_level)

        # Setup the console handler
        Logger.console_handler = logging.StreamHandler(sys.stdout)
        Logger.console_handler.setLevel(Logger._logger_level)

        # Setup the file handler (size-capped, archives to log/archive/)
        Logger.file_handler = ArchiveRotatingFileHandler(
            Logger._current_log_file_path,
            maxBytes=LOG_FILE_MAX_BYTES,
            backupCount=LOG_FILE_BACKUPS,
            encoding='utf-8'
        )
        Logger.file_handler.setLevel(Logger._logger_level)

        # Optionally add a formatter
        _format = CustomFormatter()
        Logger.string_handler.setFormatter(_format)
        Logger.console_handler.setFormatter(_format)
        Logger.file_handler.setFormatter(logging.Formatter(_format._format))

        # Add the handler to the logger
        Logger.logger.addHandler(Logger.string_handler)
        Logger.logger.addHandler(Logger.console_handler)
        Logger.logger.addHandler(Logger.file_handler)
        Logger.install_exception_hooks()

        # redirect stderr & stdout to logger, e.g. print("...")
        # would have to implement all the std func such as write() flush() etc.
        # sys.stderr = Logger
        # sys.stdout = Logger

    @staticmethod
    def remove_file_logger(delete_current_log: bool = False):
        """
        Remove the file logger to not write output to a log file
        """
        Logger.logger.removeHandler(Logger.file_handler)
        if delete_current_log and os.path.exists(Logger._current_log_file_path):
            try:
                os.remove(Logger._current_log_file_path)
            except PermissionError:
                warnings.warn(f"Could not remove {Logger._current_log_file_path}, permission denied")