Add system-level dynamic variables support in session management. Implement methods to generate and apply built-in variables for current session time, UTC time, and timezone. Update documentation to reflect new variables and enhance tests for dynamic variable handling in the UI components.
This commit is contained in:
@@ -5,6 +5,7 @@ import hashlib
|
||||
import json
|
||||
import re
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
from enum import Enum
|
||||
from typing import Optional, Dict, Any, List
|
||||
from loguru import logger
|
||||
@@ -32,6 +33,7 @@ _DYNAMIC_VARIABLE_KEY_RE = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]{0,63}$")
|
||||
_DYNAMIC_VARIABLE_PLACEHOLDER_RE = re.compile(r"\{\{\s*([a-zA-Z_][a-zA-Z0-9_]*)\s*\}\}")
|
||||
_DYNAMIC_VARIABLE_MAX_ITEMS = 30
|
||||
_DYNAMIC_VARIABLE_VALUE_MAX_CHARS = 1000
|
||||
_SYSTEM_DYNAMIC_VARIABLE_KEYS = {"system__time", "system_utc", "system_timezone"}
|
||||
|
||||
|
||||
class WsSessionState(str, Enum):
|
||||
@@ -845,6 +847,16 @@ class Session:
|
||||
rendered = _DYNAMIC_VARIABLE_PLACEHOLDER_RE.sub(_replace, str(template or ""))
|
||||
return rendered, sorted(missing)
|
||||
|
||||
def _system_dynamic_variables(self) -> Dict[str, str]:
|
||||
"""Build system-level dynamic variables for the current session timestamp."""
|
||||
local_now = datetime.now().astimezone()
|
||||
utc_now = local_now.astimezone(timezone.utc)
|
||||
return {
|
||||
"system__time": local_now.strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"system_utc": utc_now.strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"system_timezone": str(local_now.tzinfo or ""),
|
||||
}
|
||||
|
||||
def _apply_dynamic_variables(
|
||||
self,
|
||||
metadata: Dict[str, Any],
|
||||
@@ -859,7 +871,7 @@ class Session:
|
||||
"""
|
||||
merged = dict(metadata or {})
|
||||
raw_dynamic_vars = client_metadata.get("dynamicVariables")
|
||||
dynamic_vars: Dict[str, str] = {}
|
||||
dynamic_vars: Dict[str, str] = self._system_dynamic_variables()
|
||||
|
||||
if raw_dynamic_vars is not None:
|
||||
if not isinstance(raw_dynamic_vars, dict):
|
||||
@@ -888,6 +900,9 @@ class Session:
|
||||
f"'{raw_key}'. Expected ^[a-zA-Z_][a-zA-Z0-9_]{{0,63}}$"
|
||||
),
|
||||
}
|
||||
if key in _SYSTEM_DYNAMIC_VARIABLE_KEYS:
|
||||
# Reserved system variables are generated by server time context.
|
||||
continue
|
||||
if key in dynamic_vars:
|
||||
return merged, {
|
||||
"code": "protocol.dynamic_variables_invalid",
|
||||
@@ -912,12 +927,6 @@ class Session:
|
||||
template_keys.update(self._extract_dynamic_template_keys(merged.get("greeting")))
|
||||
if not template_keys:
|
||||
return merged, None
|
||||
if raw_dynamic_vars is None:
|
||||
missing = ", ".join(sorted(template_keys))
|
||||
return merged, {
|
||||
"code": "protocol.dynamic_variables_missing",
|
||||
"message": f"Missing dynamic variables for placeholders: {missing}",
|
||||
}
|
||||
|
||||
missing_keys = sorted([key for key in template_keys if key not in dynamic_vars])
|
||||
if missing_keys:
|
||||
|
||||
Reference in New Issue
Block a user