src/utils/logger.py

"""Configuração centralizada de logging do Analytical-Force.

Fornece um logger com saída para console e arquivo (``logs/analytical_force.log``).
Inclui um filtro simples que evita o vazamento acidental de segredos no log.
"""

from __future__ import annotations

import logging
import re
from logging.handlers import RotatingFileHandler
from pathlib import Path

# Diretório de logs na raiz do projeto.
_RAIZ_PROJETO = Path(__file__).resolve().parents[2]
_DIR_LOGS = _RAIZ_PROJETO / "logs"
_ARQUIVO_LOG = _DIR_LOGS / "analytical_force.log"

# Nome do logger raiz do projeto.
_NOME_LOGGER = "analytical_force"

# Padrões que NÃO devem aparecer no log (mascarados como ***).
# Cobre chaves comuns e valores tipo token/secret.
_PADROES_SENSIVEIS = [
    re.compile(r"(?i)(password\s*[=:]\s*)(\S+)"),
    re.compile(r"(?i)(token\s*[=:]\s*)(\S+)"),
    re.compile(r"(?i)(api[_-]?key\s*[=:]\s*)(\S+)"),
    re.compile(r"(?i)(secret\s*[=:]\s*)(\S+)"),
    re.compile(r"(?i)(authorization:\s*bearer\s+)(\S+)"),
]


class _FiltroSegredos(logging.Filter):
    """Filtro que mascara valores sensíveis nas mensagens de log."""

    def filter(self, record: logging.LogRecord) -> bool:  # noqa: A003
        try:
            mensagem = record.getMessage()
        except Exception:
            return True
        original = mensagem
        for padrao in _PADROES_SENSIVEIS:
            mensagem = padrao.sub(r"\1***", mensagem)
        if mensagem != original:
            # Substitui args já formatados pela versão mascarada.
            record.msg = mensagem
            record.args = None
        return True


def configurar_logger(nivel: int = logging.INFO) -> logging.Logger:
    """Configura e retorna o logger raiz do projeto.

    É idempotente: chamadas repetidas não duplicam handlers.

    Args:
        nivel: Nível de log (ex.: ``logging.INFO``, ``logging.DEBUG``).

    Returns:
        Logger configurado.
    """
    logger = logging.getLogger(_NOME_LOGGER)
    logger.setLevel(nivel)

    # Evita adicionar handlers duplicados em re-execuções/imports.
    if logger.handlers:
        return logger

    formato = logging.Formatter(
        fmt="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )
    filtro = _FiltroSegredos()

    # Handler de console.
    console = logging.StreamHandler()
    console.setFormatter(formato)
    console.addFilter(filtro)
    logger.addHandler(console)

    # Handler de arquivo com rotação (best-effort: não quebra se faltar permissão).
    try:
        _DIR_LOGS.mkdir(parents=True, exist_ok=True)
        arquivo = RotatingFileHandler(
            _ARQUIVO_LOG, maxBytes=2_000_000, backupCount=5, encoding="utf-8"
        )
        arquivo.setFormatter(formato)
        arquivo.addFilter(filtro)
        logger.addHandler(arquivo)
    except OSError:
        logger.warning("Não foi possível criar o arquivo de log em %s", _ARQUIVO_LOG)

    # Não propaga para o root logger para evitar logs duplicados.
    logger.propagate = False
    return logger


def get_logger(nome: str | None = None) -> logging.Logger:
    """Retorna um logger filho do logger do projeto.

    Args:
        nome: Sufixo do logger (ex.: ``"salesforce"``). Se None, retorna o raiz.

    Returns:
        Logger pronto para uso.
    """
    # Garante que o logger raiz esteja configurado ao menos uma vez.
    configurar_logger()
    if nome:
        return logging.getLogger(f"{_NOME_LOGGER}.{nome}")
    return logging.getLogger(_NOME_LOGGER)