"""Application logger — loguru with structured-field support.""" import os import sys from pathlib import Path from loguru import logger as _loguru_logger from app.platform.paths import log_dir as get_log_dir # Re-export as the canonical logger so imports stay uniform. logger = _loguru_logger _configured = False _console_sink_id: int | None = None _file_sink_id: int | None = None _console_level = "INFO" _json_console = False _file_logging = True _log_dir_override: Path | None = None _FMT_TEXT = ( "{time:YYYY-MM-DD HH:mm:ss.SSS} | " "{level: <8} | " "{name}:{function}:{line} - " "{message}" ) def _get_env_bool(name: str, default: bool) -> bool: value = os.getenv(name) if value is None: return default return value.strip().lower() in {"1", "true", "yes", "on"} def setup_logging( *, level: str = "INFO", file_level: str | None = None, json_console: bool = False, file_logging: bool = True, log_dir: Path | None = None, max_files: int = 7, ) -> None: """Configure loguru sinks. Safe to call multiple times (idempotent).""" global _configured, _console_sink_id, _file_sink_id global _console_level, _json_console, _file_logging, _log_dir_override logger.remove() _console_sink_id = None _file_sink_id = None fmt_json = "{time} | {level} | {name}:{function}:{line} | {message}" resolved_level = level.upper() _console_sink_id = logger.add( sys.stdout, level=resolved_level, format=fmt_json if json_console else _FMT_TEXT, colorize=not json_console, enqueue=False, backtrace=False, diagnose=False, ) if file_logging: _add_file_sink( file_level=(file_level or resolved_level).upper(), max_files=max_files, log_dir=log_dir, ) _configured = True _console_level = resolved_level _json_console = json_console _file_logging = file_logging _log_dir_override = log_dir def reload_logging( *, level: str | None = None, default_level: str = "INFO", json_console: bool = False, max_files: int = 7, file_level: str | None = None, ) -> None: """Re-configure logging from runtime values (called after config loads).""" resolved_level = (level or "").strip() or os.getenv("LOG_LEVEL", default_level) file_logging = _get_env_bool("LOG_FILE_ENABLED", True) setup_logging( level=resolved_level, file_level=file_level or resolved_level, json_console=json_console, file_logging=file_logging, max_files=max_files, ) def reload_file_logging( *, file_level: str | None = None, max_files: int = 7, ) -> None: """Re-configure only the file sink, preserving the current console output.""" global _file_sink_id, _file_logging if not _configured: reload_logging(file_level=file_level, max_files=max_files) return if _file_sink_id is not None: logger.remove(_file_sink_id) _file_sink_id = None _file_logging = _get_env_bool("LOG_FILE_ENABLED", True) if not _file_logging: return _add_file_sink( file_level=(file_level or _console_level).upper(), max_files=max_files, log_dir=_log_dir_override, ) def _add_file_sink( *, file_level: str, max_files: int, log_dir: Path | None, ) -> None: global _file_sink_id _dir = log_dir or get_log_dir() _dir.mkdir(parents=True, exist_ok=True) _file_sink_id = logger.add( str(_dir / "app_{time:YYYY-MM-DD}.log"), level=file_level, format=_FMT_TEXT, rotation="00:00", # new file every day at midnight retention=max_files, # keep the last N daily files enqueue=True, encoding="utf-8", backtrace=False, diagnose=False, ) __all__ = ["logger", "setup_logging", "reload_logging", "reload_file_logging"]