57 lines
2.2 KiB
Python
57 lines
2.2 KiB
Python
import json
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
DEFAULT_BASE_DIR = "~/.timmy/training-data/evennia"
|
|
|
|
|
|
def telemetry_dir(base_dir: str | Path = DEFAULT_BASE_DIR) -> Path:
|
|
return Path(base_dir).expanduser()
|
|
|
|
|
|
def _session_day() -> str:
|
|
return datetime.now(timezone.utc).strftime("%Y%m%d")
|
|
|
|
|
|
def event_log_path(session_id: str, base_dir: str | Path = DEFAULT_BASE_DIR) -> Path:
|
|
session_id = (session_id or "unbound").strip() or "unbound"
|
|
return telemetry_dir(base_dir) / _session_day() / f"{session_id}.jsonl"
|
|
|
|
|
|
def session_meta_path(session_id: str, base_dir: str | Path = DEFAULT_BASE_DIR) -> Path:
|
|
session_id = (session_id or "unbound").strip() or "unbound"
|
|
return telemetry_dir(base_dir) / _session_day() / f"{session_id}.meta.json"
|
|
|
|
|
|
def write_session_metadata(session_id: str, metadata: dict, base_dir: str | Path = DEFAULT_BASE_DIR) -> Path:
|
|
path = session_meta_path(session_id, base_dir)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
payload = {}
|
|
if path.exists():
|
|
try:
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
payload = {}
|
|
payload.update(metadata or {})
|
|
payload.setdefault("session_id", session_id)
|
|
payload.setdefault("event_log_path", str(event_log_path(session_id, base_dir)))
|
|
payload.setdefault("updated_at", datetime.now(timezone.utc).isoformat())
|
|
path.write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def append_event(session_id: str, event: dict, base_dir: str | Path = DEFAULT_BASE_DIR) -> Path:
|
|
path = event_log_path(session_id, base_dir)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
payload = dict(event)
|
|
payload.setdefault("timestamp", datetime.now(timezone.utc).isoformat())
|
|
with path.open("a", encoding="utf-8") as f:
|
|
f.write(json.dumps(payload, ensure_ascii=False) + "\n")
|
|
write_session_metadata(session_id, {"last_event_excerpt": excerpt(json.dumps(payload, ensure_ascii=False), 400)}, base_dir)
|
|
return path
|
|
|
|
|
|
def excerpt(text: str, limit: int = 240) -> str:
|
|
text = " ".join((text or "").split())
|
|
return text if len(text) <= limit else text[: limit - 3] + "..."
|