#!/usr/bin/env python3 """ Auto Mailer - Legge lead da CSV e manda email personalizzate. Rate-limited, GDPR-compliant (unsubscribe link, dati minimi). """ import os import csv import smtplib import time import random from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from pathlib import Path # Config - usa variabili d'ambiente per sicurezza SMTP_HOST = os.getenv("SMTP_HOST", "smtp.gmail.com") SMTP_PORT = int(os.getenv("SMTP_PORT", "587")) SMTP_USER = os.getenv("SMTP_USER", "4ms1xseller@gmail.com") SMTP_PASS = os.getenv("SMTP_PASS", "") # App Password Gmail FROM_NAME = os.getenv("FROM_NAME", "Amraj Singh Atwal") REPLY_TO = os.getenv("REPLY_TO", "4ms1xseller@gmail.com") UNSUBSCRIBE_URL = os.getenv("UNSUBSCRIBE_URL", "https://svoraj.me/unsubscribe") LEADS_DIR = Path("/home/ubuntu/GhostAgency") SENT_LOG = LEADS_DIR / "sent_emails.json" MAX_EMAILS_PER_RUN = int(os.getenv("MAX_EMAILS_PER_RUN", "10")) # Safety limit DELAY_MIN = int(os.getenv("DELAY_MIN", "30")) # secondi tra email DELAY_MAX = int(os.getenv("DELAY_MAX", "90")) # Template email SUBJECT_TEMPLATE = "Migliorare la presenza online di {nome}" BODY_TEMPLATE = """Buongiorno al team di {nome}, mi chiamo Amraj, sono uno sviluppatore web e studio informatica qui a Parma. Vi scrivo perché stavo cercando il vostro menù online e ho visto che vi appoggiate a un portale esterno (mordy.it o simili). Purtroppo, affidarsi a queste piattaforme generiche toglie molta identità al vostro locale, la pagina risulta un po' datata e spesso non rassicura i nuovi clienti che vi cercano su Google per la prima volta. Lavoro nel settore IT e realizzo siti web veloci, moderni e, soprattutto, di vostra proprietà esclusiva. Senza impegno, mi farebbe piacere aiutarvi a fare il salto di qualità. Con una spesa minima possiamo registrare il vero dominio ufficiale del vostro ristorante, creando una vetrina professionale che spazzi via la pagina di terze parti. Vi interessa parlarne per 5 minuti? Buon lavoro e a presto, Amraj Singh Atwal Sviluppatore Web a Parma - svoraj.me --- Se non vuoi ricevere altre comunicazioni: {unsubscribe} """ def load_sent_log(): if SENT_LOG.exists(): import json with open(SENT_LOG) as f: return set(json.load(f)) return set() def save_sent_log(sent_set): import json with open(SENT_LOG, 'w') as f: json.dump(list(sent_set), f) def find_latest_leads_csv(): """Trova il CSV leads più recente.""" # Cerca entrambi i pattern: ultimate e local csvs = list(LEADS_DIR.glob("leads_*_ultimate.csv")) + list(LEADS_DIR.glob("leads_*_local.csv")) if not csvs: return None return max(csvs, key=lambda p: p.stat().st_mtime) def send_email(to_email, to_name, sent_log): if to_email in sent_log: print(f"[=] Già inviato a {to_email}, salto") return False msg = MIMEMultipart() msg['From'] = f"{FROM_NAME} <{SMTP_USER}>" msg['To'] = to_email msg['Subject'] = SUBJECT_TEMPLATE.format(nome=to_name) msg['Reply-To'] = REPLY_TO msg['List-Unsubscribe'] = f"<{UNSUBSCRIBE_URL}>" body = BODY_TEMPLATE.format(nome=to_name, unsubscribe=UNSUBSCRIBE_URL) msg.attach(MIMEText(body, 'plain', 'utf-8')) try: print(f"[*] Connessione SMTP...") server = smtplib.SMTP(SMTP_HOST, SMTP_PORT) server.starttls() server.login(SMTP_USER, SMTP_PASS) server.sendmail(SMTP_USER, to_email, msg.as_string()) server.quit() print(f"[+] Inviato a {to_name} ({to_email})") return True except Exception as e: print(f"[!] Errore invio a {to_email}: {e}") return False def main(): if not SMTP_PASS: print("[!] SMTP_PASS non configurato (variabile d'ambiente)") return csv_path = find_latest_leads_csv() if not csv_path: print("[!] Nessun CSV leads trovato in GhostAgency/") return print(f"[*] Leggo lead da {csv_path.name}") sent_log = load_sent_log() sent_count = 0 with open(csv_path, newline='', encoding='utf-8') as f: reader = csv.DictReader(f) for row in reader: if sent_count >= MAX_EMAILS_PER_RUN: print(f"[=] Limite {MAX_EMAILS_PER_RUN} email/run raggiunto") break nome = row.get('Nome', '').strip() email = row.get('Email', '').strip() telefono = row.get('Telefono', '').strip() # Se non c'è email, prova a costruirla dal nome if not email or '@' not in email: # Prova pattern comuni clean = "".join(c for c in nome if c.isalnum()).lower() guessed_emails = [ f"info@{clean}.it", f"contatto@{clean}.it", f"prenotazioni@{clean}.it", f"info@{clean}.com", f"contatto@{clean}.com", ] # Quick DNS check su MX found_email = None for guess in guessed_emails: domain = guess.split('@')[1] try: import dns.resolver dns.resolver.resolve(domain, 'MX') found_email = guess break except: continue if found_email: email = found_email print(f" [*] Email indovinata: {email}") else: # Nessuna email trovata -> salva in call list per follow-up telefonico print(f" [*] Nessuna email trovata per {nome}, aggiunto a call list") # Aggiungi a lista chiamate call_list_path = LEADS_DIR / "call_list.csv" if not call_list_path.exists(): with open(call_list_path, "w", newline="", encoding="utf-8") as f: writer = csv.writer(f) writer.writerow(["Nome", "Telefono", "Indirizzo", "Categoria", "Data"]) with open(call_list_path, "a", newline="", encoding="utf-8") as f: writer = csv.writer(f) writer.writerow([nome, telefono, row.get('Indirizzo', ''), row.get('Categoria', ''), time.strftime("%Y-%m-%d")]) continue # Skip email send if send_email(email, nome, sent_log): sent_log.add(email) sent_count += 1 save_sent_log(sent_log) # Rate limiting casuale if sent_count < MAX_EMAILS_PER_RUN: delay = random.randint(DELAY_MIN, DELAY_MAX) print(f"[*] Pausa {delay}s...") time.sleep(delay) print(f"[+] Fatto. Inviate {sent_count} email nuove.") if __name__ == "__main__": main()