使用python编写一个自动化部署工具_F11 - 专业站长和开发者的学习网站

效果

起因

现在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包并启动
检测启动结果

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

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.打包

打包命令：

1	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工程的文件名