Python中如何实现定时给达到条件的用户发送短信功能

用Celery配合APScheduler或者直接用Celery Beat就能搞定。下面给你个完整的例子，用Django + Celery + Redis。

首先，安装依赖：

pip install django celery redis django-celery-beat

1. 配置Celery

在Django项目的your_project/celery.py中：

import os
from celery import Celery
from celery.schedules import crontab

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

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

# 定时任务配置
app.conf.beat_schedule = {
    'send-sms-every-minute': {
        'task': 'your_app.tasks.send_sms_to_eligible_users',
        'schedule': crontab(minute='*/1'),  # 每分钟执行一次
    },
}

2. 创建任务

在your_app/tasks.py中：

from celery import shared_task
from django.contrib.auth.models import User
from datetime import datetime, timedelta

@shared_task
def send_sms_to_eligible_users():
    # 定义条件：例如，注册超过24小时但未激活的用户
    cutoff_time = datetime.now() - timedelta(hours=24)
    
    eligible_users = User.objects.filter(
        date_joined__lt=cutoff_time,
        is_active=False
    )
    
    for user in eligible_users:
        # 这里调用你的短信发送接口
        send_sms(user.phone, "您的账户需要激活")
        print(f"已发送短信给 {user.username}")

3. 启动服务

开三个终端分别运行：

# 启动Redis
redis-server

# 启动Celery worker
celery -A your_project worker --loglevel=info

# 启动Celery beat（定时任务调度器）
celery -A your_project beat --loglevel=info

4. 短信发送函数示例

简单实现个发送函数：

import requests

def send_sms(phone_number, message):
    # 这里替换成你的短信服务商API
    url = "https://your-sms-api.com/send"
    data = {
        "phone": phone_number,
        "message": message,
        "api_key": "your_api_key"
    }
    response = requests.post(url, data=data)
    return response.status_code == 200

关键点：

• 用Celery处理异步任务，避免阻塞主程序
• Celery Beat负责定时触发
• 条件查询写在任务函数里，根据业务需求调整
• 生产环境记得配置结果后端和监控

这样就能定时扫描数据库，给符合条件的用户发短信了。根据实际需求调整查询条件和发送频率就行。

总结：用Celery + Beat实现定时任务，在任务函数里写条件查询和短信发送逻辑。