django入门到精通12 django2 + celery4自动化任务实现网站ssl证书的检测
我们在日常运维工作中如果管理的网站较多，经常会发生ssl证书过期而不能及时更新的问题，我们需要对域名证书的使用情况做检测，并且能及时知道什么时候续费证书并进行更新
这个项目只是个雏形，毕竟学了一段时间的django，用来小试牛刀，打通 前端 和 服务端及 redis,mysql的基本使用
功能如下：
a.可以添加、删除、修改网站
b.后台启用celery 任务对证书的信息进行更新
后续功能可以加入
1.如果证书过期实际小于10天发邮件报警
2.自动读取godaddy,aliyun等域名管理平台的域名信息并写入系统，还可以对二级域名进行批量检查，避免证书更新遗漏等
3.对网站的可用状态进行检查
4.对网站所在服务器的端口进行扫描汇总
...
celery原理
video 项目目录结构
# tree video
video
# 修改了默认的 vedeo 的settings.py 目录名为 config
├── config
│ ├── asgi.py
│ ├── celery.py
│ ├── __init__.py
│ ├── __pycache__
│ ├── settings.py
│ ├── urls.py
│ └── wsgi.py
├── manage.py
├── __pycache__
├── ssl_check
│ ├── admin.py
│ ├── apps.py
│ ├── check_ssl.py
│ ├── __init__.py
│ ├── migrations
│ │ ├── 0001_initial.py
│ │ ├── 0002_auto_20210222_2142.py
│ │ ├── __init__.py
│ │ └── __pycache__
│ ├── models.py
│ ├── __pycache__
│ ├── tasks.py
│ ├── tests.py
│ ├── urls.py
│ └── views.py
└── templates├── add_domain.html├── index.html└── list_domains.html
1.搭建基本的开发环境
pip install virtualenv
pip install virtualenvwrapper-win
# 创建虚拟环境 python379_django2
mkvirtualenv python379_django2
# 安装依赖
workon python379_django2
pip install celery==4.4.2
pip install eventlet==0.25.2
pip install Django==2.0.4
pip install pymysql
pip install requests
pip install redis==3.4.1
2.配置基本的数据库app等信息
video/config/settings.py
import os
from pathlib import PathBASE_DIR = Path(__file__).resolve().parent.parentSECRET_KEY = '7qh#b#@'DEBUG = True
ALLOWED_HOSTS = ['*']INSTALLED_APPS = ['django.contrib.admin','django.contrib.auth','django.contrib.contenttypes','django.contrib.sessions','django.contrib.messages','django.contrib.staticfiles','app','ssl_check',
]MIDDLEWARE = ['django.middleware.security.SecurityMiddleware','django.contrib.sessions.middleware.SessionMiddleware','django.middleware.common.CommonMiddleware','django.middleware.csrf.CsrfViewMiddleware','django.contrib.auth.middleware.AuthenticationMiddleware','django.contrib.messages.middleware.MessageMiddleware','django.middleware.clickjacking.XFrameOptionsMiddleware',
]ROOT_URLCONF = 'config.urls'
TEMPLATES = [{'BACKEND': 'django.template.backends.django.DjangoTemplates','DIRS': [os.path.join(BASE_DIR, 'templates')],'APP_DIRS': True,'OPTIONS': {'context_processors': ['django.template.context_processors.debug','django.template.context_processors.request','django.contrib.auth.context_processors.auth','django.contrib.messages.context_processors.messages',],},},
]WSGI_APPLICATION = 'config.wsgi.application'
DATABASES = {'default': {'ENGINE': 'django.db.backends.mysql','NAME': 'muke_video','USER': 'root','PASSWORD': 'root','HOST': '10.9.4.199','PORT': 3306}
}AUTH_PASSWORD_VALIDATORS = [{'NAME': 'django.contrib.auth.password_validation.UserAttributeSimilarityValidator',},{'NAME': 'django.contrib.auth.password_validation.MinimumLengthValidator',},{'NAME': 'django.contrib.auth.password_validation.CommonPasswordValidator',},{'NAME': 'django.contrib.auth.password_validation.NumericPasswordValidator',},
]LANGUAGE_CODE = 'zh-hans'TIME_ZONE = 'Asia/Shanghai'USE_I18N = True
USE_L10N = True
USE_TZ = True
STATIC_URL = '/static/'
STATICFILES_DIRS = (os.path.join(BASE_DIR, 'static'), )
CELERY_BROKER_URL = 'redis://10.9.4.199:6379/1'
CELERY_RESULT_BACKEND = 'redis://10.9.4.199:6379/2'CELERY_RESULT_SERIALIZER = 'json'
from celery.schedules import crontabCELERY_BEAT_SCHEDULE = {# 周期性任务# 'task-one': {
# 'task': 'ssl_check.tasks.print_test',# 'schedule': 5.0, # 每5秒执行一次# # 'args': ()# },'task-modify-ssl-info': {'task': 'ssl_check.tasks.modify_domain_ssl_info_task',# 'schedule': crontab(month_of_year='9', day_of_month='9', minute='*/1'), # 设置9月9日，每一分钟执行一次'schedule': crontab(minute='*/10'), # 每一分钟执行一次# 'args': ()}
}
2.celery异步任务插件的引入
在settings.py同级目录创建celery.py
# coding:utf-8
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery# 设置环境变量
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')# 注册Celery的APP，
app = Celery('config')
# 绑定配置文件
app.config_from_object('django.conf:settings', namespace='CELERY')# 自动发现各个app下的tasks.py文件
app.autodiscover_tasks()
注意 config 为你当前的django项目settings.py的目录名称
修改settings.py同级目录的 __init__.py文件
# coding:utf-8
from __future__ import absolute_import, unicode_literalsfrom .celery import app as celery_app# 导包
import pymysql# 初始化
pymysql.install_as_MySQLdb()__all__ = ['celery_app']
# 数据库models
video/ssl_check/models.py
from django.db import models# Create your models here.class Domain(models.Model):name = models.CharField(max_length=200, blank=False, unique=True)status = models.IntegerField(default=1)idc_id = models.IntegerField(default=0)company_id = models.IntegerField(default=0)ssl_start_date = models.CharField(max_length=200, blank=True, default="0000-00-00")ssl_expire_date = models.CharField(max_length=200, blank=True, default="0000-00-00")remaining_time = models.IntegerField(blank=True, default=0)def __str__(self):return "domain:{}".format(self.name)
在应用中创建tasks.py文件
video/ssl_check/tasks.py
# _*_ coding:utf-8 _*_
# __author__ == 'jack'
# __date__ == '2021-02-22 9:24 PM'import re
import time
import subprocess
from datetime import datetime
from io import StringIOfrom celery.task import task
from .models import Domain# 自定义要执行的task任务，这个测试任务
@task
def print_test():print("nict try")return "hello"def modify_domain_ssl_info(domain):print("domain_name",domain.name)print(type(domain))f = StringIO()comm = f"curl -Ivs https://{domain.name} --connect-timeout 10"print("comm:", comm)result = subprocess.getstatusoutput(comm)f.write(result[1])m = re.search('start date: (.*?)\n.*?expire date: (.*?)\n.*?common name: (.*?)\n.*?issuer: CN=(.*?)\n',f.getvalue(), re.S)start_date = m.group(1)expire_date = m.group(2)common_name = m.group(3)issuer = m.group(4)# time 字符串转时间数组start_date = time.strptime(start_date, "%b %d %H:%M:%S %Y GMT")start_date_st = time.strftime("%Y-%m-%d %H:%M:%S", start_date)# datetime 字符串转时间数组expire_date = datetime.strptime(expire_date, "%b %d %H:%M:%S %Y GMT")expire_date_st = datetime.strftime(expire_date, "%Y-%m-%d %H:%M:%S")# 剩余天数remaining = (expire_date - datetime.now()).daysprint('域名:', "domain")print('通用名:', common_name)print('开始时间:', start_date_st)print('到期时间:', expire_date_st)print(f'剩余时间: {remaining}天')print('颁发机构:', issuer)print('*' * 30)domain.remaining_time = remainingdomain.ssl_expire_date = expire_date_stdomain.ssl_start_date = start_date_st# domain.update(remaining_time=remaining, ssl_expire_date=expire_date_st, ssl_start_date=start_date_st)domain.save()f.close()time.sleep(0.5)@task
def modify_domain_ssl_info_task():domains = Domain.objects.all()print("start check domain ssl celery task")for domain in domains:modify_domain_ssl_info(domain)print("end check domain ssl celery task")return "modify ssl"
可以在settings.py里将该任务配置为定时任务(周期任务)
from celery.schedules import crontabCELERY_BEAT_SCHEDULE = {# 周期性任务# 'task-one': {
# 'task': 'ssl_check.tasks.print_test',# 'schedule': 5.0, # 每5秒执行一次# # 'args': ()# },'task-modify-ssl-info': {'task': 'ssl_check.tasks.modify_domain_ssl_info_task',# 'schedule': crontab(month_of_year='9', day_of_month='9', minute='*/1'), # 设置9月9日，每一分钟执行一次'schedule': crontab(minute='*/10'), # 每一分钟执行一次# 'args': ()}
}
同时异步任务也可以通过django的视图进行在线调用
# 这个是测试任务，可以通过django的web直接访问触发
class CetestView(View):def get(self, request, *args, **kwargs):res = print_test.delay()# 任务逻辑return JsonResponse({
'status': 'successful', 'task_id': res.task_id})# 可以通过web页面直接触发任务
class CheckSslView(View):def get(self, request):domains = Domain.objects.all()for domain in domains:self.check_ssl(domain)return redirect(reverse('list_domain'))def check_ssl(self, domain):print("domain_name",domain.name)print(type(domain))f = StringIO()comm = f"curl -Ivs https://{domain.name} --connect-timeout 10"print("comm:", comm)result = subprocess.getstatusoutput(comm)f.write(result[1])m = re.search('start date: (.*?)\n.*?expire date: (.*?)\n.*?common name: (.*?)\n.*?issuer: CN=(.*?)\n',f.getvalue(), re.S)start_date = m.group(1)expire_date = m.group(2)common_name = m.group(3)issuer = m.group(4)# time 字符串转时间数组start_date = time.strptime(start_date, "%b %d %H:%M:%S %Y GMT")start_date_st = time.strftime("%Y-%m-%d %H:%M:%S", start_date)# datetime 字符串转时间数组expire_date = datetime.strptime(expire_date, "%b %d %H:%M:%S %Y GMT")expire_date_st = datetime.strftime(expire_date, "%Y-%m-%d %H:%M:%S")# 剩余天数remaining = (expire_date - datetime.now()).daysprint('域名:', "domain")print('通用名:', common_name)print('开始时间:', start_date_st)print('到期时间:', expire_date_st)print(f'剩余时间: {remaining}天')print('颁发机构:', issuer)print('*' * 30)# 修改域名信息domain.remaining_time = remainingdomain.ssl_expire_date = expire_date_stdomain.ssl_start_date = start_date_stdomain.save()f.close()time.sleep(0.5)
视图信息 video/ssl_check/views.py
from django.shortcuts import render,reverse,redirectimport re
import time
import subprocess
from datetime import datetime
from io import StringIOfrom django.views import View
from .models import Domain
from .tasks import *
from django.http import JsonResponse'''
域名的增、删、查询 的视图
包括可以手动调取更新域名 ssl 证书信息的视图：CheckSslView'''
class DomainRegisterView(View):def get(self, request):return render(request, "add_domain.html")class AddDomainView(View):def post(self, request):domain_name = request.POST.get("domain_name", "")Domain.objects.create(name=domain_name)return redirect(reverse('list_domain'))class ListDomainView(View):def get(self, request):domains = Domain.objects.all()return render(request, 'list_domains.html', {
"domain_list":domains})class DeleteDomainView(View):def get(self, request):domain_id = request.GET.get("domain_id", "")print("domain_id={}".format(domain_id))Domain.objects.filter(id=domain_id).delete()return redirect(reverse('list_domain'))class DoChecksslTaskView(View):def get(self, request, *args, **kwargs):# 执行异步任务print('start to check domain ssl')# modify_domain_ssl_info_task.delay()# modify_domain_ssl_info_task.apply_async(args=('check',), queue='work_queue')res = modify_domain_ssl_info_task.delay()print('end do check domain ssl')# 任务逻辑return JsonResponse({
'status': 'successful', 'task_id': res.task_id})# 这个是测试任务，可以通过django的web直接访问触发
class CetestView(View):def get(self, request, *args, **kwargs):res = print_test.delay()# 任务逻辑return JsonResponse({
'status': 'successful', 'task_id': res.task_id})# 可以通过web页面直接触发任务
class CheckSslView(View):def get(self, request):domains = Domain.objects.all()for domain in domains:self.check_ssl(domain)return redirect(reverse('list_domain'))def check_ssl(self, domain):print("domain_name",domain.name)print(type(domain))f = StringIO()comm = f"curl -Ivs https://{domain.name} --connect-timeout 10"print("comm:", comm)result = subprocess.getstatusoutput(comm)f.write(result[1])m = re.search('start date: (.*?)\n.*?expire date: (.*?)\n.*?common name: (.*?)\n.*?issuer: CN=(.*?)\n',f.getvalue(), re.S)start_date = m.group(1)expire_date = m.group(2)common_name = m.group(3)issuer = m.group(4)# time 字符串转时间数组start_date = time.strptime(start_date, "%b %d %H:%M:%S %Y GMT")start_date_st = time.strftime("%Y-%m-%d %H:%M:%S", start_date)# datetime 字符串转时间数组expire_date = datetime.strptime(expire_date, "%b %d %H:%M:%S %Y GMT")expire_date_st = datetime.strftime(expire_date, "%Y-%m-%d %H:%M:%S")# 剩余天数remaining = (expire_date - datetime.now()).daysprint('域名:', "domain")print('通用名:', common_name)print('开始时间:', start_date_st)print('到期时间:', expire_date_st)print(f'剩余时间: {remaining}天')print('颁发机构:', issuer)print('*' * 30)# 修改域名信息domain.remaining_time = remainingdomain.ssl_expire_date = expire_date_stdomain.ssl_start_date = start_date_stdomain.save()f.close()time.sleep(0.5)
4.模板文件
# 添加页面
templates/add_domain.html
<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><title>add</title>
</head>
<body><form action="/domain/add/" method="post">{% csrf_token %}域名：<input type="text" name="domain_name" /> <br/><input type="submit" value="提交"></form></body>
</html>
# 列出域名信息页面
templates/list_domain.html
<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><title>域名列表</title>
</head>
<body><table border="1" style="solid-color: black" cellpadding="0" cellspacing="0" width="30%"><tr><th>域名</th><th>状态</th><th>证书启用时间</th><th>证书过期时间</th><th>证书剩余过期时间</th><th>操作</th></tr>{% for domain in domain_list %}<tr><td>{
{ domain.name }}</td><td>{
{ domain.status }}</td><td>{
{ domain.ssl_expire_date }}</td><td>{
{ domain.ssl_start_date }}</td><td>{
{ domain.remaining_time }}</td><td> <a href="/domain/delete/?domain_id={
{ domain.id }}">删除</a></td></tr>{% endfor %}</table></body>
</html>
页面
这里的delay方法就是异步方式请求，而非django默认的同步执行步骤
在manage.py的目录下启动celery服务
windows 中启动方式
(python37_django2) D:\python\django_imooc_xiaobai\muke_vedio_test\video>celery worker -A config -l info -P eventlet-------------- celery@SZ18052967C01 v4.4.2 (cliffs)
--- ***** -----
-- ******* ---- Windows-10-10.0.19041-SP0 2021-02-23 17:03:40
- *** --- * ---
- ** ---------- [config]
- ** ---------- .> app: config:0x1ca2a3bf128
- ** ---------- .> transport: redis://localhost:6379/1
- ** ---------- .> results: redis://localhost:6379/2
- *** --- * --- .> concurrency: 8 (eventlet)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ------------------- [queues].> celery exchange=celery(direct) key=celery[tasks]. ssl_check.tasks.print_test[2021-02-23 17:03:40,585: INFO/MainProcess] Connected to redis://localhost:6379/1
[2021-02-23 17:03:40,604: INFO/MainProcess] mingle: searching for neighbors
[2021-02-23 17:03:41,757: INFO/MainProcess] mingle: all alone
[2021-02-23 17:03:41,777: INFO/MainProcess] pidbox: Connected to redis://localhost:6379/1.
[2021-02-23 17:03:41,782: WARNING/MainProcess] c:\users\ws\envs\python37_django2\lib\site-packages\celery\fixups\django.py:203: UserWarning: Using settings.DEBUG leads to a memoryleak, never use this setting in production environments!leak, never use this setting in production environments!''')
[2021-02-23 17:03:41,782: INFO/MainProcess] celery@SZ18052967C01 ready.
[2021-02-23 17:03:41,784: INFO/MainProcess] Received task: ssl_check.tasks.print_test[383d9f23-1224-481f-8a3e-364b07dc2dc8]
[2021-02-23 17:03:41,785: WARNING/MainProcess] nict try
[2021-02-23 17:03:41,788: INFO/MainProcess] Received task: ssl_check.tasks.print_test[0959e985-d4ee-4439-8dd7-0a0e2377e610]
[2021-02-23 17:03:41,790: WARNING/MainProcess] nict try
[2021-02-23 17:03:41,792: INFO/MainProcess] Task ssl_check.tasks.print_test[383d9f23-1224-481f-8a3e-364b07dc2dc8] succeeded in 0.0s: 'hello'
[2021-02-23 17:03:41,793: INFO/MainProcess] Task ssl_check.tasks.print_test[0959e985-d4ee-4439-8dd7-0a0e2377e610] succeeded in 0.0s: 'hello'
[2021-02-23 17:03:41,793: INFO/MainProcess] Received task: ssl_check.tasks.print_test[f418314a-a01e-48df-834f-79220526eca2]
[2021-02-23 17:03:41,794: WARNING/MainProcess] nict try
[2021-02-23 17:03:41,795: INFO/MainProcess] Task ssl_check.tasks.print_test[f418314a-a01e-48df-834f-79220526eca2] succeeded in 0.0s: 'hello'
[2021-02-23 17:03:41,795: INFO/MainProcess] Received task: ssl_check.tasks.print_test[a6b95ed9-7a37-49ff-8445-2db250fd402b]
[2021-02-23 17:03:41,796: WARNING/MainProcess] nict try
[2021-02-23 17:03:41,796: INFO/MainProcess] Task ssl_check.tasks.print_test[a6b95ed9-7a37-49ff-8445-2db250fd402b] succeeded in 0.0s: 'hello'
[2021-02-23 17:03:41,797: INFO/MainProcess] Received task: ssl_check.tasks.print_test[0d18e1f3-7af3-4776-903b-97517c4d7d81]
[2021-02-23 17:03:41,798: WARNING/MainProcess] nict try
[2021-02-23 17:03:41,799: INFO/MainProcess] Task ssl_check.tasks.print_test[0d18e1f3-7af3-4776-903b-97517c4d7d81] succeeded in 0.0s: 'hello'
[2021-02-23 17:03:41,799: INFO/MainProcess] Received task: ssl_check.tasks.print_test[1b415bd2-e7da-49cc-810c-57c99f1b80c1]
[2021-02-23 17:03:41,800: WARNING/MainProcess] nict try
[2021-02-23 17:03:41,800: INFO/MainProcess] Task ssl_check.tasks.print_test[1b415bd2-e7da-49cc-810c-57c99f1b80c1] succeeded in 0.0s: 'hello'
linux中调用
celery -A config worker -l info
在浏览器中调用异步服务接口
127.0.0.1:8000/domain/ssl/
同时也可以在backend中查询任务结果
注意一点，redis中的key并不是单纯的task_id，而是需要加上前缀celery-task-meta-
最后，如果需要启动定时任务，就需要在manage.py所在的文件夹内单独启动beat服务
windows中调用方法
(python37_django2) D:\python\django_imooc_xiaobai\muke_vedio_test\video>celery -A config beat -l info
celery beat v4.4.2 (cliffs) is starting.
__ - ... __ - _
LocalTime -> 2021-02-23 17:15:23
Configuration ->. broker -> redis://localhost:6379/1. loader -> celery.loaders.app.AppLoader. scheduler -> celery.beat.PersistentScheduler. db -> celerybeat-schedule. logfile -> [stderr]@%INFO. maxinterval -> 5.00 minutes (300s)
[2021-02-23 17:15:23,454: INFO/MainProcess] beat: Starting...
[2021-02-23 17:15:28,555: INFO/MainProcess] Scheduler: Sending due task task-one (ssl_check.tasks.print_test)
linux中调用
celery -A config beat -l info