Отправлять массовую почту во фласке через 4 референтных сервера домена референтной почты ⇐ Python

[Anonymous]

Код: Выделить всё

import os
import re
import time
from datetime import datetime
from smtplib import SMTPRecipientsRefused

import pandas as pd
from celery import Celery
from celery.schedules import crontab
from flask import Flask, render_template, render_template_string
import mysql.connector
from flask_mail import Mail, Message

flask_app = Flask(__name__)
flask_app.template_folder = 'templates'

celery = Celery(
'tasks',
broker='redis://localhost:6380/0',
backend='redis://localhost:6380/0'
)

celery.conf.timezone = 'Asia/Kolkata'

celery.conf.beat_schedule = {
'send_worker1': {
'task': 'tasks.send_worker1',
'schedule': crontab(minute=27, hour=15),
},
'send_worker2': {
'task': 'tasks.send_worker2',
'schedule': crontab(minute=27, hour=15),
},
'send_worker3': {
'task': 'tasks.send_worker3',
'schedule': crontab(minute=15, hour=15),
},
'send_worker4': {
'task': 'tasks.send_worker4',
'schedule': crontab(minute=22, hour=16),
},
'send_mail_emp': {
'task': 'tasks.send_mail_emp',
'schedule': crontab(minute=25, hour=11),
}
}

mail_configs = {
"viskohr1": {
"MAIL_SERVER": "mail.example.com",
"MAIL_USERNAME": "example@example.com",
"MAIL_PASSWORD": "example@example",
"MAIL_DEFAULT_SENDER": "example@example.com",
"MAIL_USE_TLS": True,
"MAIL_PORT": 587
},
"viskohr2": {
"MAIL_SERVER": "mail.example.in",
"MAIL_USERNAME": "example@example.in",
"MAIL_PASSWORD": "example@example",
"MAIL_DEFAULT_SENDER": "example@example.in",
"MAIL_USE_TLS": True,
"MAIL_PORT": 587
},
"viskohr3": {
"MAIL_SERVER": "mail.example.group",
"MAIL_USERNAME": "example@example.group",
"MAIL_PASSWORD": "example@example",
"MAIL_DEFAULT_SENDER": "example@example.group",
"MAIL_USE_TLS": True,
"MAIL_PORT": 587
},
"viskohr4": {
"MAIL_SERVER": "mail.example-marketing.com",
"MAIL_USERNAME": "example@example-marketing.com",
"MAIL_PASSWORD": "example@example",
"MAIL_DEFAULT_SENDER": "example@example-marketing.com",
"MAIL_USE_TLS": True,
"MAIL_PORT": 587
},

}

# ==============================================
# Database
# ==============================================
def get_db_connection(database):
"""Return MySQL connection."""
return mysql.connector.connect(
host="xhgvjgdegfjsdfgjsdfsdjf.amazonaws.com",
database=database,
user="msdfgndsgsdg",
password="zmfld,xgmlxmfdg",
port=3306
)

def apply_mail_config(config_name):
"""Apply mail config and return Mail object."""
config = mail_configs.get(config_name)
if not config:
raise ValueError(f"Invalid mail config name:  {config_name}")

for key, value in config.items():
flask_app.config[key] = value

mail = Mail(flask_app)
# Disable SMTP debug output (suppress smtplib debug prints)
mail.debug = False
return mail

# ==============================================
# Excel Reader - Candidate Emails
# ==============================================
def get_email_candidate():
"""Read Excel and return valid email list."""
current_dir = os.getcwd()
path = os.path.join(current_dir, "Gujarat.xlsx")

if not os.path.exists(path):
raise FileNotFoundError("Excel file 'Gujarat.xlsx' not found in current directory.")

target_sheets = ["Sheet1(2)", "Sheet1"]
pattern = re.compile(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$')

valid_records = []

for sheet in target_sheets:
try:
df = pd.read_excel(path, sheet_name=sheet)
except Exception as e:
print(f"⚠️ Could not read sheet '{sheet}': {e}")
continue

if 'Email' not in df.columns or 'Name' not in df.columns:
print(f"⚠️ Missing required columns in '{sheet}'")
continue

for _, row in df.iterrows():
email = str(row['Email']).strip()
name = str(row['Name']).strip()

if pattern.fullmatch(email):
valid_records.append({"name": name, "email": email})

all_unique_valid = {rec['email']: rec for rec in valid_records}
unique_valid = sorted(all_unique_valid.values(), key=lambda x: x['email'])
print(f"✅ Total valid unique emails: {len(unique_valid)}")
return list(unique_valid)

# ==============================================
# Send Message Safely
# ==============================================
def send_message_safe(mail, msg, max_retries=5):
"""Send email with retry & rate limit handling."""
attempt = 0

while attempt < max_retries:
try:
mail.send(msg)
print("✅ Email sent successfully.")
return True
except SMTPRecipientsRefused as e:
print(f"⚠️ SMTP limit hit: {e}. Waiting 1 hour...")
time.sleep(3600)
return False
except OSError as e:
win_err = getattr(e, "winerror", None)
if win_err == 10061:
attempt += 1
wait_time = min(30, 5 * attempt)
remaining = max_retries - attempt
print(
f"⚠️ Connection refused (WinError 10061). Retrying in {wait_time}s "
f"(attempt {attempt}/{max_retries}, {remaining} retries left)..."
)
time.sleep(wait_time)
continue

print(f"⚠️ OS error while sending email: {e}. Retrying in 10s...")
time.sleep(10)
attempt += 1
except Exception as e:
attempt += 1
wait_time = min(60, 5 * attempt)
print(f"⚠️ Error while sending email: {e}.  Retrying in {wait_time}s...")
time.sleep(wait_time)

print("❌ Failed to send email after multiple retries.")
return

# ==============================================
# Email Sending Function
# ==============================================
def send_emails(unique_emails, worker_name, inst):
print(f"\n🚀 Starting {worker_name} with config: {inst}")
mail = apply_mail_config(inst)

with flask_app.app_context():
db = get_db_connection("xyz")
cursor = db.cursor(dictionary=True)

try:
cursor.execute("SELECT * FROM email_templates WHERE et_name = 'JobFair-1'")
template = cursor.fetchone()  # FIXED ✔✔

if not template:
print("❌ No email template found!")
return

print(f"📄 Using template ID: {template['et_id']}")

for user in unique_emails:
name = user["name"]
email = user["email"]

query = """select * from email_record where er_user_email=%s and er_template_id=%s"""
cursor.execute(query, (email, template["et_id"]))
existing_record = cursor.fetchone()
if existing_record:
continue

email_body = render_template_string(
template['et_template'],  # FIXED ✔✔
name=name
)

msg = Message(
subject="Gujarat’s Biggest Online Job Fair is Live. Register Now",
recipients=[email],  # testing
html=email_body
)

status=send_message_safe(mail, msg)
if status:
cursor.execute("""
INSERT INTO email_record (er_user_email, er_template_id, er_email_send_status)
VALUES (%s, %s, %s)
""", (email, template["et_id"], 1))  # FIXED ✔✔

db.commit()

print(f"✅ [{worker_name}] Sent: {email}")

except Exception as e:
print(f"❌ [{worker_name}] Error: {e}")

finally:
cursor.close()
db.close()

# 🎯 Worker 1
@celery.task
def send_worker1():
all_emails = get_email_candidate()
send_emails(all_emails[:40000], "Worker 1", "viskohr1")
print("✅ Worker 1 completed.")

# 🎯 Worker 2
@celery.task
def send_worker2():
all_emails = get_email_candidate()
send_emails(all_emails[40000:80000], "Worker 2", "viskohr2")
print("✅ Worker 2 completed.")

# 🎯 Worker 3
@celery.task
def send_worker3():
all_emails = get_email_candidate()
send_emails(all_emails[80000:120000], "Worker 3", "viskohr3")
print("✅ Worker 3 completed.")

# 🎯 Worker 4
@celery.task
def send_worker4():
all_emails = get_email_candidate()
send_emails(all_emails[120000:], "Worker 4", "viskohr4")
print("✅ Worker 4 completed.")

в этом коде я хочу отправить массовую почту. Электронная почта получается из файла Excel, а количество адресов электронной почты составляет 140 000. Поэтому я разделю на 4 функции и отправлю референтные домены электронной почты, но я буду использовать один почтовый сервер домена, поэтому никаких ошибок не возникнет, но если я использую весь почтовый сервер домена 4, то у меня возникнет много ошибок, таких как отказ в соединении и блокировка порта, поэтому, пожалуйста, решите эту проблему и дайте мне лучшие решения для этого типа проблем

Подробнее здесь: https://stackoverflow.com/questions/798 ... ain-server