""" Notification system for Google Alerts Reporter Supports Email (SMTP), WhatsApp (Twilio), and Slack """ import logging import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from email.mime.base import MIMEBase from email import encoders from pathlib import Path from typing import Optional, List, Dict import requests from datetime import datetime from .config import settings logger = logging.getLogger(__name__) class EmailNotifier: """Email notifications via SMTP.""" def __init__(self): """Initialize email notifier.""" self.enabled = getattr(settings, 'email_enabled', False) self.smtp_host = getattr(settings, 'smtp_host', '') self.smtp_port = getattr(settings, 'smtp_port', 587) self.smtp_user = getattr(settings, 'smtp_user', '') self.smtp_password = getattr(settings, 'smtp_password', '') self.from_email = getattr(settings, 'email_from', '') self.to_emails = getattr(settings, 'email_to', '').split(',') def send_report( self, subject: str, body_html: str, attachments: Optional[List[Path]] = None ) -> bool: """ Send email report with optional attachments. Args: subject: Email subject body_html: HTML body content attachments: List of file paths to attach Returns: True if sent successfully, False otherwise """ if not self.enabled: logger.info("Email notifications disabled") return False try: # Create message msg = MIMEMultipart('alternative') msg['Subject'] = subject msg['From'] = self.from_email msg['To'] = ', '.join(self.to_emails) msg['Date'] = datetime.now().strftime('%a, %d %b %Y %H:%M:%S %z') # Add HTML body msg.attach(MIMEText(body_html, 'html', 'utf-8')) # Add attachments if attachments: for file_path in attachments: if not file_path.exists(): logger.warning(f"Attachment not found: {file_path}") continue with open(file_path, 'rb') as f: part = MIMEBase('application', 'octet-stream') part.set_payload(f.read()) encoders.encode_base64(part) part.add_header( 'Content-Disposition', f'attachment; filename={file_path.name}' ) msg.attach(part) # Send email with smtplib.SMTP(self.smtp_host, self.smtp_port) as server: server.starttls() server.login(self.smtp_user, self.smtp_password) server.send_message(msg) logger.info(f"Email sent to {len(self.to_emails)} recipients") return True except Exception as e: logger.error(f"Failed to send email: {e}") return False def send_crisis_alert( self, percent_negative: float, total_items: int, negative_count: int ) -> bool: """ Send crisis alert email. Args: percent_negative: Percentage of negative alerts total_items: Total number of alerts negative_count: Count of negative alerts Returns: True if sent successfully """ crisis_level = "CRÍTICO" if percent_negative > 70 else "MODERADO" subject = f"🚨 ALERTA DE CRISE {crisis_level}: {percent_negative:.0f}% Negativos" body_html = f"""

🚨 ALERTA DE CRISE: {settings.search_query}

Nível: {crisis_level} | {datetime.now().strftime('%d/%m/%Y %H:%M')}

{percent_negative:.0f}% DOS ALERTAS SÃO NEGATIVOS

Foram identificados {negative_count} alertas negativos de um total de {total_items} notícias analisadas.

Estatísticas

{negative_count}

Negativos

{total_items - negative_count}

Outros

{percent_negative:.1f}%

Crítico

Ações Recomendadas

  1. Acesse o dashboard: https://{getattr(settings, 'domain', 'localhost')}
  2. Revise alertas negativos e identifique fontes críticas
  3. Ative plano de mitigação conforme protocolo de crise
  4. Desindexe URLs do Reclame Aqui via Search Console

Google Alerts Reporter v{settings.app_version}

Este é um alerta automático do sistema de monitoramento de crise

""" return self.send_report(subject, body_html) class WhatsAppNotifier: """WhatsApp notifications via Twilio.""" def __init__(self): """Initialize WhatsApp notifier.""" self.enabled = getattr(settings, 'whatsapp_enabled', False) self.account_sid = getattr(settings, 'twilio_account_sid', '') self.auth_token = getattr(settings, 'twilio_auth_token', '') self.from_number = getattr(settings, 'twilio_whatsapp_from', '') self.to_numbers = getattr(settings, 'twilio_whatsapp_to', '').split(',') self.threshold = getattr(settings, 'whatsapp_threshold', 70) def send_message(self, message: str) -> bool: """ Send WhatsApp message via Twilio. Args: message: Message text (max 1600 chars) Returns: True if sent successfully """ if not self.enabled: logger.info("WhatsApp notifications disabled") return False try: from twilio.rest import Client client = Client(self.account_sid, self.auth_token) for to_number in self.to_numbers: to_number = to_number.strip() msg = client.messages.create( from_=self.from_number, body=message[:1600], # Twilio limit to=to_number ) logger.info(f"WhatsApp message sent: {msg.sid}") return True except Exception as e: logger.error(f"Failed to send WhatsApp message: {e}") return False def send_crisis_alert( self, percent_negative: float, total_items: int, negative_count: int ) -> bool: """ Send crisis alert via WhatsApp. Args: percent_negative: Percentage of negative alerts total_items: Total alerts negative_count: Negative alerts count Returns: True if sent successfully """ # Only send if threshold exceeded if percent_negative < self.threshold: logger.info(f"WhatsApp threshold not met: {percent_negative:.0f}% < {self.threshold}%") return False crisis_level = "🔴 CRÍTICO" if percent_negative > 70 else "🟠 MODERADO" message = f""" 🚨 *ALERTA DE CRISE* {settings.search_query} {crisis_level} 📊 *Estatísticas:* • {negative_count} alertas negativos • {total_items} total analisados • *{percent_negative:.0f}%* são negativos ⚡ *Ação Imediata Necessária* 🔗 Acesse: https://{getattr(settings, 'domain', 'localhost')} _{datetime.now().strftime('%d/%m/%Y %H:%M')}_ """ return self.send_message(message) class SlackNotifier: """Slack notifications via webhook.""" def __init__(self): """Initialize Slack notifier.""" self.enabled = getattr(settings, 'slack_enabled', False) self.webhook_url = getattr(settings, 'slack_webhook_url', '') self.channel = getattr(settings, 'slack_channel', '#crise-alerts') self.username = getattr(settings, 'slack_username', 'Alerts Reporter Bot') def send_message( self, text: str, attachments: Optional[List[Dict]] = None ) -> bool: """ Send Slack message. Args: text: Message text attachments: Optional message attachments Returns: True if sent successfully """ if not self.enabled: logger.info("Slack notifications disabled") return False try: payload = { 'channel': self.channel, 'username': self.username, 'text': text, 'icon_emoji': ':rotating_light:' } if attachments: payload['attachments'] = attachments response = requests.post( self.webhook_url, json=payload, timeout=10 ) response.raise_for_status() logger.info("Slack message sent successfully") return True except Exception as e: logger.error(f"Failed to send Slack message: {e}") return False def send_crisis_alert( self, percent_negative: float, total_items: int, negative_count: int, negative_items: List[Dict] ) -> bool: """ Send crisis alert to Slack. Args: percent_negative: Percentage of negative alerts total_items: Total alerts negative_count: Negative count negative_items: List of negative items Returns: True if sent successfully """ crisis_level = "CRÍTICO" if percent_negative > 70 else "MODERADO" color = "#d32f2f" if percent_negative > 70 else "#f57c00" text = f":rotating_light: *ALERTA DE CRISE {crisis_level}*: {settings.search_query}" # Build top 3 negative items top_items = negative_items[:3] items_text = "\n".join([ f"• {item['title'][:80]}..." for item in top_items ]) attachments = [ { "color": color, "title": f"{percent_negative:.0f}% dos alertas são negativos", "fields": [ { "title": "Total Analisado", "value": str(total_items), "short": True }, { "title": "Negativos", "value": str(negative_count), "short": True } ], "text": f"*Top 3 Alertas Negativos:*\n{items_text}", "footer": "Google Alerts Reporter", "footer_icon": "https://platform.slack-edge.com/img/default_application_icon.png", "ts": int(datetime.now().timestamp()) } ] return self.send_message(text, attachments) class NotificationManager: """Manages all notification channels.""" def __init__(self): """Initialize notification manager.""" self.email = EmailNotifier() self.whatsapp = WhatsAppNotifier() self.slack = SlackNotifier() def send_crisis_alert( self, sentiments: Dict, total_items: int, negative_items: List[Dict] ) -> Dict[str, bool]: """ Send crisis alert via all enabled channels. Args: sentiments: Sentiment counts total_items: Total items negative_items: List of negative items Returns: Dict with status of each channel """ percent_negative = (sentiments['neg'] / total_items * 100) if total_items > 0 else 0 negative_count = sentiments['neg'] results = {} # Email if self.email.enabled: results['email'] = self.email.send_crisis_alert( percent_negative, total_items, negative_count ) # WhatsApp if self.whatsapp.enabled: results['whatsapp'] = self.whatsapp.send_crisis_alert( percent_negative, total_items, negative_count ) # Slack if self.slack.enabled: results['slack'] = self.slack.send_crisis_alert( percent_negative, total_items, negative_count, negative_items ) logger.info(f"Notifications sent: {results}") return results def send_daily_report( self, report_path: Path ) -> bool: """ Send daily report via email. Args: report_path: Path to PDF report Returns: True if sent successfully """ subject = f"Relatório Diário - {settings.search_query} - {datetime.now().strftime('%d/%m/%Y')}" body_html = f"""

Relatório Diário de Monitoramento

{settings.search_query}

Segue em anexo o relatório diário de análise de crise.

Data: {datetime.now().strftime('%d/%m/%Y')}

Acessar Dashboard

""" return self.email.send_report(subject, body_html, [report_path])