logging_setup.py

"""Logging — 로테이팅 파일 + in-memory ring buffer + 레벨 필터.

UI 의 "로그" 페이지가 `recent()` 결과를 폴링해서 표시.
파일은 `logs/app.log` (1MB 단위 롤링, 최근 3개 보존).
"""
from __future__ import annotations

import logging
import logging.handlers
import time
from collections import deque
from pathlib import Path
from threading import Lock

LOG_DIR = Path(__file__).resolve().parent / "logs"
LOG_DIR.mkdir(exist_ok=True)
LOG_FILE = LOG_DIR / "app.log"

_BUFFER: deque = deque(maxlen=600)
_buf_lock = Lock()

LEVEL_RANK = {"DEBUG": 0, "INFO": 1, "WARNING": 2, "ERROR": 3, "CRITICAL": 4}


class MemoryHandler(logging.Handler):
    """In-memory ring — UI 가 즉시 가져갈 수 있게 dict 형태로 저장."""

    def emit(self, record: logging.LogRecord) -> None:
        try:
            msg = record.getMessage()
            if record.exc_info:
                # 간단히 한 줄 요약만
                msg += f" | {record.exc_info[0].__name__}: {record.exc_info[1]}"
        except Exception:
            msg = "<format error>"
        with _buf_lock:
            _BUFFER.append({
                "ts":      time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(record.created)) +
                           f".{int(record.msecs):03d}",
                "ts_unix": record.created,
                "level":   record.levelname,
                "logger":  record.name,
                "module":  record.module,
                "line":    record.lineno,
                "message": msg,
            })


def setup(level: str = "INFO") -> None:
    root = logging.getLogger()
    root.setLevel(level.upper())
    # 중복 등록 방지
    for h in list(root.handlers):
        if isinstance(h, MemoryHandler) or (
            isinstance(h, logging.handlers.RotatingFileHandler)
            and getattr(h, "baseFilename", "") == str(LOG_FILE)
        ):
            root.removeHandler(h)

    fmt = logging.Formatter("%(asctime)s [%(levelname)s] %(name)s: %(message)s")

    fh = logging.handlers.RotatingFileHandler(
        LOG_FILE, maxBytes=1024 * 1024, backupCount=3, encoding="utf-8",
    )
    fh.setFormatter(fmt)
    root.addHandler(fh)

    mh = MemoryHandler()
    root.addHandler(mh)

    logging.getLogger("werkzeug").setLevel("WARNING")  # 요청 로그 줄임


def set_level(level: str) -> str:
    level = level.upper()
    if level not in LEVEL_RANK:
        raise ValueError(f"unknown level: {level}")
    logging.getLogger().setLevel(level)
    return level


def get_level() -> str:
    return logging.getLevelName(logging.getLogger().level)


def recent(level: str | None = None, limit: int = 200, contains: str | None = None) -> list[dict]:
    """최근 로그 조회. level= 'INFO' 면 INFO 이상, contains= 부분문자열 필터."""
    with _buf_lock:
        items = list(_BUFFER)
    if level:
        threshold = LEVEL_RANK.get(level.upper(), 0)
        items = [r for r in items if LEVEL_RANK.get(r["level"], 0) >= threshold]
    if contains:
        c = contains.lower()
        items = [r for r in items if c in r["message"].lower() or c in r["logger"].lower()]
    return items[-limit:]


def buffer_size() -> int:
    return len(_BUFFER)