如何自动检测Django用户订阅是否到期并修改对应模型字段

前置优化（可选） #

你当前views.py中不同套餐的处理逻辑重复度过高，可先通过字典映射简化代码，降低后续维护成本：

# 先定义套餐映射规则，后续加新套餐直接改这个字典就行
PLAN_CONFIG = {
    '7days': {'days':7, 'price':2000},
    '14days': {'days':14, 'price':3500},
    '21days': {'days':21, 'price':5000}
}

if request.method == 'POST':
    agreement = Contract()
    plan = request.user.plan
    # 取对应配置，默认走21天
    config = PLAN_CONFIG.get(plan, PLAN_CONFIG['21days'])
    current_time = datetime.now()
    
    agreement.user = request.user
    agreement.contract_status = True
    agreement.date_of_acceptance = current_time
    agreement.date_of_expiration = current_time + timedelta(days=config['days'])
    agreement.save()

    # 发送用户邮件
    user_template = render_to_string('contract_email.html', {
        'name': request.user.full_name, 
        'email': request.user.email, 
        'plan': plan,
        'price': config['price'], 
        'accept': agreement.date_of_acceptance, 
        'expire':agreement.date_of_expiration
    })
    user_email = EmailMessage(
        'Copy of Contract',
        user_template,
        settings.EMAIL_HOST_USER,
        [request.user.email],
    )
    user_email.fail_silently = False
    user_email.content_subtype = 'html'
    user_email.send()

    # 发送管理员邮件
    admin_template = render_to_string('contract_email.html', {
        'name': request.user.full_name, 
        'email': request.user.email, 
        'plan': plan,
        'price': config['price'], 
        'accept': agreement.date_of_acceptance, 
        'expire':agreement.date_of_expiration
    })
    admin_email = EmailMessage(
        'Copy of Customer Contract',
        admin_template,
        settings.EMAIL_HOST_USER,
        ['owner-email'],
    )
    admin_email.fail_silently = False
    admin_email.content_subtype = 'html'
    admin_email.send()

    return redirect('contract')
else:
    return render(request, 'contract.html')

核心功能实现方案 #

以下三种方案适配不同项目场景，可按需选择：

方案1：Django自定义管理命令 + 系统定时任务（优先推荐，轻量无额外依赖） #

无需引入新的中间件，开发成本最低，适合绝大多数中小项目：

1. 在你存放Contract模型的APP目录下，创建层级目录 management/commands/，目录下新建空文件 __init__.py 用于标识为Python包
2. 同目录下新建 expire_contracts.py 文件，写入以下代码：

from django.core.management.base import BaseCommand
from django.utils import timezone
from yourapp.models import Contract # 替换成你自己的app名

class Command(BaseCommand):
    help = '将所有到期的订阅合约状态设置为失效'

    def handle(self, *args, **options):
        # 筛选所有已到期且仍处于生效状态的合约
        expired_contracts = Contract.objects.filter(
            date_of_expiration__lt = timezone.now(),
            contract_status = True
        )
        # 批量更新状态，支持扩展其他触发操作
        update_count = expired_contracts.update(contract_status = False)
        # 这里可以加其他扩展逻辑，比如给到期用户发提醒邮件、记录日志等
        self.stdout.write(self.style.SUCCESS(f'成功更新了{update_count}份到期合约的状态'))

3. 手动执行测试：运行命令 python manage.py expire_contracts 验证功能是否正常
4. 配置系统定时任务自动执行：
   • Linux系统可配置crontab，示例为每天凌晨1点执行一次：

   0 1 * * * /usr/bin/python3 /你的项目根目录绝对路径/manage.py expire_contracts >> /var/log/contract_expire.log 2>&1
   

   • Windows系统可通过「任务计划程序」配置对应定时任务，执行同样的命令即可

方案2：Celery定时任务（适合已引入Celery的项目，扩展性更强） #

如果你的项目已经用了Celery做异步任务，可直接用Celery Beat实现定时检测，还支持为单个用户创建延迟任务，到期自动触发，无需全局扫表：

1. 编写Celery任务：

from celery import shared_task
from django.utils import timezone
from yourapp.models import Contract

@shared_task
def check_expired_contracts():
    expired_contracts = Contract.objects.filter(
        date_of_expiration__lt = timezone.now(),
        contract_status = True
    )
    expired_contracts.update(contract_status = False)
    # 扩展其他操作
    return f'更新了{expired_contracts.count()}份到期合约'

2. 配置Celery Beat定时规则，示例每天凌晨1点执行：

# 项目celery.py配置
from celery.schedules import crontab

app.conf.beat_schedule = {
    'check-expired-contracts-daily': {
        'task': 'yourapp.tasks.check_expired_contracts',
        'schedule': crontab(hour=1, minute=0),
    },
}

*如果需要更高实时性，也可以在用户同意协议保存合约后，直接为该用户创建延迟任务，指定到到期时间点执行状态更新操作，无需定时扫表。

方案3：请求触发被动检测（零额外配置，适合小型个人项目） #

无需配置定时任务，在用户登录、访问需要订阅权限的接口时，先校验当前用户的合约状态：

# 可以封装成装饰器或者中间件全局使用
def check_contract_status(user):
    try:
        contract = Contract.objects.get(user=user)
        if contract.contract_status and contract.date_of_expiration < timezone.now():
            contract.contract_status = False
            contract.save()
            # 扩展其他触发操作
            return False
        return contract.contract_status
    except Contract.DoesNotExist:
        return False

该方案缺点是如果用户长期不登录，状态不会自动更新，仅适合对状态实时性要求不高的场景。

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