110 lines
4.4 KiB
Python
110 lines
4.4 KiB
Python
from contextlib import contextmanager
|
||
import inspect
|
||
import logging.config
|
||
import logging, os, sys, platform
|
||
try:
|
||
# Only available on Linux with systemd
|
||
from systemd.journal import JournalHandler # type: ignore
|
||
HAS_JOURNAL = True
|
||
except Exception:
|
||
HAS_JOURNAL = False
|
||
|
||
|
||
class MaxLevelFilter(logging.Filter):
|
||
"""Allow records up to a certain level (inclusive)."""
|
||
def __init__(self, level): super().__init__(); self.level = level
|
||
def filter(self, record): return record.levelno <= self.level
|
||
|
||
|
||
class EnsureContext(logging.Filter):
|
||
"""Always provide 'context' so the formatter never KeyErrors."""
|
||
def filter(self, record: logging.LogRecord) -> bool:
|
||
if not hasattr(record, "context"):
|
||
record.context = ""
|
||
return True
|
||
|
||
def setup_logging(level: int = logging.INFO, syslog_id: str = "energy-scrapers") -> None:
|
||
"""Use journald if available; otherwise split stdout/stderr (Windows-friendly)."""
|
||
root = logging.getLogger()
|
||
root.handlers.clear()
|
||
root.setLevel(level)
|
||
|
||
fmt = logging.Formatter("%(asctime)s %(levelname)s [%(name)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S")
|
||
|
||
if HAS_JOURNAL and platform.system() == "Linux":
|
||
# Native journald handler (preserves levels/metadata)
|
||
h = JournalHandler(SYSLOG_IDENTIFIER=syslog_id)
|
||
h.setFormatter(logging.Formatter("%(message)s")) # journald adds timestamp/level
|
||
root.addHandler(h)
|
||
else:
|
||
# Portable fallback: INFO and below -> stdout, WARNING+ -> stderr
|
||
h_out = logging.StreamHandler(sys.stdout)
|
||
h_out.setLevel(logging.DEBUG)
|
||
h_out.addFilter(MaxLevelFilter(logging.INFO))
|
||
h_out.setFormatter(fmt)
|
||
|
||
h_err = logging.StreamHandler(sys.stderr)
|
||
h_err.setLevel(logging.WARNING)
|
||
h_err.setFormatter(fmt)
|
||
|
||
root.addHandler(h_out)
|
||
root.addHandler(h_err)
|
||
|
||
# Optional: make sure Python doesn’t buffer output (useful on Windows/services)
|
||
os.environ.setdefault("PYTHONUNBUFFERED", "1")
|
||
|
||
def _fmt_ctx(ctx: dict) -> str:
|
||
# Build a single bracketed context block for the formatter: "[k=v a=b] "
|
||
if not ctx:
|
||
return ""
|
||
kv = " ".join(f"{k}={v}" for k, v in ctx.items() if v is not None)
|
||
return f"[{kv}] "
|
||
|
||
|
||
class HasLogger:
|
||
"""Mixin with explicit logger initialization. No __init__, no MRO dependency."""
|
||
|
||
def __init__(self):
|
||
self.log = None
|
||
self._log_ctx = None
|
||
self._base_logger = None
|
||
|
||
def init_logger(self, *, context: dict | None = None, name: str | None = None) -> None:
|
||
"""Initialize the logger explicitly; call from your class __init__."""
|
||
base_name = name or self.__class__.__name__
|
||
self._base_logger = logging.getLogger(base_name)
|
||
self._log_ctx: dict = dict(context or {})
|
||
self.log = logging.LoggerAdapter(self._base_logger, {"context": _fmt_ctx(self._log_ctx)})
|
||
|
||
def set_logger_context(self, **context) -> None:
|
||
"""Persistently merge new context into this instance and refresh adapter."""
|
||
if not hasattr(self, "_base_logger"): # safety if init_logger was forgotten
|
||
self.init_logger()
|
||
self._log_ctx.update({k: v for k, v in context.items() if v is not None})
|
||
self.log = logging.LoggerAdapter(self._base_logger, {"context": _fmt_ctx(self._log_ctx)})
|
||
|
||
def child_logger(self, **extra) -> logging.LoggerAdapter:
|
||
"""Temporary adapter with additional context (does not mutate instance context)."""
|
||
if not hasattr(self, "_base_logger"):
|
||
self.init_logger()
|
||
merged = self._log_ctx.copy()
|
||
merged.update({k: v for k, v in extra.items() if v is not None})
|
||
return logging.LoggerAdapter(self._base_logger, {"context": _fmt_ctx(merged)})
|
||
|
||
@contextmanager
|
||
def scoped_log(self, **extra):
|
||
"""Context manager yielding a temporary adapter with extra context."""
|
||
yield self.child_logger(**extra)
|
||
|
||
|
||
def function_logger(name: str | None = None, **context):
|
||
"""Return a LoggerAdapter that follows the HasLogger format for free functions."""
|
||
# Derive a readable default name: "<module>.<function>"
|
||
if name is None:
|
||
frame = inspect.currentframe().f_back # caller
|
||
func = frame.f_code.co_name
|
||
mod = frame.f_globals.get("__name__", "app")
|
||
name = f"{mod}.{func}"
|
||
h = HasLogger()
|
||
h.init_logger(name=name, context=context)
|
||
return h.log |