使用python編寫一個自動化部署工具

更新時間：2024年12月10日 15:31:27 作者：軟件測試雜談

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

效果

起因

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

實現(xiàn)步驟

開發(fā)準備

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

步驟

1. 導入依賴

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

import os #用于-提取文件名  
import re #用于-正則表達式  
import time #用于-線程休眠  
import paramiko #用于-遠程執(zhí)行l(wèi)inux命令  
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. 輸入校驗

部署畢竟是件嚴謹?shù)氖虑椋覀冊黾觽€部署密鑰校驗，我的這個部署密鑰承擔了以下的功能

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

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

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

    # 部署環(huán)境  pro代表生成環(huán)境，test代表測試環(huán)境
    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，也定義了一些基本的變量。密鑰是一串由短線連接的字符串，短線前的代碼用以區(qū)分環(huán)境，短線后的代碼用以區(qū)分模塊或項目。另外上面代碼中的package_name是打包時的包名（即profiles.profile.id），一般配置在springboot項目pom文件中的編輯模塊，類似下面這樣:

3. 連接linux服務器

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

#檢查密鑰格式
def check_deploy_sign(deploy_site):  
    #確保密鑰只能是以下4個之一才能繼續(xù)往下操作，否則無限循環(huán)輸入 或 退出程序
    if deploy_site != 'pro-main' and deploy_site != 'pro-manage' and deploy_site != 'test-main' and deploy_site != 'test-manage': 
        #校驗失敗，一直校驗
        new_deploy_site = input("錯誤：請?zhí)顚懖渴鹈荑€：")  
        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('失?。翰渴鸱掌鳂俗R有誤')  
    # 連接遠程服務器  
    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:  
    # 服務器環(huán)境信息的加密字符串，包含各服務器的 ip和密碼  
    service_password_pro = 'asdatrgsd=='  
    service_password_test = 'sgherfhdf=='  
    service_host_pro = 'jfhgfvdcfdtr=='  
    service_host_test = 'jutyrbfvret=='
    
    
    deploy_sign = input("提示：請?zhí)顚懖渴鹈荑€：")  
    deploy_sign = check_deploy_sign(deploy_sign)  

    # 部署環(huán)境  pro代表生成環(huán)境，test代表測試環(huán)境
    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和密碼我們用加密的密文顯示，解密的密鑰就是手動輸入的部署密鑰
當一段邏輯執(zhí)行完成后，通過bar(0.1)來顯示進度條進度，alive_bar的第一個參數(shù)就是步驟總數(shù)

4. 部署工具主邏輯

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

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

import os #用于-提取文件名
import re #用于-正則表達式
import time #用于-線程休眠
import paramiko #用于-遠程執(zhí)行l(wèi)inux命令
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("錯誤：請?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í)行遠程命令
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í)行遠程命令
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('失?。翰渴鸱掌鳂俗R有誤')
    # 連接遠程服務器
    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('失?。簀ar未找到')


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


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

    # 部署環(huán)境
    deploy_server = deploy_sign.split('-')[0]
    # 部署模塊
    deploy_site = deploy_sign.split('-')[1]
    # 部署環(huán)境對應服務正式的名字
    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 + "模塊打包成功")

        # 查詢進程，如果查不到 就不執(zhí)行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)
    # 關(guān)閉連接
    shell.close()
    ssh.close()

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

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

5.打包

打包命令：

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

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

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

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

您可能感興趣的文章: