Source code for robonix_api.scribe_logger

# SPDX-License-Identifier: MulanPSL-2.0
"""scribe_logger — Python-side unified logging facade.

Mirrors `system/scribe/src/lib.rs` semantics: single ``log()`` entry
point, lazy init, dual sink (stderr + per-tag JSON-lines file).

Usage::

    from robonix_api import scribe_logger

    scribe_logger.info("atlas", "control plane ready")
    scribe_logger.log(scribe_logger.Level.WARN, "my_driver", "sensor timeout")
"""

from __future__ import annotations

import json
import logging
import os
import sys
import threading
import time
from enum import Enum
from pathlib import Path
from typing import Dict, Optional, TextIO  # noqa: UP035


[docs] class Level(Enum): """Severity, ordered DEBUG < INFO < WARN < ERROR.""" DEBUG = "D" INFO = "I" WARN = "W" ERROR = "E"
# ── layout constants (match Rust format.rs) ───────────────────────── _TAG_WIDTH = 24 # ── global lazy state ──────────────────────────────────────────────── _lock: threading.Lock = threading.Lock() _log_dir: Optional[Path] = None _writers: Dict[str, TextIO] = {} # Whether to also mirror each record to stderr in console format. Suppressed # when an orchestrator (rbnx) provisioned $SCRIBE_LOG_DIR: rbnx already pipes # the process's stdout/stderr back into Scribe under the same tag, so a stderr # echo would be re-ingested and every line would appear twice in the per-tag # file. With SCRIBE_LOG_DIR set, the direct file write below is authoritative. # When unset (standalone dev, or a docker container whose in-container log dir # isn't collected) the echo stays on — there it is the only way the output # reaches a human or rbnx's pipe. _console_echo: bool = True def _ensure_init() -> None: """Lazily initialise log directory and internal state on first call.""" global _log_dir, _console_echo # noqa: PLW0603 if _log_dir is not None: return with _lock: if _log_dir is not None: return _console_echo = "SCRIBE_LOG_DIR" not in os.environ _log_dir = Path(os.environ.get("SCRIBE_LOG_DIR", "./logs")) _log_dir.mkdir(parents=True, exist_ok=True) def _decompose_ts(ts_ns: int) -> tuple[int, int, int, int, int, int]: """Nanosecond UNIX timestamp → (month, day, hour, min, sec, ms).""" secs = ts_ns // 1_000_000_000 nanos_rem = ts_ns % 1_000_000_000 ms = nanos_rem // 1_000_000 # localtime t = time.localtime(secs) return (t.tm_mon, t.tm_mday, t.tm_hour, t.tm_min, t.tm_sec, ms) def _format_console(ts_ns: int, level: Level, tag: str, msg: str) -> str: """Render a logcat-style console line (matches Rust format.rs).""" month, day, hour, minute, sec, ms = _decompose_ts(ts_ns) tag_display = tag[: _TAG_WIDTH - 1] + "…" if len(tag) > _TAG_WIDTH else tag.ljust(_TAG_WIDTH) return ( f"{month:02}-{day:02} {hour:02}:{minute:02}:{sec:02}.{ms:03}" f" {level.value} {tag_display} {msg}\n" ) def _ts_fmt(ts_ns: int) -> str: """Nanosecond UNIX timestamp → 'YYYY-MM-DD HH:MM:SS.nnnnnnnnn' (local time).""" secs, nsec = divmod(ts_ns, 1_000_000_000) t = time.localtime(secs) return ( f"{t.tm_year:04}-{t.tm_mon:02}-{t.tm_mday:02} " f"{t.tm_hour:02}:{t.tm_min:02}:{t.tm_sec:02}.{nsec:09}" ) def _format_json(ts_ns: int, level: Level, tag: str, msg: str) -> str: """Render a JSON line (matches Rust LogRecord serialisation).""" rec = { "ts": _ts_fmt(ts_ns), "level": level.name.lower(), "tag": tag, "msg": msg, } return json.dumps(rec, ensure_ascii=False) def _sanitize_tag(tag: str) -> str: """Replace non-safe characters to prevent path traversal.""" return "".join(c if c.isascii() and (c.isalnum() or c in "._-") else "_" for c in tag) def _get_writer(tag: str) -> Optional[TextIO]: """Return (creating if needed) the append-mode file handle for *tag*. Returns ``None`` if the log directory is unwritable or the file can't be opened — the caller treats file writes as best-effort. """ global _writers # noqa: PLW0603 if tag not in _writers: safe = _sanitize_tag(tag) path = _log_dir / f"{safe}.log" try: _writers[tag] = open(str(path), "a", encoding="utf-8") # noqa: SIM115 except OSError: _writers[tag] = None # type: ignore[assignment] return _writers[tag] # ── public API ───────────────────────────────────────────────────────
[docs] def log(level: Level, tag: str, msg: str) -> None: """Log one record — the **only** entry point. The first call transparently creates ``$SCRIBE_LOG_DIR`` (default ``./logs``). Errors are best-effort (stderr / disk full are silently ignored). """ _ensure_init() ts_ns = time.time_ns() # console → stderr (skipped when rbnx owns the per-tag file; see _console_echo) if _console_echo: try: console_line = _format_console(ts_ns, level, tag, msg) sys.stderr.write(console_line) sys.stderr.flush() except Exception: # noqa: BLE001 pass # file → $SCRIBE_LOG_DIR/{tag}.log try: json_line = _format_json(ts_ns, level, tag, msg) with _lock: writer = _get_writer(tag) if writer is not None: writer.write(json_line + "\n") writer.flush() except Exception: # noqa: BLE001 pass
[docs] def debug(tag: str, msg: str) -> None: """Convenience: :attr:`Level.DEBUG`.""" log(Level.DEBUG, tag, msg)
[docs] def info(tag: str, msg: str) -> None: """Convenience: :attr:`Level.INFO`.""" log(Level.INFO, tag, msg)
[docs] def warn(tag: str, msg: str) -> None: """Convenience: :attr:`Level.WARN`.""" log(Level.WARN, tag, msg)
[docs] def error(tag: str, msg: str) -> None: """Convenience: :attr:`Level.ERROR`.""" log(Level.ERROR, tag, msg)
# ── stdlib `logging` → Scribe bridge ───────────────────────────────── _STDLIB_LEVEL_MAP: Dict[int, Level] = { logging.DEBUG: Level.DEBUG, logging.INFO: Level.INFO, logging.WARNING: Level.WARN, logging.ERROR: Level.ERROR, logging.CRITICAL: Level.ERROR, } class _StdlibBridgeHandler(logging.Handler): """A stdlib ``logging.Handler`` that re-emits every record into Scribe. Carries a fixed component *tag* so any library's logging (the package's own, plus transitive deps that use stdlib ``logging``) lands in that component's ``rbnx logs -t <tag>`` stream. Best-effort: a failure to forward never propagates back into the caller's logging path. """ def __init__(self, tag: str) -> None: super().__init__() self._tag = tag def emit(self, record: logging.LogRecord) -> None: try: level = _STDLIB_LEVEL_MAP.get(record.levelno, Level.INFO) log(level, self._tag, record.getMessage()) except Exception: # noqa: BLE001 pass
[docs] def install_stdlib_bridge( tag: str, *, level: int = logging.INFO, logger: Optional[logging.Logger] = None, replace_existing_handlers: bool = True, ) -> logging.Logger: """Route stdlib ``logging`` through Scribe so packages need no own sink. Attaches a :class:`_StdlibBridgeHandler` (tagged *tag*) to *logger* (the root logger by default, so transitive-dependency logs are captured too) and raises its level to *level*. When *replace_existing_handlers* is true, every other handler already on that logger — ``StreamHandler`` (stdout/ stderr), ``FileHandler`` (a per-package log file), a ``rich`` handler from fastmcp/uvicorn, anything — is removed first, so the bridge is the *only* sink. This enforces the repo rule "no package owns a log file or duplicate stdout sink; everything goes through Scribe" (Scribe already mirrors to stderr itself, so console output is not lost). Idempotent and re-taggable: exactly one bridge handler ever exists on the logger, and the most recent call's *tag* wins (so the framework can re-tag with the provider id at bootstrap after a package set a provisional tag at import). Returns the configured logger. """ target = logger if logger is not None else logging.getLogger() target.setLevel(level) for handler in list(target.handlers): if isinstance(handler, _StdlibBridgeHandler): # Drop any prior bridge so the new tag replaces the old one. target.removeHandler(handler) elif replace_existing_handlers: target.removeHandler(handler) target.addHandler(_StdlibBridgeHandler(tag)) return target