"""Email rendering, personalization, SMTP delivery, and campaign runner for MailPulse."""
import base64
import html
import re
import smtplib
import ssl
import threading
import time
import uuid
from email.header import Header
from email.mime.base import MIMEBase
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email import encoders
from email.utils import formataddr
from db import Database, utc_now
VAR_PATTERN = re.compile(r"\{\{\s*([a-zA-Z0-9_áéíóúÁÉÍÓÚñÑ]+)(?:\|([^}]*))?\s*\}\}")
def html_to_plain_text(raw_html: str) -> str:
"""Convert HTML email content into a readable plain-text fallback."""
if not raw_html:
return ""
text = re.sub(r"<(br|BR)\s*/?>", "\n", raw_html)
text = re.sub(r"", "\n", text)
text = re.sub(r"<[^>]+>", "", text)
text = html.unescape(text)
lines = [line.strip() for line in text.splitlines()]
cleaned = "\n".join(line for line in lines if line)
return cleaned
def build_contact_context(contact: dict = None, extra_vars: dict = None) -> dict:
"""Build dictionary of personalization variables from contact and extra overrides."""
ctx = {
"nombre": "",
"first_name": "",
"apellido": "",
"last_name": "",
"nombre_completo": "",
"email": "",
"empresa": "",
"company": "",
"telefono": "",
"phone": "",
}
if contact:
fn = (contact.get("first_name") or "").strip()
ln = (contact.get("last_name") or "").strip()
full = f"{fn} {ln}".strip()
comp = (contact.get("company") or "").strip()
phone = (contact.get("phone") or "").strip()
email = (contact.get("email") or "").strip()
ctx.update(
{
"nombre": fn,
"first_name": fn,
"apellido": ln,
"last_name": ln,
"nombre_completo": full or fn or email,
"email": email,
"empresa": comp,
"company": comp,
"telefono": phone,
"phone": phone,
}
)
custom = contact.get("custom_fields") or {}
if isinstance(custom, dict):
for k, v in custom.items():
ctx[str(k).lower()] = "" if v is None else str(v)
if extra_vars:
for k, v in extra_vars.items():
if v is not None and str(v) != "":
ctx[str(k).lower()] = str(v)
return ctx
def render_template_string(template: str, context: dict) -> str:
"""Replace {{variable}} or {{variable|fallback}} placeholders."""
if not template:
return ""
def replacer(match):
key = match.group(1).lower()
fallback = match.group(2) if match.group(2) is not None else ""
val = context.get(key, "")
return val if val else fallback
return VAR_PATTERN.sub(replacer, template)
def inject_tracking_and_footer(
html_body: str,
tracking_id: str,
base_url: str = "",
track_opens: bool = True,
unsubscribe_contact_id: str = "",
) -> str:
"""Append open-tracking pixel and optional unsubscribe footer to HTML body."""
parts = [html_body or ""]
if unsubscribe_contact_id and base_url:
unsub_url = f"{base_url.rstrip('/')}/unsubscribe/{unsubscribe_contact_id}"
parts.append(
f'"
)
if track_opens and tracking_id and base_url:
pixel_url = f"{base_url.rstrip('/')}/t/o/{tracking_id}.gif"
parts.append(f'
')
return "\n".join(parts)
class MailerService:
"""Handles 1-to-1 sending, mass campaign execution, and SMTP verification."""
def __init__(self, db: Database):
self.db = db
self._active_campaigns = set()
self._lock = threading.Lock()
def test_smtp_connection(self, smtp_cfg: dict) -> dict:
"""Verify SMTP connectivity and credentials."""
mode = smtp_cfg.get("mode", "sandbox")
if mode == "sandbox":
return {
"ok": True,
"mode": "sandbox",
"message": "Modo Simulación Local activo. Los correos se registran en la Bandeja de Salida sin envío externo.",
}
host = (smtp_cfg.get("host") or "").strip()
port = int(smtp_cfg.get("port") or 587)
security = (smtp_cfg.get("security") or "tls").lower()
username = (smtp_cfg.get("username") or "").strip()
password = smtp_cfg.get("password") or ""
if not host:
return {"ok": False, "message": "Debes ingresar el servidor SMTP (Host)."}
try:
if security == "ssl":
context = ssl.create_default_context()
server = smtplib.SMTP_SSL(host, port, timeout=12, context=context)
else:
server = smtplib.SMTP(host, port, timeout=12)
server.ehlo()
if security == "tls":
context = ssl.create_default_context()
server.starttls(context=context)
server.ehlo()
if username and password:
server.login(username, password)
server.quit()
return {"ok": True, "mode": "smtp", "message": f"Conexión SMTP exitosa con {host}:{port} ({security.upper()})."}
except Exception as exc:
return {"ok": False, "mode": "smtp", "message": f"Error SMTP: {exc}"}
def _build_mime_message(
self,
from_name: str,
from_email: str,
to_email: str,
to_name: str,
subject: str,
html_body: str,
text_body: str,
reply_to: str = "",
cc: str = "",
bcc: str = "",
attachments: list = None,
) -> MIMEMultipart:
has_attachments = bool(attachments)
root = MIMEMultipart("mixed") if has_attachments else MIMEMultipart("alternative")
root["Subject"] = str(Header(subject, "utf-8"))
root["From"] = formataddr((from_name, from_email)) if from_name else from_email
root["To"] = formataddr((to_name, to_email)) if to_name else to_email
if reply_to:
root["Reply-To"] = reply_to
if cc:
root["Cc"] = cc
alt = MIMEMultipart("alternative") if has_attachments else root
plain = text_body or html_to_plain_text(html_body)
alt.attach(MIMEText(plain, "plain", "utf-8"))
alt.attach(MIMEText(html_body or "", "html", "utf-8"))
if has_attachments:
root.attach(alt)
for att in attachments:
filename = att.get("filename") or "adjunto"
content_b64 = att.get("content_base64") or ""
mime_type = att.get("mime_type") or "application/octet-stream"
if "," in content_b64 and content_b64.startswith("data:"):
content_b64 = content_b64.split(",", 1)[1]
try:
raw_bytes = base64.b64decode(content_b64)
except Exception:
continue
maintype, subtype = (mime_type.split("/", 1) + ["octet-stream"])[:2]
part = MIMEBase(maintype, subtype)
part.set_payload(raw_bytes)
encoders.encode_base64(part)
part.add_header("Content-Disposition", "attachment", filename=filename)
root.attach(part)
return root
def _dispatch_smtp(self, smtp_cfg: dict, mime_msg: MIMEMultipart, recipients: list) -> None:
host = (smtp_cfg.get("host") or "").strip()
port = int(smtp_cfg.get("port") or 587)
security = (smtp_cfg.get("security") or "tls").lower()
username = (smtp_cfg.get("username") or "").strip()
password = smtp_cfg.get("password") or ""
from_email = smtp_cfg.get("from_email")
if not host:
raise ValueError("Servidor SMTP no configurado (Host vacío).")
if security == "ssl":
context = ssl.create_default_context()
with smtplib.SMTP_SSL(host, port, timeout=15, context=context) as server:
if username and password:
server.login(username, password)
server.sendmail(from_email, recipients, mime_msg.as_string())
else:
with smtplib.SMTP(host, port, timeout=15) as server:
server.ehlo()
if security == "tls":
context = ssl.create_default_context()
server.starttls(context=context)
server.ehlo()
if username and password:
server.login(username, password)
server.sendmail(from_email, recipients, mime_msg.as_string())
def send_single_email(
self,
to_email: str,
subject: str,
html_content: str,
text_content: str = "",
to_name: str = "",
cc: str = "",
bcc: str = "",
sender_id: str = None,
contact_id: str = None,
extra_vars: dict = None,
attachments: list = None,
send_type: str = "individual",
campaign_id: str = "",
track_opens: bool = True,
base_url: str = "",
include_unsubscribe: bool = False,
) -> dict:
"""Render, send (or simulate), and persist an email log entry."""
to_email = (to_email or "").strip()
if not to_email or "@" not in to_email:
raise ValueError("El correo de destino no es válido.")
smtp_cfg = self.db.get_smtp_config(sender_id)
if not smtp_cfg:
raise ValueError("No hay ningún perfil de remitente configurado.")
contact = None
if contact_id:
contact = self.db.get_contact(contact_id=contact_id)
elif to_email:
contact = self.db.get_contact(email=to_email)
if not to_name and contact:
to_name = f"{contact.get('first_name', '')} {contact.get('last_name', '')}".strip()
ctx = build_contact_context(contact, extra_vars=extra_vars)
if not ctx["email"]:
ctx["email"] = to_email
if not ctx["nombre"] and to_name:
ctx["nombre"] = to_name.split()[0]
ctx["nombre_completo"] = to_name
rendered_subject = render_template_string(subject, ctx)
rendered_html = render_template_string(html_content, ctx)
rendered_text = render_template_string(text_content, ctx) if text_content else html_to_plain_text(rendered_html)
tracking_id = uuid.uuid4().hex
final_html = inject_tracking_and_footer(
rendered_html,
tracking_id=tracking_id,
base_url=base_url,
track_opens=track_opens,
unsubscribe_contact_id=(contact["id"] if (include_unsubscribe and contact) else ""),
)
mode = smtp_cfg.get("mode", "sandbox")
status = "simulated"
error_msg = ""
if mode == "smtp":
mime_msg = self._build_mime_message(
from_name=smtp_cfg.get("from_name", ""),
from_email=smtp_cfg.get("from_email", ""),
to_email=to_email,
to_name=to_name,
subject=rendered_subject,
html_body=final_html,
text_body=rendered_text,
reply_to=smtp_cfg.get("reply_to", ""),
cc=cc,
bcc=bcc,
attachments=attachments,
)
all_rcpts = [to_email]
for extra_field in (cc, bcc):
if extra_field:
all_rcpts.extend([addr.strip() for addr in extra_field.split(",") if addr.strip()])
try:
self._dispatch_smtp(smtp_cfg, mime_msg, all_rcpts)
status = "sent"
except Exception as exc:
status = "failed"
error_msg = str(exc)
att_meta = [{"filename": a.get("filename", "adjunto"), "mime_type": a.get("mime_type", "")} for a in (attachments or [])]
log = self.db.create_email_log(
{
"tracking_id": tracking_id,
"send_type": send_type,
"campaign_id": campaign_id,
"contact_id": contact["id"] if contact else (contact_id or ""),
"sender_id": smtp_cfg["id"],
"from_email": smtp_cfg.get("from_email", ""),
"to_email": to_email,
"to_name": to_name,
"cc": cc,
"bcc": bcc,
"subject": rendered_subject,
"html_body": final_html,
"text_body": rendered_text,
"attachments": att_meta,
"status": status,
"error_message": error_msg,
}
)
return log
def start_campaign_async(self, campaign_id: str, base_url: str = "") -> dict:
"""Launch mass campaign sending in a background thread."""
with self._lock:
if campaign_id in self._active_campaigns:
return {"started": False, "message": "La campaña ya se está enviando."}
campaign = self.db.get_campaign(campaign_id)
if not campaign:
raise ValueError("Campaña no encontrada.")
if not campaign.get("list_id"):
raise ValueError("La campaña no tiene una lista de contactos asignada.")
self._active_campaigns.add(campaign_id)
thread = threading.Thread(
target=self._run_campaign_worker,
args=(campaign_id, base_url),
daemon=True,
)
thread.start()
return {"started": True, "campaign_id": campaign_id}
def _run_campaign_worker(self, campaign_id: str, base_url: str = ""):
try:
campaign = self.db.get_campaign(campaign_id)
if not campaign:
return
contacts = self.db.list_contacts(list_id=campaign["list_id"], status="subscribed")
total = len(contacts)
self.db.update_campaign_progress(
campaign_id,
status="sending",
total_recipients=total,
sent_count=0,
failed_count=0,
)
sent_count = 0
failed_count = 0
throttle_sec = max(0.0, float(campaign.get("throttle_ms") or 0) / 1000.0)
track_opens = bool(campaign.get("track_opens", 1))
for idx, contact in enumerate(contacts):
log = self.send_single_email(
to_email=contact["email"],
subject=campaign["subject"],
html_content=campaign["html_content"],
text_content=campaign.get("text_content", ""),
to_name=f"{contact.get('first_name', '')} {contact.get('last_name', '')}".strip(),
sender_id=campaign.get("sender_id"),
contact_id=contact["id"],
send_type="campaign",
campaign_id=campaign_id,
track_opens=track_opens,
base_url=base_url,
include_unsubscribe=True,
)
if log["status"] in ("sent", "simulated"):
sent_count += 1
else:
failed_count += 1
self.db.update_campaign_progress(
campaign_id,
sent_count=sent_count,
failed_count=failed_count,
)
if throttle_sec > 0 and idx < total - 1:
time.sleep(throttle_sec)
final_status = "sent" if (sent_count > 0 or total == 0) else "failed"
self.db.update_campaign_progress(
campaign_id,
status=final_status,
sent_count=sent_count,
failed_count=failed_count,
sent_at=utc_now(),
)
finally:
with self._lock:
self._active_campaigns.discard(campaign_id)