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 Path BASE_DIR = Path(__file__).resolve().parent.parent SECRET_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 crontab CELERY_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_literals from .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 StringIO from 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()).days print(‘域名:‘, "domain") print(‘通用名:‘, common_name) print(‘开始时间:‘, start_date_st) print(‘到期时间:‘, expire_date_st) print(f‘剩余时间: {remaining}天‘) print(‘颁发机构:‘, issuer) print(‘*‘ * 30) domain.remaining_time = remaining domain.ssl_expire_date = expire_date_st domain.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 crontab CELERY_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()).days print(‘域名:‘, "domain") print(‘通用名:‘, common_name) print(‘开始时间:‘, start_date_st) print(‘到期时间:‘, expire_date_st) print(f‘剩余时间: {remaining}天‘) print(‘颁发机构:‘, issuer) print(‘*‘ * 30) # 修改域名信息 domain.remaining_time = remaining domain.ssl_expire_date = expire_date_st domain.ssl_start_date = start_date_st domain.save() f.close() time.sleep(0.5)
视图信息 video/ssl_check/views.py
from django.shortcuts import render,reverse,redirect import re import time import subprocess from datetime import datetime from io import StringIO from 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()).days print(‘域名:‘, "domain") print(‘通用名:‘, common_name) print(‘开始时间:‘, start_date_st) print(‘到期时间:‘, expire_date_st) print(f‘剩余时间: {remaining}天‘) print(‘颁发机构:‘, issuer) print(‘*‘ * 30) # 修改域名信息 domain.remaining_time = remaining domain.ssl_expire_date = expire_date_st domain.ssl_start_date = start_date_st domain.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 memory leak, 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