#!/usr/bin/env python3
"""
Ghost Agency Mailer - Invia email via Brevo API (v3).
Non richiede IP whitelist come SMTP.
Salva stato in SQLite, rate-limit 25/h con jitter, tracking completo.
"""
import csv
import os
import requests
import sqlite3
import random
import time
from datetime import datetime
from pathlib import Path
# Config
BREVO_API_KEY = os.environ.get("BREVO_API_KEY")
FROM_EMAIL = os.environ.get("FROM_EMAIL", "voipamraj@gmail.com")
FROM_NAME = "Amraj Singh"
BASE_URL = os.environ.get("BASE_URL", "https://mockup.svoraj.it")
DB_PATH = Path("leads.db")
CSV_PATH = Path("leads_raw.csv")
SEND_PER_HOUR = 25
JITTER_MIN, JITTER_MAX = 60, 180 # 1-3 minuti tra email
if not BREVO_API_KEY:
print("❌ BREVO_API_KEY deve essere impostata come variabile d'ambiente")
print(" Ottienila da: https://app.brevo.com/settings/keys/api")
exit(1)
BREVO_API_URL = "https://api.brevo.com/v3/smtp/email"
def init_db():
con = sqlite3.connect(DB_PATH)
con.execute("""
CREATE TABLE IF NOT EXISTS leads (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT, category TEXT, phone TEXT, email TEXT,
city TEXT, maps_url TEXT, mockup_url TEXT,
status TEXT DEFAULT 'new',
sent_at TEXT, followed_up_at TEXT, replied_at TEXT,
opened_at TEXT, clicked_at TEXT, bounced_at TEXT,
notes TEXT
)
""")
con.execute("CREATE INDEX IF NOT EXISTS idx_email ON leads(email)")
con.commit()
return con
def load_csv(con):
"""Carica lead da CSV nel DB se non presenti"""
seen = {row[0] for row in con.execute("SELECT email FROM leads WHERE email IS NOT NULL")}
if not CSV_PATH.exists():
print(f"⚠️ {CSV_PATH} non trovato")
return
with open(CSV_PATH, newline="", encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
email = (row.get("email") or "").strip().lower()
if not email or email in seen:
continue
if any(x in email for x in ("example.com", "test@", "noreply", "no-reply")):
continue
name = row.get("name", "").strip()
slug = "".join(c if c.isalnum() else "-" for c in name.lower()).strip("-")
mockup_url = f"{BASE_URL}/{slug}" if slug else BASE_URL
con.execute("""
INSERT INTO leads(name, category, phone, email, city, maps_url, mockup_url)
VALUES (?, ?, ?, ?, ?, ?, ?)
""", (
name,
row.get("category", ""),
row.get("phone", ""),
email,
row.get("city", ""),
row.get("maps_url", ""),
mockup_url
))
con.commit()
print(f"✅ Lead caricati/aggiornati in DB")
def send_email_api(lead):
"""Invia email via Brevo API v3"""
name = lead["name"] or "la vostra attività"
subject = f"Ho preparato un sito per {name}"
html_content = f"""
Ciao,
mi chiamo {FROM_NAME}, faccio siti web per {lead['category'] or 'attività locali'}.
Ho visto che {name} non ha un sito e ne ho preparato uno io per voi:
👉 Vedi il mockup del sito
Se vi piace, lo mettiamo online sul vostro dominio in 24 ore — 350€ una tantum, nessun abbonamento.
Se non vi serve, ignorate questa email, non vi scrivo più.
{FROM_NAME}
Ghost Agency
{FROM_EMAIL}
"""
text_content = f"""Ciao,
mi chiamo {FROM_NAME}, faccio siti web per {lead['category'] or 'attività locali'}.
Ho visto che {name} non ha un sito e ne ho preparato uno io per voi:
{lead['mockup_url']}
Se vi piace, lo mettiamo online sul vostro dominio in 24 ore — 350€ una tantum, nessun abbonamento.
Se non vi serve, ignorate questa email, non vi scrivo più.
{FROM_NAME}
{FROM_EMAIL}
"""
payload = {
"sender": {"name": FROM_NAME, "email": FROM_EMAIL},
"to": [{"email": lead["email"], "name": lead["name"]}],
"subject": subject,
"htmlContent": html_content,
"textContent": text_content,
"tags": ["ghost-agency", "cold-outreach"]
}
headers = {
"accept": "application/json",
"content-type": "application/json",
"api-key": BREVO_API_KEY
}
resp = requests.post(BREVO_API_URL, json=payload, headers=headers, timeout=30)
resp.raise_for_status()
return resp.json()
def main():
con = init_db()
load_csv(con)
rows = con.execute("""
SELECT id, name, category, phone, email, city, mockup_url
FROM leads
WHERE status = 'new' AND email IS NOT NULL AND email != ''
LIMIT ?
""", (SEND_PER_HOUR,)).fetchall()
if not rows:
print("[i] Nessun nuovo lead da contattare")
return
print(f"📧 Invio {len(rows)} email via Brevo API (max {SEND_PER_HOUR}/ora)...")
for i, (lid, name, cat, phone, email, city, mockup) in enumerate(rows, 1):
lead = {"name": name, "category": cat, "phone": phone, "email": email, "city": city, "mockup_url": mockup}
try:
result = send_email_api(lead)
con.execute("""
UPDATE leads SET status = 'sent', sent_at = ? WHERE id = ?
""", (datetime.utcnow().isoformat(), lid))
con.commit()
print(f"[+] {i}/{len(rows)} → {email} (messageId: {result.get('messageId', 'N/A')})")
except Exception as e:
con.execute("UPDATE leads SET status = 'error', notes = ? WHERE id = ?", (str(e), lid))
con.commit()
print(f"[!] {email}: {e}")
if i < len(rows):
wait = random.randint(JITTER_MIN, JITTER_MAX)
print(f" ⏳ Attesa {wait}s...")
time.sleep(wait)
if __name__ == "__main__":
main()