Spaces:
Running
Running
File size: 4,730 Bytes
d660fa9 f4542ce d660fa9 f4542ce d660fa9 98f0a17 f4542ce 98f0a17 f4542ce 98f0a17 f4542ce 98f0a17 f4542ce | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 | """log — centralized logging setup for the whole project.
Every module imports get_logger(__name__) here so format, destinations,
and level stay consistent across the engine, planner, and server.
Architecture: a dependency of every backend module; no business logic.
Design: logs land in backend/output/logs with timestamps for post-hoc
analysis of long runs. Console output is off by default — the admin log
terminal (src/core/log_relay.py) mirrors every line to the live log file
and the admin-only /api/logs endpoints instead.
"""
from __future__ import annotations
import logging
import logging.config
from datetime import datetime
from pathlib import Path
from typing import Optional
# Path config lives in one place: src/config.py
from src.config import LOG_DIR, LOG_LEVEL as _DEFAULT_LEVEL_NAME
DEFAULT_LEVEL = getattr(logging, _DEFAULT_LEVEL_NAME.upper(), logging.INFO)
LOG_FORMAT = "%(asctime)s | %(levelname)-8s | %(name)s | %(message)s"
DATE_FORMAT = "%Y-%m-%d %H:%M:%S"
_configured = False
def _build_log_path(run_id: Optional[str] = None) -> Path:
LOG_DIR.mkdir(parents=True, exist_ok=True)
stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"{stamp}_{run_id}.log" if run_id else f"{stamp}.log"
return LOG_DIR / filename
def setup_logging(
level: int = DEFAULT_LEVEL,
run_id: Optional[str] = None,
console: bool = False,
file: bool = True,
max_bytes: int = 5 * 1024 * 1024,
backup_count: int = 5,
) -> Optional[Path]:
"""
Configures the ROOT logger once. Idempotent -- a second call is a no-op,
so accidentally calling it from more than one entrypoint (e.g. both a
test script and an imported module) will never duplicate handlers or
duplicate every log line.
level -- root log level (module-specific loggers can still be
set louder/quieter individually if you ever need that)
run_id -- optional tag folded into the log filename, e.g. a
persona name or simulation id, so a run's logs are
easy to find in output/logs/
console -- also stream logs to stdout (default off; watch the
admin log terminal instead)
file -- also write logs to output/logs/<timestamp>[_<run_id>].log
max_bytes /
backup_count -- rotation settings for the file handler
Returns the Path of the log file created, or None if file=False.
"""
global _configured
if _configured:
return None
handlers: dict = {}
log_path: Optional[Path] = None
if console:
handlers["console"] = {
"class": "logging.StreamHandler",
"formatter": "standard",
"stream": "ext://sys.stdout",
"level": level,
}
if file:
log_path = _build_log_path(run_id)
handlers["file"] = {
"class": "logging.handlers.RotatingFileHandler",
"formatter": "standard",
"filename": str(log_path),
"maxBytes": max_bytes,
"backupCount": backup_count,
"encoding": "utf-8",
"level": level,
}
config = {
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"standard": {
"format": LOG_FORMAT,
"datefmt": DATE_FORMAT,
},
},
"handlers": handlers,
"root": {
"level": level,
"handlers": list(handlers.keys()),
},
# Quiet down noisy third-party loggers so your own log lines don't
# get buried -- add to this as you bring in more SDKs.
"loggers": {
"google_genai": {"level": "WARNING", "propagate": True},
"httpx": {"level": "WARNING", "propagate": True},
"httpcore": {"level": "WARNING", "propagate": True},
},
}
logging.config.dictConfig(config)
_configured = True
# Mirror every emitted line into the admin log terminal's buffer and the
# live log file, regardless of the console/file toggles above.
try:
from src.core.log_relay import install_relay
install_relay()
except Exception:
pass
if log_path:
logging.getLogger(__name__).info("[log] logging initialized -> %s", log_path)
return log_path
def get_logger(name: str) -> logging.Logger:
"""
Thin wrapper so call sites read `from src.core.log import get_logger`
instead of raw `logging.getLogger`. Lazily calls setup_logging() with
defaults if nothing has configured logging yet.
"""
if not _configured:
setup_logging()
return logging.getLogger(name) |