如何用Python实现自动化运维？Paramiko实战

import paramiko
import os
import sys
import time
import json # 假设服务器配置放在JSON文件里

# 配置信息，实际应用中可以从配置文件读取
# 为了演示，我们直接定义一个列表
SERVER_CONFIGS = [
    {
        "hostname": "your_server_ip_or_domain_1",
        "username": "your_ssh_user",
        "port": 22,
        "private_key_path": "~/.ssh/id_rsa" # 请替换为你的私钥路径
    },
    {
        "hostname": "your_server_ip_or_domain_2",
        "username": "your_ssh_user",
        "port": 22,
        "private_key_path": "~/.ssh/id_rsa"
    }
]

LOCAL_APP_ARCHIVE = './my_app_v1.0.tar.gz' # 本地待上传的应用包
REMOTE_DEPLOY_DIR = '/opt/deploy/my_app' # 远程服务器上的部署目录
REMOTE_APP_NAME = 'my_app_extracted' # 解压后的应用目录名

def run_remote_command(client, command, description=""):
    """在远程服务器执行命令并打印输出"""
    print(f"  -> {description}: Executing '{command}'...")
    stdin, stdout, stderr = client.exec_command(command)
    stdout_str = stdout.read().decode().strip()
    stderr_str = stderr.read().decode().strip()
    exit_status = stdout.channel.recv_exit_status() # 获取命令退出状态码

    if stdout_str:
        print(f"     STDOUT: {stdout_str}")
    if stderr_str:
        print(f"     STDERR: {stderr_str}")

    if exit_status != 0:
        print(f"     [ERROR] Command failed with exit status {exit_status}.")
        return False
    else:
        print(f"     [SUCCESS] Command completed.")
        return True

def deploy_application_to_server(server_info):
    """
    连接到指定服务器并执行部署流程
    """
    hostname = server_info['hostname']
    username = server_info['username']
    port = server_info['port']
    private_key_path = os.path.expanduser(server_info['private_key_path'])

    print(f"\n--- Starting deployment to {hostname} ({username}@{hostname}:{port}) ---")

    client = paramiko.SSHClient()
    # 生产环境请使用 client.load_system_host_keys() 或 client.load_host_keys()
    # 并手动管理 known_hosts 文件，避免 AutoAddPolicy 的安全风险
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    try:
        # 加载私钥
        private_key = paramiko.RSAKey.from_private_key_file(private_key_path)
        client.connect(hostname, port=port, username=username, pkey=private_key, timeout=15)
        print(f"Successfully connected to {hostname}.")

        # 1. 确保远程部署目录存在
        if not run_remote_command(client, f"mkdir -p {REMOTE_DEPLOY_DIR}", "Creating remote deployment directory"):
            return False

        # 2. 上传应用包
        print(f"  -> Uploading application package '{LOCAL_APP_ARCHIVE}' to '{REMOTE_DEPLOY_DIR}'...")
        sftp = client.open_sftp()
        remote_archive_path = os.path.join(REMOTE_DEPLOY_DIR, os.path.basename(LOCAL_APP_ARCHIVE))
        sftp.put(LOCAL_APP_ARCHIVE, remote_archive_path)
        sftp.close()
        print("  -> Application package uploaded successfully.")

        # 3. 解压并部署应用
        # 注意：这里假设tar包解压后会在REMOTE_DEPLOY_DIR下生成一个名为REMOTE_APP_NAME的目录
        commands = [
            f"cd {REMOTE_DEPLOY_DIR}",
            f"tar -xzf {os.path.basename(LOCAL_APP_ARCHIVE)}",
            f"rm {os.path.basename(LOCAL_APP_ARCHIVE)}", # 清理上传的tar包
            f"pip install -r {REMOTE_APP_NAME}/requirements.txt", # 安装依赖
            # 假设你的应用有一个systemd服务叫my_app_service
            f"systemctl restart my_app_service" # 重启服务
        ]

        for cmd in commands:
            if not run_remote_command(client, cmd, "Deployment step"):
                print(f"[CRITICAL] Deployment failed on {hostname} at command: {cmd}")
                return False

        print(f"--- Deployment to {hostname} completed successfully ---")
        return True

    except paramiko.AuthenticationException:
        print(f"[ERROR] Authentication failed for {username}@{hostname}. Check your private key or username.")
    except paramiko.SSHException as e:
        print(f"[ERROR] SSH connection error to {hostname}: {e}")
    except Exception as e:
        print(f"[ERROR] An unexpected error occurred during deployment to {hostname}: {e}")
    finally:
        if client:
            client.close()
    return False

if __name__ == "__main__":
    # 模拟创建一个本地应用包，以便脚本可以运行
    if not os.path.exists(LOCAL_APP_ARCHIVE):
        print(f"Creating dummy application archive for demonstration: {LOCAL_APP_ARCHIVE}")
        os.makedirs(f"{REMOTE_APP_NAME}", exist_ok=True)
        with open(f"{REMOTE_APP_NAME}/requirements.txt", "w") as f:
            f.write("Flask\nrequests\n")
        with open(f"{REMOTE_APP_NAME}/app.py", "w") as f:
            f