import logging import os from logging.handlers import RotatingFileHandler from pathlib import Path # Rotation policy shared by every log file this backend creates: # 5 MB per file, keep 3 rotations (logs/brain.log, brain.log.1, .2, .3). # Tune both via env vars if you need larger logs on the robot. _ROT_MAX_BYTES = int(os.environ.get("MARCUS_LOG_MAX_BYTES", 5_000_000)) _ROT_BACKUP_COUNT = int(os.environ.get("MARCUS_LOG_BACKUP_COUNT", 3)) def _rotating_handler(path: str) -> RotatingFileHandler: """FileHandler with size-based rotation — prevents unbounded growth.""" return RotatingFileHandler( path, maxBytes=_ROT_MAX_BYTES, backupCount=_ROT_BACKUP_COUNT, encoding="utf-8", ) class Logs: def __init__(self, default_log_level=logging.DEBUG, main_log_file="main.log"): self.default_log_level = default_log_level self.log_format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s' self.base_dir = str(Path(__file__).resolve().parents[1]) # The canonical log directory is "logs" (lowercase) — matches what # every module writing via stdlib logging expects. "Logs" (capital L) # was historically used by a parallel implementation and is gone. self.default_logs_dir = os.path.join(self.base_dir, "logs") self.fallback_log_dir = self._choose_fallback_log_dir() self.mainloggerfile = self.resolve_log_path(main_log_file) self.logger = None # Initialize the main logger self.main_logger = logging.getLogger("MainLogger") self.main_logger.setLevel(self.default_log_level) self.main_logger.propagate = False # Prevent logging from printing to terminal if self.main_logger.hasHandlers(): self.main_logger.handlers.clear() # Remove any StreamHandler (to avoid console logs) for handler in list(self.main_logger.handlers): if isinstance(handler, logging.StreamHandler): self.main_logger.removeHandler(handler) os.makedirs(os.path.dirname(self.mainloggerfile), exist_ok=True) main_handler = _rotating_handler(self.mainloggerfile) main_handler.setFormatter(logging.Formatter(self.log_format)) main_handler.setLevel(self.default_log_level) self.main_logger.addHandler(main_handler) def _choose_fallback_log_dir(self): env_dir = os.environ.get("MARCUS_LOG_DIR", "").strip() candidates = [] if env_dir: candidates.append(env_dir) candidates.extend( [ self.default_logs_dir, os.path.join(os.path.expanduser("~"), ".marcus_logs"), "/tmp/marcus_logs", ] ) for d in candidates: try: os.makedirs(d, exist_ok=True) test = os.path.join(d, ".write_test") with open(test, "a", encoding="utf-8"): pass try: os.remove(test) except Exception: pass return os.path.abspath(d) except Exception: continue return os.path.abspath("/tmp") @staticmethod def _normalize_log_name(name): base = os.path.basename(str(name or "").strip()) or "main" while base.lower().endswith(".log.log"): base = base[:-4] if not base.lower().endswith(".log"): base += ".log" return base def _is_writable_path(self, full_path): parent = os.path.dirname(full_path) try: os.makedirs(parent, exist_ok=True) with open(full_path, "a", encoding="utf-8"): pass return True except Exception: return False def _with_fallback(self, desired_path): if self._is_writable_path(desired_path): return os.path.abspath(desired_path) fallback_path = os.path.join(self.fallback_log_dir, os.path.basename(desired_path)) if self._is_writable_path(fallback_path): return os.path.abspath(fallback_path) return os.path.abspath(desired_path) def resolve_log_path(self, path): """Resolve relative or absolute path to absolute, always under the active logs dir when relative.""" normalized_name = self._normalize_log_name(path) if os.path.isabs(str(path)): full_path = os.path.abspath(str(path)) else: full_path = os.path.join(self.fallback_log_dir, normalized_name) return self._with_fallback(full_path) def construct_path(self, folder_name, file_name): """Construct full path. Relative folders are centralized under the active logs dir.""" normalized_name = self._normalize_log_name(file_name) if os.path.isabs(folder_name): full_path = os.path.join(folder_name, normalized_name) else: full_path = os.path.join(self.fallback_log_dir, normalized_name) return self._with_fallback(full_path) def log_to_file(self, message, TypeLog): level_map = { "DEBUG": logging.DEBUG, "INFO": logging.INFO, "WARNING": logging.WARNING, "ERROR": logging.ERROR, "CRITICAL": logging.CRITICAL } log_level = level_map.get(TypeLog.upper(), logging.WARNING) self.main_logger.log(log_level, message) def LogEngine(self, folder_name, log_name): """Set up a named logger and resolve the file path correctly.""" full_path = self.construct_path(folder_name, log_name) self.logger = logging.getLogger(log_name) self.logger.setLevel(self.default_log_level) self.logger.propagate = False # Prevent printing to terminal # Clear existing FileHandlers for handler in self.logger.handlers[:]: if isinstance(handler, logging.FileHandler): self.logger.removeHandler(handler) handler = _rotating_handler(full_path) handler.setFormatter(logging.Formatter(self.log_format)) handler.setLevel(self.default_log_level) self.logger.addHandler(handler) def LogsMessages(self, message, message_type="info", folder_name=None, file_name=None): if folder_name and file_name: full_path = self.construct_path(folder_name, file_name) temp_logger = logging.getLogger(f"{folder_name}_{file_name}") temp_logger.setLevel(self.default_log_level) temp_logger.propagate = False # Prevent printing to terminal # Re-use the existing handler if it's already attached to the # same file (by absolute path). Prevents handler accumulation # when this function is called from long-running loops. if not any(isinstance(h, logging.FileHandler) and h.baseFilename == full_path for h in temp_logger.handlers): handler = _rotating_handler(full_path) handler.setFormatter(logging.Formatter(self.log_format)) temp_logger.addHandler(handler) getattr(temp_logger, message_type.lower(), temp_logger.warning)(message) elif self.logger: log_method = getattr(self.logger, message_type.lower(), self.logger.warning) log_method(message) else: self.log_to_file(message, message_type.upper()) def print_and_log(self, message, message_type="info", folder_name=None, file_name=None): self.LogsMessages(message, message_type, folder_name, file_name) print(message) # ============================== # Usage Example # ============================== if __name__ == "__main__": logger = Logs() logger.LogEngine("ExxxxampleLogger", "ExampleLogger.log") logger.LogsMessages("This is a hidden message") logger.print_and_log("This is a test message.", message_type="info") # You can also directly specify folder and file for a log message logger.print_and_log("Direct log to folder", message_type="info", folder_name="CustomLogs", file_name="event.log")