#!/usr/bin/env python3 import sys import argparse import time import base64 from datetime import datetime import requests import mysql.connector from concurrent.futures import ThreadPoolExecutor, as_completed from config import DB_HOST, DB_USERNAME, DB_PASSWORD, DB_DATABASE, EMAIL_LIMIT_PER_HOUR from sendPulseAuth import get_access_token from emailTemplates import get_email_template, get_email_text from conteudo import get_conteudo def processar_envio(row, acao, ordem, token, conteudo): # Monta o corpo do e-mail email_from_corpo = get_email_template( conteudo['title'], conteudo['mainTitle'], conteudo['description'], conteudo['additionalInfo'], conteudo['liveLink'], conteudo['date'], conteudo['time'], conteudo['imageAlt'], conteudo['imageSrc'], "", # partialCongresso vazio row['REGISTRO_ID'], acao # trackerAcao ) email_from_corpo_alt = get_email_text( conteudo['title'], conteudo['description'], conteudo['additionalInfo'], conteudo['liveLink'], conteudo['imageSrc'], conteudo.get('congressoText', "") ) email_data = { "html": base64.b64encode(email_from_corpo.encode("utf-8")).decode("utf-8"), "text": email_from_corpo_alt, "subject": conteudo['title'], "from": { "email": "no-reply@nexix.com.br", "name": "5 Motivos para Mudar Seu Prato Ainda Hoje" }, "to": [ { "email": row['EMAIL'], "name": row['NOME'] } ] } headers = { 'Authorization': f'Bearer {token}', 'Content-Type': 'application/json' } resultado = {"email": row['EMAIL']} try: response = requests.post('https://api.sendpulse.com/smtp/emails', headers=headers, json={"email": email_data}) response.raise_for_status() resultado["status"] = "SUCESSO" except Exception as e: resultado["status"] = f"Erro: {str(e)}" # Registra o envio no banco de dados (cada thread abre sua própria conexão) try: conn = mysql.connector.connect( host=DB_HOST, user=DB_USERNAME, password=DB_PASSWORD, database=DB_DATABASE ) cursor = conn.cursor() insert_query = "INSERT INTO enviarEmails (acao, acao_id, registro_id, enviado_em) VALUES (%s, %s, %s, NOW())" cursor.execute(insert_query, (acao, ordem, row['REGISTRO_ID'])) conn.commit() cursor.close() conn.close() except mysql.connector.Error as err: resultado["status"] += f" | Erro no DB: {err}" return resultado def enviar_emails(acao, ordem, emails_param, token, conteudo): total_enviados_geral = 0 # Envia os e-mails fornecidos via parâmetro, se houver if emails_param: print(f"\nEnviando e-mails informados via parâmetro: {emails_param}") for email in emails_param: # Cria um registro "fake" com os campos necessários. registro = { 'EMAIL': email, 'NOME': email.split('@')[0].capitalize(), 'REGISTRO_ID': f'param_{email}' # identificador customizado para envios por parâmetro } resultado = processar_envio(registro, acao, ordem, token, conteudo) print(f"Enviado para: {resultado['email']} - {resultado['status']}") if "SUCESSO" in resultado["status"]: total_enviados_geral += 1 print(f"\nTotal de e-mails enviados via parâmetro: {len(emails_param)}\n") # A partir daqui, continua com a consulta no banco para recuperar os demais e-mails while True: # Consulta os registros pendentes para envio try: conn = mysql.connector.connect( host=DB_HOST, user=DB_USERNAME, password=DB_PASSWORD, database=DB_DATABASE ) except mysql.connector.Error as err: sys.exit("Falha na conexão com o banco de dados: " + str(err)) cursor = conn.cursor(dictionary=True) if acao == "Teste": query = "SELECT nome AS NOME, email AS EMAIL, id AS REGISTRO_ID FROM consolidado WHERE email = 'marcusthiago@gmail.com' LIMIT 1" cursor.execute(query) else: query = f"""SELECT nome AS NOME, email AS EMAIL, id AS REGISTRO_ID FROM consolidado WHERE id NOT IN (SELECT registro_id FROM enviarEmails WHERE acao = %s AND acao_id = %s) AND email <> '' AND status = 'Ativo' ORDER BY id DESC LIMIT {EMAIL_LIMIT_PER_HOUR}""" cursor.execute(query, (acao, ordem)) rows = cursor.fetchall() cursor.close() conn.close() if not rows: print(f"\nNão há mais e-mails para enviar para a ação '{acao}' e ordem '{ordem}'.") break lote_size = len(rows) inicio_lote = time.time() print(f"\nIniciando novo lote de {lote_size} e-mails para a ação '{acao}' e ordem '{ordem}' às {datetime.fromtimestamp(inicio_lote).strftime('%H:%M:%S')}.") emails_enviados_lote = 0 max_workers = 10 # Ajuste conforme a capacidade do ambiente with ThreadPoolExecutor(max_workers=max_workers) as executor: if acao == "Teste": futuros = [executor.submit(processar_envio, rows[0], acao, ordem, token, conteudo)] else: futuros = [executor.submit(processar_envio, row, acao, ordem, token, conteudo) for row in rows] for futuro in as_completed(futuros): res = futuro.result() print(f"Enviado para: {res['email']} - {res['status']}") if "SUCESSO" in res["status"]: emails_enviados_lote += 1 fim_lote = time.time() tempo_lote = fim_lote - inicio_lote total_enviados_geral += emails_enviados_lote print(f"\nLote concluído: {emails_enviados_lote} e-mails enviados com sucesso neste lote, finalizado às {datetime.fromtimestamp(fim_lote).strftime('%H:%M:%S')}.") print(f"Tempo gasto neste lote: {int(tempo_lote)} segundos.") print(f"Total acumulado de e-mails enviados: {total_enviados_geral}.\n") # Se estiver em modo Teste, processa apenas um lote if acao == "Teste": break # Se o lote enviado foi um lote completo, aguarda 1 hora antes de iniciar o próximo. if lote_size >= EMAIL_LIMIT_PER_HOUR: proxima_execucao = datetime.fromtimestamp(time.time() + 3600).strftime("%d/%m/%Y %H:%M") print(f"Lote completo enviado. Próxima execução iniciará em 1 hora, às {proxima_execucao}.\n") time.sleep(3600) else: print("Lote incompleto detectado. Não há mais e-mails pendentes.") break print(f"\nEnvio finalizado. Total de e-mails enviados nesta execução: {total_enviados_geral}") def main(): parser = argparse.ArgumentParser(description='Envio de e-mails em lotes com máximo de 6000 por hora') parser.add_argument('--acao', type=str, default='Live', help='Ação a ser executada (ex.: Teste, Live, etc.)') parser.add_argument('--ordem', type=str, default='2024-07-24', help='Ordem da ação') parser.add_argument('--emails', type=str, default='', help='Endereços de e-mail separados por vírgula para serem enviados antes da consulta ao banco de dados') args = parser.parse_args() # Divide a string em uma lista de e-mails, se fornecida if args.emails: lista_emails = [email.strip() for email in args.emails.split(',') if email.strip()] else: lista_emails = [] token = get_access_token() conteudo = get_conteudo() enviar_emails(args.acao, args.ordem, lista_emails, token, conteudo) if __name__ == '__main__': main()