File size: 2,616 Bytes
b404de3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 | import json
import logging
from logging.handlers import RotatingFileHandler
from pathlib import Path
class Ansi:
RESET = "\033[0m"
BOLD = "\033[1m"
RED = "\033[31m"
GREEN = "\033[32m"
YELLOW = "\033[33m"
CYAN = "\033[36m"
MAGENTA = "\033[35m"
def setup_logger(log_file: str = "logs/cloudenv_steps.log") -> logging.Logger:
logger = logging.getLogger("cloudenv.step")
logger.setLevel(logging.DEBUG)
if logger.handlers:
return logger
Path(log_file).parent.mkdir(parents=True, exist_ok=True)
console = logging.StreamHandler()
class ColorFormatter(logging.Formatter):
def __init__(self):
super().__init__("%(message)s")
def format(self, record):
if not hasattr(record, "episode_id"):
return super().format(record)
status = record.__dict__.get("status", "UNKNOWN")
if status == "SUCCESS":
status_col = f"{Ansi.GREEN}SUCCESS{Ansi.RESET}"
elif status == "ACCESS_DENIED":
status_col = f"{Ansi.YELLOW}ACCESS_DENIED{Ansi.RESET}"
else:
status_col = f"{Ansi.RED}ERROR{Ansi.RESET}"
return (
f"{Ansi.BOLD}{Ansi.CYAN}[STEP]{Ansi.RESET} "
f"ep={Ansi.YELLOW}{record.episode_id}{Ansi.RESET} "
f"sid={Ansi.YELLOW}{record.task_id}{Ansi.RESET} "
f"step={Ansi.MAGENTA}{record.step}{Ansi.RESET} "
f"{record.service}.{record.operation} "
f"=> {status_col}"
)
console.setFormatter(ColorFormatter())
logger.addHandler(console)
file_handler = RotatingFileHandler(
log_file,
# 5MB
maxBytes=5 * 1024 * 1024,
backupCount=5,
)
class JsonFormatter(logging.Formatter):
def __init__(self):
super().__init__("%(message)s")
def format(self, record):
payload = {
"episode_id": record.__dict__.get("episode_id"),
"task_id": record.__dict__.get("task_id"),
"step": record.__dict__.get("step"),
"service": record.__dict__.get("service"),
"operation": record.__dict__.get("operation"),
"status": record.__dict__.get("status"),
"input": record.__dict__.get("input"),
"output": record.__dict__.get("output"),
}
return json.dumps(payload)
file_handler.setFormatter(JsonFormatter())
logger.addHandler(file_handler)
return logger
logger = setup_logger()
|