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)
