server/utils.py

import logging
import logging.handlers
import os
import yaml
from pathlib import Path

try:
    import psutil as _psutil
    _PSUTIL = True
except ImportError:
    _PSUTIL = False

_logging_configured = False


def configure_logging() -> None:
    """Configure root logger once: console + rotating file. Call from main.py only."""
    global _logging_configured
    if _logging_configured:
        return

    log_dir = Path("logs")
    log_dir.mkdir(exist_ok=True)

    fmt = logging.Formatter(
        "%(asctime)s  %(levelname)-8s  %(name)-30s  %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )

    console = logging.StreamHandler()
    console.setFormatter(fmt)

    file_handler = logging.handlers.RotatingFileHandler(
        log_dir / "prism.log",
        maxBytes=5 * 1024 * 1024,  # 5 MB
        backupCount=3,
        encoding="utf-8",
    )
    file_handler.setFormatter(fmt)

    level = getattr(logging, os.getenv("LOG_LEVEL", "INFO").upper(), logging.INFO)
    root = logging.getLogger()
    root.setLevel(level)
    root.addHandler(console)
    root.addHandler(file_handler)

    # Quiet noisy third-party loggers
    for noisy in (
        "httpx", "httpcore", "chromadb", "urllib3", "multipart",
        "huggingface_hub", "sentence_transformers", "transformers",
    ):
        logging.getLogger(noisy).setLevel(logging.WARNING)

    _logging_configured = True


def setup_logger(name: str) -> logging.Logger:
    """Return named logger. configure_logging() must be called before first use."""
    return logging.getLogger(name)


def load_config(config_path: str = "config.yaml") -> dict:
    """Load config.yaml and return as dict."""
    with open(config_path, "r") as f:
        return yaml.safe_load(f)


def count_tokens(text: str) -> int:
    """Approximate token count: len(text.split()) * 1.3"""
    return int(len(text.split()) * 1.3)


def log_memory_mb(logger: logging.Logger, label: str) -> float:
    """Log current process RSS in MB. Returns MB (0 if psutil unavailable)."""
    if not _PSUTIL:
        return 0.0
    rss = _psutil.Process().memory_info().rss / 1024 / 1024
    logger.info("MEM [%s] RSS=%.1fMB", label, rss)
    return rss