Spaces:
Running
Running
| """log_relay — mirror of every project log line for the admin log terminal. | |
| Attaches a second handler to the ROOT logger: each emitted record is | |
| formatted with the project's standard format and appended to a bounded | |
| in-memory ring buffer (polled by the admin-only /api/logs endpoints) and | |
| to a live file backend/output/logs/live.log, so the same output also | |
| survives restarts and stays viewable without the Space console. | |
| Architecture: installed once by src/core/log.py setup_logging(); consumed | |
| by Odin.py's admin-gated endpoints and the frontend LogTerminal panel. | |
| Design: stdlib-only, a one-way mirror — never changes existing handlers, | |
| levels, or the file rotation policy. | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| import threading | |
| from collections import deque | |
| from itertools import count | |
| from src.config import LOG_DIR | |
| from src.core.log import DATE_FORMAT, LOG_FORMAT | |
| MAX_LINES = 2000 | |
| LIVE_LOG_PATH = LOG_DIR / "live.log" | |
| _seq = count(1) | |
| _buffer: deque = deque(maxlen=MAX_LINES) | |
| _latest_seq = 0 | |
| _lock = threading.Lock() | |
| _installed = False | |
| class LogRelayHandler(logging.Handler): | |
| """Appends each formatted record to the ring buffer and the live file.""" | |
| def __init__(self, level: int = logging.NOTSET) -> None: | |
| super().__init__(level) | |
| self._formatter = logging.Formatter(LOG_FORMAT, DATE_FORMAT) | |
| def emit(self, record: logging.LogRecord) -> None: | |
| global _latest_seq | |
| try: | |
| text = self._formatter.format(record) | |
| except Exception: | |
| text = f"{record.name} | {record.getMessage()}" | |
| with _lock: | |
| _latest_seq = next(_seq) | |
| _buffer.append({"seq": _latest_seq, "level": record.levelname, "text": text}) | |
| try: | |
| LOG_DIR.mkdir(parents=True, exist_ok=True) | |
| with open(LIVE_LOG_PATH, "a", encoding="utf-8") as fh: | |
| fh.write(text + "\n") | |
| except OSError: | |
| pass | |
| def install_relay() -> None: | |
| """Attach the relay handler to the root logger exactly once.""" | |
| global _installed | |
| if _installed: | |
| return | |
| logging.getLogger().addHandler(LogRelayHandler()) | |
| _installed = True | |
| def relay_lines(since: int = 0) -> dict: | |
| """Return relayed entries with seq > since, plus the latest seq as cursor. | |
| The cursor tracks the last emitted sequence, so polling stays | |
| incremental even after the buffer is cleared. | |
| """ | |
| with _lock: | |
| lines = [entry for entry in _buffer if entry["seq"] > since] | |
| return {"lines": lines, "next": _latest_seq} | |
| def clear_relay() -> None: | |
| """Drop the in-memory buffer (the live file is intentionally kept).""" | |
| with _lock: | |
| _buffer.clear() |