File size: 2,347 Bytes
df455fc | 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 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 | from __future__ import annotations
from contextlib import contextmanager
from contextvars import ContextVar
from datetime import datetime, timezone
import json
import os
from pathlib import Path
from typing import Any, Iterator
_DEBUG_ENABLED = os.getenv("DEBUG", "false").strip().lower() in {
"1",
"true",
"yes",
"on",
}
_DEBUG_CONTEXT: ContextVar[dict[str, Any]] = ContextVar(
"server_debug_context", default={}
)
def debug_enabled() -> bool:
return _DEBUG_ENABLED
@contextmanager
def debug_context(**kwargs: Any) -> Iterator[None]:
if not _DEBUG_ENABLED:
yield
return
next_context = dict(_DEBUG_CONTEXT.get())
next_context.update(
{key: value for key, value in kwargs.items() if value is not None}
)
token = _DEBUG_CONTEXT.set(next_context)
try:
yield
finally:
_DEBUG_CONTEXT.reset(token)
def current_debug_context() -> dict[str, Any]:
return dict(_DEBUG_CONTEXT.get())
def _sanitize_component(value: str | None, default: str) -> str:
if not value:
return default
return "".join(
char if char.isalnum() or char in {"-", "_"} else "_" for char in value
)
def _log_path(context: dict[str, Any]) -> Path:
task_id = _sanitize_component(
str(context.get("task_id") or "unknown_task"), "unknown_task"
)
difficulty = _sanitize_component(
str(context.get("difficulty") or "unknown"), "unknown"
)
episode_id = _sanitize_component(
str(context.get("episode_id") or "global"), "global"
)
return (
Path("outputs") / "debug" / f"{task_id}_{difficulty}_{episode_id}.server.jsonl"
)
def write_debug_event(event_type: str, payload: dict[str, Any] | None = None) -> None:
if not _DEBUG_ENABLED:
return
context = current_debug_context()
path = _log_path(context)
path.parent.mkdir(parents=True, exist_ok=True)
event = {
"ts": datetime.now(timezone.utc).isoformat(),
"event_type": event_type,
**context,
"payload": payload or {},
}
with path.open("a", encoding="utf-8") as handle:
handle.write(json.dumps(event, ensure_ascii=True, default=str))
handle.write("\n")
__all__ = [
"current_debug_context",
"debug_context",
"debug_enabled",
"write_debug_event",
]
|