JBON_DATA

AUTOMATION

Email Automation Done Right

Email remains a core business communication channel. Automating email processing—reading, parsing, responding, forwarding—saves countless hours but requires careful design.

Common Use Cases

  • Order confirmations from suppliers
  • Report distribution and notifications
  • Document extraction from attachments
  • Ticket creation from support emails
  • Approval workflows

Architecture Options

1. Polling Approach

import imaplib
import email
from email.header import decode_header

def check_inbox():
    """Poll mailbox for new messages."""
    mail = imaplib.IMAP4_SSL("imap.server.com")
    mail.login(username, password)
    mail.select("inbox")
    
    status, messages = mail.search(None, "UNSEEN")
    
    for num in messages[0].split():
        status, data = mail.fetch(num, "(RFC822)")
        msg = email.message_from_bytes(data[0][1])
        
        process_email(msg)
        
        # Mark as processed
        mail.store(num, '+FLAGS', '\\Seen')
    
    mail.logout()

2. Event-Driven (Microsoft Graph)

from azure.identity import DefaultAzureCredential
from msgraph import GraphServiceClient

async def process_new_emails():
    """Using webhooks for real-time processing."""
    credential = DefaultAzureCredential()
    client = GraphServiceClient(credential)
    
    messages = await client.me.messages.get()
    
    for message in messages.value:
        await handle_message(message)
        await message.mark_as_read()

Parsing Email Content

def extract_order_details(email_body: str) -> dict:
    """Extract structured data from email text."""
    import re
    
    patterns = {
        'order_number': r'Order\s*#?\s*(\d+)',
        'total': r'Total:\s*\$?([\d,]+\.?\d*)',
        'date': r'Date:\s*(\d{1,2}/\d{1,2}/\d{4})'
    }
    
    result = {}
    for field, pattern in patterns.items():
        match = re.search(pattern, email_body, re.IGNORECASE)
        if match:
            result[field] = match.group(1)
    
    return result

Attachment Handling

def save_attachments(msg, output_dir):
    """Extract and save email attachments."""
    for part in msg.walk():
        if part.get_content_maintype() == 'multipart':
            continue
        
        filename = part.get_filename()
        if filename:
            filepath = Path(output_dir) / filename
            with open(filepath, 'wb') as f:
                f.write(part.get_payload(decode=True))
            
            yield filepath

Error Handling

  • Connection failures: Retry with backoff
  • Parse errors: Route to exception queue
  • Duplicate processing: Idempotency checks
  • Rate limits: Throttling mechanisms

Security Considerations

  1. Use OAuth instead of passwords where possible
  2. Validate sender addresses for critical workflows
  3. Scan attachments before processing
  4. Log all actions for audit trails
  5. Encrypt credentials at rest

Monitoring

Track these metrics:

  • Emails processed per hour
  • Parse success rate
  • Average processing latency
  • Exception queue depth

Email automation is deceptively complex. Start simple and add sophistication based on real needs.

← Back to Blog