Email Automation Bot with Scheduling and Attachments

1) Project Overview

What it does:

This Email Automation Bot reads a recipient list (CSV), composes personalized emails (templated message with variables), attaches files per recipient (optional), and sends emails either immediately or at scheduled times. It supports retries on failure, logs each send attempt, and can run continuously to trigger scheduled deliveries.

Real-world use cases:

Sending newsletters or reports with attachments.
Automated invoices, reminders, or confirmations.
Scheduled email campaigns for marketing, onboarding, or system alerts.

Technical goals:

Safe SMTP integration (TLS/SSL, credentials via environment).
Scheduling using a lightweight scheduler.
Robust error handling, retries, and logging.
Templating for personalization and attachment handling.
Test mode using a local debug SMTP server.

2) Key Technologies & Libraries

Python 3.8+
Standard library: smtplib, ssl, email, csv, logging, os, pathlib, threading, time, datetime, mimetypes, traceback
Third-party: schedule (for scheduling tasks). Install with:

pip install schedule

We intentionally use minimal dependencies to keep the project portable. You can optionally add jinja2 for richer templating.

3) Learning Outcomes

Securely connect to SMTP servers using TLS/SSL and credentials via environment variables.
Construct multipart emails with attachments using Python’s email package.
Schedule recurring or one-off tasks using schedule and run them in background threads.
Implement retries (exponential backoff) and structured logging.
Design a reproducible automation pipeline and test it locally safely.

4) Step-by-Step Explanation (high-level)

Project scaffold: create folders data/ and outputs/.
Environment variables: set SMTP_HOST, SMTP_PORT, SMTP_USER, SMTP_PASS, SMTP_USE_TLS (true/false), and optional TEST_MODE=true to use local debug SMTP.
Prepare recipients CSV: data/recipients.csv with columns: email, name, subject, message, attachment, send_time (HH:MM or blank for immediate).
Email composition: build EmailMessage, set headers, add text, attach files if any (detect MIME type), and send through smtplib.SMTP or SMTP_SSL.
Scheduling: schedule daily jobs at send_time; for blank send_time, send immediately in a batch.
Logging & retries: log success/failure to outputs/email_log.csv and retry failed sends.
Local testing: run a debug SMTP server and point SMTP_HOST/PORT to localhost:1025.

5) Full Working and Verified Python Code

Save as email_automation_bot.py. This script is self-contained and includes code to create sample data (recipient CSV and sample attachment) if they don't exist so you can run immediately.

"""
email_automation_bot.py

Email Automation Bot with scheduling and attachments.

Usage:
    1) Create a virtual environment and install schedule:
       pip install schedule

    2) Provide SMTP settings via environment variables:
       - SMTP_HOST (e.g., smtp.gmail.com or localhost for testing)
       - SMTP_PORT (e.g., 587 for TLS, 465 for SSL, or 1025 for local debug server)
       - SMTP_USER (SMTP username or email)
       - SMTP_PASS (SMTP password or app password)
       - SMTP_USE_TLS ("true" or "false") - whether to use STARTTLS (port 587)
       - TEST_MODE ("true"/"false") optional - if true, use debug SMTP server defaults

    3) Run:
       python email_automation_bot.py

    Notes:
      - For local testing without sending real emails, run a debug SMTP server:
        Python 3.7+: Use aiosmtpd installed:
          pip install aiosmtpd
          python -m aiosmtpd -n -l localhost:1025
        Or with older Python (may be removed in some versions):
          python -m smtpd -c DebuggingServer -n localhost:1025
"""

from __future__ import annotations
import os
import csv
import ssl
import time
import logging
import mimetypes
import threading
import traceback
from pathlib import Path
from datetime import datetime
from email.message import EmailMessage
import smtplib

try:
    import schedule
except Exception as e:
    raise RuntimeError("Missing dependency 'schedule'. Install with: pip install schedule") from e

# -------------------- Configuration --------------------
BASE_DIR = Path(__file__).parent.resolve()
DATA_DIR = BASE_DIR / "data"
OUTPUT_DIR = BASE_DIR / "outputs"
DATA_DIR.mkdir(exist_ok=True)
OUTPUT_DIR.mkdir(exist_ok=True)

RECIPIENTS_CSV = DATA_DIR / "recipients.csv"
LOG_CSV = OUTPUT_DIR / "email_log.csv"
SAMPLE_ATTACHMENT = DATA_DIR / "sample_report.txt"

# Email sending config from environment
SMTP_HOST = os.getenv("SMTP_HOST", "localhost")
SMTP_PORT = int(os.getenv("SMTP_PORT", "1025"))
SMTP_USER = os.getenv("SMTP_USER", "")
SMTP_PASS = os.getenv("SMTP_PASS", "")
SMTP_USE_TLS = os.getenv("SMTP_USE_TLS", "false").lower() == "true"
TEST_MODE = os.getenv("TEST_MODE", "false").lower() == "true"

# Retry settings
MAX_RETRIES = 3
RETRY_DELAY_SECONDS = 5

# Logging config
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)-7s | %(message)s",
    handlers=[
        logging.StreamHandler(),
        logging.FileHandler(OUTPUT_DIR / "email_bot.log", encoding="utf-8"),
    ],
)

# -------------------- Utility helpers --------------------
def create_sample_files_if_missing():
    """
    Create a sample recipients.csv and a small attachment so you can test immediately.
    recipients.csv columns:
      email,name,subject,message,attachment,send_time
    send_time format: HH:MM (24h). Blank means send immediately.
    """
    if not SAMPLE_ATTACHMENT.exists():
        SAMPLE_ATTACHMENT.write_text("Sample Report\n\nThis is a sample attachment generated for testing the Email Automation Bot.\n", encoding="utf-8")
        logging.info(f"Created sample attachment: {SAMPLE_ATTACHMENT}")

    if not RECIPIENTS_CSV.exists():
        sample_rows = [
            {
                "email": "test1@example.com",
                "name": "Alice",
                "subject": "Monthly Report - Alice",
                "message": "Hello {{name}},\n\nPlease find attached the monthly report.\n\nRegards,\nAutomated Bot",
                "attachment": str(SAMPLE_ATTACHMENT),
                "send_time": "",
            },
            {
                "email": "test2@example.com",
                "name": "Bob",
                "subject": "Reminder - Bob",
                "message": "Hi {{name}},\nThis is a scheduled reminder.\n\nBest,\nBot",
                "attachment": "",
                "send_time": (datetime.now() + timedelta(minutes=1)).strftime("%H:%M"),
            }
        ]
        with open(RECIPIENTS_CSV, "w", newline='', encoding="utf-8") as f:
            writer = csv.DictWriter(f, fieldnames=["email","name","subject","message","attachment","send_time"])
            writer.writeheader()
            for r in sample_rows:
                writer.writerow(r)
        logging.info(f"Created sample recipients file: {RECIPIENTS_CSV}")

def safe_guess_mime_type(path: Path):
    t, _ = mimetypes.guess_type(str(path))
    if not t:
        return "application/octet-stream"
    return t

def log_send_attempt(row: dict, status: str, error: str = ""):
    """
    Append a send log to outputs/email_log.csv
    """
    fieldnames = ["timestamp","email","subject","send_time_configured","status","error"]
    exists = LOG_CSV.exists()
    with open(LOG_CSV, "a", newline='', encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        if not exists:
            writer.writeheader()
        writer.writerow({
            "timestamp": datetime.now().isoformat(),
            "email": row.get("email",""),
            "subject": row.get("subject",""),
            "send_time_configured": row.get("send_time",""),
            "status": status,
            "error": error[:200] if error else "",
        })

# -------------------- Email composition and send --------------------
def compose_email(row: dict) -> EmailMessage:
    msg = EmailMessage()
    sender = SMTP_USER if SMTP_USER else f"no-reply@{SMTP_HOST}"
    msg["From"] = sender
    msg["To"] = row["email"]
    msg["Subject"] = row["subject"] or "No Subject"

    body = row.get("message", "")
    body = body.replace("{{name}}", row.get("name",""))
    msg.set_content(body)

    attachment_path = row.get("attachment", "").strip()
    if attachment_path:
        p = Path(attachment_path)
        if p.exists() and p.is_file():
            mime_type = safe_guess_mime_type(p)
            maintype, subtype = mime_type.split("/", 1)
            with open(p, "rb") as af:
                data = af.read()
                msg.add_attachment(data, maintype=maintype, subtype=subtype, filename=p.name)
        else:
            logging.warning(f"Attachment not found for {row.get('email')}: {attachment_path}")
    return msg

def send_email_message(msg: EmailMessage, use_tls: bool = SMTP_USE_TLS) -> None:
    host = SMTP_HOST
    port = SMTP_PORT
    if port == 465:
        context = ssl.create_default_context()
        with smtplib.SMTP_SSL(host, port, context=context) as server:
            if SMTP_USER and SMTP_PASS:
                server.login(SMTP_USER, SMTP_PASS)
            server.send_message(msg)
    else:
        with smtplib.SMTP(host, port, timeout=30) as server:
            server.ehlo()
            if use_tls:
                context = ssl.create_default_context()
                server.starttls(context=context)
                server.ehlo()
            if SMTP_USER and SMTP_PASS:
                server.login(SMTP_USER, SMTP_PASS)
            server.send_message(msg)

def send_with_retries(row: dict) -> None:
    tries = 0
    last_error = ""
    while tries <= MAX_RETRIES:
        try:
            msg = compose_email(row)
            send_email_message(msg)
            logging.info(f"Sent email to {row['email']} (subject: {row.get('subject')})")
            log_send_attempt(row, status="SENT", error="")
            return
        except Exception as e:
            last_error = traceback.format_exc()
            tries += 1
            logging.warning(f"Send attempt {tries} failed for {row.get('email')}: {e}")
            if tries > MAX_RETRIES:
                logging.error(f"Failed to send to {row.get('email')} after {MAX_RETRIES} retries.")
                log_send_attempt(row, status="FAILED", error=str(e))
                return
            backoff = RETRY_DELAY_SECONDS * (2 ** (tries - 1))
            logging.info(f"Retrying after {backoff}s...")
            time.sleep(backoff)

# -------------------- CSV reading and scheduling --------------------
def read_recipients(csv_path: Path = RECIPIENTS_CSV) -> list[dict]:
    rows = []
    if not csv_path.exists():
        logging.error(f"Recipients file not found: {csv_path}")
        return rows
    with open(csv_path, newline='', encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for r in reader:
            row = {k.strip(): (v.strip() if isinstance(v, str) else v) for k, v in r.items()}
            rows.append(row)
    return rows

def schedule_jobs(rows: list[dict]):
    immediate = []
    scheduled = []
    for row in rows:
        st = (row.get("send_time") or "").strip()
        if st:
            try:
                datetime.strptime(st, "%H:%M")
                scheduled.append(row)
            except ValueError:
                logging.warning(f"Invalid send_time for {row.get('email')}: '{st}'. Sending immediately.")
                immediate.append(row)
        else:
            immediate.append(row)

    if immediate:
        t = threading.Thread(target=send_batch, args=(immediate,), daemon=True)
        t.start()

    for row in scheduled:
        hhmm = row["send_time"]
        def make_job(r):
            return lambda: send_with_retries(r)
        job = make_job(row)
        schedule.every().day.at(hhmm).do(job)
        logging.info(f"Scheduled daily email to {row.get('email')} at {hhmm}")

def send_batch(rows: list[dict]):
    for row in rows:
        send_with_retries(row)

# -------------------- Scheduler runner --------------------
def run_scheduler_forever():
    logging.info("Scheduler started. Waiting for scheduled jobs...")
    while True:
        schedule.run_pending()
        time.sleep(1)

# -------------------- Main --------------------
from datetime import timedelta

def main():
    logging.info("Starting Email Automation Bot")
    create_sample_files_if_missing()

    if TEST_MODE:
        logging.warning("TEST_MODE enabled — ensure you have a local debug SMTP server running on localhost:1025.")
        global SMTP_HOST, SMTP_PORT, SMTP_USER, SMTP_PASS, SMTP_USE_TLS
        SMTP_HOST = os.getenv("SMTP_HOST", "localhost")
        SMTP_PORT = int(os.getenv("SMTP_PORT", "1025"))
        SMTP_USER = os.getenv("SMTP_USER", "")
        SMTP_PASS = os.getenv("SMTP_PASS", "")
        SMTP_USE_TLS = False

    rows = read_recipients()
    if not rows:
        logging.error("No recipients to process. Exiting.")
        return

    schedule_jobs(rows)

    try:
        run_scheduler_forever()
    except KeyboardInterrupt:
        logging.info("Shutting down Email Automation Bot (KeyboardInterrupt).")

if __name__ == "__main__":
    main()

Notes about the code
• Security: SMTP credentials are taken from environment variables — do not hard-code passwords.
• Test mode: set TEST_MODE=true and run a local debug SMTP server to verify message composition without sending real emails.
• Templating: simple {{name}} placeholder replacement is implemented. You can replace with Jinja2 for richer templates.
• Attachments: file paths in CSV are attached if present and valid; missing attachments are logged but do not stop sending.
• Scheduling: uses schedule.every().day.at("HH:MM").do(...) — scheduled tasks run daily at the configured time.
• Retries: 3 attempts with exponential backoff (5s, 10s, 20s ...). This is configurable.
• Logs: both console and outputs/email_bot.log; send attempts are appended to outputs/email_log.csv.

6) Sample Output or How to Test Locally

Local debug SMTP server (safe testing)
pip install aiosmtpd python -m aiosmtpd -n -l localhost:1025
python -m smtpd -c DebuggingServer -n localhost:1025
Set environment and run the bot in TEST_MODE:
export TEST_MODE=true python email_automation_bot.py
The bot logs successes to outputs/email_log.csv.
Using a real SMTP (e.g., Gmail)
export SMTP_HOST="smtp.gmail.com" export SMTP_PORT="587" export SMTP_USER="your.email@gmail.com" export SMTP_PASS="your_app_password" export SMTP_USE_TLS="true" python email_automation_bot.py
Sample recipients CSV
email,name,subject,message,attachment,send_time test1@example.com,Alice,Monthly Report - Alice,"Hello {{name}},\nPlease find attached the monthly report.",data/sample_report.txt,
Outputs
- outputs/email_bot.log — runtime logs
- outputs/email_log.csv — records of each send attempt with status SENT/FAILED
- If running a local debug SMTP server, you'll see email content printed in that server's terminal.

7) Possible Enhancements

Rich templating — replace simple placeholders with jinja2 templates (HTML emails with both plain text and HTML parts).
Batch parallelism — use thread pool or asyncio for sending large batches concurrently.
OAuth2 for SMTP — implement OAuth2 for providers that require it (e.g., Gmail modern auth).
Web UI / API — expose a small Flask/FastAPI UI to upload recipient CSVs, schedule jobs, and view logs.
Database — replace CSV logs with a SQLite/Postgres DB for robust audit and retries.
Monitoring & alerts — integrate with Prometheus/Alertmanager or send Slack/Email alerts on repeated failures.
Rate limiting — implement per-provider rate limiting and backoff to comply with sending policies.
Attachment generation — generate PDF reports on the fly and attach them dynamically.
Templated A/B testing — send different subject/body variants and record open/clicks.

8) Summary

This project delivers a production-ready Email Automation Bot with scheduling, attachments, robust error handling, and test-friendly configuration. It’s portable, secure, and easy to extend for real-world workflows.

python Topics

python Tutorial