发布：2023/10/24 22:23:35作者：管理员 来源：本站 浏览次数：296

这篇文章主要为大家详细介绍了如何利用Python编写一个自动化部署工具，文中的示例代码讲解详细，具有一定的学习价值，需要的可以了解一下

   效果

   起因

   实现步骤

       开发准备

       步骤

效果

起因

现在springboot项目的自动化部署已经非常普遍，有用Jenkins的，有用git钩子函数的，有用docker的...等等。这段时间在玩python，想着用python实现自动化部署，即能锻炼下编码能力，又方便运维。于是开始着手写了一个exe程序，可直接在任何windows电脑上运行（不具备python环境的windows电脑也可以运行）。有兴趣的小伙伴可以跟着代码一起练一练噢，写的详细一点，对python新手也很友好。

实现步骤

开发准备

   具有python基本环境和ide的windows或macOS电脑一台

   安装打包工具pip install pyinstaller

   一点小小的python基础

步骤

1. 导入依赖

新建一个py文件，可以把它命名为 deployment.py（名字随意哈，什么名儿都可以），然后把下面的库导入语句copy到此py文件中

import os #用于-提取文件名  

import re #用于-正则表达式  

import time #用于-线程休眠  

import paramiko #用于-远程执行linux命令  

from alive_progress import alive_bar #用于-进度条工具类  

from cryptography.fernet import Fernet #用于-加解密代码  

import base64 #用于-加解密代码  

import hashlib #用于-加解密代码

在导入依赖的时候，可能有些依赖咱们的电脑上之前没下载过，不要紧，只需要在pycharm中按 alt+enter就可以自动导入了，PyCharm跟Idea的快捷键一模一样，可以按Idea的习惯使用。而且在python中还不用配置maven或pom文件，非常方便。

2. 输入校验

部署毕竟是件严谨的事情，我们增加个部署密钥校验，我的这个部署密钥承担了以下的功能

   确保部署的安全性，不是谁拿到这个exe程序都能运行的（哼~傲娇）

   密钥字符串用-分割开，前面的区分环境，后面的区分项目或模块。

   如果同学们不需要区分项目子模块，就不需要搞这么复杂，随便定义一个密钥就好了

import os #用于-提取文件名  

import re #用于-正则表达式  

import time #用于-线程休眠  

import paramiko #用于-远程执行linux命令  

from alive_progress import alive_bar #用于-进度条工具类  

from cryptography.fernet import Fernet #用于-加解密代码  

import base64 #用于-加解密代码  

import hashlib #用于-加解密代码  

#检查密钥格式

def check_deploy_sign(deploy_site):  

   #确保密钥只能是以下4个之一才能继续往下操作，否则无限循环输入 或 退出程序

   if deploy_site != 'pro-main' and deploy_site != 'pro-manage' and deploy_site != 'test-main' and deploy_site != 'test-manage':

       #校验失败，一直校验

       new_deploy_site = input("错误：请填写部署密钥：")  

       check_deploy_sign(new_deploy_site)  

    #校验成功，退出

    return deploy_site  

try:  

   deploy_sign = input("提示：请填写部署密钥：")  

   deploy_sign = check_deploy_sign(deploy_sign)  

   # 部署环境  pro代表生成环境，test代表测试环境

   deploy_server = deploy_sign.split('-')[0]  

   # 部署模块或项目 manage代表manage模块，main代表main模块，

   deploy_site = deploy_sign.split('-')[1]  

   # 打包时的包名，三目运算符

   package_name = 'production' if deploy_server == 'pro' else 'staging'

except Exception as e:  

   print(f"异常: {str(e)}")

上面的代码中 增加了全局的异常处理，类似Java的try catch，也定义了一些基本的变量。密钥是一串由短线连接的字符串，短线前的代码用以区分环境，短线后的代码用以区分模块或项目。另外上面代码中的package_name是打包时的包名（即profiles.profile.id），一般配置在springboot项目pom文件中的编辑模块，类似下面这样:

3. 连接linux服务器

import os #用于-提取文件名  

import re #用于-正则表达式  

import time #用于-线程休眠  

import paramiko #用于-远程执行linux命令  

from alive_progress import alive_bar #用于-进度条工具类  

from cryptography.fernet import Fernet #用于-加解密代码  

import base64 #用于-加解密代码  

import hashlib #用于-加解密代码  

#检查密钥格式

def check_deploy_sign(deploy_site):  

   #确保密钥只能是以下4个之一才能继续往下操作，否则无限循环输入 或 退出程序

   if deploy_site != 'pro-main' and deploy_site != 'pro-manage' and deploy_site != 'test-main' and deploy_site != 'test-manage':

       #校验失败，一直校验

       new_deploy_site = input("错误：请填写部署密钥：")  

       check_deploy_sign(new_deploy_site)  

    #校验成功，退出

    return deploy_site  

# 连接服务器  

def connect_service(deploy_server):

   server_password = ''  

   server_host = ''  

   sign = hashlib.sha256(deploy_server.encode()).digest()  

   sign = base64.urlsafe_b64encode(sign)  

   if deploy_server == 'pro':  

       server_password = decrypt_str(sign, service_password_pro)  

       server_host = decrypt_str(sign, service_host_pro)  

   elif deploy_server == 'test':  

       server_password = decrypt_str(sign, service_password_test)  

       server_host = decrypt_str(sign, service_host_test)  

   else:  

       raise Exception('失败：部署服务器标识有误')  

   # 连接远程服务器  

   ssh = paramiko.SSHClient()  

   ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())  

   ssh.connect(server_host, username='root', password=server_password)  

   return ssh  

# 解密密码  

def decrypt_str(key, encrypted_password):  

   f = Fernet(key)  

   decrypted_password = f.decrypt(encrypted_password).decode()  

   return decrypted_password

try:  

   # 服务器环境信息的加密字符串，包含各服务器的 ip和密码  

   service_password_pro = 'asdatrgsd=='

   service_password_test = 'sgherfhdf=='

   service_host_pro = 'jfhgfvdcfdtr=='

   service_host_test = 'jutyrbfvret=='

   deploy_sign = input("提示：请填写部署密钥：")  

   deploy_sign = check_deploy_sign(deploy_sign)  

   # 部署环境  pro代表生成环境，test代表测试环境

   deploy_server = deploy_sign.split('-')[0]  

   # 部署模块或项目 manage代表manage模块，main代表main模块，

   deploy_site = deploy_sign.split('-')[1]  

   # 打包时的包名，三目运算符

   package_name = 'production' if deploy_server == 'pro' else 'staging'

   #进度条

   with alive_bar(7, force_tty=True, title="进度") as bar:  

       # 连接服务器  

       ssh = connect_service(deploy_server)  

       bar(0.1)  

       print("完成-服务器连接成功")  

       time.sleep(0.5)

except Exception as e:  

   print(f"异常: {str(e)}")

在连接服务器之前，我们加个进度条显示，方便查看部署到哪一步了，要点讲解：

   with alive_bar 中放的事需要进度条显示的步骤，connect_service是连接服务器的方法

   主机的ip和密码我们用加密的密文显示，解密的密钥就是 手动输入的部署密钥

   当一段逻辑执行完成后，通过bar(0.1)来显示进度条进度，alive_bar的第一个参数就是步骤总数

4. 部署工具主逻辑

代码要点讲解： 下面的代码是工程的全部代码，主要包含了以下逻辑

   连接服务器

   进入到项目工程目录，拉取git代码

   编译公共依赖的代码（有的项目不一定有公共模块，可酌情删减）

   编译打包程序代码

   杀死旧进程

   寻找编译好的程序jar包并启动

   检测启动结果

import os #用于-提取文件名

import re #用于-正则表达式

import time #用于-线程休眠

import paramiko #用于-远程执行linux命令

from alive_progress import alive_bar #用于-进度条工具类

from cryptography.fernet import Fernet #用于-加解密代码

import base64 #用于-加解密代码

import hashlib #用于-加解密代码

def check_deploy_sign(deploy_site):

   if deploy_site != 'pro-main' and deploy_site != 'pro-manage' and deploy_site != 'test-main' and deploy_site != 'test-manage':

       new_deploy_site = input("错误：请填写部署密钥：")

       check_deploy_sign(new_deploy_site)

   return deploy_site

# 解密密码

def decrypt_str(key, encrypted_password):

   f = Fernet(key)

   decrypted_password = f.decrypt(encrypted_password).decode()

   return decrypted_password

# 执行远程命令

def execute_command(ssh, command):

   stdin, stdout, stderr = ssh.exec_command(command)

   stdout.channel.recv_exit_status()  # 等待命令执行完毕

   output = stdout.read().decode('utf-8')

   time.sleep(0.5)

   return output

# 执行远程命令

def execute_command_shell(shell, command, endword):

   shell.send(command + '\n')

   output = ''

   while True:

       while shell.recv_ready():

           recv = shell.recv(1024).decode('utf-8', errors='ignore')

           output += recv

       if endword == '# ':

           if output.endswith('$ ') or output.endswith('# '):

               break

       elif endword in output:

           break

   time.sleep(0.5)

   return output

# 连接服务器

def connect_service(deploy_server):  

   server_password = ''

   server_host = ''

   sign = hashlib.sha256(deploy_server.encode()).digest()

   sign = base64.urlsafe_b64encode(sign)

   if deploy_server == 'pro':

       server_password = decrypt_str(sign, service_password_pro)

       server_host = decrypt_str(sign, service_host_pro)

   elif deploy_server == 'test':

       server_password = decrypt_str(sign, service_password_test)

       server_host = decrypt_str(sign, service_host_test)

   else:

       raise Exception('失败：部署服务器标识有误')

   # 连接远程服务器

   ssh = paramiko.SSHClient()

   ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

   ssh.connect(server_host, username='root', password=server_password)

   return ssh

# 查询进程

def query_process(ssh, process_name):  

   process_id = ''

   command = f"ps -ef | grep {process_name}-system-master. | grep -v grep"

   process_output = execute_command(ssh, command)

   if process_output:

       # 提取进程ID并杀死进程

       process_id = process_output.split("    ")[1]

   return process_id

# 杀掉进程

def kill_process(ssh, process_id):  

   command = f"kill -9 {process_id}"

   output = execute_command(ssh, command)

   return output

# 寻找编译好的jar包

def find_jarname(output):

   match = re.search(r"Building jar: .+?\/(.+?\.jar)", output)

   if match:

       jar_filepath = match.group(1)

       jar_filename = os.path.basename(jar_filepath)

       return jar_filename

   else:

       raise Exception('失败：jar未找到')

try:

   service_password_pro = 'asdatrgsd=='

   service_password_test = 'sgherfhdf=='

   service_host_pro = 'jfhgfvdcfdtr=='

   service_host_test = 'jutyrbfvret=='

   deploy_sign = input("提示：请填写部署密钥：")

   deploy_sign = check_deploy_sign(deploy_sign)

   # 部署环境

   deploy_server = deploy_sign.split('-')[0]

   # 部署模块

   deploy_site = deploy_sign.split('-')[1]

   # 部署环境对应服务正式的名字

   package_name = 'production' if deploy_server == 'pro' else 'staging'

   with alive_bar(7, force_tty=True, title="进度") as bar:

       # 连接服务器

       ssh = connect_service(deploy_server)

       bar(0.1)

       print("完成-服务器连接成功")

       time.sleep(0.5)

       # 拉取代码

       shell = ssh.invoke_shell()

       execute_command_shell(shell, 'cd /root/build/x-system','#')

       execute_command_shell(shell, 'git pull','#')

       bar(0.2)

       print("完成-git代码拉取成功")

       # 编译代码

       execute_command_shell(shell, 'cd /root/build/x-system/modules', '#')

       execute_command_shell(shell, 'mvn clean install', 'BUILD SUCCESS')

       bar(0.4)

       print("完成-公共模块编译成功")

       # 打包代码

       execute_command_shell(shell, 'cd /root/build/x-system/webapps/' + deploy_site + '-system ', '#')

       output=execute_command_shell(shell, 'mvn clean package -P ' + package_name, 'BUILD SUCCESS')

       bar(0.6)

       print("完成-" + deploy_site + "模块打包成功")

       # 查询进程，如果查不到 就不执行kill命令

       pid = query_process(ssh, deploy_site)

       if pid != '':

           kill_process(ssh, pid)

           print("完成-旧程序进程已被杀掉，等待启动")

       else:

           print("完成-旧程序PID未找到，直接启动")

       bar(0.7)

       # 启动jar

       jar_name = find_jarname(output)

       execute_command_shell(shell, 'cd /root/build/x-system/webapps/' + deploy_site + '-system/target', '#')

       execute_command_shell(shell, 'nohup java -jar ' + jar_name + '>log.out  2>&1 & ', '#')

       bar(0.8)

       print("完成-程序正在启动中...")

       # 查看日志确认服务启动成功

       log_path = '/var/log/x-system/' + deploy_site + '-system' if deploy_server == 'pro' else '/var/log/x-system/' + deploy_site + '-system-staging'

       execute_command_shell(shell, 'cd '+log_path, '#')

       execute_command_shell(shell, 'tail -200f '+deploy_site+'-system-info.log', 'TomcatWebServer:206 - Tomcat started on port(s)')

       bar(1)

       print("完成-程序启动成功")

except Exception as e:

   print(f"异常: {str(e)}")

finally:

   time.sleep(10)

   # 关闭连接

   shell.close()

   ssh.close()

代码用try catch finally包裹，如果过程中出现任何异常，都输出错误原因 一些提示：

   每个人的项目服务器的路径都不同，我只是提供个例子，不可盲目复制运行

   每个人项目的名字也不同，我在文中出现类似 manage和main，是我项目模块中的名字，只是个例子，不可盲目复制

5.打包

打包命令：

   pyinstaller --onefile --icon 太空人.ico --add-data ".\grapheme_break_property.json;grapheme\data"  --name 远程部署 deployment.py

打包命令中的几个参数解释一下：

   --onefile ：将项目工程文件输出在同一个可执行文件中即exe中

   --icon 太空人.ico ：exe的图标是一个ico的图片

   --add-data ".\grapheme_break_property.json;grapheme\data" ： 打包时 grapheme_break_property这个依赖找不到，导致打包失败，就手动添加一下

   --name 远程部署 ：exe的名字（注意不需要带.exe后缀）

   deployment.py ：python工程的文件名