from fastapi import FastAPI, UploadFile, File, Form, HTTPException, BackgroundTasks, Request from fastapi.responses import HTMLResponse, JSONResponse, Response from fastapi.middleware.cors import CORSMiddleware import pandas as pd import smtplib import re import json import os import io import uuid import base64 import time import asyncio import httpx from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from datetime import datetime, date from urllib.parse import quote from typing import Optional, List from pydantic import BaseModel # ───────────────────────────────────────────────────────────────────────────── app = FastAPI( title="EmailBlast Pro API", description="Production email automation — per-row branding, HTML templates, open tracking, round-robin SMTP, warm-up mode", version="2.0.0", docs_url="/swagger", # default Swagger moved to /swagger redoc_url="/redoc", ) ALLOWED_ORIGINS = [ "https://aero-woad.vercel.app", # Vercel frontend "http://localhost:3000", # local dev "http://localhost:5500", # VS Code Live Server ] app.add_middleware( CORSMiddleware, allow_origins=ALLOWED_ORIGINS, allow_origin_regex=r"https://.*\.hf\.space", # allow any HF Space (self-calls) allow_methods=["*"], allow_headers=["*"], ) # ── In-memory stores ────────────────────────────────────────────────────────── jobs: dict = {} # job_id -> job dict open_events: dict = {} # job_id -> list of {email, ts} email_regex = re.compile(r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$") # ───────────────────────────────────────────────────────────────────────────── # TEMPLATES # ───────────────────────────────────────────────────────────────────────────── TEMPLATES = { "plain": lambda ctx: ( f"Hello,\n\n" f"Thank you for your interest in {ctx['from_name']}. {ctx['body_extra']}\n\n" f"Best regards,\n{ctx['from_name']}\n{ctx['sender_address']}\n\n" f"---\nUnsubscribe: {ctx['unsub']}" ), "promo": lambda ctx: f"""

{ctx['from_name']}

{ctx['subject']}

Hello,

{ctx['body_extra']}

{f'{ctx["cta_text"]}' if ctx.get("cta_url") else ''}

Best regards,
{ctx['from_name']}
{ctx['sender_address']}

Unsubscribe · {ctx['sender_address']}
""", "newsletter": lambda ctx: f"""

{ctx['from_name']}

Newsletter · {datetime.now().strftime('%B %Y')}

Hello,

{ctx['body_extra']}

Unsubscribe · {ctx['sender_address']}
""", } # ───────────────────────────────────────────────────────────────────────────── # WARM-UP SCHEDULE (emails per session based on day number) # ───────────────────────────────────────────────────────────────────────────── WARMUP_SCHEDULE = [5, 10, 20, 30, 50, 75, 100, 150, 200, 300, 400] def warmup_limit(day: int) -> int: idx = max(0, min(day - 1, len(WARMUP_SCHEDULE) - 1)) return WARMUP_SCHEDULE[idx] # ───────────────────────────────────────────────────────────────────────────── # CORE BLAST RUNNER # ───────────────────────────────────────────────────────────────────────────── def run_blast( job_id: str, df: pd.DataFrame, smtp_accounts: list, # [{"user":..., "pass":...}, ...] subject: str, default_from_name: str, sender_address: str, body_extra: str, max_emails: int, sleep_interval: float, template_name: str, cta_url: str, cta_text: str, warmup_day: int, webhook_url: str, base_url: str, ): job = jobs[job_id] job["status"] = "running" sent_set = set() # Apply warm-up cap if enabled if warmup_day > 0: cap = warmup_limit(warmup_day) max_emails = min(max_emails, cap) job["log"].append(f"🌡 Warm-up day {warmup_day} — capped at {cap} emails") # Round-robin SMTP connection pool servers = [] for acc in smtp_accounts: try: srv = smtplib.SMTP("smtp.gmail.com", 587) srv.starttls() srv.login(acc["user"], acc["pass"]) servers.append({"server": srv, "user": acc["user"]}) job["log"].append(f"✓ SMTP connected: {acc['user']}") except Exception as e: job["log"].append(f"✗ SMTP failed {acc['user']}: {e}") if not servers: job["status"] = "failed" job["log"].append("No SMTP connections available — aborting.") return smtp_idx = 0 sent_count = 0 for _, row in df.iterrows(): if sent_count >= max_emails: job["log"].append("Reached limit — stopping.") break to_email = row.get("Email", "") if pd.isna(to_email) or not str(to_email).strip(): continue to_email = str(to_email).strip() if not email_regex.match(to_email): continue if to_email in sent_set: continue # Per-row overrides from_name = row.get("FromName", default_from_name) if pd.isna(from_name) or not str(from_name).strip(): from_name = default_from_name from_name = str(from_name).strip() reply_to = row.get("ReplyTo", None) if pd.isna(reply_to) or not str(reply_to).strip(): reply_to = None else: reply_to = str(reply_to).strip() row_body = row.get("BodyExtra", body_extra) if pd.isna(row_body) or not str(row_body).strip(): row_body = body_extra row_cta_url = row.get("CTAUrl", cta_url) if pd.isna(row_cta_url) or not str(row_cta_url).strip(): row_cta_url = cta_url # Tracking pixel URL pixel = f"{base_url}/track/open/{job_id}/{quote(to_email)}" unsub = f"{base_url}/unsubscribe/{job_id}/{quote(to_email)}" ctx = { "from_name": from_name, "subject": subject, "body_extra": row_body, "sender_address": sender_address, "cta_url": row_cta_url, "cta_text": cta_text, "pixel": pixel, "unsub": unsub, } tmpl_fn = TEMPLATES.get(template_name, TEMPLATES["plain"]) is_html = template_name != "plain" if is_html: body = tmpl_fn(ctx) mime_type = "html" else: body = tmpl_fn(ctx) mime_type = "plain" # Round-robin SMTP acc = servers[smtp_idx % len(servers)] smtp_idx += 1 gmail_user = acc["user"] msg = MIMEMultipart("alternative") msg["From"] = f"{from_name} <{gmail_user}>" msg["To"] = to_email msg["Subject"] = subject if reply_to: msg["Reply-To"] = reply_to msg.attach(MIMEText(body, mime_type)) try: acc["server"].sendmail(gmail_user, to_email, msg.as_string()) sent_count += 1 sent_set.add(to_email) job["sent"] = sent_count job["log"].append( f"[{datetime.now().strftime('%H:%M:%S')}] ✓ {to_email} via {gmail_user.split('@')[0]}… as '{from_name}'" ) except Exception as e: job["errors"] += 1 job["log"].append(f"[{datetime.now().strftime('%H:%M:%S')}] ✗ {to_email} — {e}") time.sleep(sleep_interval) for acc in servers: try: acc["server"].quit() except Exception: pass job["status"] = "done" job["sent"] = sent_count job["log"].append(f"✅ Done — {sent_count} sent, {job['errors']} errors.") # Fire webhook if configured if webhook_url and webhook_url.startswith("http"): try: import urllib.request data = json.dumps({ "job_id": job_id, "sent": sent_count, "errors": job["errors"], "status": "done" }).encode() req = urllib.request.Request( webhook_url, data=data, headers={"Content-Type": "application/json"}, method="POST" ) urllib.request.urlopen(req, timeout=5) job["log"].append(f"📡 Webhook fired → {webhook_url}") except Exception as e: job["log"].append(f"⚠ Webhook failed: {e}") # ───────────────────────────────────────────────────────────────────────────── # ROUTES # ───────────────────────────────────────────────────────────────────────────── @app.get("/", response_class=HTMLResponse) async def root(): with open("index.html") as f: return f.read() @app.get("/health") async def health(): return { "ok": True, "message": "EmailBlast Pro API is running.", "version": "2.0.0", "features": ["html-templates", "open-tracking", "round-robin-smtp", "warmup-mode", "webhooks", "per-row-branding"], "endpoints": ["/health", "/send", "/status/{job_id}", "/track/open/{job_id}/{email}", "/analytics/{job_id}", "/docs", "/swagger"] } from fastapi.responses import RedirectResponse @app.get("/docs", include_in_schema=False) async def docs_redirect(): """Redirect /docs to the Vercel documentation page.""" return RedirectResponse(url="https://aero-woad.vercel.app/email.html") # ── Read HF Space secrets once at startup ──────────────────────────────────── _DEFAULT_GMAIL_USER = os.getenv("GMAIL_USER", "") _DEFAULT_GMAIL_PASS = os.getenv("GMAIL_PASS", "") @app.get("/config") async def config(): """Returns non-sensitive config so the UI can pre-fill fields.""" return { "gmail_user": _DEFAULT_GMAIL_USER, "has_pass": bool(_DEFAULT_GMAIL_PASS), } @app.post("/send") async def send_emails( request: Request, background_tasks: BackgroundTasks, file: UploadFile = File(...), # SMTP — falls back to HF Space secrets if left blank gmail_user: str = Form(default=""), gmail_pass: str = Form(default=""), # Optional extra SMTP accounts for round-robin (JSON array string) extra_accounts: str = Form("[]"), # Campaign settings subject: str = Form("Hello from Your Business"), default_from_name: str = Form("Your Business"), sender_address: str = Form("Your Business, City, Country"), body_extra: str = Form("We'd love to share our latest updates with you."), max_emails: int = Form(50), sleep_interval: float = Form(5.0), template_name: str = Form("plain"), # plain | promo | newsletter cta_url: str = Form(""), cta_text: str = Form("Learn More"), warmup_day: int = Form(0), # 0 = disabled, 1-11+ = day in schedule webhook_url: str = Form(""), ): content = await file.read() try: if file.filename.endswith(".csv"): df = pd.read_csv(io.BytesIO(content)) else: df = pd.read_excel(io.BytesIO(content)) except Exception as e: raise HTTPException(status_code=400, detail=f"Could not parse file: {e}") if "Email" not in df.columns: raise HTTPException(status_code=400, detail=f"File must have 'Email' column. Found: {list(df.columns)}") # Fall back to HF Space secrets if the UI left the fields blank if not gmail_user: gmail_user = _DEFAULT_GMAIL_USER if not gmail_pass: gmail_pass = _DEFAULT_GMAIL_PASS if not gmail_user or not gmail_pass: raise HTTPException(status_code=400, detail="No Gmail credentials provided and no GMAIL_USER/GMAIL_PASS secrets configured.") # Build SMTP list smtp_accounts = [{"user": gmail_user, "pass": gmail_pass}] try: extras = json.loads(extra_accounts) smtp_accounts.extend(extras) except Exception: pass base_url = str(request.base_url).rstrip("/") job_id = str(uuid.uuid4())[:8] jobs[job_id] = {"status": "queued", "sent": 0, "total": len(df), "errors": 0, "log": []} open_events[job_id] = [] background_tasks.add_task( run_blast, job_id, df, smtp_accounts, subject, default_from_name, sender_address, body_extra, max_emails, sleep_interval, template_name, cta_url, cta_text, warmup_day, webhook_url, base_url ) return { "job_id": job_id, "total_rows": len(df), "smtp_accounts": len(smtp_accounts), "template": template_name, "warmup_day": warmup_day, "message": f"Job queued — poll /status/{job_id}" } @app.get("/status/{job_id}") async def job_status(job_id: str): if job_id not in jobs: raise HTTPException(status_code=404, detail="Job not found") j = jobs[job_id] opens = len(open_events.get(job_id, [])) open_rate = round((opens / j["sent"] * 100), 1) if j["sent"] > 0 else 0 return { "job_id": job_id, "status": j["status"], "sent": j["sent"], "total": j["total"], "errors": j["errors"], "opens": opens, "open_rate": f"{open_rate}%", "log": j["log"][-80:], } @app.get("/analytics/{job_id}") async def analytics(job_id: str): if job_id not in jobs: raise HTTPException(status_code=404, detail="Job not found") j = jobs[job_id] evts = open_events.get(job_id, []) return { "job_id": job_id, "sent": j["sent"], "opens": len(evts), "open_rate": f"{round(len(evts)/j['sent']*100,1) if j['sent'] else 0}%", "open_events": evts[-100:], } # ── Tracking pixel ──────────────────────────────────────────────────────────── PIXEL_GIF = base64.b64decode( "R0lGODlhAQABAIAAAAAAAP///yH5BAEAAAAALAAAAAABAAEAAAIBRAA7" ) @app.get("/track/open/{job_id}/{email}") async def track_open(job_id: str, email: str): if job_id in open_events: open_events[job_id].append({"email": email, "ts": datetime.now().isoformat()}) if job_id in jobs: jobs[job_id].setdefault("opens", 0) jobs[job_id]["opens"] = len(open_events[job_id]) return Response(content=PIXEL_GIF, media_type="image/gif") # ── Unsubscribe page ────────────────────────────────────────────────────────── @app.get("/unsubscribe/{job_id}/{email}", response_class=HTMLResponse) async def unsubscribe(job_id: str, email: str): return f"""
✓

Unsubscribed

{email} has been removed from this list.

"""