python毕业设计-基于Django+channels+gojs实现的代码发布系统设计与实现+使用说明.zip

# @Time : 2022/6/4 9:22 # @Author : kongsh # @File : consumers.py import json import time import threading import os import subprocess import shutil import traceback from channels.generic.websocket import WebsocketConsumer from channels.exceptions import StopConsumer from asgiref.sync import async_to_sync from django.conf import settings from web import models from utils.git_util import GitRepository from utils.ssh import SSHProxy consumer_object_list = [] def get_node(task_obj, task_id): """返回节点数据""" node_obj_list = [] queryset = models.Node.objects.filter(task_id=task_id) if queryset: return queryset # 没有的话在数据库创建节点 start_node = models.Node.objects.create(text="开始", task_id=task_id) node_obj_list.append(start_node) # 判断第1个钩子是否有自定义脚本：任务单对象.before_download_script if task_obj.before_download_script: start_node = models.Node.objects.create(text='下载前', task_id=task_id, parent=start_node) node_obj_list.append(start_node) download_node = models.Node.objects.create(text="下载", task_id=task_id, parent=start_node) node_obj_list.append(download_node) # 判断第2个钩子是否有自定义脚本：任务单对象.after_download_script if task_obj.after_download_script: download_node = models.Node.objects.create(text='下载后', task_id=task_id, parent=download_node) node_obj_list.append(download_node) upload_node = models.Node.objects.create(text="上传", task_id=task_id, parent=download_node) node_obj_list.append(upload_node) task_obj = models.Deploy.objects.filter(id=task_id).first() for server_obj in task_obj.project.servers.all(): row = models.Node.objects.create(text=server_obj.hostname, task_id=task_id, parent=upload_node, server=server_obj) node_obj_list.append(row) # 判断是否有发布前钩子 if task_obj.before_deploy_script: row = models.Node.objects.create(text="发布前", task_id=task_id, parent=row, server=server_obj) node_obj_list.append(row) deploy_node = models.Node.objects.create(text="发布", task_id=task_id, parent=row, server=server_obj) node_obj_list.append(deploy_node) # 判断是否有发布后钩子 if task_obj.after_deploy_script: row = models.Node.objects.create(text="发布后", task_id=task_id, parent=deploy_node, server=server_obj) node_obj_list.append(row) return node_obj_list def convert_obj_to_gojs(node_obj_list): """将对象列表转换为gojs识别的json格式""" node_list = [] for node in node_obj_list: tmp = { 'key': str(node.id), 'text': node.text, 'color': node.status, 'stroke': 'white', } if node.parent: tmp['parent'] = str(node.parent.id) node_list.append(tmp) return node_list class PublishConsumer(WebsocketConsumer): def websocket_connect(self, message): """ 请求websocket链接时自动触发 :param message: :return: """ # 与客户端建立链接 self.accept() # 将自己加入到task_id群里，接收以后发来的消息 # 【固定写法】获取url中传递的参数,如果url中定义的是关键字参数，就用kwargs,否则就用args task_id = self.scope['url_route']['kwargs'].get('task_id') async_to_sync(self.channel_layer.group_add)(task_id, self.channel_name) # 用户打开页面时，如果已经创建好节点则直接返回 queryset = models.Node.objects.filter(task_id=task_id) if queryset: node_list = convert_obj_to_gojs(queryset) self.send(text_data=json.dumps({'code': 'init', 'data': node_list})) def websocket_receive(self, message): """ 客户端发送消息时自动触发 :param message: :return: """ # print("客户端消息：", message) task_id = self.scope['url_route']['kwargs'].get('task_id') task_obj = models.Deploy.objects.filter(id=task_id).first() text = message.get('text') if text == 'init': # node_list = [ # {'key': "start", 'text': '开始', 'figure': 'Ellipse'}, # {'key': "download", 'parent': 'start', 'text': '下载代码', 'link_text': '执行中...'}, # {'key': "compile", 'parent': 'download', 'text': '本地编译'}, # ] # 第1步:数据库没有的话先创建节点，有的话直接读取 node_obj_list = get_node(task_obj, task_id) # 第2步:根据对象列表封装好指定格式数据，以json格式返回给gojs node_list = convert_obj_to_gojs(node_obj_list) # self.send(text_data=json.dumps({'code': 'init', 'data': node_list})) # 第1个参数是群号； # 第2个参数是给这个群内的每个用户发送消息的函数； # 第3个参数是发送的消息 async_to_sync(self.channel_layer.group_send)(task_id, {'type': 'xxx.yyy', 'message': {'code': 'init', 'data': node_list}}) if text == 'deploy': # 代码发布 # channels的缺陷，有sleep的时候，颜色在最后统一展示，不能逐个展示，需要将执行函数放到另外一个线程 task = threading.Thread(target=self.deploy, args=(task_id, task_obj)) task.start() def xxx_yyy(self, event): """发送消息函数，会给每个用户都发送message字段中的内容""" message = event['message'] self.send(json.dumps(message)) def websocket_disconnect(self, message): """断开websocket链接时触发""" task_id = self.scope['url_route']['kwargs'].get('task_id') # 群聊里剔除离开的用户,self.channel_name相当于给每个人创建的唯一id async_to_sync(self.channel_layer.group_discard)(task_id, self.channel_name) raise StopConsumer() def _change_node_status(self, text, task_id, server=None, color='green'): """ 修改指定节点的状态和颜色 :param text: :param server:主机 :param task_id: :param color:修改后的颜色 :return: """ if server: node = models.Node.objects.filter(text=text, task_id=task_id, server=server).first() else: node = models.Node.objects.filter(text=text, task_id=task_id).first() node.status = color node.save() # 群发 async_to_sync(self.channel_layer.group_send)(task_id, {'type': 'xxx.yyy', 'message': {'code': 'update', 'node_id': node.id, 'color': node.status}}) def _send_log(self, task_id, info): """实时输出日志信息""" # 群发 async_to_sync(self.channel_layer.group_send)(task_id, {'type': 'xxx.yyy', 'me