descarga_factura_transporte/logger_manager.py

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

LOG_DIR = Path(__file__).parent / "logs"
LOG_FILE = LOG_DIR / "descarga_factura_transporte.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) as sub:
            yield sub

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

    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)

class _StreamToLogger:
    """Redirige escrituras de un stream (stdout/stderr) al logger, línea por línea."""

    def __init__(self, logger: logging.Logger, level: int) -> None:
        self._logger = logger
        self._level = level
        self._buffer = ""

    def write(self, message: str) -> None:
        self._buffer += message
        while "\n" in self._buffer:
            line, self._buffer = self._buffer.split("\n", 1)
            if line:
                self._logger.log(self._level, line)

    def flush(self) -> None:
        pass


# ============================================================================
def setup_logging(level=None, console=True, capture_print=True):
    """
    Configura el logging del proceso: archivo rotado a medianoche (30 días
    de historial) y, opcionalmente, consola. Si ``capture_print`` es True
    (por defecto), también redirige ``print()`` (stdout/stderr) al log, de
    forma que cualquier salida existente en el proyecto quede registrada
    en archivo sin tener que migrar cada ``print`` a ``logger.info``.
    """
    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: list[logging.Handler] = [file_handler]
    if console:
        console_handler = logging.StreamHandler(sys.stdout)
        console_handler.setFormatter(formatter)
        handlers.append(console_handler)

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

    if capture_print:
        stdout_logger = logging.getLogger("stdout")
        stderr_logger = logging.getLogger("stderr")
        sys.stdout = _StreamToLogger(stdout_logger, logging.INFO)
        sys.stderr = _StreamToLogger(stderr_logger, logging.ERROR)

    _LOG_CONFIGURED = True

# ============================================================================
def get_logger(name="descarga_factura_transporte"):
    setup_logging()
    return TaskLogger(logging.getLogger(name))