"""
Notification Services
This module contains services for sending notifications through various
channels including Telegram, email, webhooks, and other messaging platforms.
It provides a unified interface for notification delivery with retry logic
and error handling.
"""
import requests
import json
import time
import logging
from typing import Optional, Dict, Any, List
from datetime import datetime, timedelta
from django.conf import settings
from django.utils import timezone
from django.core.mail import send_mail
from django.template.loader import render_to_string
from .models import (
NotificationChannel, NotificationTemplate, Notification,
NotificationRule
)
class TelegramService:
"""
Service for sending notifications via Telegram Bot API.
This service handles Telegram bot integration, message formatting,
and delivery with automatic retry logic for failed messages.
Attributes:
logger (Logger): Logger instance for recording operations
bot_token (str): Telegram bot token from configuration
default_chat_id (str): Default chat ID for notifications
api_base_url (str): Base URL for Telegram Bot API
"""
def __init__(self, bot_token: Optional[str] = None, chat_id: Optional[str] = None):
"""
Initialize the Telegram service.
Args:
bot_token (str, optional): Telegram bot token, uses settings if not provided
chat_id (str, optional): Default chat ID, uses settings if not provided
"""
# Set up logging for Telegram operations
self.logger = logging.getLogger('stream_processor.telegram')
# Get configuration from settings or parameters
self.bot_token = bot_token or settings.TELEGRAM_CONFIG.get('BOT_TOKEN', '')
self.default_chat_id = chat_id or settings.TELEGRAM_CONFIG.get('CHAT_ID', '')
# Telegram Bot API base URL
self.api_base_url = f"https://api.telegram.org/bot{self.bot_token}"
# Validate configuration
self._validate_config()
def _validate_config(self) -> None:
"""
Validate Telegram configuration.
Raises:
ValueError: If required configuration is missing
"""
if not self.bot_token:
self.logger.warning("Telegram bot token not configured - notifications disabled")
return
if len(self.bot_token) < 20:
raise ValueError("Invalid Telegram bot token format")
self.logger.info("Telegram service initialized successfully")
def send_message(
self,
message: str,
chat_id: Optional[str] = None,
parse_mode: str = 'HTML',
disable_web_page_preview: bool = True
) -> Dict[str, Any]:
"""
Send a text message via Telegram.
Args:
message (str): Message text to send
chat_id (str, optional): Target chat ID, uses default if not provided
parse_mode (str): Message parse mode (HTML, Markdown, or None)
disable_web_page_preview (bool): Disable link previews
Returns:
Dict[str, Any]: API response data
Raises:
requests.RequestException: If API request fails
"""
# Skip if bot token is not configured
if not self.bot_token:
self.logger.info("Telegram bot token not configured, skipping notification")
return {"status": "skipped", "reason": "bot_token_not_configured"}
# Use provided chat_id or default
target_chat_id = chat_id or self.default_chat_id
if not target_chat_id:
self.logger.warning("No chat ID provided and no default configured")
return {"status": "skipped", "reason": "no_chat_id"}
# Prepare API endpoint and payload
url = f"{self.api_base_url}/sendMessage"
payload = {
'chat_id': target_chat_id,
'text': message,
'parse_mode': parse_mode,
'disable_web_page_preview': disable_web_page_preview
}
try:
self.logger.debug(f"Sending Telegram message to chat {target_chat_id}")
# Send POST request to Telegram API
response = requests.post(
url,
json=payload,
timeout=30,
headers={'Content-Type': 'application/json'}
)
# Check response status
if response.status_code == 200:
result = response.json()
if result.get('ok'):
self.logger.info(f"Telegram message sent successfully: {result['result']['message_id']}")
return result
else:
error_msg = result.get('description', 'Unknown error')
self.logger.error(f"Telegram API error: {error_msg}")
raise requests.RequestException(f"Telegram API error: {error_msg}")
else:
self.logger.error(f"HTTP error {response.status_code}: {response.text}")
raise requests.RequestException(f"HTTP {response.status_code}: {response.text}")
except requests.Timeout:
self.logger.error("Telegram API request timed out")
raise
except requests.RequestException as e:
self.logger.error(f"Failed to send Telegram message: {e}")
raise
def send_photo(
self,
photo_path: str,
caption: str = '',
chat_id: Optional[str] = None
) -> Dict[str, Any]:
"""
Send a photo with caption via Telegram.
Args:
photo_path (str): Path to the photo file
caption (str): Photo caption text
chat_id (str, optional): Target chat ID
Returns:
Dict[str, Any]: API response data
"""
target_chat_id = chat_id or self.default_chat_id
if not target_chat_id:
raise ValueError("No chat ID provided and no default configured")
url = f"{self.api_base_url}/sendPhoto"
try:
with open(photo_path, 'rb') as photo_file:
files = {'photo': photo_file}
data = {
'chat_id': target_chat_id,
'caption': caption
}
response = requests.post(url, files=files, data=data, timeout=60)
if response.status_code == 200:
result = response.json()
if result.get('ok'):
self.logger.info(f"Telegram photo sent successfully")
return result
else:
error_msg = result.get('description', 'Unknown error')
raise requests.RequestException(f"Telegram API error: {error_msg}")
else:
raise requests.RequestException(f"HTTP {response.status_code}: {response.text}")
except FileNotFoundError:
self.logger.error(f"Photo file not found: {photo_path}")
raise
except Exception as e:
self.logger.error(f"Failed to send Telegram photo: {e}")
raise
def format_jingle_detection_message(
self,
jingle_name: str,
similarity: float,
iframe_path: str,
iframe_url: str,
detection_time: datetime
) -> str:
"""
Format a jingle detection message for Telegram.
Args:
jingle_name (str): Name of detected jingle
similarity (float): Similarity score
iframe_path (str): Path to extracted frame
iframe_url (str): URL to view the frame
detection_time (datetime): When detection occurred
Returns:
str: Formatted Telegram message
"""
return (
"🎵 Jingle Detected 🎵\n\n"
f"Name: {jingle_name}\n"
f"Similarity: {similarity:.3f}\n"
f"Frame Path: {iframe_path}\n"
f"Frame URL: View Frame\n"
f"Detection Time: {detection_time.strftime('%Y-%m-%d %H:%M:%S')}"
)
def format_ad_break_message(
self,
channel_name: str,
start_time: datetime,
end_time: Optional[datetime] = None,
duration: Optional[int] = None,
is_estimated: bool = False
) -> str:
"""
Format an ad break message for Telegram.
Args:
channel_name (str): Name of the channel
start_time (datetime): Ad break start time
end_time (datetime, optional): Ad break end time
duration (int, optional): Duration in seconds
is_estimated (bool): Whether duration is estimated
Returns:
str: Formatted Telegram message
"""
if end_time and duration:
duration_type = "Estimated" if is_estimated else "Actual"
return (
"🚨 Ad Break Detected 🚨\n\n"
f"Channel: {channel_name}\n"
f"Start Time: {start_time.strftime('%Y-%m-%d %H:%M:%S')}\n"
f"End Time: {end_time.strftime('%Y-%m-%d %H:%M:%S')}\n"
f"{duration_type} Duration: {duration} seconds"
)
else:
return (
"🔊 Ad Break Started 🔊\n\n"
f"Channel: {channel_name}\n"
f"Start Time: {start_time.strftime('%Y-%m-%d %H:%M:%S')}\n"
f"Status: In Progress..."
)
class EmailService:
"""
Service for sending email notifications.
This service handles email delivery using Django's email framework
with support for HTML templates and attachments.
"""
def __init__(self):
"""Initialize the email service."""
self.logger = logging.getLogger('stream_processor.email')
def send_email(
self,
subject: str,
message: str,
recipients: List[str],
html_message: Optional[str] = None,
from_email: Optional[str] = None
) -> bool:
"""
Send an email notification.
Args:
subject (str): Email subject line
message (str): Plain text message content
recipients (List[str]): List of recipient email addresses
html_message (str, optional): HTML message content
from_email (str, optional): Sender email address
Returns:
bool: True if email was sent successfully
"""
try:
sent_count = send_mail(
subject=subject,
message=message,
from_email=from_email,
recipient_list=recipients,
html_message=html_message,
fail_silently=False
)
self.logger.info(f"Email sent to {sent_count} recipients")
return sent_count > 0
except Exception as e:
self.logger.error(f"Failed to send email: {e}")
return False
class WebhookService:
"""
Service for sending webhook notifications.
This service handles HTTP webhook delivery with configurable
endpoints, headers, and payload formats.
"""
def __init__(self):
"""Initialize the webhook service."""
self.logger = logging.getLogger('stream_processor.webhooks')
def send_webhook(
self,
url: str,
data: Dict[str, Any],
headers: Optional[Dict[str, str]] = None,
method: str = 'POST',
timeout: int = 30
) -> bool:
"""
Send a webhook notification.
Args:
url (str): Webhook endpoint URL
data (Dict[str, Any]): Data to send in the webhook
headers (Dict[str, str], optional): HTTP headers
method (str): HTTP method (POST, PUT, PATCH)
timeout (int): Request timeout in seconds
Returns:
bool: True if webhook was sent successfully
"""
try:
# Prepare headers
webhook_headers = {'Content-Type': 'application/json'}
if headers:
webhook_headers.update(headers)
# Send webhook request
response = requests.request(
method=method.upper(),
url=url,
json=data,
headers=webhook_headers,
timeout=timeout
)
if response.status_code in [200, 201, 202, 204]:
self.logger.info(f"Webhook sent successfully to {url}")
return True
else:
self.logger.error(f"Webhook failed with status {response.status_code}: {response.text}")
return False
except Exception as e:
self.logger.error(f"Failed to send webhook to {url}: {e}")
return False
class NotificationService:
"""
Main notification service that coordinates different notification channels.
This service provides a unified interface for sending notifications
through various channels with automatic channel selection, template
rendering, and retry logic.
"""
def __init__(self):
"""Initialize the notification service."""
self.logger = logging.getLogger('stream_processor.notifications')
# Initialize individual services
self.telegram = TelegramService()
self.email = EmailService()
self.webhook = WebhookService()
def send_notification(
self,
template_type: str,
context: Dict[str, Any],
channel_name: Optional[str] = None,
recipient: Optional[str] = None
) -> bool:
"""
Send a notification using the specified template and context.
Args:
template_type (str): Type of notification template to use
context (Dict[str, Any]): Template variables for rendering
channel_name (str, optional): Specific channel to use
recipient (str, optional): Specific recipient override
Returns:
bool: True if notification was sent successfully
"""
try:
# Find applicable notification rules
rules = NotificationRule.objects.filter(
event_type=template_type,
is_active=True
).order_by('-priority')
if not rules.exists():
self.logger.warning(f"No notification rules found for event type: {template_type}")
return False
success_count = 0
# Process each applicable rule
for rule in rules:
if rule.can_trigger() and rule.check_conditions(context):
# Send notification using this rule
if self._send_rule_notification(rule, context, recipient):
success_count += 1
rule.mark_triggered()
return success_count > 0
except Exception as e:
self.logger.error(f"Failed to send notification: {e}")
return False
def _send_rule_notification(
self,
rule: NotificationRule,
context: Dict[str, Any],
recipient: Optional[str] = None
) -> bool:
"""
Send a notification based on a specific rule.
Args:
rule (NotificationRule): Notification rule to apply
context (Dict[str, Any]): Template rendering context
recipient (str, optional): Recipient override
Returns:
bool: True if notification was sent successfully
"""
try:
# Render message from template
subject = rule.template.render_subject(context)
message = rule.template.render_message(context)
# Determine recipient
target_recipient = recipient
if not target_recipient:
if rule.channel.channel_type == 'telegram':
target_recipient = rule.channel.get_telegram_config().get('chat_id')
elif rule.channel.channel_type == 'email':
recipients = rule.channel.get_email_config().get('recipients', [])
target_recipient = recipients[0] if recipients else None
if not target_recipient:
self.logger.error(f"No recipient available for rule: {rule.name}")
return False
# Create notification record
notification = Notification.objects.create(
channel=rule.channel,
template=rule.template,
recipient=target_recipient,
subject=subject,
message=message,
context_data=context,
status='pending'
)
# Send via appropriate channel
return self._deliver_notification(notification)
except Exception as e:
self.logger.error(f"Failed to send rule notification: {e}")
return False
def _deliver_notification(self, notification: Notification) -> bool:
"""
Deliver a notification via its configured channel.
Args:
notification (Notification): Notification to deliver
Returns:
bool: True if delivery was successful
"""
try:
channel = notification.channel
if channel.channel_type == 'telegram':
# Send via Telegram
config = channel.get_telegram_config()
result = self.telegram.send_message(
message=notification.message,
chat_id=notification.recipient
)
# Mark as sent
message_id = result.get('result', {}).get('message_id')
notification.mark_sent(external_id=str(message_id) if message_id else None)
return True
elif channel.channel_type == 'email':
# Send via email
success = self.email.send_email(
subject=notification.subject,
message=notification.message,
recipients=[notification.recipient]
)
if success:
notification.mark_sent()
return True
else:
notification.mark_failed("Email delivery failed")
return False
elif channel.channel_type == 'webhook':
# Send via webhook
config = channel.get_webhook_config()
success = self.webhook.send_webhook(
url=config.get('url'),
data={
'subject': notification.subject,
'message': notification.message,
'context': notification.context_data,
'timestamp': notification.scheduled_at.isoformat()
},
headers=config.get('headers', {}),
method=config.get('method', 'POST')
)
if success:
notification.mark_sent()
return True
else:
notification.mark_failed("Webhook delivery failed")
return False
else:
self.logger.error(f"Unsupported channel type: {channel.channel_type}")
notification.mark_failed(f"Unsupported channel type: {channel.channel_type}")
return False
except Exception as e:
self.logger.error(f"Failed to deliver notification {notification.id}: {e}")
notification.mark_failed(str(e))
return False
def retry_failed_notifications(self, max_age_hours: int = 24) -> int:
"""
Retry failed notifications that are eligible for retry.
Args:
max_age_hours (int): Maximum age of notifications to retry
Returns:
int: Number of notifications retried
"""
cutoff_time = timezone.now() - timedelta(hours=max_age_hours)
# Find failed notifications that can be retried
failed_notifications = Notification.objects.filter(
status='failed',
created_at__gte=cutoff_time
)
retry_count = 0
for notification in failed_notifications:
if notification.can_retry():
notification.status = 'pending'
notification.save()
if self._deliver_notification(notification):
retry_count += 1
if retry_count > 0:
self.logger.info(f"Retried {retry_count} failed notifications")
return retry_count