How to run (quick)

Create a folder and paste files as above.

python -m venv venv && source venv/bin/activate (or venvScriptsactivate on Windows)

pip install -r requirements.txt

Copy .env.example → .env and fill SMTP settings (Mailtrap or Gmail app password)

For local tests, Mailtrap is easiest; or use Gmail with an app password and allow SMTP.

python app.py

Open http://localhost:5000 and submit a test lead.

Visit tracking links printed in console or use the track/open/<id> and track/click/<id> endpoints to simulate opens/clicks.

Admin campaign example (curl):

curl -X POST http://localhost:5000/admin/send_campaign
-H "Content-Type: application/json"
-d '{"q":"all","subject":"Hello from curl","body":"<p>Campaign body</p>"}'

Where to extend (next steps)

Replace smtplib with SendGrid/Mailgun SDK for scalable sends.

Use a message queue (Redis + RQ/Celery) for background tasks.

Persist events more robustly (email provider webhooks for real opens/clicks).

Add templates, A/B testing, analytics dashboard (Plotly/Chart.js).

GDPR/unsubscribe management and double opt-in flows.

Use real tracking pixel (1x1 image) to record opens and unique user agents/IPs.

8) app.py — main Flask app
import os
from flask import Flask, render_template_string, request, redirect, jsonify, send_file
from models import db, Lead
from dotenv import load_dotenv
from email_utils import render_template_string as render_tpl, send_email
from tasks import start_scheduler, send_welcome
import csv
from io import StringIO
from datetime import datetime

load_dotenv()

def create_app():
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = os.environ.get("DATABASE_URL", "sqlite:///marketing.db")
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.secret_key = os.environ.get("SECRET_KEY", "dev")
db.init_app(app)

@app.before_first_request
def init_db():
db.create_all()
# start scheduler
start_scheduler(app)

@app.route("/")
def home():
tpl = open("templates/lead_form.html").read()
return render_template_string(tpl)

@app.route("/leads", methods=["POST"])
def create_lead():
email = request.form.get("email")
name = request.form.get("name")
if not email:
return "Email required", 400
existing = Lead.query.filter_by(email=email).first()
if existing:
return "Already subscribed", 400
lead = Lead(email=email, name=name)
db.session.add(lead)
db.session.commit()

# Send welcome email (sync for simplicity) — can be queued/backgrounded
tpl = open("templates/welcome_email.html").read()
tracking_link = f"http://localhost:5000/track/open/{lead.id}"
html = render_tpl(tpl, name=name or "friend", tracking_link=tracking_link)
try:
send_email(lead.email, "Welcome!", html)
lead.last_email_sent = datetime.utcnow()
db.session.commit()
except Exception as e:
print("Email send failed:", e)

return redirect("/thanks")

@app.route("/thanks")
def thanks():
return "<h3>Thanks! Check your inbox (or spam) for a welcome email.</h3>"

# Simulated tracking endpoints
@app.route("/track/open/<int:lead_id>")
def track_open(lead_id):
lead = Lead.query.get(lead_id)
if lead:
lead.opened = True
lead.add_tag("opened")
db.session.commit()
# redirect to a real landing page; for demo just show text
return f"Tracked open for {lead.email}. Thank you!"
return "Not found", 404

@app.route("/track/click/<int:lead_id>")
def track_click(lead_id):
lead = Lead.query.get(lead_id)
if lead:
lead.clicked = True
lead.add_tag("clicked")
lead.stage = "nurtured"
db.session.commit()
return f"Tracked click for {lead.email}"
return "Not found", 404

# Basic segmentation API
@app.route("/api/segment")
def segment():
# simple examples: ?q=opened, ?q=nurtured, ?q=tag:clicked
q = request.args.get("q", "")
if q == "opened":
leads = Lead.query.filter_by(opened=True).all()
elif q == "nurtured":
leads = Lead.query.filter_by(stage="nurtured").all()
elif q.startswith("tag:"):
t = q.split(":",1)[1]
leads = Lead.query.filter(Lead.tags.like(f"%{t}%")).all()
else:
leads = Lead.query.all()
return jsonify([{"id": l.id, "email": l.email, "name": l.name, "stage": l.stage, "tags": l.tags} for l in leads])

# export CSV of leads
@app.route("/export/leads.csv")
def export_leads():
leads = Lead.query.all()
si = StringIO()
cw = csv.writer(si)
cw.writerow(["id","email","name","created_at","stage","tags","opened","clicked"])
for l in leads:
cw.writerow([l.id, l.email, l.name, l.created_at.isoformat(), l.stage, l.tags or "", l.opened, l.clicked])
output = si.getvalue()
return (output, 200, {
"Content-Type": "text/csv",
"Content-Disposition": "attachment; filename=leads.csv"
})

# small admin to trigger a manual campaign to a segment
@app.route("/admin/send_campaign", methods=["POST"])
def send_campaign():
data = request.json or {}
q = data.get("q", "all") # e.g., "opened", "all", "tag:clicked"
subject = data.get("subject", "A campaign message")
body = data.get("body", "<p>Hello from campaign</p>")
if q == "all":
leads = Lead.query.all()
elif q == "opened":
leads = Lead.query.filter_by(opened=True).all()
elif q.startswith("tag:"):
t = q.split(":",1)[1]
leads = Lead.query.filter(Lead.tags.like(f"%{t}%")).all()
else:
leads = Lead.query.all()

sent = 0
for l in leads:
try:
send_email(l.email, subject, body)
sent += 1
except Exception as e:
print("Failed to send to", l.email, e)
return jsonify({"sent": sent})

return app

if __name__ == "__main__":
app = create_app()
app.run(debug=True)

6) templates/welcome_email.html
<!doctype html>
<html>
<body>
<h1>Welcome, {{ name }}!</h1>
<p>Thanks for signing up. We're glad to have you.</p>
<p><a href="{{ tracking_link }}">Click here to visit and track</a></p>
<p>— Your Company</p>
</body>
</html>

7) tasks.py (scheduler + drip)
from apscheduler.schedulers.background import BackgroundScheduler
from datetime import datetime, timedelta
from models import db, Lead
from email_utils import render_template_string, send_email
from jinja2 import Template
from sqlalchemy import select
import os

scheduler = BackgroundScheduler()

def send_welcome(lead: Lead, app):
with app.app_context():
tpl = open("templates/welcome_email.html").read()
tracking_link = f"http://localhost:5000/track/open/{lead.id}"
html = render_template_string(tpl, name=lead.name or "friend", tracking_link=tracking_link)
subject = "Welcome!"
send_email(lead.email, subject, html)
lead.last_email_sent = datetime.utcnow()
db.session.commit()
print(f"Welcome email sent to {lead.email}")

def weekly_drip_job(app):
with app.app_context():
# Example: send drip to leads in stage 'new' who didn't open yet
cutoff = datetime.utcnow() - timedelta(days=3)
leads = Lead.query.filter(Lead.stage=="new").all()
for lead in leads:
# simple rule: send second email if last_email_sent older than 2 days
if not lead.last_email_sent or (lead.last_email_sent < cutoff):
# create a simple email
html = f"<p>Hey {lead.name or 'there'}, here's another helpful resource.</p><p><a href='/track/open/{lead.id}'>Read more</a></p>"
send_email(lead.email, "More resources", html)
lead.last_email_sent = datetime.utcnow()
db.session.commit()
print("Drip sent to", lead.email)

def start_scheduler(app):
# send weekly drip every 1 minute for demo (change to hours/days in prod)
scheduler.add_job(lambda: weekly_drip_job(app), 'interval', minutes=1, id='drip_job', replace_existing=True)
scheduler.start()


Note: for demo we run the drip every 1 minute. In real life use days=7 or cron schedules.

3) models.py
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime

db = SQLAlchemy()

class Lead(db.Model):
__tablename__ = "leads"
id = db.Column(db.Integer, primary_key=True)
email = db.Column(db.String(255), nullable=False, unique=True)
name = db.Column(db.String(255))
created_at = db.Column(db.DateTime, default=datetime.utcnow)
stage = db.Column(db.String(50), default="new") # new, nurtured, customer
tags = db.Column(db.String(255)) # comma-separated tags
last_email_sent = db.Column(db.DateTime, nullable=True)
opened = db.Column(db.Boolean, default=False) # simulated
clicked = db.Column(db.Boolean, default=False) # simulated

def add_tag(self, t):
ts = set((self.tags or "").split(",")) if self.tags else set()
ts.add(t)
self.tags = ",".join(x for x in ts if x)

4) email_utils.py
import os
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from jinja2 import Template
from dotenv import load_dotenv

load_dotenv()

SMTP_HOST = os.environ.get("SMTP_HOST")
SMTP_PORT = int(os.environ.get("SMTP_PORT", 587))
SMTP_USER = os.environ.get("SMTP_USER")
SMTP_PASS = os.environ.get("SMTP_PASS")
FROM_EMAIL = os.environ.get("FROM_EMAIL")

def render_template_string(template_str: str, **ctx):
return Template(template_str).render(**ctx)

def send_email(to_email: str, subject: str, html_body: str, plain_body: str = None):
msg = MIMEMultipart("alternative")
msg["Subject"] = subject
msg["From"] = FROM_EMAIL
msg["To"] = to_email

part1 = MIMEText(plain_body or "Please view HTML email", "plain")
part2 = MIMEText(html_body, "html")
msg.attach(part1)
msg.attach(part2)

# send via SMTP
with smtplib.SMTP(SMTP_HOST, SMTP_PORT) as server:
server.ehlo()
server.starttls()
server.login(SMTP_USER, SMTP_PASS)
server.sendmail(FROM_EMAIL, [to_email], msg.as_string())

5) templates/lead_form.html
<!doctype html>
<html>
<head><meta charset="utf-8"><title>Sign up</title></head>
<body>
<h1>Join our mailing list</h1>
<form action="/leads" method="post">
<label>Name: <input type="text" name="name"/></label><br/>
<label>Email: <input type="email" name="email" required/></label><br/>
<button type="submit">Subscribe</button>
</form>
</body>
</html>

It implements a simple flow:

Collect leads via a web form

Store leads in SQLite (SQLAlchemy)

Send a welcome email immediately

Run a scheduled drip/nurture campaign (APScheduler)

Track basic opens/clicks (simulated) and basic segmentation

Export leads as CSV

Everything below runs locally. No external paid services required (you can swap in SendGrid/Mailgun later).
Project overview (files)
marketing-automation-sample/
├─ requirements.txt
├─ .env.example
├─ app.py # Flask app + routes
├─ models.py # SQLAlchemy models
├─ email_utils.py # send_email helper (smtplib)
├─ tasks.py # scheduled drip tasks (APScheduler)
├─ templates/
│ ├─ lead_form.html
│ └─ welcome_email.html
└─ README.md # run instructions (below)

1) requirements.txt
Flask==2.3.3
SQLAlchemy==2.0.22
Flask-SQLAlchemy==3.0.3
APScheduler==3.10.1
python-dotenv==1.0.0
Jinja2==3.1.2

2) .env.example
# copy to .env and edit
FLASK_ENV=development
SECRET_KEY=replace_with_random
DATABASE_URL=sqlite:///marketing.db

# SMTP settings - use Gmail app password or Mailtrap for local testing
SMTP_HOST=smtp.gmail.com
SMTP_PORT=587
SMTP_USER=youremail@gmail.com
SMTP_PASS=your_smtp_password_or_app_pass
FROM_EMAIL=Your Company <youremail@gmail.com>