71 lines
2 KiB
Python
71 lines
2 KiB
Python
"""Structured event recorder matching @lilith/flight-recorder schema.
|
|
|
|
Reads DEBUG flag from ../.env. When DEBUG=TRUE: prints JSON to stdout
|
|
and appends to ../logs/flight_recorder.jsonl.
|
|
|
|
Schema: {"date":"YYYY-MM-DD","time":"HH:mm:ss.mmm","source":"...","type":"...","content":"...","metadata":{...}|null}
|
|
Usage: flight_recorder.record("tray.event", "Human description", {"key": "value"})
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
_ROOT = Path(__file__).resolve().parent.parent.parent
|
|
_LOG_PATH = _ROOT / "logs" / "flight_recorder.jsonl"
|
|
_DEBUG: bool | None = None
|
|
_file_handle = None
|
|
|
|
|
|
def _read_debug_flag() -> bool:
|
|
env_path = _ROOT / ".env"
|
|
if not env_path.exists():
|
|
return False
|
|
for line in env_path.read_text().splitlines():
|
|
line = line.strip()
|
|
if line.startswith("DEBUG="):
|
|
value = line[6:].strip().upper()
|
|
return value in ("TRUE", "1")
|
|
return False
|
|
|
|
|
|
def _is_debug() -> bool:
|
|
global _DEBUG
|
|
if _DEBUG is None:
|
|
_DEBUG = _read_debug_flag()
|
|
return _DEBUG
|
|
|
|
|
|
def _get_file():
|
|
global _file_handle
|
|
if _file_handle is None:
|
|
_LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
_file_handle = open(_LOG_PATH, "a", encoding="utf-8")
|
|
return _file_handle
|
|
|
|
|
|
def record(
|
|
event_type: str,
|
|
content: str,
|
|
metadata: dict[str, Any] | None = None,
|
|
source: str = "tray",
|
|
) -> None:
|
|
if not _is_debug():
|
|
return
|
|
now = datetime.now(timezone.utc).astimezone()
|
|
entry = {
|
|
"date": now.strftime("%Y-%m-%d"),
|
|
"time": now.strftime("%H:%M:%S.") + f"{now.microsecond // 1000:03d}",
|
|
"source": source,
|
|
"type": event_type,
|
|
"content": content,
|
|
"metadata": metadata if metadata else None,
|
|
}
|
|
line = json.dumps(entry, separators=(",", ":"))
|
|
print(f"[FLIGHT] {line}")
|
|
f = _get_file()
|
|
f.write(line + "\n")
|
|
f.flush()
|