descarga_cpe_sovos/logger_manager.py

import logging
import contextvars
import os
from contextlib import contextmanager
from logging.handlers import TimedRotatingFileHandler
from pathlib import Path

LOG_DIR = Path(__file__).parent / "logs"
LOG_FILE = LOG_DIR / "descarga-cpe.log"
_LOG_CONFIGURED = False

_stack: contextvars.ContextVar[list] = contextvars.ContextVar("log_stack", default=[])
# ============================================================================

class TreeFormatter(logging.Formatter):
    def format(self, record):
        return f"{self.formatTime(record)} ─── {record.getMessage()}"

    def formatTime(self, record, datefmt=None):
        import datetime
        return datetime.datetime.fromtimestamp(record.created).strftime("%Y-%m-%d %H:%M:%S")

# ============================================================================
def _build_prefix(stack, is_last):
    if not stack:
        return ""
    parts = []
    for sib_is_last in stack[:-1]:
        parts.append("    " if sib_is_last else "│   ")
    parts.append("└── " if is_last else "├── ")
    return "".join(parts)

# ============================================================================
class _TaskContext:
    def __init__(self, logger, msg, emoji, is_last):
        self._logger = logger
        self._msg = msg
        self._emoji = emoji
        self._is_last = is_last

    def __enter__(self):
        stack = _stack.get()
        prefix = _build_prefix(stack, self._is_last)
        emoji_str = f"{self._emoji} " if self._emoji else ""
        self._logger._logger.info(f"{prefix}{emoji_str}{self._msg}")
        _stack.set(stack + [self._is_last])
        return TaskLogger(self._logger._logger)

    def __exit__(self, exc_type, exc_val, exc_tb):
        _stack.set(_stack.get()[:-1])

# ============================================================================
class TaskLogger:
    def __init__(self, logger):
        self._logger = logger

    @contextmanager
    def task(self, msg, emoji=""):
        with _TaskContext(self, msg, emoji, is_last=False):
            yield

    @contextmanager
    def last_task(self, msg, emoji=""):
        with _TaskContext(self, msg, emoji, is_last=True):
            yield

    def info(self, msg, *args, _last=False, **kwargs):
        self._log(logging.INFO, msg, args, kwargs, _last)

    def warning(self, msg, *args, _last=False, **kwargs):
        self._log(logging.WARNING, msg, args, kwargs, _last)

    def error(self, msg, *args, _last=False, **kwargs):
        self._log(logging.ERROR, msg, args, kwargs, _last)

    def _log(self, level, msg, args, kwargs, is_last):
        stack = _stack.get()
        prefix = _build_prefix(stack, is_last)
        self._logger.log(level, f"{prefix}{msg}", *args, **kwargs)

# ============================================================================
def setup_logging(level=None, console=True):
    global _LOG_CONFIGURED
    if _LOG_CONFIGURED:
        return

    if level is None:
        level_name = os.getenv("LOG_LEVEL", "INFO").upper()
        level = getattr(logging, level_name, logging.INFO)

    LOG_DIR.mkdir(exist_ok=True)

    formatter = TreeFormatter()

    file_handler = TimedRotatingFileHandler(
        LOG_FILE,
        when="midnight",
        interval=1,
        backupCount=30,
        encoding="utf-8",
    )
    file_handler.setFormatter(formatter)

    handlers = [file_handler]
    if console:
        console_handler = logging.StreamHandler()
        console_handler.setFormatter(formatter)
        handlers.append(console_handler)

    logging.basicConfig(level=level, handlers=handlers, force=True)
    _LOG_CONFIGURED = True

# ============================================================================
def get_logger(name="descarga-cpe"):
    setup_logging()
    return TaskLogger(logging.getLogger(name))