快捷導(dǎo)航

使用python編寫一個(gè)自動(dòng)化部署工具

更新時(shí)間：2024年12月10日 15:31:27 作者：軟件測(cè)試雜談

這篇文章主要為大家詳細(xì)介紹了如何使用python編寫一個(gè)自動(dòng)化部署工具,文中的示例代碼講解詳細(xì),感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下

效果

起因

現(xiàn)在springboot項(xiàng)目的自動(dòng)化部署已經(jīng)非常普遍，有用Jenkins的，有用git鉤子函數(shù)的，有用docker的...等等。這段時(shí)間在玩python，想著用python實(shí)現(xiàn)自動(dòng)化部署，即能鍛煉下編碼能力，又方便運(yùn)維。于是開始著手寫了一個(gè)exe程序，可直接在任何windows電腦上運(yùn)行（不具備python環(huán)境的windows電腦也可以運(yùn)行）。有興趣的小伙伴可以跟著代碼一起練一練噢，寫的詳細(xì)一點(diǎn)，對(duì)python新手也很友好。

實(shí)現(xiàn)步驟

開發(fā)準(zhǔn)備

具有python基本環(huán)境和ide的windows或macOS電腦一臺(tái)
安裝打包工具pip install pyinstaller
一點(diǎn)小小的python基礎(chǔ)

步驟

1. 導(dǎo)入依賴

新建一個(gè)py文件，可以把它命名為 deployment.py（名字隨意哈，什么名兒都可以），然后把下面的庫導(dǎo)入語句copy到此py文件中

import os #用于-提取文件名  
import re #用于-正則表達(dá)式  
import time #用于-線程休眠  
import paramiko #用于-遠(yuǎn)程執(zhí)行l(wèi)inux命令  
from alive_progress import alive_bar #用于-進(jìn)度條工具類  
from cryptography.fernet import Fernet #用于-加解密代碼  
import base64 #用于-加解密代碼  
import hashlib #用于-加解密代碼

在導(dǎo)入依賴的時(shí)候，可能有些依賴咱們的電腦上之前沒下載過，不要緊，只需要在pycharm中按 alt+enter就可以自動(dòng)導(dǎo)入了，PyCharm跟Idea的快捷鍵一模一樣，可以按Idea的習(xí)慣使用。而且在python中還不用配置maven或pom文件，非常方便。

2. 輸入校驗(yàn)

部署畢竟是件嚴(yán)謹(jǐn)?shù)氖虑?，我們?cè)黾觽€(gè)部署密鑰校驗(yàn)，我的這個(gè)部署密鑰承擔(dān)了以下的功能

確保部署的安全性，不是誰拿到這個(gè)exe程序都能運(yùn)行的（哼~傲嬌）
密鑰字符串用-分割開，前面的區(qū)分環(huán)境，后面的區(qū)分項(xiàng)目或模塊。
如果同學(xué)們不需要區(qū)分項(xiàng)目子模塊，就不需要搞這么復(fù)雜，隨便定義一個(gè)密鑰就好了

import os #用于-提取文件名  
import re #用于-正則表達(dá)式  
import time #用于-線程休眠  
import paramiko #用于-遠(yuǎn)程執(zhí)行l(wèi)inux命令  
from alive_progress import alive_bar #用于-進(jìn)度條工具類  
from cryptography.fernet import Fernet #用于-加解密代碼  
import base64 #用于-加解密代碼  
import hashlib #用于-加解密代碼  

#檢查密鑰格式
def check_deploy_sign(deploy_site):  
    #確保密鑰只能是以下4個(gè)之一才能繼續(xù)往下操作，否則無限循環(huán)輸入 或 退出程序
    if deploy_site != 'pro-main' and deploy_site != 'pro-manage' and deploy_site != 'test-main' and deploy_site != 'test-manage': 
        #校驗(yàn)失敗，一直校驗(yàn)
        new_deploy_site = input("錯(cuò)誤：請(qǐng)?zhí)顚懖渴鹈荑€：")  
        check_deploy_sign(new_deploy_site)  
     #校驗(yàn)成功，退出
     return deploy_site  
  
  
try:  
    deploy_sign = input("提示：請(qǐng)?zhí)顚懖渴鹈荑€：")  
    deploy_sign = check_deploy_sign(deploy_sign)  

    # 部署環(huán)境  pro代表生成環(huán)境，test代表測(cè)試環(huán)境
    deploy_server = deploy_sign.split('-')[0]  
    # 部署模塊或項(xiàng)目 manage代表manage模塊，main代表main模塊， 
    deploy_site = deploy_sign.split('-')[1]  
    # 打包時(shí)的包名，三目運(yùn)算符
    package_name = 'production' if deploy_server == 'pro' else 'staging'  

except Exception as e:  
    print(f"異常: {str(e)}")

上面的代碼中增加了全局的異常處理，類似Java的try catch，也定義了一些基本的變量。密鑰是一串由短線連接的字符串，短線前的代碼用以區(qū)分環(huán)境，短線后的代碼用以區(qū)分模塊或項(xiàng)目。另外上面代碼中的package_name是打包時(shí)的包名（即profiles.profile.id），一般配置在springboot項(xiàng)目pom文件中的編輯模塊，類似下面這樣:

3. 連接linux服務(wù)器

import os #用于-提取文件名  
import re #用于-正則表達(dá)式  
import time #用于-線程休眠  
import paramiko #用于-遠(yuǎn)程執(zhí)行l(wèi)inux命令  
from alive_progress import alive_bar #用于-進(jìn)度條工具類  
from cryptography.fernet import Fernet #用于-加解密代碼  
import base64 #用于-加解密代碼  
import hashlib #用于-加解密代碼  

#檢查密鑰格式
def check_deploy_sign(deploy_site):  
    #確保密鑰只能是以下4個(gè)之一才能繼續(xù)往下操作，否則無限循環(huán)輸入 或 退出程序
    if deploy_site != 'pro-main' and deploy_site != 'pro-manage' and deploy_site != 'test-main' and deploy_site != 'test-manage': 
        #校驗(yàn)失敗，一直校驗(yàn)
        new_deploy_site = input("錯(cuò)誤：請(qǐng)?zhí)顚懖渴鹈荑€：")  
        check_deploy_sign(new_deploy_site)  
     #校驗(yàn)成功，退出
     return deploy_site  
     
# 連接服務(wù)器  
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('失敗：部署服務(wù)器標(biāo)識(shí)有誤')  
    # 連接遠(yuǎn)程服務(wù)器  
    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:  
    # 服務(wù)器環(huán)境信息的加密字符串，包含各服務(wù)器的 ip和密碼  
    service_password_pro = 'asdatrgsd=='  
    service_password_test = 'sgherfhdf=='  
    service_host_pro = 'jfhgfvdcfdtr=='  
    service_host_test = 'jutyrbfvret=='
    
    
    deploy_sign = input("提示：請(qǐng)?zhí)顚懖渴鹈荑€：")  
    deploy_sign = check_deploy_sign(deploy_sign)  

    # 部署環(huán)境  pro代表生成環(huán)境，test代表測(cè)試環(huán)境
    deploy_server = deploy_sign.split('-')[0]  
    # 部署模塊或項(xiàng)目 manage代表manage模塊，main代表main模塊， 
    deploy_site = deploy_sign.split('-')[1]  
    # 打包時(shí)的包名，三目運(yùn)算符
    package_name = 'production' if deploy_server == 'pro' else 'staging'  
    #進(jìn)度條
    with alive_bar(7, force_tty=True, title="進(jìn)度") as bar:  
        # 連接服務(wù)器  
        ssh = connect_service(deploy_server)  
        bar(0.1)  
        print("完成-服務(wù)器連接成功")  
        time.sleep(0.5)
except Exception as e:  
    print(f"異常: {str(e)}")

在連接服務(wù)器之前，我們加個(gè)進(jìn)度條顯示，方便查看部署到哪一步了，要點(diǎn)講解：

with alive_bar 中放的事需要進(jìn)度條顯示的步驟，connect_service是連接服務(wù)器的方法
主機(jī)的ip和密碼我們用加密的密文顯示，解密的密鑰就是手動(dòng)輸入的部署密鑰
當(dāng)一段邏輯執(zhí)行完成后，通過bar(0.1)來顯示進(jìn)度條進(jìn)度，alive_bar的第一個(gè)參數(shù)就是步驟總數(shù)

4. 部署工具主邏輯

代碼要點(diǎn)講解：下面的代碼是工程的全部代碼，主要包含了以下邏輯

連接服務(wù)器
進(jìn)入到項(xiàng)目工程目錄，拉取git代碼
編譯公共依賴的代碼（有的項(xiàng)目不一定有公共模塊，可酌情刪減）
編譯打包程序代碼
殺死舊進(jìn)程
尋找編譯好的程序jar包并啟動(dòng)
檢測(cè)啟動(dòng)結(jié)果

import os #用于-提取文件名
import re #用于-正則表達(dá)式
import time #用于-線程休眠
import paramiko #用于-遠(yuǎn)程執(zhí)行l(wèi)inux命令
from alive_progress import alive_bar #用于-進(jìn)度條工具類
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("錯(cuò)誤：請(qǐng)?zhí)顚懖渴鹈荑€：")
        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

# 執(zhí)行遠(yuǎn)程命令
def execute_command(ssh, command):
    stdin, stdout, stderr = ssh.exec_command(command)
    stdout.channel.recv_exit_status()  # 等待命令執(zhí)行完畢
    output = stdout.read().decode('utf-8')
    time.sleep(0.5)
    return output

# 執(zhí)行遠(yuǎn)程命令
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

# 連接服務(wù)器
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('失敗：部署服務(wù)器標(biāo)識(shí)有誤')
    # 連接遠(yuǎn)程服務(wù)器
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect(server_host, username='root', password=server_password)
    return ssh

# 查詢進(jìn)程
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:
        # 提取進(jìn)程ID并殺死進(jìn)程
        process_id = process_output.split("    ")[1]
    return process_id

# 殺掉進(jìn)程
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('失?。簀ar未找到')


try:
    service_password_pro = 'asdatrgsd=='
    service_password_test = 'sgherfhdf=='
    service_host_pro = 'jfhgfvdcfdtr=='
    service_host_test = 'jutyrbfvret=='


    deploy_sign = input("提示：請(qǐng)?zhí)顚懖渴鹈荑€：")
    deploy_sign = check_deploy_sign(deploy_sign)

    # 部署環(huán)境
    deploy_server = deploy_sign.split('-')[0]
    # 部署模塊
    deploy_site = deploy_sign.split('-')[1]
    # 部署環(huán)境對(duì)應(yīng)服務(wù)正式的名字
    package_name = 'production' if deploy_server == 'pro' else 'staging'

    with alive_bar(7, force_tty=True, title="進(jìn)度") as bar:
        # 連接服務(wù)器
        ssh = connect_service(deploy_server)
        bar(0.1)
        print("完成-服務(wù)器連接成功")
        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 + "模塊打包成功")

        # 查詢進(jìn)程，如果查不到 就不執(zhí)行kill命令
        pid = query_process(ssh, deploy_site)
        if pid != '':
            kill_process(ssh, pid)
            print("完成-舊程序進(jìn)程已被殺掉，等待啟動(dòng)")
        else:
            print("完成-舊程序PID未找到，直接啟動(dòng)")
        bar(0.7)


        # 啟動(dòng)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("完成-程序正在啟動(dòng)中...")


        # 查看日志確認(rèn)服務(wù)啟動(dòng)成功
        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("完成-程序啟動(dòng)成功")
except Exception as e:
    print(f"異常: {str(e)}")

finally:
    time.sleep(10)
    # 關(guān)閉連接
    shell.close()
    ssh.close()

代碼用try catch finally包裹，如果過程中出現(xiàn)任何異常，都輸出錯(cuò)誤原因一些提示：

每個(gè)人的項(xiàng)目服務(wù)器的路徑都不同，我只是提供個(gè)例子，不可盲目復(fù)制運(yùn)行
每個(gè)人項(xiàng)目的名字也不同，我在文中出現(xiàn)類似 manage和main，是我項(xiàng)目模塊中的名字，只是個(gè)例子，不可盲目復(fù)制

5.打包

打包命令：

pyinstaller --onefile --icon 太空人.ico --add-data ".\grapheme_break_property.json;grapheme\data"  --name 遠(yuǎn)程部署 deployment.py

打包命令中的幾個(gè)參數(shù)解釋一下：

--onefile ：將項(xiàng)目工程文件輸出在同一個(gè)可執(zhí)行文件中即exe中
--icon 太空人.ico ：exe的圖標(biāo)是一個(gè)ico的圖片
--add-data ".\grapheme_break_property.json;grapheme\data" ：打包時(shí) grapheme_break_property這個(gè)依賴找不到，導(dǎo)致打包失敗，就手動(dòng)添加一下
--name 遠(yuǎn)程部署：exe的名字（注意不需要帶.exe后綴）
deployment.py ：python工程的文件名

以上就是使用python編寫一個(gè)自動(dòng)化部署工具的詳細(xì)內(nèi)容，更多關(guān)于python自動(dòng)化部署的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: