""" Email Tasks - ServiceManagerWeb Workers Tareas asíncronas para envío de emails y notificaciones """ from celery import current_task from celery.exceptions import Retry from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from email.mime.base import MIMEBase from email import encoders import smtplib import ssl from typing import Dict, List, Optional, Any from jinja2 import Template, Environment, BaseLoader import structlog from app.celery import celery_app from app.core.config import get_settings from app.core.logging import get_logger settings = get_settings() logger = get_logger(__name__) class MemoryLoader(BaseLoader): """Jinja2 loader for templates from memory/database.""" def __init__(self, templates: Dict[str, str]): self.templates = templates def get_source(self, environment, template): if template in self.templates: source = self.templates[template] return source, None, lambda: True raise TemplateNotFoundError(template) @celery_app.task( bind=True, autoretry_for=(Exception,), retry_kwargs={'max_retries': 3, 'countdown': 60}, time_limit=120, soft_time_limit=90 ) def send_email_task( self, to_email: str, subject: str, html_content: str, text_content: Optional[str] = None, from_email: Optional[str] = None, from_name: Optional[str] = None, attachments: Optional[List[Dict[str, Any]]] = None, tenant_id: Optional[str] = None, correlation_id: Optional[str] = None ) -> Dict[str, Any]: """ Send email using SMTP. Args: to_email: Recipient email address subject: Email subject html_content: HTML content text_content: Plain text content (optional) from_email: Sender email (optional, uses default) from_name: Sender name (optional) attachments: List of attachment dicts tenant_id: Tenant ID for logging correlation_id: Correlation ID for tracking Returns: Dict with send result """ task_logger = logger.bind( task_id=self.request.id, task_name=self.name, tenant_id=tenant_id, correlation_id=correlation_id ) task_logger.info( "Starting email send task", to_email=to_email, subject=subject ) try: # Prepare email msg = MIMEMultipart('alternative') msg['Subject'] = subject msg['From'] = f"{from_name or settings.DEFAULT_FROM_NAME} <{from_email or settings.DEFAULT_FROM_EMAIL}>" msg['To'] = to_email # Add text content if text_content: text_part = MIMEText(text_content, 'plain', 'utf-8') msg.attach(text_part) # Add HTML content html_part = MIMEText(html_content, 'html', 'utf-8') msg.attach(html_part) # Add attachments if attachments: for attachment in attachments: part = MIMEBase('application', 'octet-stream') part.set_payload(attachment['content']) encoders.encode_base64(part) part.add_header( 'Content-Disposition', f'attachment; filename= {attachment["filename"]}' ) msg.attach(part) # Send email context = ssl.create_default_context() with smtplib.SMTP(settings.SMTP_HOST, settings.SMTP_PORT) as server: if settings.SMTP_USE_TLS: server.starttls(context=context) if settings.SMTP_USER and settings.SMTP_PASSWORD: server.login(settings.SMTP_USER, settings.SMTP_PASSWORD) server.send_message(msg) task_logger.info( "Email sent successfully", to_email=to_email, subject=subject ) return { "success": True, "to_email": to_email, "subject": subject, "sent_at": current_task.request.eta or "now" } except Exception as exc: task_logger.error( "Failed to send email", to_email=to_email, subject=subject, error=str(exc), exc_info=True ) # Check if we should retry if self.request.retries < self.max_retries: task_logger.info( "Retrying email send", retry_count=self.request.retries + 1, max_retries=self.max_retries ) raise self.retry(countdown=60 * (2 ** self.request.retries)) return { "success": False, "to_email": to_email, "subject": subject, "error": str(exc) } @celery_app.task( bind=True, time_limit=300, soft_time_limit=240 ) def send_templated_email_task( self, to_email: str, template_name: str, context: Dict[str, Any], tenant_id: Optional[str] = None, correlation_id: Optional[str] = None ) -> Dict[str, Any]: """ Send email using a template. Args: to_email: Recipient email template_name: Template identifier context: Template context variables tenant_id: Tenant ID correlation_id: Correlation ID Returns: Dict with send result """ task_logger = logger.bind( task_id=self.request.id, task_name=self.name, tenant_id=tenant_id, correlation_id=correlation_id ) task_logger.info( "Starting templated email task", to_email=to_email, template_name=template_name ) try: # TODO: Fetch template from database # For now, use mock templates templates = { "ticket_created": { "subject": "Nuevo ticket #{{ ticket_number }}: {{ subject }}", "html": """

Nuevo ticket creado

Hola {{ user_name }},

Se ha creado un nuevo ticket:

Ver ticket

Saludos,
Equipo de Soporte

""", "text": """ Nuevo ticket creado Hola {{ user_name }}, Se ha creado un nuevo ticket: Número: #{{ ticket_number }} Asunto: {{ subject }} Prioridad: {{ priority }} Ver ticket: {{ ticket_url }} Saludos, Equipo de Soporte """ }, "ticket_assigned": { "subject": "Ticket #{{ ticket_number }} asignado a ti", "html": """

Ticket asignado

Hola {{ agent_name }},

Se te ha asignado el ticket:

Ver ticket

""", "text": """ Ticket asignado Hola {{ agent_name }}, Se te ha asignado el ticket: Número: #{{ ticket_number }} Asunto: {{ subject }} Cliente: {{ customer_name }} Prioridad: {{ priority }} Ver ticket: {{ ticket_url }} """ } } if template_name not in templates: raise ValueError(f"Template '{template_name}' not found") template_data = templates[template_name] # Render templates env = Environment(loader=MemoryLoader({ f"{template_name}_subject": template_data["subject"], f"{template_name}_html": template_data["html"], f"{template_name}_text": template_data["text"] })) subject_template = env.get_template(f"{template_name}_subject") html_template = env.get_template(f"{template_name}_html") text_template = env.get_template(f"{template_name}_text") subject = subject_template.render(**context) html_content = html_template.render(**context) text_content = text_template.render(**context) # Send email using the basic send task return send_email_task.apply_async( kwargs={ "to_email": to_email, "subject": subject, "html_content": html_content, "text_content": text_content, "tenant_id": tenant_id, "correlation_id": correlation_id } ).get() except Exception as exc: task_logger.error( "Failed to send templated email", to_email=to_email, template_name=template_name, error=str(exc), exc_info=True ) return { "success": False, "to_email": to_email, "template_name": template_name, "error": str(exc) } @celery_app.task( bind=True, time_limit=600, soft_time_limit=540 ) def send_bulk_email_task( self, email_list: List[Dict[str, Any]], tenant_id: Optional[str] = None, correlation_id: Optional[str] = None ) -> Dict[str, Any]: """ Send bulk emails. Args: email_list: List of email dicts with to_email, subject, content tenant_id: Tenant ID correlation_id: Correlation ID Returns: Dict with bulk send results """ task_logger = logger.bind( task_id=self.request.id, task_name=self.name, tenant_id=tenant_id, correlation_id=correlation_id ) total_emails = len(email_list) task_logger.info(f"Starting bulk email task", total_emails=total_emails) results = [] for i, email_data in enumerate(email_list): try: result = send_email_task.apply_async( kwargs={ **email_data, "tenant_id": tenant_id, "correlation_id": correlation_id } ).get() results.append(result) # Update task progress current_task.update_state( state='PROGRESS', meta={'current': i + 1, 'total': total_emails} ) except Exception as exc: task_logger.error( "Failed to send bulk email item", index=i, email_data=email_data, error=str(exc) ) results.append({ "success": False, "to_email": email_data.get("to_email"), "error": str(exc) }) # Calculate stats successful = sum(1 for r in results if r.get("success")) failed = total_emails - successful task_logger.info( "Bulk email task completed", total=total_emails, successful=successful, failed=failed ) return { "total": total_emails, "successful": successful, "failed": failed, "results": results }