83 lines
2.7 KiB
Python
83 lines
2.7 KiB
Python
"""Per-simulation capture of the same logs shown in the backend terminal."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import threading
|
|
from contextvars import ContextVar
|
|
from pathlib import Path
|
|
|
|
|
|
# Context variables are inherited by asyncio tasks, including the concurrent
|
|
# LLM calls created during a simulation. This keeps concurrent simulations from
|
|
# writing each other's records to the same file.
|
|
CURRENT_SIMULATION_ID: ContextVar[str | None] = ContextVar(
|
|
"current_simulation_id", default=None
|
|
)
|
|
|
|
PROJECT_LOGS_DIR = Path(__file__).resolve().parents[3] / "logs"
|
|
|
|
|
|
class ChineseSimulationFormatter(logging.Formatter):
|
|
"""Readable Chinese metadata while preserving the original message."""
|
|
|
|
LEVELS = {
|
|
"DEBUG": "调试",
|
|
"INFO": "信息",
|
|
"WARNING": "警告",
|
|
"ERROR": "错误",
|
|
"CRITICAL": "严重错误",
|
|
}
|
|
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
timestamp = f"{self.formatTime(record, '%Y-%m-%d %H:%M:%S')},{int(record.msecs):03d}"
|
|
level = self.LEVELS.get(record.levelname, record.levelname)
|
|
line = f"{timestamp} [模拟日志] [{level}] [{record.name}] {record.getMessage()}"
|
|
if record.exc_info:
|
|
line += "\n" + self.formatException(record.exc_info)
|
|
return line
|
|
|
|
|
|
class SimulationLogHandler(logging.Handler):
|
|
"""Write terminal-formatted records to ``logs/<simulation_id>.log``."""
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__(level=logging.INFO)
|
|
self._write_lock = threading.Lock()
|
|
|
|
def emit(self, record: logging.LogRecord) -> None:
|
|
simulation_id = CURRENT_SIMULATION_ID.get()
|
|
if not simulation_id:
|
|
return
|
|
|
|
try:
|
|
PROJECT_LOGS_DIR.mkdir(parents=True, exist_ok=True)
|
|
path = PROJECT_LOGS_DIR / f"{simulation_id}.log"
|
|
line = self.format(record)
|
|
with self._write_lock:
|
|
with path.open("a", encoding="utf-8") as handle:
|
|
handle.write(line + "\n")
|
|
except Exception:
|
|
# Logging must never break a simulation if the disk is unavailable.
|
|
self.handleError(record)
|
|
|
|
|
|
def install_simulation_log_handler() -> None:
|
|
"""Install the capture handler once on the process root logger."""
|
|
root = logging.getLogger()
|
|
if any(isinstance(handler, SimulationLogHandler) for handler in root.handlers):
|
|
return
|
|
|
|
handler = SimulationLogHandler()
|
|
handler.setFormatter(ChineseSimulationFormatter())
|
|
root.addHandler(handler)
|
|
|
|
|
|
def bind_simulation(simulation_id: str):
|
|
"""Bind a simulation ID to the current asyncio task."""
|
|
return CURRENT_SIMULATION_ID.set(simulation_id)
|
|
|
|
|
|
def unbind_simulation(token) -> None:
|
|
CURRENT_SIMULATION_ID.reset(token)
|