116 lines
3.8 KiB
Python
116 lines
3.8 KiB
Python
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 / "validador-documentos.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="validador-documentos"):
|
|
setup_logging()
|
|
return TaskLogger(logging.getLogger(name)) |