本站首页 收藏本站
广告位联系

APP正在开发中...
返回顶部
您的位置： 主页 > 脚本 > python >

收藏
分享到

使用python编写一个自动化部署工具

python 来源：互联网 作者：佚名 发布时间：2024-12-11 22:33:34 人浏览
摘要

效果 起因 现在springboot项目的自动化部署已经非常普遍，有用Jenkins的，有用git钩子函数的，有用docker的...等等。这段时间在玩python，想着用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文件中

1

2

3

4

5

6

7

8

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程序都能运行的（哼~傲娇）
  • 密钥字符串用-分割开，前面的区分环境，后面的区分项目或模块。
  • 如果同学们不需要区分项目子模块，就不需要搞这么复杂，随便定义一个密钥就好了

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

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服务器

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

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

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

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工程的文件名
版权声明 : 本文内容来源于互联网或用户自行发布贡献，该文观点仅代表原作者本人。本站仅提供信息存储空间服务和不拥有所有权，不承担相关法律责任。如发现本站有涉嫌抄袭侵权， 违法违规的内容， 请发送邮件至2530232025#qq.cn(#换@)举报，一经查实，本站将立刻删除。

您可能感兴趣的文章 :

原文链接 :
相关文章
  • 利用Python定位Span标签中文字
    在开始之前，需要确保安装了必要的Python库。requests库用于发送HTTP请求，获取网页内容；BeautifulSoup库用于解析HTML文档，提取所需信息。 可

  • 使用python编写一个自动化部署工具

    使用python编写一个自动化部署工具
    效果 起因 现在springboot项目的自动化部署已经非常普遍，有用Jenkins的，有用git钩子函数的，有用docker的...等等。这段时间在玩python，想着用
  • Python中的下划线“_”们介绍
    随便拿一份Python代码，几乎都可以看到很多_的身影。 在Python中，下划线（_）有多种用途和含义，具体取决于它们的位置和使用方式。在这
  • OpenCV-Python给图像去除水印多种方法
    去除水印的过程与添加水印相反，它涉及到图像修复、颜色匹配和区域填充等技术。OpenCV-Python 提供了多种方法来处理不同类型的水印，包括

  • Python连接和操作Elasticsearch

    Python连接和操作Elasticsearch
    一、服务器端配置 在开始之前，确保你的 Elasticsearch 服务已经在服务器上正确安装和配置。 以下是一些基本的配置步骤： 1. 修改 Elasticse

  • Python ArcPy实现栅格图像文件由HDF格式批量转换为

    Python ArcPy实现栅格图像文件由HDF格式批量转换为
    首先，来看看我们想要实现的需求。 在一个名为HDF的文件夹下，有五个子文件夹；每一个子文件夹中，都存储了大量的.hdf格式的栅格遥感影
  • python随机种子ranrandom seed的使用介绍
    在Python中启用随机种子（random seed）是为了确保你的随机数生成过程是可重复的。通过设置随机种子，你可以保证每次运行代码时生成的随机

  • Numpy判断数组是否全0的三种方法

    Numpy判断数组是否全0的三种方法
    1numpy.any() numpy.any()函数用于检查一个numpy数字是否存在任何一个非0元素，因此将numpy.any()的结果取反即得numpy数组是否全0的结果。例如： 1
  • python实现字符串逆序输出的几种方法
    方法一：使用切片（Slicing） 1 2 3 4 5 6 def reverse_string(s): return s[::-1] s=str(input(请输入字符串：)) reversed_string=reverse_string(s) print(reversed_string) 在
  • python删除目录的三种方法介绍
    一、os.rmdir(path) 删除目录 path，path必须是个空目录，否则抛出OSError异常。 二、os.removedirs(path) 递归地删除目录。要求每一级目录都为空，才
推荐阅读
最新更新
友情链接
  • 本站所有内容来源于互联网或用户自行发布，本站仅提供信息存储空间服务，不拥有版权，不承担法律责任。如有侵犯您的权益，请您联系站长处理！
  • Copyright © 2017-2022 F11.CN All Rights Reserved. F11站长开发者网 版权所有 | 苏ICP备2022031554号-1 | 51LA统计