I Automated 100 Cold Emails a Day With Python — Here's the Complete System

┌─────────────┐    ┌──────────────┐    ┌─────────────┐
│  Prospect    │───▶│  Personalize  │───▶│   Send via   │
│  Database    │    │  Engine       │    │   SMTP       │
└─────────────┘    └──────────────┘    └─────────────┘
       │                  │                    │
       ▼                  ▼                    ▼
  CSV/SQLite        Template +              Rate Limiter
  with metadata      Variables             + BCC Track

import csv
import sqlite3
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class Prospect:
    email: str
    first_name: str
    company: str
    role: str
    industry: str
    pain_point: str  # The #1 thing their industry struggles with
    last_contacted: Optional[str] = None
    replied: bool = False
    template_id: Optional[str] = None

class ProspectDB:
    """Simple SQLite prospect manager. No ORM needed."""

    def __init__(self, db_path: str = "prospects.db"):
        self.conn = sqlite3.connect(db_path)
        self.conn.row_factory = sqlite3.Row
        self._init_db()

    def _init_db(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS prospects (
                email TEXT PRIMARY KEY,
                first_name TEXT,
                company TEXT,
                role TEXT,
                industry TEXT,
                pain_point TEXT,
                last_contacted TEXT,
                replied INTEGER DEFAULT 0,
                template_id TEXT,
                bounce INTEGER DEFAULT 0
            )
        """)
        self.conn.commit()

    def add_from_csv(self, csv_path: str) -> int:
        """Bulk import from CSV. Returns count of new prospects."""
        added = 0
        with open(csv_path) as f:
            reader = csv.DictReader(f)
            for row in reader:
                try:
                    self.conn.execute(
                        "INSERT OR IGNORE INTO prospects VALUES (?,?,?,?,?,?,?,?,0,0)",
                        (row['email'], row['first_name'], row['company'],
                         row['role'], row['industry'], row['pain_point'],
                         None, None)
                    )
                    added += 1
                except sqlite3.Error:
                    continue
        self.conn.commit()
        return added

    def get_uncontacted(self, limit: int = 100) -> List[Prospect]:
        """Get prospects we have not emailed yet."""
        rows = self.conn.execute(
            "SELECT * FROM prospects WHERE last_contacted IS NULL AND bounce = 0 LIMIT ?",
            (limit,)
        ).fetchall()
        return [Prospect(**dict(r)) for r in rows]

    def mark_contacted(self, email: str, template_id: str):
        from datetime import datetime
        self.conn.execute(
            "UPDATE prospects SET last_contacted = ?, template_id = ? WHERE email = ?",
            (datetime.now().isoformat(), template_id, email)
        )
        self.conn.commit()

    def mark_bounced(self, email: str):
        self.conn.execute(
            "UPDATE prospects SET bounce = 1 WHERE email = ?", (email,)
        )
        self.conn.commit()

from string import Template
from typing import Dict

class PersonalizationEngine:
    """Rule-based personalization. No AI, no latency, no API costs."""

    INDUSTRY_HOOKS = {
        "saas": "churn is eating your MRR",
        "agency": "client acquisition is feast-or-famine",  
        "ecommerce": "cart abandonment is bleeding revenue",
        "devtools": "developer adoption is a distribution problem",
        "fintech": "compliance overhead slows shipping",
        "healthtech": "HIPAA makes everything 3x harder",
    }

    ROLE_VALUES = {
        "ceo": "save 10+ hours/week on manual tasks",
        "cto": "cut infrastructure costs by 40%",
        "founder": "automate the 80% of work that does not move the needle",
        "vp_engineering": "ship 2x faster without adding headcount",
        "head_of_growth": "automate lead qualification so your team closes more",
    }

    def personalize(self, template: str, prospect: Prospect) -> str:
        """Replace variables with prospect-specific content."""
        variables = {
            "first_name": prospect.first_name,
            "company": prospect.company,
            "role": prospect.role,
            "industry": prospect.industry,
            "pain_point": prospect.pain_point,
            "industry_hook": self.INDUSTRY_HOOKS.get(
                prospect.industry.lower().replace(" ", ""), 
                f"{prospect.industry} has unique automation opportunities"
            ),
            "role_value": self.ROLE_VALUES.get(
                prospect.role.lower().replace(" ", "_"),
                "automate repetitive tasks and reclaim your time"
            ),
        }

        return Template(template).safe_substitute(variables)

    def generate_subject(self, prospect: Prospect) -> str:
        """Generate personalized subject lines. Short = better."""
        templates = [
            f"{prospect.first_name}, quick question about {prospect.company}",
            f"re: {prospect.pain_point} at {prospect.company}",
            f"{prospect.industry} automation for {prospect.company}",
        ]
        idx = hash(prospect.email) % len(templates)
        return templates[idx]

import smtplib
import time
import random
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from datetime import datetime, timedelta
from typing import Dict

class ColdEmailSender:
    """Rate-limited SMTP sender. Respects inbox provider limits."""

    HOURLY_LIMIT = 20
    DAILY_LIMIT = 100
    MIN_DELAY = 45   # seconds between emails
    MAX_DELAY = 120  # random delay ceiling

    def __init__(self, smtp_host: str, smtp_port: int, 
                 username: str, password: str):
        self.smtp_host = smtp_host
        self.smtp_port = smtp_port
        self.username = username
        self.password = password
        self.sent_this_hour = 0
        self.sent_today = 0
        self.hour_start = datetime.now()

    def _check_limits(self) -> bool:
        """Enforce rate limits. Returns False if we should stop."""
        now = datetime.now()

        if now - self.hour_start >= timedelta(hours=1):
            self.sent_this_hour = 0
            self.hour_start = now

        if self.sent_today >= self.DAILY_LIMIT:
            print(f"Daily limit reached ({self.DAILY_LIMIT}). Stopping.")
            return False

        if self.sent_this_hour >= self.HOURLY_LIMIT:
            print("Hourly limit reached. Waiting...")
            sleep_time = 3600 - (now - self.hour_start).seconds
            time.sleep(sleep_time + 30)
            self.sent_this_hour = 0
            self.hour_start = datetime.now()

        return True

    def _send_one(self, to: str, subject: str, body: str) -> bool:
        """Send a single email. Returns True on success."""
        msg = MIMEMultipart()
        msg['From'] = self.username
        msg['To'] = to
        msg['Subject'] = subject
        msg.attach(MIMEText(body, 'plain'))

        try:
            with smtplib.SMTP(self.smtp_host, self.smtp_port) as server:
                server.starttls()
                server.login(self.username, self.password)
                server.send_message(msg)
            return True
        except smtplib.SMTPException as e:
            print(f"SMTP error for {to}: {e}")
            return False

    def send_campaign(self, prospects: list, 
                      body_template: str,
                      engine: PersonalizationEngine) -> dict:
        """Send personalized emails to a list of prospects."""
        stats = {"sent": 0, "failed": 0, "skipped": 0}

        for prospect in prospects:
            if not self._check_limits():
                break

            subject = engine.generate_subject(prospect)
            body = engine.personalize(body_template, prospect)

            if self._send_one(prospect.email, subject, body):
                stats["sent"] += 1
                self.sent_this_hour += 1
                self.sent_today += 1
            else:
                stats["failed"] += 1

            delay = random.uniform(self.MIN_DELAY, self.MAX_DELAY)
            time.sleep(delay)

        return stats

def run_daily_campaign():
    """Run once per day via cron or scheduled task."""
    db = ProspectDB("prospects.db")
    engine = PersonalizationEngine()
    sender = ColdEmailSender(
        smtp_host="smtp.gmail.com",
        smtp_port=587,
        username="you@yourdomain.com",
        password="your_app_password"
    )

    prospects = db.get_uncontacted(limit=100)

    if not prospects:
        print("No new prospects to contact.")
        return

    template = """Hi $first_name,

Noticed $company is in $industry — been doing some work there and $industry_hook.

I built an automation that $role_value. Took about 2 hours to set up, has been running for 3 months straight.

Would it be useful for $company? Happy to share how it works.

Best,
Alex"""

    stats = sender.send_campaign(prospects, template, engine)

    for p in prospects[:stats["sent"]]:
        db.mark_contacted(p.email, "v1_outreach")

    print(f"Campaign complete: {stats}")

if __name__ == "__main__":
    run_daily_campaign()

import logging
from collections import Counter

class CampaignMonitor:
    """Track deliverability issues before they tank your domain."""

    def __init__(self):
        self.bounces = Counter()
        self.complaints = 0
        self.logger = logging.getLogger("campaign")

    def check_health(self) -> bool:
        """Returns False if you should STOP sending."""
        total = sum(self.bounces.values())
        if total == 0:
            return True

        bounce_rate = self.bounces.get("hard", 0) / total
        if bounce_rate > 0.05:
            self.logger.error(
                f"Hard bounce rate {bounce_rate:.1%} exceeds 5%. STOP SENDING."
            )
            return False

        if self.complaints > 3:
            self.logger.error(
                f"{self.complaints} spam complaints. Review your list."
            )
            return False

        return True