Email validator en python

En estos días tenía una db de email en donde tenía que validar estructura de direcciones estuvieran bien y a la vez que el dominio existiera, sin validar si user está activo en dicho servidor de correo. Por ejemplo:

usuario @gmail.con

Este email está mal escrito porque se debe marcar dicho email para tratar de arreglar manualmente a usuario@dominio, ahora también hay que validar si dominio gmail.con existe, de no existir se trata de corregir o se elimina email.

Manualmente, debería quedar así: [email protected]

Ahora bien, buscando código anteriormente en google o cualquier buscador, uno facilmente se demoraba 1día o 2 buscando código, adaptando y haciendo pruebas.

Esta vez quisimos probar ChatGPT o Bard, explicándole que necesitábamos y en el lenguaje que debíamos usar. Ahora con estas herramientas se ahorra uno cantidad de tiempo y código reutilizado … Ah, esto es para cosas buenas, pero ni imaginar para lo malo que se puede usar.

Le preguntamos al AI lo siguiente:

I need create a python 3.12 script, to validate emails is validate address and if exist with best library on python. Also I have .csv with header (info, first name, last name, email), I need import and review line by line. If possible enable this process with 10 threads or config more threard. Export file .csv with email and info if it’s ok or some error on column

import pandas as pd
import sys
import concurrent.futures
from email_validator import validate_email, EmailNotValidError
from datetime import datetime

def validate_email_address(email):
try:
validate_email(email)
return True, None
except EmailNotValidError as e:
return False, str(e)

def process_row(row):
info, first_name, last_name, email = row
is_valid, error_message = validate_email_address(email)
return info, email, is_valid, error_message

def process_csv(input_file):
df = pd.read_csv(input_file)
total_rows = len(df)
processed_count = 0
with concurrent.futures.ThreadPoolExecutor(max_workers=100) as executor:
results = []
for result in executor.map(process_row, df.itertuples(index=False)):
processed_count = 1
print(f"Processed {processed_count} out of {total_rows} rows.")
results.append(result)
return results, df.columns.tolist()

def write_output(output_file, header, results):
df = pd.DataFrame(results, columns=['Info', 'Email', 'Is Valid', 'Error Message'])
df.to_csv(output_file, index=False)

#init
if __name__ == "__main__":
if len(sys.argv) != 3:
print("Usage: validate-email.py input_file.csv output_file.csv")
else:
start_time = datetime.now()
print('Processing...')
input_file = sys.argv[1]
output_file = sys.argv[2]
results, header = process_csv(input_file)
write_output(output_file, header, results)
time_elapsed = datetime.now() - start_time
print('Time elapsed (hh:mm:ss.ms) {}'.format(time_elapsed))
validate-email.py

*Solo modificamos code para agregar «time elapsed», para saber cuando demoró el proceso

Un ejemplo de .csv

INFO,FIRST_NAME,LAST_NAME,EMAIL
CO,Nombre1,Apellido1,[email protected]
CR,Nombre2,Apellido2,[email protected]
input.csv

Software Utilizado y comandos.

  • En powershell ir a CD c:\python312\, luego instalar librerias – pip install email-validator pandas
  • En powershell en misma folder colocar validate-email.py y input.csv y ejecutar: python validate-email.py C:\python312\input.csv C:\python312\output.csv
  • Revisar archivo de salida en C:\python312

Comentarios

Deja una respuesta

Tu dirección de correo electrónico no será publicada. Los campos obligatorios están marcados con *