File size: 976 Bytes
0fcfe1c
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""Lightweight telemetry helper for BLUX-cA."""

from __future__ import annotations

import json
import os
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict

TELEMETRY_ENV = "BLUX_CA_TELEMETRY"
DEFAULT_TELEMETRY_PATH = Path.home() / ".config" / "blux-ca" / "telemetry.jsonl"


def _is_enabled() -> bool:
    return os.environ.get(TELEMETRY_ENV, "on").lower() not in {"0", "off", "false"}


def emit(event: str, payload: Dict[str, Any] | None = None) -> None:
    """Record a telemetry event when telemetry is enabled."""

    if not _is_enabled():
        return

    record = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "event": event,
        "payload": payload or {},
    }

    path = DEFAULT_TELEMETRY_PATH
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("a", encoding="utf-8") as handle:
        handle.write(json.dumps(record, ensure_ascii=False) + "\n")


__all__ = ["emit"]