基于Django+Celery实现群组会议前定时邮件通知的问题

现有Group模型 #

class Group(models.Model):
   leader = models.ForeignKey(User, on_delete=models.CASCADE)
   name = models.CharField(max_length=55)
   description = models.TextField()
   joined = models.ManyToManyField(User, blank=True)
   start_time = models.TimeField(null=True)
   end_time = models.TimeField(null=True)
   email_list = ArrayField(
        models.CharField(max_length=255, blank=True),
        blank=True,
        default=list,
    )

需求 #

1. 自动在群组start_time前30分钟，给所有joined该群组的用户发送邮件提醒（例如start_time为13:00，则12:30发送）
2. 后续支持让群组leader自定义提醒提前时长（如10分钟、30分钟、1小时）

已完成内容 #

• 已实现由用户操作（创建、更新、加入、退出、删除群组）触发的Celery任务，运行正常
• 已编写发送提醒邮件的Celery任务：

from celery import shared_task
from celery.utils.log import get_task_logger
from django.core.mail import send_mail
from my_chaburah.settings import NOTIFICATION_EMAIL
from django.template.loader import render_to_string

@shared_task(name='start_group_notification_task')
def start_group_notification_task(recipients):
    logger.info('sent email to whole group that group is starting')
    for recipient in recipients:
        send_mail (
                    'group starting',
                    'group starting',
                    NOTIFICATION_EMAIL,
                    [recipient],
                    fail_silently=False
                )

现有配置文件 #

myapp/celery.py #

import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'my_group.settings')

app = Celery('my_group')

app.config_from_object('django.conf:settings', namespace='CELERY')

app.autodiscover_tasks()

@app.task(bind=True, ignore_result=True)
def debug_task(self):
    print(f'Request: {self.request!r}')

group/tasks.py #

from celery import shared_task
from celery.utils.log import get_task_logger
from django.core.mail import send_mail
from my_chaburah.settings import NOTIFICATION_EMAIL
from django.template.loader import render_to_string

logger = get_task_logger(__name__)

困惑点 #

1. 如何通过Celery Beat周期性任务调用已编写的邮件发送任务
2. 如何查询出需要触发提醒的群组（即当前时间加上提前时长等于群组start_time的群组）
3. 与Group相关的Celery任务应该放在哪个位置

解决方案 #

1. 修改Group模型，支持自定义提醒提前时长 #

在Group模型中添加字段存储提前提醒的分钟数，用可选值限制范围：

from django.db import models

class Group(models.Model):
    # 原有字段保持不变
    leader = models.ForeignKey(User, on_delete=models.CASCADE)
    name = models.CharField(max_length=55)
    description = models.TextField()
    joined = models.ManyToManyField(User, blank=True)
    start_time = models.TimeField(null=True)
    end_time = models.TimeField(null=True)
    email_list = ArrayField(
        models.CharField(max_length=255, blank=True),
        blank=True,
        default=list,
    )
    # 新增字段：自定义提前提醒时长，默认30分钟
    REMINDER_CHOICES = [
        (10, '10分钟'),
        (30, '30分钟'),
        (60, '1小时'),
    ]
    reminder_offset = models.IntegerField(choices=REMINDER_CHOICES, default=30)

修改后执行数据库迁移：

python manage.py makemigrations
python manage.py migrate

2. 编写周期性扫描任务（放在group/tasks.py中） #

这个任务每分钟运行一次，查询符合提醒条件的群组并调用邮件发送任务：

from datetime import datetime, timedelta
from django.utils import timezone
from .models import Group
from django.db.models import F, ExpressionWrapper, TimeField
from django.db.models.functions import ExtractHour, ExtractMinute

@shared_task(name='scan_groups_for_reminders')
def scan_groups_for_reminders():
    # 获取当前时区时间
    now = timezone.localtime(timezone.now())
    current_time = now.time()

    # 计算当前时间加上群组提醒时长后的目标时间，匹配群组start_time
    groups = Group.objects.filter(
        start_time__isnull=False,
        start_time=ExpressionWrapper(
            (ExtractHour(current_time)*3600 + ExtractMinute(current_time)*60 + F('reminder_offset')*60) % 86400,
            output_field=TimeField()
        )
    )

    for group in groups:
        # 收集所有收件人邮箱并去重
        user_emails = list(group.joined.values_list('email', flat=True).distinct())
        all_recipients = list(set(user_emails + group.email_list))
        
        if all_recipients:
            start_group_notification_task.delay(all_recipients)
            logger.info(f"触发群组{group.name}的提醒邮件，发送给{len(all_recipients)}个收件人")

3. 配置Celery Beat，让扫描任务每分钟运行 #

在Django的settings.py中添加Celery Beat配置：

CELERY_BEAT_SCHEDULE = {
    'scan-groups-for-reminders-every-minute': {
        'task': 'group.tasks.scan_groups_for_reminders',
        'schedule': 60.0,  # 每分钟运行一次
    },
}

4. 调整邮件发送任务（统一放在group/tasks.py中） #

优化邮件内容渲染，统一任务管理：

# group/tasks.py中完善邮件发送任务
@shared_task(name='start_group_notification_task')
def start_group_notification_task(recipients):
    logger.info(f'给{len(recipients)}个收件人发送群组开始提醒')
    # 使用模板渲染HTML邮件内容
    html_content = render_to_string('group_reminder_email.html', {
        'subject': '您的群组即将开始',
        'message': '您加入的群组马上就要开始了，请做好准备！'
    })
    for recipient in recipients:
        send_mail(
            '您的群组即将开始',
            '您加入的群组马上就要开始了，请做好准备！',
            NOTIFICATION_EMAIL,
            [recipient],
            fail_silently=False,
            html_message=html_content
        )

5. 启动Celery服务 #

确保Celery Worker和Beat同时运行：

# 启动Celery Worker
celery -A my_group worker --loglevel=info
# 启动Celery Beat
celery -A my_group beat --loglevel=info

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