Coverage for integrations / channels / response / templates.py: 46.7%

1""" 

2TemplateEngine - Response formatting with variable substitution. 

3 

4Provides template rendering with support for common variables like 

5{model}, {provider}, {identity.name}, {user.name}, {channel}, {timestamp}. 

6""" 

7 

8import re 

9import time 

10from datetime import datetime 

11from typing import Optional, Dict, Any, Callable, List 

12from dataclasses import dataclass, field 

13import logging 

14 

15logger = logging.getLogger(__name__) 

16 

17 

18@dataclass 

19class Identity: 

20 """Bot identity information.""" 

21 name: str = "Assistant" 

22 description: str = "" 

23 avatar_url: str = "" 

24 metadata: Dict[str, Any] = field(default_factory=dict) 

25 

26 

27@dataclass 

28class User: 

29 """User information.""" 

30 name: str = "User" 

31 id: str = "" 

32 display_name: str = "" 

33 metadata: Dict[str, Any] = field(default_factory=dict) 

34 

35 

36@dataclass 

37class TemplateContext: 

38 """Context for template rendering.""" 

39 model: str = "" 

40 provider: str = "" 

41 identity: Identity = field(default_factory=Identity) 

42 user: User = field(default_factory=User) 

43 channel: str = "" 

44 timestamp: Optional[datetime] = None 

45 custom_vars: Dict[str, Any] = field(default_factory=dict) 

46 

47 def to_dict(self) -> Dict[str, Any]: 

48 """Convert context to flat dictionary for variable substitution.""" 

49 ts = self.timestamp or datetime.now() 

50 

51 return { 

52 "model": self.model, 

53 "provider": self.provider, 

54 "identity.name": self.identity.name, 

55 "identity.description": self.identity.description, 

56 "identity.avatar_url": self.identity.avatar_url, 

57 "user.name": self.user.name or self.user.display_name, 

58 "user.id": self.user.id, 

59 "user.display_name": self.user.display_name or self.user.name, 

60 "channel": self.channel, 

61 "timestamp": ts.strftime("%Y-%m-%d %H:%M:%S"), 

62 "timestamp.date": ts.strftime("%Y-%m-%d"), 

63 "timestamp.time": ts.strftime("%H:%M:%S"), 

64 "timestamp.iso": ts.isoformat(), 

65 "timestamp.unix": str(int(ts.timestamp())), 

66 **self.custom_vars 

67 } 

68 

69 

70@dataclass 

71class TemplateConfig: 

72 """Configuration for the template engine.""" 

73 prefix: str = "" 

74 suffix: str = "" 

75 default_format: str = "{content}" 

76 escape_html: bool = False 

77 strict_mode: bool = False # Raise error on missing variables 

78 missing_var_placeholder: str = "" # What to show for missing vars 

79 

80 

81class TemplateEngine: 

82 """ 

83 Template engine for response formatting. 

84 

85 Supports variable substitution using {variable} syntax, 

86 with support for nested variables like {identity.name}. 

87 """ 

88 

89 # Pattern to match {variable} or {variable.subvar} 

90 VAR_PATTERN = re.compile(r'\{([a-zA-Z_][a-zA-Z0-9_]*(?:\.[a-zA-Z_][a-zA-Z0-9_]*)*)\}') 

91 

92 def __init__(self, config: Optional[TemplateConfig] = None): 

93 """ 

94 Initialize the TemplateEngine. 

95 

96 Args: 

97 config: Optional configuration for template behavior. 

98 """ 

99 self._config = config or TemplateConfig() 

100 self._context = TemplateContext() 

101 self._custom_filters: Dict[str, Callable[[str], str]] = {} 

102 self._templates: Dict[str, str] = {} 

103 

104 @property 

105 def config(self) -> TemplateConfig: 

106 """Get the template configuration.""" 

107 return self._config 

108 

109 @property 

110 def context(self) -> TemplateContext: 

111 """Get the current template context.""" 

112 return self._context 

113 

114 def set_context(self, context: TemplateContext) -> None: 

115 """Set the template context.""" 

116 self._context = context 

117 

118 def update_context(self, **kwargs) -> None: 

119 """Update specific context fields.""" 

120 for key, value in kwargs.items(): 

121 if hasattr(self._context, key): 

122 setattr(self._context, key, value) 

123 else: 

124 self._context.custom_vars[key] = value 

125 

126 def set_model(self, model: str) -> None: 

127 """Set the model name.""" 

128 self._context.model = model 

129 

130 def set_provider(self, provider: str) -> None: 

131 """Set the provider name.""" 

132 self._context.provider = provider 

133 

134 def set_identity(self, identity: Identity) -> None: 

135 """Set the bot identity.""" 

136 self._context.identity = identity 

137 

138 def set_user(self, user: User) -> None: 

139 """Set the user information.""" 

140 self._context.user = user 

141 

142 def set_channel(self, channel: str) -> None: 

143 """Set the channel name.""" 

144 self._context.channel = channel 

145 

146 def set_prefix(self, prefix: str) -> None: 

147 """Set the response prefix.""" 

148 self._config.prefix = prefix 

149 

150 def set_suffix(self, suffix: str) -> None: 

151 """Set the response suffix.""" 

152 self._config.suffix = suffix 

153 

154 def set_variable(self, name: str, value: Any) -> None: 

155 """Set a custom variable.""" 

156 self._context.custom_vars[name] = value 

157 

158 def get_variable(self, name: str) -> Optional[Any]: 

159 """Get a variable value.""" 

160 context_dict = self._context.to_dict() 

161 return context_dict.get(name, self._context.custom_vars.get(name)) 

162 

163 def register_template(self, name: str, template: str) -> None: 

164 """Register a named template for later use.""" 

165 self._templates[name] = template 

166 

167 def get_template(self, name: str) -> Optional[str]: 

168 """Get a registered template by name.""" 

169 return self._templates.get(name) 

170 

171 def list_templates(self) -> List[str]: 

172 """List all registered template names.""" 

173 return list(self._templates.keys()) 

174 

175 def register_filter(self, name: str, func: Callable[[str], str]) -> None: 

176 """ 

177 Register a custom filter function. 

178 

179 Filters can be applied using {variable|filter} syntax. 

180 """ 

181 self._custom_filters[name] = func 

182 

183 def render(self, template: str, context: Optional[TemplateContext] = None, 

184 extra_vars: Optional[Dict[str, Any]] = None) -> str: 

185 """ 

186 Render a template with variable substitution. 

187 

188 Args: 

189 template: The template string with {variable} placeholders. 

190 context: Optional context to use (defaults to stored context). 

191 extra_vars: Additional variables to include. 

192 

193 Returns: 

194 Rendered template string. 

195 

196 Raises: 

197 ValueError: If strict_mode is True and a variable is missing. 

198 """ 

199 ctx = context or self._context 

200 var_dict = ctx.to_dict() 

201 

202 if extra_vars: 

203 var_dict.update(extra_vars) 

204 

205 def replace_var(match): 

206 var_name = match.group(1) 

207 

208 # Check for filter syntax: {var|filter} 

209 if '|' in var_name: 

210 var_name, filter_name = var_name.split('|', 1) 

211 value = var_dict.get(var_name) 

212 if value is not None and filter_name in self._custom_filters: 

213 return self._custom_filters[filter_name](str(value)) 

214 

215 value = var_dict.get(var_name) 

216 

217 if value is None: 

218 if self._config.strict_mode: 

219 raise ValueError(f"Missing template variable: {var_name}") 

220 return self._config.missing_var_placeholder or match.group(0) 

221 

222 result = str(value) 

223 if self._config.escape_html: 

224 result = self._escape_html(result) 

225 return result 

226 

227 return self.VAR_PATTERN.sub(replace_var, template) 

228 

229 def render_named(self, template_name: str, context: Optional[TemplateContext] = None, 

230 extra_vars: Optional[Dict[str, Any]] = None) -> str: 

231 """ 

232 Render a registered template by name. 

233 

234 Args: 

235 template_name: Name of the registered template. 

236 context: Optional context to use. 

237 extra_vars: Additional variables to include. 

238 

239 Returns: 

240 Rendered template string. 

241 

242 Raises: 

243 KeyError: If template name is not found. 

244 """ 

245 template = self._templates.get(template_name) 

246 if template is None: 

247 raise KeyError(f"Template not found: {template_name}") 

248 return self.render(template, context, extra_vars) 

249 

250 def format_response(self, content: str, context: Optional[TemplateContext] = None, 

251 extra_vars: Optional[Dict[str, Any]] = None) -> str: 

252 """ 

253 Format a response with prefix, suffix, and variable substitution. 

254 

255 Args: 

256 content: The main response content. 

257 context: Optional context to use. 

258 extra_vars: Additional variables to include. 

259 

260 Returns: 

261 Formatted response string. 

262 """ 

263 ctx = context or self._context 

264 vars_dict = {"content": content} 

265 if extra_vars: 

266 vars_dict.update(extra_vars) 

267 

268 # Render prefix 

269 prefix = "" 

270 if self._config.prefix: 

271 prefix = self.render(self._config.prefix, ctx, vars_dict) 

272 

273 # Render suffix 

274 suffix = "" 

275 if self._config.suffix: 

276 suffix = self.render(self._config.suffix, ctx, vars_dict) 

277 

278 # Render content with format template 

279 formatted = self.render(self._config.default_format, ctx, vars_dict) 

280 

281 return f"{prefix}{formatted}{suffix}" 

282 

283 def _escape_html(self, text: str) -> str: 

284 """Escape HTML special characters.""" 

285 replacements = { 

286 '&': '&', 

287 '<': '&lt;', 

288 '>': '&gt;', 

289 '"': '&quot;', 

290 "'": '&#39;' 

291 } 

292 for char, escaped in replacements.items(): 

293 text = text.replace(char, escaped) 

294 return text 

295 

296 def get_available_variables(self) -> List[str]: 

297 """Get list of available variable names.""" 

298 return list(self._context.to_dict().keys()) 

299 

300 def validate_template(self, template: str) -> Dict[str, Any]: 

301 """ 

302 Validate a template string. 

303 

304 Returns: 

305 Dictionary with validation results. 

306 """ 

307 variables = self.VAR_PATTERN.findall(template) 

308 available = set(self._context.to_dict().keys()) 

309 

310 missing = [v for v in variables if v not in available and v not in self._context.custom_vars] 

311 

312 return { 

313 "valid": len(missing) == 0, 

314 "variables_used": variables, 

315 "missing_variables": missing, 

316 "available_variables": list(available) 

317 } 

318 

319 def create_context( 

320 self, 

321 model: str = "", 

322 provider: str = "", 

323 identity_name: str = "Assistant", 

324 user_name: str = "User", 

325 channel: str = "", 

326 **custom_vars 

327 ) -> TemplateContext: 

328 """ 

329 Create a new template context with common values. 

330 

331 Args: 

332 model: Model name. 

333 provider: Provider name. 

334 identity_name: Bot identity name. 

335 user_name: User name. 

336 channel: Channel name. 

337 **custom_vars: Additional custom variables. 

338 

339 Returns: 

340 New TemplateContext instance. 

341 """ 

342 return TemplateContext( 

343 model=model, 

344 provider=provider, 

345 identity=Identity(name=identity_name), 

346 user=User(name=user_name), 

347 channel=channel, 

348 timestamp=datetime.now(), 

349 custom_vars=custom_vars 

350 ) 

351 

352 def clone(self) -> 'TemplateEngine': 

353 """Create a copy of this template engine.""" 

354 new_engine = TemplateEngine(TemplateConfig( 

355 prefix=self._config.prefix, 

356 suffix=self._config.suffix, 

357 default_format=self._config.default_format, 

358 escape_html=self._config.escape_html, 

359 strict_mode=self._config.strict_mode, 

360 missing_var_placeholder=self._config.missing_var_placeholder 

361 )) 

362 new_engine._context = TemplateContext( 

363 model=self._context.model, 

364 provider=self._context.provider, 

365 identity=Identity( 

366 name=self._context.identity.name, 

367 description=self._context.identity.description, 

368 avatar_url=self._context.identity.avatar_url 

369 ), 

370 user=User( 

371 name=self._context.user.name, 

372 id=self._context.user.id, 

373 display_name=self._context.user.display_name 

374 ), 

375 channel=self._context.channel, 

376 timestamp=self._context.timestamp, 

377 custom_vars=dict(self._context.custom_vars) 

378 ) 

379 new_engine._custom_filters = dict(self._custom_filters) 

380 new_engine._templates = dict(self._templates) 

381 return new_engine 

382 

383 # Built-in filters 

384 @staticmethod 

385 def filter_upper(value: str) -> str: 

386 """Convert to uppercase.""" 

387 return value.upper() 

388 

389 @staticmethod 

390 def filter_lower(value: str) -> str: 

391 """Convert to lowercase.""" 

392 return value.lower() 

393 

394 @staticmethod 

395 def filter_title(value: str) -> str: 

396 """Convert to title case.""" 

397 return value.title() 

398 

399 @staticmethod 

400 def filter_strip(value: str) -> str: 

401 """Strip whitespace.""" 

402 return value.strip() 

403 

404 def register_default_filters(self) -> None: 

405 """Register common built-in filters.""" 

406 self.register_filter("upper", self.filter_upper) 

407 self.register_filter("lower", self.filter_lower) 

408 self.register_filter("title", self.filter_title) 

409 self.register_filter("strip", self.filter_strip) 

410 

411 def get_stats(self) -> dict: 

412 """Get statistics about the template engine.""" 

413 return { 

414 "registered_templates": len(self._templates), 

415 "custom_filters": len(self._custom_filters), 

416 "custom_variables": len(self._context.custom_vars), 

417 "prefix_set": bool(self._config.prefix), 

418 "suffix_set": bool(self._config.suffix) 

419 }