AareCommon/src/aarecommon/config/logger.py

import logging
import logging.config
import os
from pathlib import Path

import yaml


class IgnoreSuccessfulStatusAccessFilter(logging.Filter):
    # Should be disabled if debugging status calls.
    # In particular if this is causing server/GUI lag
    _SUPPRESSED_SUCCESS_MESSAGES = (
        '"GET /status HTTP/1.1" 200',
        '"GET /sample/reference_tools HTTP/1.1" 200',
        '"GET /sample/spreadsheet HTTP/1.1" 200',
        '"GET /local_contact/device_state HTTP/1.1" 200',
        '"POST /scan/smart_params HTTP/1.1" 200',
    )

    _SUPPRESSED_EXPECTED_403_MESSAGES = (
        '"GET /sse/face_detection HTTP/1.1" 403',
        '"GET /sse/automation_progress HTTP/1.1" 403',
    )

    def filter(self, record: logging.LogRecord) -> bool:
        message = record.getMessage()
        return not any(
            entry in message
            for entry in (
                *self._SUPPRESSED_SUCCESS_MESSAGES,
                *self._SUPPRESSED_EXPECTED_403_MESSAGES,
            )
        )


def get_uvicorn_logging_config() -> dict:
    return {
        "version": 1,
        "disable_existing_loggers": False,
        "filters": {
            "ignore_successful_status_access": {
                "()": "aarecommon.logger_config.IgnoreSuccessfulStatusAccessFilter"
            }
        },
        "formatters": {
            "default": {
                "()": "uvicorn.logging.DefaultFormatter",
                "fmt": "%(levelprefix)s %(message)s",
                "use_colors": True,
            }
        },
        "handlers": {
            "default": {
                "formatter": "default",
                "class": "logging.StreamHandler",
                "stream": "ext://sys.stdout",
            },
            "access": {
                "formatter": "default",
                "class": "logging.StreamHandler",
                "stream": "ext://sys.stdout",
                "filters": ["ignore_successful_status_access"],
            },
        },
        "loggers": {
            "uvicorn": {"handlers": ["default"], "level": "DEBUG", "propagate": False},
            "uvicorn.access": {"handlers": ["access"], "level": "INFO", "propagate": False},
        },
    }


def get_config_path(env: str = "dev") -> Path:
    module_dir = Path(__file__).parent
    config_filename = f"logging_{env}.yaml"
    config_path = module_dir / "logging_configs" / config_filename

    if not config_path.exists():
        raise FileNotFoundError(f"Logging config not found: {config_path}")

    return config_path


# TODO fix logging.
def setup_logger(
    name="aareDAQ", base_dir: str | None = f"~/tmp/mxlogs/", config_path: str | None = None
):
    # switch to production mode using: $ APP_ENV=prod python main.py
    env = os.getenv("APP_ENV", "dev")  # default: dev

    if config_path is None:
        config_file = get_config_path(env)
    else:
        config_file = Path(config_path)

    with open(config_file, "r") as f:
        config = yaml.safe_load(f.read())

    # Force base directory to be under the user's home dir
    effective_base = base_dir + f"{name}" or f"/tmp/logs/mxlogs/{name}"
    effective_base = os.path.abspath(os.path.expanduser(os.path.expandvars(effective_base)))

    # Ensure directories for file handlers exist
    handlers = config.get("handlers", {})

    for h in handlers.values():
        filename = h.get("filename")
        if not filename:
            continue
        abs_filename = os.path.join(effective_base, os.path.basename(filename))

        h["filename"] = abs_filename
        log_dir = os.path.dirname(abs_filename)
        if log_dir and not os.path.exists(log_dir):
            os.makedirs(log_dir, exist_ok=True)

    logging.config.dictConfig(config)
    logging.getLogger("redis_lock").setLevel(logging.WARNING)
    logging.getLogger("urllib3").setLevel(logging.WARNING)
    logging.getLogger("matplotlib.font_manager").setLevel(logging.WARNING)
    logging.getLogger("aaredaq").setLevel(logging.DEBUG)  # DAQ
    logging.getLogger("aaregui").setLevel(logging.INFO)
    return logging.getLogger(name)


def find_existing_formatter(
    default_fmt="%(asctime)s [%(levelname)s] %(name)s: %(message)s", default_datefmt=None
):
    root = logging.getLogger()
    for h in root.handlers:
        if h.formatter:
            return h.formatter

    for logger_name, logger in logging.Logger.manager.loggerDict.items():
        if isinstance(logger, logging.Logger):
            for h in logger.handlers:
                if h.formatter:
                    return h.formatter

    return logging.Formatter(default_fmt, datefmt=default_datefmt)


def attach_to_logger(logger_name: str = "", handler: logging.Handler = None):
    logging.getLogger(logger_name).addHandler(handler)


if __name__ == "__main__":
    os.environ["APP_ENV"] = "dev"  # or "prod"
    # os.environ["LOG_DIR"] = "/tmp/mxlogs"  # optional; matches your setup code

    log = setup_logger("aareDAQ", base_dir="/tmp/test_logs")  # use the configured logger name

    log.debug("debug smoke test")
    log.error("error smoke test")
    log.info("info smoke test")
    log.warning("warning smoke test")
    log.critical("critical smoke test")

    for h in log.handlers:
        if hasattr(h, "flush"):
            h.flush()

    for h in log.handlers:
        log.debug("debug test")
        log.error("error test")
        print(type(h), getattr(h, "baseFilename", None), h.level)