import os import logging import inspect from logging.handlers import RotatingFileHandler from rich.logging import RichHandler DEFAULT_UNIFIED_LOGFILE = "logs/agentm.log" DEFAULT_LOG_LEVEL = os.environ.get("AGENTM_LOG_LEVEL", "DEBUG") def get_logger(name="AGENTM", level=None, max_bytes=5 * 1024 * 1024, backup_count=5): logger = logging.getLogger(name) if logger.handlers: return logger # Already configured level = level or DEFAULT_LOG_LEVEL if isinstance(level, str): level = getattr(logging, level.upper(), logging.INFO) logger.setLevel(level) # 🎛 Rich Console Handler console_handler = RichHandler( rich_tracebacks=True, markup=True, show_time=True, show_level=True, show_path=False, ) logger.addHandler(console_handler) # 🛡 Unified File Logger log_dir = os.getenv("AGENTM_LOG_DIR", "logs") try: os.makedirs(log_dir, exist_ok=True) unified_path = os.path.join(log_dir, "agentm.log") file_handler = RotatingFileHandler(unified_path, maxBytes=max_bytes, backupCount=backup_count, encoding="utf-8") file_formatter = logging.Formatter("[%(asctime)s] [%(levelname)s] [%(name)s] %(message)s", "%Y-%m-%d %H:%M:%S") file_handler.setFormatter(file_formatter) logger.addHandler(file_handler) except Exception: logger.warning("⚠️ File logging disabled: could not create log directory or file.") return logger def get_module_logger(level=None): module_name = inspect.stack()[1].frame.f_globals.get("__name__", "unknown") return get_logger(name="AGENTM", level=level) # Use unified logger name def log_with_caller(level: str, message: str): stack = inspect.stack() callee = stack[1] callee_func = callee.function callee_module = callee.frame.f_globals.get("__name__", "unknown") caller_func = "unknown" caller_module = "unknown" for frame in stack[2:]: if frame.function not in {"wrapper", "inner", ""}: caller_func = frame.function caller_module = frame.frame.f_globals.get("__name__", "unknown") break logger = get_logger("AGENTM") full_message = ( f"{message} ← {callee_module}.{callee_func} " f"→ called by {caller_module}.{caller_func}" ) getattr(logger, level.lower())(full_message) def set_global_log_level(level: str): resolved_level = getattr(logging, level.upper(), logging.INFO) logging.getLogger().setLevel(resolved_level) for name in logging.root.manager.loggerDict: logging.getLogger(name).setLevel(resolved_level) log_with_caller("info", f"🔧 Global log level set to {level.upper()}")