""" Structured logging setup for Code of Conquest. Environment-aware behavior: - dev → colorful console output (human-friendly) - test/prod → JSON logs (machine-friendly) Features: - Includes logger name, level, filename, and line number - Unified stdlib + structlog integration - Respects LOG_LEVEL from environment - Works with both ChatGPT/Ollama components and Web/TUI layers """ from __future__ import annotations import logging import sys from dataclasses import dataclass from typing import Optional import structlog from structlog.dev import ConsoleRenderer from structlog.processors import JSONRenderer, TimeStamper, CallsiteParameterAdder, CallsiteParameter from structlog.stdlib import ProcessorFormatter @dataclass class _ConfiguredState: configured: bool = False level_name: str = "INFO" _state = _ConfiguredState() def _level_from_name(name: str) -> int: mapping = { "CRITICAL": logging.CRITICAL, "ERROR": logging.ERROR, "WARNING": logging.WARNING, "INFO": logging.INFO, "DEBUG": logging.DEBUG, "NOTSET": logging.NOTSET, } return mapping.get((name or "INFO").upper(), logging.INFO) def _shared_processors(): """ Processors common to both console and JSON pipelines. Adds level, logger name, and callsite metadata. """ return [ structlog.stdlib.add_log_level, structlog.stdlib.add_logger_name, CallsiteParameterAdder( parameters=[ CallsiteParameter.FILENAME, CallsiteParameter.LINENO, CallsiteParameter.FUNC_NAME, ] ), structlog.processors.StackInfoRenderer(), structlog.processors.format_exc_info, ] def _console_renderer(): """Pretty colored logs for development.""" return ConsoleRenderer() def _json_renderer(): """Machine-friendly JSON logs for test/prod.""" return JSONRenderer(sort_keys=True) def configure_logging(settings=None) -> None: """ Configure structlog + stdlib logging once for the process. """ if _state.configured: return if settings is None: from app.core.utils.settings import get_settings # lazy import settings = get_settings() env = settings.env.value level_name = settings.log_level or "INFO" level = _level_from_name(level_name) # Choose renderers if env == "dev": renderer = _console_renderer() foreign_pre_chain = _shared_processors() else: renderer = _json_renderer() foreign_pre_chain = _shared_processors() + [ TimeStamper(fmt="iso", utc=True, key="ts") ] # stdlib -> structlog bridge handler = logging.StreamHandler(sys.stdout) handler.setLevel(level) handler.setFormatter( ProcessorFormatter( processor=renderer, foreign_pre_chain=foreign_pre_chain, ) ) root = logging.getLogger() root.handlers.clear() root.setLevel(level) root.addHandler(handler) # Quiet noisy libs logging.getLogger("urllib3").setLevel(logging.WARNING) logging.getLogger("httpx").setLevel(logging.WARNING) logging.getLogger("uvicorn").setLevel(logging.INFO) # structlog pipeline structlog.configure( processors=[ structlog.contextvars.merge_contextvars, structlog.stdlib.filter_by_level, *foreign_pre_chain, ProcessorFormatter.wrap_for_formatter, # hand off to renderer ], wrapper_class=structlog.stdlib.BoundLogger, context_class=dict, logger_factory=structlog.stdlib.LoggerFactory(), cache_logger_on_first_use=True, ) _state.configured = True _state.level_name = level_name def get_logger(name: Optional[str] = None) -> structlog.stdlib.BoundLogger: """ Retrieve a structlog logger. """ if name is None: return structlog.get_logger() return structlog.get_logger(name)