热门标签
热门文章
- 1关于ROS中对moveit和gazebo联动出现的问题进行解决_rosnoetic让moveit和gazebo联动
- 2数据结构-线性表_线性表的顺序存储结构和链式存储结构分别是
- 3了解哈希冲突_hash冲突产生的原因
- 4『App自动化测试之Appium应用篇』| uiautomator + accessibility_id定位方法完全使用攻略_appium中bounds [915,1941][1023,2049]右下角是那部分
- 5nlp的四大任务什么_nlp四大基本任务
- 6【动手学深度学习】使用块的网络（VGG）的研究详情
- 7人脸识别4-百度商用方案调研_人脸识别有源和无源的区别
- 8基于SpringBoot+Vue的酒店管理系统设计与实现_基于springboot+vue 酒店客
- 9MySQL8.0窗口函数_mysql窗口函数lead
- 10应用层协议：VPN协议
当前位置: article > 正文
用python写一个自动化部署工具_python 实现的windows下的工具
作者：羊村懒王 | 2024-06-12 05:07:47
赞
踩
python 实现的windows下的工具
效果
起因
现在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 #用于-加解密代码
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
在导入依赖的时候，可能有些依赖咱们的电脑上之前没下载过，不要紧，只需要在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)}")
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
上面的代码中 增加了全局的异常处理，类似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)}")
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
在连接服务器之前，我们加个进度条显示，方便查看部署到哪一步了，要点讲解：
- 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()
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 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
代码用try catch finally包裹，如果过程中出现任何异常，都输出错误原因 一些提示：
- 每个人的项目服务器的路径都不同，我只是提供个例子，不可盲目复制运行
- 每个人项目的名字也不同，我在文中出现类似 manage和main，是我项目模块中的名字，只是个例子，不可盲目复制
5.打包
打包命令：
pyinstaller --onefile --icon 太空人.ico --add-data ".\grapheme_break_property.json;grapheme\data" --name 远程部署 deployment.py
- 1
- 2
打包命令中的几个参数解释一下：
- –onefile ：将项目工程文件输出在同一个可执行文件中即exe中
- –icon 太空人.ico ：exe的图标是一个ico的图片
- –add-data “.\grapheme_break_property.json;grapheme\data” ： 打包时 grapheme_break_property这个依赖找不到，导致打包失败，就手动添加一下
- –name 远程部署 ：exe的名字（注意不需要带.exe后缀）
- deployment.py ：python工程的文件名
结语
python很好玩，希望大家玩的开心！
如果你对Python感兴趣，想要学习python，这里给大家分享一份Python全套学习资料，都是我自己学习时整理的，希望可以帮到你，一起加油！
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/羊村懒王/article/detail/706500
推荐阅读
相关标签
Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。
