""" Template Engine for notification templates """ import logging import re from typing import Dict, Any, List, Optional from datetime import datetime import uuid from models import NotificationTemplate, NotificationChannel, NotificationCategory logger = logging.getLogger(__name__) class TemplateEngine: """Manages and renders notification templates""" def __init__(self): self.templates = {} # In-memory storage for demo self._load_default_templates() async def load_templates(self): """Load templates from storage""" # In production, would load from database logger.info(f"Loaded {len(self.templates)} templates") def _load_default_templates(self): """Load default system templates""" default_templates = [ NotificationTemplate( id="welcome", name="Welcome Email", channel=NotificationChannel.EMAIL, category=NotificationCategory.SYSTEM, subject_template="Welcome to {{app_name}}!", body_template=""" Hi {{user_name}}, Welcome to {{app_name}}! We're excited to have you on board. Here are some things you can do to get started: - Complete your profile - Explore our features - Connect with other users If you have any questions, feel free to reach out to our support team. Best regards, The {{app_name}} Team """, variables=["user_name", "app_name"] ), NotificationTemplate( id="password_reset", name="Password Reset", channel=NotificationChannel.EMAIL, category=NotificationCategory.SECURITY, subject_template="Password Reset Request", body_template=""" Hi {{user_name}}, We received a request to reset your password for {{app_name}}. Click the link below to reset your password: {{reset_link}} This link will expire in {{expiry_hours}} hours. If you didn't request this, please ignore this email or contact support. Best regards, The {{app_name}} Team """, variables=["user_name", "app_name", "reset_link", "expiry_hours"] ), NotificationTemplate( id="order_confirmation", name="Order Confirmation", channel=NotificationChannel.EMAIL, category=NotificationCategory.TRANSACTION, subject_template="Order #{{order_id}} Confirmed", body_template=""" Hi {{user_name}}, Your order #{{order_id}} has been confirmed! Order Details: - Total: {{order_total}} - Items: {{item_count}} - Estimated Delivery: {{delivery_date}} You can track your order status at: {{tracking_link}} Thank you for your purchase! Best regards, The {{app_name}} Team """, variables=["user_name", "app_name", "order_id", "order_total", "item_count", "delivery_date", "tracking_link"] ), NotificationTemplate( id="sms_verification", name="SMS Verification", channel=NotificationChannel.SMS, category=NotificationCategory.SECURITY, body_template="Your {{app_name}} verification code is: {{code}}. Valid for {{expiry_minutes}} minutes.", variables=["app_name", "code", "expiry_minutes"] ), NotificationTemplate( id="push_reminder", name="Push Reminder", channel=NotificationChannel.PUSH, category=NotificationCategory.UPDATE, body_template="{{reminder_text}}", variables=["reminder_text"] ), NotificationTemplate( id="in_app_alert", name="In-App Alert", channel=NotificationChannel.IN_APP, category=NotificationCategory.SYSTEM, body_template="{{alert_message}}", variables=["alert_message"] ), NotificationTemplate( id="weekly_digest", name="Weekly Digest", channel=NotificationChannel.EMAIL, category=NotificationCategory.MARKETING, subject_template="Your Weekly {{app_name}} Digest", body_template=""" Hi {{user_name}}, Here's what happened this week on {{app_name}}: 📊 Stats: - New connections: {{new_connections}} - Messages received: {{messages_count}} - Activities completed: {{activities_count}} 🔥 Trending: {{trending_items}} 💡 Tip of the week: {{weekly_tip}} See you next week! The {{app_name}} Team """, variables=["user_name", "app_name", "new_connections", "messages_count", "activities_count", "trending_items", "weekly_tip"] ), NotificationTemplate( id="friend_request", name="Friend Request", channel=NotificationChannel.IN_APP, category=NotificationCategory.SOCIAL, body_template="{{sender_name}} sent you a friend request. {{personal_message}}", variables=["sender_name", "personal_message"] ) ] for template in default_templates: self.templates[template.id] = template async def create_template(self, template: NotificationTemplate) -> str: """Create a new template""" if not template.id: template.id = str(uuid.uuid4()) # Validate template if not self._validate_template(template): raise ValueError("Invalid template format") # Extract variables from template template.variables = self._extract_variables(template.body_template) if template.subject_template: template.variables.extend(self._extract_variables(template.subject_template)) template.variables = list(set(template.variables)) # Remove duplicates # Store template self.templates[template.id] = template logger.info(f"Created template: {template.id}") return template.id async def update_template(self, template_id: str, template: NotificationTemplate) -> bool: """Update an existing template""" if template_id not in self.templates: return False # Validate template if not self._validate_template(template): raise ValueError("Invalid template format") # Update template template.id = template_id template.updated_at = datetime.now() # Re-extract variables template.variables = self._extract_variables(template.body_template) if template.subject_template: template.variables.extend(self._extract_variables(template.subject_template)) template.variables = list(set(template.variables)) self.templates[template_id] = template logger.info(f"Updated template: {template_id}") return True async def get_template(self, template_id: str) -> Optional[NotificationTemplate]: """Get a template by ID""" return self.templates.get(template_id) async def get_all_templates(self) -> List[NotificationTemplate]: """Get all templates""" return list(self.templates.values()) async def delete_template(self, template_id: str) -> bool: """Delete a template""" if template_id in self.templates: del self.templates[template_id] logger.info(f"Deleted template: {template_id}") return True return False async def render_template(self, template: NotificationTemplate, variables: Dict[str, Any]) -> str: """Render a template with variables""" if not template: raise ValueError("Template not provided") # Start with body template rendered = template.body_template # Replace variables for var_name in template.variables: placeholder = f"{{{{{var_name}}}}}" value = variables.get(var_name, f"[{var_name}]") # Default to placeholder if not provided # Convert non-string values to string if not isinstance(value, str): value = str(value) rendered = rendered.replace(placeholder, value) # Clean up extra whitespace rendered = re.sub(r'\n\s*\n', '\n\n', rendered.strip()) return rendered async def render_subject(self, template: NotificationTemplate, variables: Dict[str, Any]) -> Optional[str]: """Render a template subject with variables""" if not template or not template.subject_template: return None rendered = template.subject_template # Replace variables for var_name in self._extract_variables(template.subject_template): placeholder = f"{{{{{var_name}}}}}" value = variables.get(var_name, f"[{var_name}]") if not isinstance(value, str): value = str(value) rendered = rendered.replace(placeholder, value) return rendered def _validate_template(self, template: NotificationTemplate) -> bool: """Validate template format""" if not template.name or not template.body_template: return False # Check for basic template syntax try: # Check for balanced braces open_count = template.body_template.count("{{") close_count = template.body_template.count("}}") if open_count != close_count: return False if template.subject_template: open_count = template.subject_template.count("{{") close_count = template.subject_template.count("}}") if open_count != close_count: return False return True except Exception as e: logger.error(f"Template validation error: {e}") return False def _extract_variables(self, template_text: str) -> List[str]: """Extract variable names from template text""" if not template_text: return [] # Find all {{variable_name}} patterns pattern = r'\{\{(\w+)\}\}' matches = re.findall(pattern, template_text) return list(set(matches)) # Return unique variable names async def get_templates_by_channel(self, channel: NotificationChannel) -> List[NotificationTemplate]: """Get templates for a specific channel""" return [t for t in self.templates.values() if t.channel == channel] async def get_templates_by_category(self, category: NotificationCategory) -> List[NotificationTemplate]: """Get templates for a specific category""" return [t for t in self.templates.values() if t.category == category] async def clone_template(self, template_id: str, new_name: str) -> str: """Clone an existing template""" original = self.templates.get(template_id) if not original: raise ValueError(f"Template {template_id} not found") # Create new template new_template = NotificationTemplate( id=str(uuid.uuid4()), name=new_name, channel=original.channel, category=original.category, subject_template=original.subject_template, body_template=original.body_template, variables=original.variables.copy(), metadata=original.metadata.copy(), is_active=True, created_at=datetime.now() ) self.templates[new_template.id] = new_template logger.info(f"Cloned template {template_id} to {new_template.id}") return new_template.id