USER
tengo este codigo viejo que funcionaba con postmarker:
import pandas as pd
import logging
import re
from datetime import datetime
from postmarker.core import PostmarkClient
import base64
import os
from bs4 import BeautifulSoup
# Configuración del logger
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# Configuración de Postmark
api_token = '<TRUFFLEHOG_REDACTED_POSTMARK>'
postmark = PostmarkClient(server_token=api_token)
def encode_image_to_base64(image_path):
with open(image_path, "rb") as image_file:
return base64.b64encode(image_file.read()).decode('utf-8')
def send_email(to_email, subject, body, preview=False):
try:
if preview:
# Cambiar destinatario para el correo de prueba
to_email = 'ismael.gallardo@adbosch.com'
response = postmark.emails.send(
<PRESIDIO_ANONYMIZED_EMAIL_ADDRESS>',
To=to_email,
<PRESIDIO_ANONYMIZED_EMAIL_ADDRESS>', # Agregar en copia
Subject=subject,
HtmlBody=body
)
logging.info(f"Correo enviado a: {to_email} con copia a <PRESIDIO_ANONYMIZED_EMAIL_ADDRESS>")
except Exception as e:
logging.error(f"Falló el envío de correo a {to_email}: {e}")
def is_valid_email(email):
regex = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
return re.match(regex, email) is not None
def load_data_frames(file_path='C:\\Users\\ismael.gallardo\\Documents\\mails\\Plantilla.xlsx'):
try:
offers_df = pd.read_excel(file_path, sheet_name='Sheet1')
clients_df = pd.read_excel(file_path, sheet_name='Sheet2')
logging.info("Datos cargados exitosamente desde el archivo Excel.")
return offers_df, clients_df
except Exception as e:
logging.error(f"Error al cargar los datos desde el archivo Excel: {e}")
raise
def html_to_text(html_content):
soup = BeautifulSoup(html_content, 'html.parser')
return soup.get_text()
def process_clients_and_prepare_emails(current_month, offers_df, clients_df):
month_names_es = ['Enero', 'Febrero', 'Marzo', 'Abril', 'Mayo', 'Junio',
'Julio', 'Agosto', 'Septiembre', 'Octubre', 'Noviembre', 'Diciembre']
month_names_ca = ['gener', 'febrer', 'març', 'abril', 'maig', 'juny',
'juliol', 'agost', 'setembre', 'octubre', 'novembre', 'desembre']
month_str_ca = month_names_ca[current_month - 1]
month_columns = [col for col in offers_df.columns if month_names_es[current_month - 1].upper() in col]
logging.info(f"Procesando ofertas para el mes: {month_names_es[current_month - 1]}")
signature_image_path = 'C:\\Users\\ismael.gallardo\\Documents\\mails\\Firma Daniela.PNG'
signature_image_base64 = encode_image_to_base64(signature_image_path)
email_details_list = [] # Almacenar detalles de correos
for vendor_name, vendor_group in clients_df.groupby('Nombre Vendedor'):
vendor_email = vendor_group.iloc[0]['Correo Vendedor']
for group_name, client_group in vendor_group.groupby('Grupo'):
applicable_offers = offers_df[offers_df['GRUPO'] == group_name]
offers_summary = {"1_15": [], "16_30": []}
# Generar el resumen de ofertas basado en los X de la quincena correspondiente
for _, offer_row in applicable_offers.iterrows():
for month_column in month_columns:
if str(offer_row[month_column]).strip().upper() == 'X':
offer_item = f"{offer_row['OFERTAS']}"
if "1_15" in month_column:
offers_summary["1_15"].append(offer_item)
elif "16_30" in month_column:
offers_summary["16_30"].append(offer_item)
if not (offers_summary["1_15"] or offers_summary["16_30"]):
continue # No enviar correo si no hay ofertas para el grupo en el mes actual
# Comienzo del correo con encabezado y lista de clientes
email_body = (
f'<div style="font-family: Calibri, sans-serif; font-size: 11px;">'
f"<p>Benvolgut/da {vendor_name},</p>"
f"<p>Els teus clients:</p><ul>"
)
for _, client in client_group.iterrows():
client_name = client['Nombre Cliente']
email_body += f"<li>{client_name}</li>"
email_body += "</ul>"
# Descuentos y ofertas
email_body += f"<p>Tenen els següents descomptes per formar part del grup {group_name}:</p><ul style='list-style-position: inside;'>"
for period, offers in offers_summary.items():
if offers:
if period == "1_15":
email_body += f"<li>Primera quinzena de {month_str_ca}:<ul><li>{'</li><li>'.join(offers)}</li></ul></li>"
elif period == "16_30":
email_body += f"<li>Segona quinzena de {month_str_ca}:<ul><li>{'</li><li>'.join(offers)}</li></ul></li>"
email_body += "</ul>"
# Cierre del correo con el saludo y la firma
email_body += (
"<p>Salutacions,</p>"
f'<img src="data:image/png;base64,{signature_image_base64}" alt="Firma Daniela" /></div>'
)
email_subject = f"Descomptes {month_str_ca} {group_name}"
text_body = html_to_text(email_body)
if is_valid_email(vendor_email):
email_details_list.append((vendor_email, email_subject, email_body, text_body, group_name))
return email_details_list
def save_email_summary_to_excel(email_details_list, month_str_ca):
data = {
'Número de Correo': list(range(1, len(email_details_list) + 1)),
'Correo Vendedor': [details[0] for details in email_details_list],
'Grupo': [details[4] for details in email_details_list],
}
df_summary = pd.DataFrame(data)
file_path = f'C:\\Users\\ismael.gallardo\\Documents\\mails\\resumen_mails_{month_str_ca}.xlsx'
df_summary.to_excel(file_path, index=False)
logging.info(f"Resumen de correos guardado en: {file_path}")
def save_email_summary_to_txt(email_details_list, month_str_ca, summary_content):
directory = f'C:\\Users\\ismael.gallardo\\Documents\\mails\\resumen_mails_{month_str_ca}'
os.makedirs(directory, exist_ok=True)
file_path = os.path.join(directory, f'resumen_mails_{month_str_ca}.txt')
index = "Índice de correos electrónicos:\n"
emails = summary_content.split('\n' + '-' * 40 + '\n')
for i, email in enumerate(emails, start=1):
index += f"{i}. Correo {i}\n"
with open(file_path, 'w', encoding='utf-8') as file:
file.write(index + '\n\n' + summary_content)
logging.info(f"Resumen de correos guardado en: {file_path}")
def main():
current_month = datetime.now().month
offers_df, clients_df = load_data_frames()
email_details_list = process_clients_and_prepare_emails(current_month, offers_df, clients_df)
email_summary_content = "" # Cadena para almacenar el contenido del resumen en texto
# Mostrar detalles de los correos en la consola
for idx, (to_email, subject, body, text_body, group_name) in enumerate(email_details_list, start=1):
formatted_email = f"Correo {idx}:\nPara: {to_email}\nAsunto: {subject}\nCuerpo:\n{text_body}\n{'='*40}\n"
email_summary_content += formatted_email
print(formatted_email)
# Enviar el primer correo solo a ismael.gallardo@adbosch.com para revisión
if email_details_list:
first_email = email_details_list[0]
send_email(first_email[0], first_email[1], first_email[2], preview=True)
# Guardar resumen en Excel y TXT
month_str_ca = datetime.now().strftime('%B').lower()
if email_details_list: # Solo guarda si hay correos a enviar
save_email_summary_to_excel(email_details_list, month_str_ca)
save_email_summary_to_txt(email_details_list, month_str_ca, email_summary_content)
# Preguntar al usuario si desea continuar con el envío
confirm = input("¿Desea enviar estos correos? (s/n): ").strip().lower()
if confirm == 's':
for to_email, subject, body, _, _ in email_details_list:
send_email(to_email, subject, body)
if __name__ == "__main__":
main()
Quiero que me adaptes la parte de conection y envio que aqui esta correcta de postmarker a mi codigo nuevo que es el bueno esta actualmente mailsender. El correo nuevo de postmarker que tienes que adaptar a los cambios en el codigo es <PRESIDIO_ANONYMIZED_EMAIL_ADDRESS> y el api token <TRUFFLEHOG_REDACTED_POSTMARK>
import time
import pandas as pd
import logging
import re
from datetime import datetime
import base64
import os
from bs4 import BeautifulSoup
import requests
# Configuración del logger
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# Credenciales de API MailerSend usando el token y dominio de prueba
api_token = 'mlsn.058734777500ea5e78651f97f3a898a149ffd91c7b35d3691cd7b017e06df72f'
base_url = "https://api.mailersend.com/v1/email"
def encode_image_to_base64(image_path):
with open(image_path, "rb") as image_file:
return base64.b64encode(image_file.read()).decode('utf-8')
def send_email(to_email, subject, html_content, text_content):
email_data = {
'from': {'email': 'noreply@trial-neqvygm8mod40p7w.mlsender.net', 'name': 'Trial Sender'},
'to': [{'email': to_email}],
'subject': subject,
'html': html_content,
'text': text_content
}
headers = {
"Authorization": f"Bearer {api_token}",
"Content-Type": "application/json"
}
response = requests.post(base_url, headers=headers, json=email_data)
if response.status_code == 202:
logging.info(f"Correo enviado a: {to_email}")
else:
logging.error(f"No se pudo enviar el correo a {to_email}. Código de estado: {response.status_code}. Detalles: {response.text}")
def is_valid_email(email):
regex = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
return re.match(regex, email) is not None
def load_data_frames(file_path='C:\\Users\\ismael.gallardo\\Documents\\mails\\Plantilla.xlsx'):
try:
offers_df = pd.read_excel(file_path, sheet_name='Sheet1')
clients_df = pd.read_excel(file_path, sheet_name='Sheet2')
logging.info("Datos cargados exitosamente desde el archivo Excel.")
return offers_df, clients_df
except Exception as e:
logging.error(f"Error al cargar los datos desde el archivo Excel: {e}")
raise
def html_to_text(html_content):
soup = BeautifulSoup(html_content, 'html.parser')
return soup.get_text()
def process_clients_and_prepare_emails(current_month, offers_df, clients_df):
month_names_es = ['Enero', 'Febrero', 'Marzo', 'Abril', 'Mayo', 'Junio',
'Julio', 'Agosto', 'Septiembre', 'Octubre', 'Noviembre', 'Diciembre']
month_names_ca = ['gener', 'febrer', 'març', 'abril', 'maig', 'juny',
'juliol', 'agost', 'setembre', 'octubre', 'novembre', 'desembre']
month_str_ca = month_names_ca[current_month - 1]
month_columns = [col for col in offers_df.columns if month_names_es[current_month - 1].upper() in col]
logging.info(f"Procesando ofertas para el mes: {month_names_es[current_month - 1]}")
signature_image_path = 'C:\\Users\\ismael.gallardo\\Documents\\mails\\Firma Daniela.PNG'
signature_image_base64 = encode_image_to_base64(signature_image_path)
email_details_list = [] # Almacenar detalles de correos
for vendor_name, vendor_group in clients_df.groupby('Nombre Vendedor'):
vendor_email = vendor_group.iloc[0]['Correo Vendedor']
for group_name, client_group in vendor_group.groupby('Grupo'):
applicable_offers = offers_df[offers_df['GRUPO'] == group_name]
offers_summary = {"1_15": [], "16_30": []}
for _, offer_row in applicable_offers.iterrows():
for month_column in month_columns:
if str(offer_row[month_column]).strip().upper() == 'X':
offer_item = f"{offer_row['OFERTAS']}"
if "1_15" in month_column:
offers_summary["1_15"].append(offer_item)
elif "16_30" in month_column:
offers_summary["16_30"].append(offer_item)
if not (offers_summary["1_15"] or offers_summary["16_30"]):
continue
email_body = (
f'<div style="font-family: Calibri, sans-serif; font-size: 11px;">'
f"<p>Benvolgut/da {vendor_name},</p>"
f"<p>Els teus clients:</p><ul>"
)
for _, client in client_group.iterrows():
client_name = client['Nombre Cliente']
email_body += f"<li>{client_name}</li>"
email_body += "</ul>"
email_body += f"<p>Tenen els següents descomptes per formar part del grup {group_name}:</p><ul style='list-style-position: inside;'>"
for period, offers in offers_summary.items():
if offers:
if period == "1_15":
email_body += f"<li>Primera quinzena de {month_str_ca}:<ul><li>{'</li><li>'.join(offers)}</li></ul></li>"
elif period == "16_30":
email_body += f"<li>Segona quinzena de {month_str_ca}:<ul><li>{'</li><li>'.join(offers)}</li></ul></li>"
email_body += "</ul>"
email_body += (
"<p><em>Les marques subjectes als descomptes de les campanyes són les indicades a les condicions disponibles al DRIVE-Clients Globals.</em></p>"
"<p>Salutacions,</p>"
f'<img src="data:image/png;base64,{signature_image_base64}" alt="Firma Daniela" /></div>'
)
email_subject = f"Descomptes {month_str_ca} {group_name}"
text_body = html_to_text(email_body)
if is_valid_email(vendor_email):
email_details_list.append((vendor_email, email_subject, email_body, text_body, group_name))
return email_details_list
def save_email_summary_to_excel(email_details_list, month_str_ca):
data = {
'Número de Correo': list(range(1, len(email_details_list) + 1)),
'Correo Vendedor': [details[0] for details in email_details_list],
'Grupo': [details[4] for details in email_details_list],
}
df_summary = pd.DataFrame(data)
file_path = f'C:\\Users\\ismael.gallardo\\Documents\\mails\\resumen_mails_{month_str_ca}.xlsx'
df_summary.to_excel(file_path, index=False)
logging.info(f"Resumen de correos guardado en: {file_path}")
def save_email_summary_to_txt(email_details_list, month_str_ca, summary_content):
directory = f'C:\\Users\\ismael.gallardo\\Documents\\mails\\resumen_mails_{month_str_ca}'
os.makedirs(directory, exist_ok=True)
file_path = os.path.join(directory, f'resumen_mails_{month_str_ca}.txt')
index = "Índice de correos electrónicos:\n"
emails = summary_content.split('\n' + '-' * 40 + '\n')
for i, email in enumerate(emails, start=1):
index += f"{i}. Correo {i}\n"
with open(file_path, 'w', encoding='utf-8') as file:
file.write(index + '\n\n' + summary_content)
logging.info(f"Resumen de correos guardado en: {file_path}")
def main():
current_month = datetime.now().month
offers_df, clients_df = load_data_frames()
email_details_list = process_clients_and_prepare_emails(current_month, offers_df, clients_df)
email_summary_content = "" # Cadena para almacenar el contenido del resumen en texto
# Mostrar detalles de los correos en la consola
for idx, (to_email, subject, body, text_body, group_name) in enumerate(email_details_list, start=1):
formatted_email = f"Correo {idx}:\nPara: {to_email}\nAsunto: {subject}\nCuerpo:\n{text_body}\n{'='*40}\n"
email_summary_content += formatted_email
print(formatted_email)
# Guardar resumen en Excel y TXT
month_str_ca = datetime.now().strftime('%B').lower()
if email_details_list: # Solo guarda si hay correos a enviar
save_email_summary_to_excel(email_details_list, month_str_ca)
save_email_summary_to_txt(email_details_list, month_str_ca, email_summary_content)
# Preguntar al usuario si desea continuar con el envío
confirm = input("¿Desea enviar estos correos? (s/n): ").strip().lower()
if confirm == 's':
for to_email, subject, body, plain_text, _ in email_details_list:
send_email(to_email, subject, body, plain_text)
time.sleep(45) # Espera 45 segundos entre cada envío de correo
if __name__ == "__main__":
main()
ASSISTANT
Para adaptar tu código para que use Postmark en lugar de MailerSend, debes reemplazar la parte relevante de la conexión y envío de correos electrónicos. Esto implica utilizar la biblioteca de Postmark para enviar correos en tu script actual. A continuación, te muestro cómo puedes hacer eso:
```python
import time
import pandas as pd
import logging
import re
from datetime import datetime
from postmarker.core import PostmarkClient
import base64
import os
from bs4 import BeautifulSoup
# Configuración del logger
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# Configuración de Postmark
api_token = '<TRUFFLEHOG_REDACTED_POSTMARK>'
postmark = PostmarkClient(server_token=api_token)
def encode_image_to_base64(image_path):
with open(image_path, "rb") as image_file:
return base64.b64encode(image_file.read()).decode('utf-8')
def send_email(to_email, subject, html_content, text_content):
try:
response = postmark.emails.send(
<PRESIDIO_ANONYMIZED_EMAIL_ADDRESS>',
To=to_email,
Subject=subject,
HtmlBody=html_content,
TextBody=text_content
)
logging.info(f"Correo enviado a: {to_email}")
except Exception as e:
logging.error(f"Falló el envío de correo a {to_email}: {e}")
def is_valid_email(email):
regex = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
return re.match(regex, email) is not None
def load_data_frames(file_path='C:\\Users\\ismael.gallardo\\Documents\\mails\\Plantilla.xlsx'):
try:
offers_df = pd.read_excel(file_path, sheet_name='Sheet1')
clients_df = pd.read_excel(file_path, sheet_name='Sheet2')
logging.info("Datos cargados exitosamente desde el archivo Excel.")
return offers_df, clients_df
except Exception as e:
logging.error(f"Error al cargar los datos desde el archivo Excel: {e}")
raise
def html_to_text(html_content):
soup = BeautifulSoup(html_content, 'html.parser')
return soup.get_text()
def process_clients_and_prepare_emails(current_month, offers_df, clients_df):
month_names_es = ['Enero', 'Febrero', 'Marzo', 'Abril', 'Mayo', 'Junio',
'Julio', 'Agosto', 'Septiembre', 'Octubre', 'Noviembre', 'Diciembre']
month_names_ca = ['gener', 'febrer', 'març', 'abril', 'maig', 'juny',
'juliol', 'agost', 'setembre', 'octubre', 'novembre', 'desembre']
month_str_ca = month_names_ca[current_month - 1]
month_columns = [col for col in offers_df.columns if month_names_es[current_month - 1].upper() in col]
logging.info(f"Procesando ofertas para el mes: {month_names_es[current_month - 1]}")
signature_image_path = 'C:\\Users\\ismael.gallardo\\Documents\\mails\\Firma Daniela.PNG'
signature_image_base64 = encode_image_to_base64(signature_image_path)
email_details_list = []
for vendor_name, vendor_group in clients_df.groupby('Nombre Vendedor'):
vendor_email = vendor_group.iloc[0]['Correo Vendedor']
for group_name, client_group in vendor_group.groupby('Grupo'):
applicable_offers = offers_df[offers_df['GRUPO'] == group_name]
offers_summary = {"1_15": [], "16_30": []}
for _, offer_row in applicable_offers.iterrows():
for month_column in month_columns:
if str(offer_row[month_column]).strip().upper() == 'X':
offer_item = f"{offer_row['OFERTAS']}"
if "1_15" in month_column:
offers_summary["1_15"].append(offer_item)
elif "16_30" in month_column:
offers_summary["16_30"].append(offer_item)
if not (offers_summary["1_15"] or offers_summary["16_30"]):
continue
email_body = (
f'<div style="font-family: Calibri, sans-serif; font-size: 11px;">'
f"<p>Benvolgut/da {vendor_name},</p>"
f"<p>Els teus clients:</p><ul>"
)
for _, client in client_group.iterrows():
client_name = client['Nombre Cliente']
email_body += f"<li>{client_name}</li>"
email_body += "</ul>"
email_body += f"<p>Tenen els següents descomptes per formar part del grup {group_name}:</p><ul style='list-style-position: inside;'>"
for period, offers in offers_summary.items():
if offers:
if period == "1_15":
email_body += f"<li>Primera quinzena de {month_str_ca}:<ul><li>{'</li><li>'.join(offers)}</li></ul></li>"
elif period == "16_30":
email_body += f"<li>Segona quinzena de {month_str_ca}:<ul><li>{'</li><li>'.join(offers)}</li></ul></li>"
email_body += "</ul>"
email_body += (
"<p><em>Les marques subjectes als descomptes de les campanyes són les indicades a les condicions disponibles al DRIVE-Clients Globals.</em></p>"
"<p>Salutacions,</p>"
f'<img src="data:image/png;base64,{signature_image_base64}" alt="Firma Daniela" /></div>'
)
email_subject = f"Descomptes {month_str_ca} {group_name}"
text_body = html_to_text(email_body)
if is_valid_email(vendor_email):
email_details_list.append((vendor_email, email_subject, email_body, text_body, group_name))
return email_details_list
def save_email_summary_to_excel(email_details_list, month_str_ca):
data = {
'Número de Correo': list(range(1, len(email_details_list) + 1)),
'Correo Vendedor': [details[0] for details in email_details_list],
'Grupo': [details[4] for details in email_details_list],
}
df_summary = pd.DataFrame(data)
file_path = f'C:\\Users\\ismael.gallardo\\Documents\\mails\\resumen_mails_{month_str_ca}.xlsx'
df_summary.to_excel(file_path, index=False)
logging.info(f"Resumen de correos guardado en: {file_path}")
def save_email_summary_to_txt(email_details_list, month_str_ca, summary_content):
directory = f'C:\\Users\\ismael.gallardo\\Documents\\mails\\resumen_mails_{month_str_ca}'
os.makedirs(directory, exist_ok=True)
file_path = os.path.join(directory, f'resumen_mails_{month_str_ca}.txt')
index = "Índice de correos electrónicos:\n"
emails = summary_content.split('\n' + '-' * 40 + '\n')
for i, email in enumerate(emails, start=1):
index += f"{i}. Correo {i}\n"
with open(file_path, 'w', encoding='utf-8') as file:
file.write(index + '\n\n' + summary_content)
logging.info(f"Resumen de correos guardado en: {file_path}")
def main():
current_month = datetime.now().month
offers_df, clients_df = load_data_frames()
email_details_list = process_clients_and_prepare_emails(current_month, offers_df, clients_df)
email_summary_content = ""
for idx, (to_email, subject, body, text_body, group_name) in enumerate(email_details_list, start=1):
formatted_email = f"Correo {idx}:\nPara: {to_email}\nAsunto: {subject}\nCuerpo:\n{text_body}\n{'='*40}\n"
email_summary_content += formatted_email
print(formatted_email)
# Guardar resumen en Excel y TXT
month_str_ca = datetime.now().strftime('%B').lower()
if email_details_list:
save_email_summary_to_excel(email_details_list, month_str_ca)
save_email_summary_to_txt(email_details_list, month_str_ca, email_summary_content)
confirm = input("¿Desea enviar estos correos? (s/n): ").strip().lower()
if confirm == 's':
for to_email, subject, body, plain_text, _ in email_details_list:
send_email(to_email, subject, body, plain_text)
time.sleep(45)
if __name__ == "__main__":
main()
```
### Cambios Clave:
- **Postmark Initialization**: Se ha sustituido la configuración anterior de MailerSend por la inicialización de `PostmarkClient` usando el nuevo token API.
- **Send Email Function**: Adapté la lógica del envío de correos para usar la biblioteca de Postmark y enviarlo desde `<PRESIDIO_ANONYMIZED_EMAIL_ADDRESS>`.
- **Error Handling**: Se incluyó un bloque `try-except` para capturar y registrar errores en el envío de correos electrónicos.
Este script ahora usa Postmark para enviar correos en lugar de MailerSend y está configurado para trabajar con tu nuevo API token y dirección de correo electrónico. Asegúrate de tener la biblioteca `postmarker` instalada en tu entorno de Python antes de ejecutar el script.