EvoScientist/EvoScientist/logging_config.py

"""Shared logging configuration helpers."""

from __future__ import annotations

import logging
import os
import sys
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, TextIO

DEFAULT_LOG_RETENTION_DAYS = 30
DEFAULT_LOG_FORMAT = "%(asctime)s [%(levelname)s] %(name)s: %(message)s"
DEFAULT_LOG_DATE_FORMAT = "%Y-%m-%d %H:%M:%S"
MANAGED_HANDLER_ATTR = "_evoscientist_managed_handler"


def resolve_log_level(level: int | str | None, default: int = logging.INFO) -> int:
    """Resolve a logging level from config or environment input."""
    if isinstance(level, int):
        return level
    raw = str(level or "").strip()
    if not raw:
        return default
    if raw.isdigit():
        return int(raw)
    normalized = raw.upper()
    if normalized == "WARN":
        normalized = "WARNING"
    resolved = logging.getLevelNamesMapping().get(normalized)
    return resolved if isinstance(resolved, int) else default


def _mark_managed(handler: logging.Handler, kind: str) -> logging.Handler:
    setattr(handler, MANAGED_HANDLER_ATTR, kind)
    return handler


def _managed_kind(handler: logging.Handler) -> str | None:
    kind = getattr(handler, MANAGED_HANDLER_ATTR, None)
    return kind if isinstance(kind, str) else None


def remove_managed_handlers(
    logger: logging.Logger | None = None,
    *,
    kinds: set[str] | None = None,
) -> None:
    """Remove handlers installed by this module without touching external ones."""
    target = logger or logging.getLogger()
    for handler in target.handlers[:]:
        kind = _managed_kind(handler)
        if kind and (kinds is None or kind in kinds):
            target.removeHandler(handler)
            handler.close()


def _standard_formatter() -> logging.Formatter:
    return logging.Formatter(DEFAULT_LOG_FORMAT, datefmt=DEFAULT_LOG_DATE_FORMAT)


class DailyLogFileHandler(logging.FileHandler):
    """File handler that writes the active log to a date-based filename."""

    def __init__(
        self,
        log_dir: str | Path,
        *,
        prefix: str = "evoscientist",
        retention_days: int = DEFAULT_LOG_RETENTION_DAYS,
        encoding: str = "utf-8",
        utc: bool = False,
    ) -> None:
        self.log_dir = Path(log_dir).expanduser()
        self.prefix = prefix
        self.retention_days = max(1, retention_days)
        self.utc = utc
        self.log_dir.mkdir(parents=True, exist_ok=True)
        super().__init__(self._dated_log_path(), encoding=encoding, delay=True)

    @property
    def active_log_path(self) -> Path:
        """Return the active log path for the current date."""
        return self._dated_log_path()

    def _dated_log_path(self) -> Path:
        now = datetime.now(UTC if self.utc else None)
        return self.log_dir / f"{self.prefix}-{now:%Y-%m-%d}.log"

    def emit(self, record: logging.LogRecord) -> None:
        try:
            expected = str(self.active_log_path)
            if self.baseFilename != expected:
                if self.stream:
                    self.stream.close()
                    self.stream = None
                self.baseFilename = expected
            self._delete_expired_logs()
            super().emit(record)
        except OSError:
            self.handleError(record)

    def getFilesToDelete(self) -> list[str]:
        candidates = sorted(self.log_dir.glob(f"{self.prefix}-????-??-??.log"))
        if len(candidates) <= self.retention_days:
            return []
        return [str(path) for path in candidates[: -self.retention_days]]

    def _delete_expired_logs(self) -> None:
        for path in self.getFilesToDelete():
            try:
                os.remove(path)
            except OSError:
                pass


def default_log_dir() -> Path:
    """Return the default runtime log directory."""
    env_dir = os.environ.get("EVOSCIENTIST_LOG_DIR")
    if env_dir:
        return Path(env_dir).expanduser()

    from EvoScientist.paths import DATA_DIR

    return DATA_DIR / "logs"


def configure_daily_file_logging(
    logger: logging.Logger | None = None,
    *,
    log_dir: str | Path | None = None,
    prefix: str = "evoscientist",
    level: int | str = logging.INFO,
    retention_days: int = DEFAULT_LOG_RETENTION_DAYS,
) -> DailyLogFileHandler:
    """Attach a daily file handler, replacing older matching handlers."""
    target = logger or logging.getLogger()
    resolved_level = resolve_log_level(level, default=logging.INFO)
    retention_days = max(1, int(retention_days))
    resolved_dir = Path(log_dir).expanduser() if log_dir else default_log_dir()

    for handler in target.handlers[:]:
        if (
            isinstance(handler, DailyLogFileHandler)
            and handler.prefix == prefix
            and handler.log_dir == resolved_dir
        ):
            target.removeHandler(handler)
            handler.close()

    handler = DailyLogFileHandler(
        resolved_dir,
        prefix=prefix,
        retention_days=retention_days,
    )
    _mark_managed(handler, "file")
    handler.setLevel(resolved_level)
    handler.setFormatter(_standard_formatter())
    target.addHandler(handler)
    if target.level == logging.NOTSET or target.level > resolved_level:
        target.setLevel(resolved_level)
    return handler


def configure_console_logging(
    logger: logging.Logger | None = None,
    *,
    level: int | str | None = logging.INFO,
    stream: TextIO | None = None,
    replace: bool = True,
) -> logging.StreamHandler:
    """Attach a standard console handler for non-interactive entry points."""
    target = logger or logging.getLogger()
    resolved_level = resolve_log_level(level, default=logging.INFO)
    if replace:
        remove_managed_handlers(target, kinds={"console", "rich"})

    handler = logging.StreamHandler(stream or sys.stderr)
    _mark_managed(handler, "console")
    handler.setLevel(resolved_level)
    handler.setFormatter(_standard_formatter())
    target.addHandler(handler)
    target.setLevel(resolved_level)
    return handler


def configure_rich_console_logging(
    logger: logging.Logger | None = None,
    *,
    level: int | str | None = logging.INFO,
    console: Any = None,
    replace: bool = True,
    dim_warnings: bool = False,
    show_time: bool | None = None,
    show_path: bool | None = None,
    show_level: bool | None = None,
) -> logging.Handler:
    """Attach a Rich console handler for interactive CLI output."""
    from rich.logging import RichHandler
    from rich.markup import escape

    target = logger or logging.getLogger()
    resolved_level = resolve_log_level(level, default=logging.INFO)
    verbose = resolved_level <= logging.DEBUG
    if replace:
        remove_managed_handlers(target, kinds={"console", "rich"})

    class DimWarningHandler(RichHandler):
        def emit(self, record: logging.LogRecord) -> None:
            if dim_warnings and record.levelno == logging.WARNING and console is not None:
                console.print(
                    "[dim yellow]\u26a0\ufe0f  Warning:[/dim yellow] "
                    f"[dim]{escape(record.getMessage())}[/dim]"
                )
                return
            super().emit(record)

    handler = DimWarningHandler(
        console=console,
        show_time=verbose if show_time is None else show_time,
        show_path=verbose if show_path is None else show_path,
        show_level=verbose if show_level is None else show_level,
    )
    _mark_managed(handler, "rich")
    handler.setLevel(resolved_level)
    target.addHandler(handler)
    target.setLevel(resolved_level)
    return handler


def configure_logging(
    logger: logging.Logger | None = None,
    *,
    level: int | str | None = logging.INFO,
    log_dir: str | Path | None = None,
    retention_days: int = DEFAULT_LOG_RETENTION_DAYS,
    prefix: str = "evoscientist",
    console: bool = True,
    file: bool = True,
    replace_managed: bool = True,
) -> list[logging.Handler]:
    """Configure standard EvoScientist console and daily file logging."""
    target = logger or logging.getLogger()
    resolved_level = resolve_log_level(level, default=logging.INFO)
    if replace_managed:
        remove_managed_handlers(target, kinds={"console", "rich", "file"})

    handlers: list[logging.Handler] = []
    if console:
        handlers.append(
            configure_console_logging(target, level=resolved_level, replace=False)
        )
    if file:
        handlers.append(
            configure_daily_file_logging(
                target,
                log_dir=log_dir,
                prefix=prefix,
                level=resolved_level,
                retention_days=retention_days,
            )
        )
    target.setLevel(resolved_level)
    return handlers


def configure_logging_from_settings(
    logger: logging.Logger | None = None,
    *,
    default_level: int = logging.INFO,
    prefix: str = "evoscientist",
    console: bool = True,
    file: bool = True,
) -> list[logging.Handler]:
    """Configure logging from EvoScientist settings and environment overrides."""
    level: int | str | None = os.environ.get("EVOSCIENTIST_LOG_LEVEL")
    log_dir: str | Path | None = os.environ.get("EVOSCIENTIST_LOG_DIR") or None
    retention_days = int(
        os.environ.get("EVOSCIENTIST_LOG_RETENTION_DAYS", DEFAULT_LOG_RETENTION_DAYS)
    )

    try:
        from EvoScientist.config import get_effective_config

        cfg = get_effective_config()
        level = level or getattr(cfg, "log_level", None)
        log_dir = log_dir or getattr(cfg, "log_dir", None) or None
        retention_days = int(
            getattr(cfg, "log_retention_days", DEFAULT_LOG_RETENTION_DAYS)
        )
    except Exception:
        level = level or default_level

    return configure_logging(
        logger,
        level=resolve_log_level(level, default=default_level),
        log_dir=log_dir,
        retention_days=retention_days,
        prefix=prefix,
        console=console,
        file=file,
    )