Flask实现每周自动化用户邮件推送方案咨询

1. 配置Celery定时任务（Beat） #

Celery自带的Beat组件是实现定时触发的核心，只需配置定时规则并启动Beat进程即可：

• 先在Celery配置中添加定时任务规则，指定每周一的执行时间（比如早9点）
• 启动Celery Beat和Worker，Beat负责触发定时任务，Worker负责执行任务

配置示例：

from celery.schedules import crontab
from celery import Celery

celery = Celery('app')
celery.conf.update(
    broker_url='你的Broker地址（如redis://localhost:6379/0）',
    result_backend='你的Backend地址',
    beat_schedule={
        'weekly-user-summary-task': {
            'task': 'app.tasks.send_weekly_summary_emails',
            'schedule': crontab(hour=9, minute=0, day_of_week=1),  # 每周一9点执行
        },
    }
)

启动命令：

# 启动Celery Worker
celery -A app.celery worker -l info
# 启动Celery Beat（仅需一个实例）
celery -A app.celery beat -l info

2. 编写行为统计与邮件触发逻辑 #

定时任务触发后，需要完成「统计用户上周行为」「生成邮件内容」「调用已有异步邮件任务」三个步骤：

• 先计算上周的时间范围（自然周：上周一早0点到周日晚23:59）
• 遍历用户（大用户量建议分页查询），统计每个用户的帖子、评论数量
• 调用你已有的send_user_email异步任务发送邮件

任务函数示例：

from datetime import datetime, timedelta
from app.models import User, Post, Comment
from app.tasks import send_user_email  # 你已有的异步邮件发送任务
from sqlalchemy import func

def get_last_week_range():
    today = datetime.today()
    # 计算上周一开始时间
    last_week_start = today - timedelta(days=today.weekday() + 7)
    last_week_start = last_week_start.replace(hour=0, minute=0, second=0)
    # 计算上周结束时间
    last_week_end = last_week_start + timedelta(days=6, hours=23, minutes=59, seconds=59)
    return last_week_start, last_week_end

@celery.task
def send_weekly_summary_emails():
    start, end = get_last_week_range()
    # 分页查询用户，避免一次性加载大量数据
    page = 1
    per_page = 100
    while True:
        users = User.query.paginate(page=page, per_page=per_page)
        if not users.items:
            break
        for user in users.items:
            # 用聚合查询提升统计效率
            post_count = Post.query.filter(
                Post.author_id == user.id,
                Post.created_at.between(start, end)
            ).with_entities(func.count(Post.id)).scalar() or 0
            
            comment_count = Comment.query.filter(
                Comment.author_id == user.id,
                Comment.created_at.between(start, end)
            ).with_entities(func.count(Comment.id)).scalar() or 0
            
            # 生成邮件内容
            email_body = f"""
            你好{user.username}：
            这是你上周的应用行为汇总：
            - 发布帖子数：{post_count}
            - 评论数：{comment_count}
            感谢使用我们的应用！
            """
            # 调用异步邮件任务
            send_user_email.delay(user.email, "上周应用行为汇总", email_body)
        page += 1

3. 关键优化与注意事项 #

• 性能优化：大用户量场景下必须用分页查询用户，统计时使用数据库聚合函数（如func.count）而非ORM的count()，减少内存占用和查询时间
• 避免重复发送：给User表添加last_weekly_summary_sent字段，记录上次发送时间，发送前校验是否已发送本周邮件，防止Beat重启导致重复执行
• 错误处理：在邮件任务中添加异常捕获，记录发送失败的用户和原因，避免单个用户发送失败中断整个任务流程；可对失败任务设置重试机制
• 测试验证：开发阶段可将定时规则改为每分钟执行（crontab(minute='*/1')），快速验证统计逻辑和邮件发送流程，确认无误后再改为每周一
• 日志监控：在任务中添加日志记录（如执行时间、处理用户数、成功/失败数量），生产环境可搭配Celery Flower监控任务状态

4. 部署注意事项 #

• 用Supervisor或Systemd管理Celery Worker和Beat进程，确保后台稳定运行
• 分布式部署时，Beat仅需启动一个实例，避免重复触发定时任务
• 定期检查Broker（如Redis）的运行状态，防止任务堆积

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