404 lines
12 KiB
Python
404 lines
12 KiB
Python
"""
|
|
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": """
|
|
<h2>Nuevo ticket creado</h2>
|
|
<p>Hola {{ user_name }},</p>
|
|
<p>Se ha creado un nuevo ticket:</p>
|
|
<ul>
|
|
<li><strong>Número:</strong> #{{ ticket_number }}</li>
|
|
<li><strong>Asunto:</strong> {{ subject }}</li>
|
|
<li><strong>Prioridad:</strong> {{ priority }}</li>
|
|
</ul>
|
|
<p><a href="{{ ticket_url }}">Ver ticket</a></p>
|
|
<p>Saludos,<br>Equipo de Soporte</p>
|
|
""",
|
|
"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": """
|
|
<h2>Ticket asignado</h2>
|
|
<p>Hola {{ agent_name }},</p>
|
|
<p>Se te ha asignado el ticket:</p>
|
|
<ul>
|
|
<li><strong>Número:</strong> #{{ ticket_number }}</li>
|
|
<li><strong>Asunto:</strong> {{ subject }}</li>
|
|
<li><strong>Cliente:</strong> {{ customer_name }}</li>
|
|
<li><strong>Prioridad:</strong> {{ priority }}</li>
|
|
</ul>
|
|
<p><a href="{{ ticket_url }}">Ver ticket</a></p>
|
|
""",
|
|
"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
|
|
} |