""" Mailjet Email Service Module Handles sending emails via Mailjet API with deliverability improvements """ import base64 import hashlib import time import logging import asyncio from typing import List, Optional, Dict, Any from pathlib import Path from mailjet_rest import Client from app.core.settings import settings logger = logging.getLogger(__name__) # Assets folder for inlined images ASSETS_FOLDER = Path("assets") class MailjetEmailService: """Service for sending emails via Mailjet API with spam prevention features""" def __init__(self): self.api_key = settings.MAILJET_API_KEY self.api_secret = settings.MAILJET_SECRET_KEY self.from_email = settings.SMTP_FROM self.from_name = settings.SMTP_FROM_NAME self.client = Client(auth=(self.api_key, self.api_secret), version='v3.1') def _get_inlined_attachments(self) -> List[Dict[str, str]]: """Prepare inlined attachments (logo, etc.) from assets folder""" inlined_attachments = [] # Common images to inline if they exist images = [ ("email_marketing.png", "logo", "image/png"), ("image1.jpg", "image1", "image/jpeg"), ("image2.png", "image2", "image/png") ] for filename, cid, content_type in images: file_path = ASSETS_FOLDER / filename if file_path.exists(): try: with open(file_path, 'rb') as f: img_data = f.read() b64_content = base64.b64encode(img_data).decode('utf-8') inlined_attachments.append({ "ContentType": content_type, "Filename": filename, "ContentID": cid, "Base64Content": b64_content }) except Exception as e: logger.error(f"Failed to process image {filename}: {e}") return inlined_attachments def _generate_custom_id(self, recipient: str) -> str: """Generate a unique CustomID for tracking""" unique_id = hashlib.md5(f"{recipient}{time.time()}".encode()).hexdigest()[:12] return f"email-mkt-{unique_id}" async def send_email( self, recipient: str, subject: str, body: str, is_html: bool = False ) -> bool: """ Send an email to a single recipient with deliverability headers Args: recipient: Email address of recipient subject: Subject line of email body: Body content of email is_html: Whether body is HTML content Returns: bool: True if email sent successfully, False otherwise """ # Create text fallback if body is HTML text_part = body if not is_html else f"{subject}\n\nPlease view this message in a modern email client to see the full content." html_part = body if is_html else None # Build message payload message_payload = { "From": { "Email": self.from_email, "Name": self.from_name }, "ReplyTo": { "Email": self.from_email, "Name": self.from_name }, "To": [ { "Email": recipient, "Name": recipient.split('@')[0] } ], "Subject": subject, "TextPart": text_part, "CustomID": self._generate_custom_id(recipient), "Headers": { "List-Unsubscribe": f"" } } if html_part: message_payload["HTMLPart"] = html_part # Add inlined attachments if any inlined = self._get_inlined_attachments() if inlined: message_payload["InlinedAttachments"] = inlined data = {'Messages': [message_payload]} try: loop = asyncio.get_event_loop() result = await loop.run_in_executor( None, lambda: self.client.send.create(data=data) ) if result.status_code == 200: logger.info(f"Email sent successfully to {recipient} via Mailjet (Improved)") return True else: logger.error(f"Mailjet error ({result.status_code}): {result.json()}") return False except Exception as e: logger.error(f"Failed to send email to {recipient} via Mailjet: {str(e)}") return False async def send_bulk_email( self, recipients: List[str], subject: str, body: str, is_html: bool = False ) -> dict: """ Send email to multiple recipients with deliverability headers Args: recipients: List of recipient email addresses subject: Subject line of email body: Body content of email is_html: Whether body is HTML content Returns: dict: Dictionary with success and failure counts """ results = { 'successful': 0, 'failed': 0, 'failed_emails': [] } inlined = self._get_inlined_attachments() # Mailjet allows up to 50 messages per bulk request chunk_size = 40 for i in range(0, len(recipients), chunk_size): chunk = recipients[i:i + chunk_size] messages = [] for recipient in chunk: text_part = body if not is_html else f"{subject}\n\nPlease view this message in a modern email client." msg = { "From": { "Email": self.from_email, "Name": self.from_name }, "ReplyTo": { "Email": self.from_email, "Name": self.from_name }, "To": [ { "Email": recipient, "Name": recipient.split('@')[0] } ], "Subject": subject, "TextPart": text_part, "CustomID": self._generate_custom_id(recipient), "Headers": { "List-Unsubscribe": f"" } } if is_html: msg["HTMLPart"] = body if inlined: msg["InlinedAttachments"] = inlined messages.append(msg) data = {'Messages': messages} try: loop = asyncio.get_event_loop() result = await loop.run_in_executor( None, lambda: self.client.send.create(data=data) ) if result.status_code == 200: payload = result.json() for msg_result in payload.get('Messages', []): if msg_result.get('Status') == 'success': results['successful'] += 1 else: results['failed'] += 1 else: logger.error(f"Mailjet bulk error ({result.status_code}): {result.json()}") results['failed'] += len(chunk) results['failed_emails'].extend(chunk) except Exception as e: logger.error(f"Failed to send bulk emails via Mailjet: {str(e)}") results['failed'] += len(chunk) results['failed_emails'].extend(chunk) return results # Create singleton instance mailjet_service = MailjetEmailService()