98 lines
2.7 KiB
Python
98 lines
2.7 KiB
Python
import json
|
|
import logging
|
|
import logging.config
|
|
import os
|
|
from pathlib import Path
|
|
|
|
|
|
def _as_bool(val: str | None, default: bool = False) -> bool:
|
|
if val is None:
|
|
return default
|
|
return val.strip().lower() in ("1", "true", "yes", "on")
|
|
|
|
|
|
def _level_from_env() -> int:
|
|
level_name = os.getenv("LOG_LEVEL", "INFO").upper()
|
|
return getattr(logging, level_name, logging.INFO)
|
|
|
|
|
|
def setup_logging() -> None:
|
|
level = _level_from_env()
|
|
to_file = _as_bool(os.getenv("LOG_TO_FILE"), True)
|
|
log_file = os.getenv("LOG_FILE", "logs/mathema.log")
|
|
as_json = _as_bool(os.getenv("LOG_JSON"), False)
|
|
|
|
if to_file:
|
|
Path(log_file).parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
if as_json:
|
|
fmt = '%(message)s'
|
|
else:
|
|
fmt = ("%(asctime)s | %(levelname)-8s | %(name)s | "
|
|
"task=%(taskName)s | %(message)s")
|
|
|
|
console_handler = {
|
|
"class": "logging.StreamHandler",
|
|
"level": level,
|
|
"formatter": "json" if as_json else "default",
|
|
}
|
|
|
|
file_handler = {
|
|
"class": "logging.FileHandler",
|
|
"level": level,
|
|
"filename": log_file,
|
|
"encoding": "utf-8",
|
|
"formatter": "json" if as_json else "default",
|
|
}
|
|
|
|
handlers = {"console": console_handler}
|
|
if to_file:
|
|
handlers["file"] = file_handler
|
|
|
|
formatters = {
|
|
"default": {
|
|
"format": fmt,
|
|
"datefmt": "%Y-%m-%d %H:%M:%S",
|
|
},
|
|
"json": {
|
|
"()": "logging.Formatter",
|
|
"format": "%(message)s",
|
|
},
|
|
}
|
|
|
|
config = {
|
|
"version": 1,
|
|
"disable_existing_loggers": False,
|
|
"formatters": formatters,
|
|
"handlers": handlers,
|
|
"root": {
|
|
"level": level,
|
|
"handlers": list(handlers.keys()),
|
|
},
|
|
"loggers": {
|
|
"mathema": {"level": level, "propagate": True},
|
|
"asyncio": {"level": "WARNING"},
|
|
"gymnasium": {"level": "WARNING"},
|
|
"pygame": {"level": "WARNING"},
|
|
"neo4j": {"level": "WARNING"},
|
|
},
|
|
}
|
|
|
|
logging.config.dictConfig(config)
|
|
|
|
if as_json:
|
|
class JsonFilter(logging.Filter):
|
|
def filter(self, record: logging.LogRecord) -> bool:
|
|
payload = {
|
|
"ts": record.created,
|
|
"level": record.levelname,
|
|
"logger": record.name,
|
|
"task": getattr(record, "taskName", None),
|
|
"msg": record.getMessage(),
|
|
}
|
|
record.msg = json.dumps(payload, ensure_ascii=False)
|
|
return True
|
|
|
|
for h in logging.getLogger().handlers:
|
|
h.addFilter(JsonFilter())
|