如何基于数据库存储的DateTime，使用Django/Celery发送邮件？

1. 数据库查询：筛选待处理记录 #

首先要精准筛选出需要处理的到期或即将到期的记录，重点解决日期格式的兼容问题：

处理字符串类型的日期字段 #

如果你的send_email_at是字符串格式（如2022/07/12），查询时需要先转换为数据库可识别的日期类型再比较：

• MySQL：

  SELECT * FROM your_table 
  WHERE STR_TO_DATE(send_email_at, '%Y/%m/%d') > NOW() AND is_sent = 0;
  

• PostgreSQL：

  SELECT * FROM your_table 
  WHERE TO_DATE(send_email_at, 'YYYY/MM/DD') > CURRENT_TIMESTAMP AND is_sent = 0;
  

优化建议 #

强烈建议将send_email_at字段改为DATE或DATETIME类型，这样无需格式转换，查询效率更高，还能避免字符串格式不一致导致的错误。此时查询语句简化为：

SELECT * FROM your_table WHERE send_email_at > NOW() AND is_sent = 0;

2. 定时任务：周期性触发检查 #

核心是通过定时任务周期性执行检查逻辑，常用两种方式：

系统级定时任务 #

适合轻量场景，直接用操作系统自带的定时工具：

• Linux：使用crontab，比如每小时执行一次检查脚本：

  0 * * * * /usr/bin/python3 /path/to/your/email_reminder.py
  

• Windows：使用「任务计划程序」，设置触发周期和执行的脚本路径。

应用内定时任务 #

如果是Web应用，推荐用框架自带的定时组件：

• Python：Celery Beat
• Java：Quartz
• Node.js：node-schedule
  这类工具支持更灵活的调度规则（如精准到分钟的触发），还能和业务逻辑深度集成。

3. 邮件发送与状态更新 #

每次定时任务触发时，执行以下流程：

1. 查询当前时间已到期且未发送的记录
2. 逐条发送邮件
3. 发送成功后标记记录为已处理（避免重复发送）

以下是简化的Python实现示例：

import mysql.connector
import smtplib
from email.mime.text import MIMEText
from datetime import datetime

# 数据库连接配置
db_config = {
    "host": "your_db_host",
    "user": "your_db_user",
    "password": "your_db_pass",
    "database": "your_db_name"
}

# SMTP邮件配置
smtp_config = {
    "server": "smtp.example.com",
    "port": 587,
    "user": "your_email@example.com",
    "password": "your_email_password"
}

def send_reminder_emails():
    db = mysql.connector.connect(**db_config)
    cursor = db.cursor()

    # 获取当前时间，匹配数据库日期格式
    now = datetime.now().strftime('%Y/%m/%d')
    # 查询今日到期且未发送的记录
    query = """
    SELECT id, recipient_email, reminder_content 
    FROM your_table 
    WHERE STR_TO_DATE(send_email_at, '%Y/%m/%d') = %s AND is_sent = 0
    """
    cursor.execute(query, (now,))
    records = cursor.fetchall()

    for record in records:
        record_id, to_email, content = record
        # 构造邮件
        msg = MIMEText(content, 'plain', 'utf-8')
        msg['Subject'] = "您的提醒邮件"
        msg['From'] = smtp_config["user"]
        msg['To'] = to_email

        # 发送邮件并更新状态
        try:
            with smtplib.SMTP(smtp_config["server"], smtp_config["port"]) as server:
                server.starttls()
                server.login(smtp_config["user"], smtp_config["password"])
                server.send_message(msg)
            
            # 标记为已发送
            update_query = "UPDATE your_table SET is_sent = 1 WHERE id = %s"
            cursor.execute(update_query, (record_id,))
            db.commit()
        except Exception as e:
            print(f"记录ID {record_id} 邮件发送失败：{str(e)}")
            # 可添加日志记录或重试逻辑

    cursor.close()
    db.close()

if __name__ == "__main__":
    send_reminder_emails()

4. 关键注意事项 #

• 避免重复发送：必须添加is_sent（布尔/整型）字段标记发送状态，每次只处理未发送的记录。
• 日期格式一致性：确保数据库中send_email_at的格式统一，禁止混合使用/和-分隔符。
• 重试机制：邮件发送失败时，记录错误日志，可设置重试次数（比如失败后1小时再试）。
• 性能优化：给send_email_at和is_sent字段添加联合索引，数据量大时采用分页查询，避免一次性加载过多数据。

内容的提问来源于stack exchange，提问作者Mark Kiraly