准备工作

如果使用SSH连接的话，首先在本地生成SSH密钥对，有两种方式都可以用

生成ras密钥对
ssh-keygen -t rsa -b 4096 -C "your_email@example.com"

生成ed25519密钥对
ssh-keygen -t ed25519 -C "your_email@example.com"

生成过程
$ ssh-keygen -t ed25519 -C "user@company.com"
Enter file in which to save the key (/home/username/.ssh/id_ed25519): [回车使用默认路径]
Enter passphrase (empty for no passphrase): [可选：输入密码保护私钥]
Enter same passphrase again: [再次输入]

使用

源码中，有SSH连接和密码连接，建议和我一样使用SSH来弄，这样子安全一点

源码包：下载源码

配置好后，把证书移进cert目录里面，然后运行代码就好

import os  # os模块，处理文件
import subprocess
import sys
import shutil
import time

try:
    import paramiko  # sftp插件
except ImportError as error:
    response = input("paramiko 未安装。是否现在安装？ (y/n): ")
    if response.lower().strip() in ['y', 'yes', '是']:
        print("正在尝试自动安装paramiko插件...")
        try:
            subprocess.check_call((sys.executable, "-m", "pip", "install", "paramiko"))  # 安装paramiko插件
            print("paramiko 安装成功！")
            import paramiko  # sftp插件

        except subprocess.CalledProcessError as error:
            print("paramiko 安装失败！请手动运行 'pip install paramiko' 安装")

        except Exception as e:
            print(f"安装过程中发生未知错误: {e}")
            sys.exit(1)  # 安装失败，退出脚本
    else:
        print("请手动运行 'pip install paramiko' 安装")


# 上传文件
def file_upload(host, port, username, privateKey, upload_path):
    global current_directory  # 添加成全局，最后用
    ssh = paramiko.SSHClient()  # 创建SSH客户端
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())  # 自动添加主机密钥
    private_key = paramiko.RSAKey(filename=privateKey)  # 加载RSA私钥
    # 如果是 Ed25519 密钥，使用下面这行：
    # private_key = paramiko.Ed25519Key(filename=privateKey)  # 加载Ed25519私钥

    try:
        current_directory = os.path.join(os.path.dirname(os.path.abspath(__file__)), "cert")  # 获取证书文件目录
        # 进度条特效
        progress_bar(10)
        print(f"获取证书文件目录：{current_directory}")

        all_items = os.listdir(current_directory)  # 获取文件夹中所有文件和文件夹名

        # '.idea', 'main.py' 这两个文件不上传
        result_files = [
            item for item in all_items
            if os.path.isfile(os.path.join(current_directory, item)) and item not in {'.idea', 'main.py'}
        ]

        # 如果 result_files 为空，抛出异常
        if not result_files:
            raise ValueError("未找到有效的上传文件：result_files 为空")

        # 进度条特效
        progress_bar(13)
        print(f"获取要上传的文件：{result_files}")

        ssh.connect(hostname=host, port=port, username=username, pkey=private_key)
        # 进度条特效
        progress_bar(14)
        print("服务器连接成功")

        sftp = ssh.open_sftp()  # 创建SFTP会话
        # 进度条特效
        progress_bar(11)
        print("会话创建成功")

        for file in result_files:
            local_path = os.path.join(current_directory, file)
            remote_path = os.path.join(upload_path, file)
            if not os.path.isfile(local_path):
                raise FileNotFoundError(f"本地文件不存在：{local_path}")
            sftp.put(local_path, remote_path)
            print(f"文件上传成功：{local_path} -----> {remote_path}")

        # 进度条特效
        progress_bar(13)
        print("文件上传完成")

        """
        重启服务器nginx
        """
        command = "sudo /usr/local/bin/nginx -t"
        stdin, stdout, stderr = ssh.exec_command(command)
        print(stderr.read().decode('utf-8'))

        command = "sudo /usr/local/bin/nginx -s reload"
        stdin, stdout, stderr = ssh.exec_command(command)

        # 检查命令退出状态
        exit_status = stdout.channel.recv_exit_status()
        if exit_status == 0:
            print("✅ nginx 配置成功！")
        else:
            print(f"❌ nginx 配置失败！退出状态码: {exit_status}")

    except paramiko.PasswordRequiredException:
        raise Exception("私钥受密码保护，请提供密码")
    except paramiko.SSHException as e:
        print(f"验证失败：{e}")
    except Exception as e:
        print(f"上传失败：{e}")
    finally:
        # 关闭连接
        sftp.close()
        ssh.close()
        # 进度条特效
        progress_bar(15)
        print("连接已关闭")

        # 清空证书文件夹
        for filename in os.listdir(current_directory):
            file_path = os.path.join(current_directory, filename)
            try:
                if os.path.isfile(file_path) or os.path.islink(file_path):
                    os.unlink(file_path)  # 删除文件或符号链接
                elif os.path.isdir(file_path):
                    shutil.rmtree(file_path)  # 删除目录树
                print("本地证书已经删除")
            except Exception as e:
                print(f'删除 {file_path} 失败: {e}')


# 进度条特效
def progress_bar(sum):
    print(), print()  # 换行
    for i in range(5, sum + 1):
        bar = '█' * i + '.' * (sum - i)
        print(f"\r进度: {i}/{sum} [{bar}]", end="", flush=True)
        time.sleep(0.1)
    print()  # 换行


if __name__ == '__main__':
    host = '47.120.4.193'  # 要上传的服务器地址
    port = 22  # 要上传的服务器端口
    username = 'lyj'  # 要上传的服务器用户名
    privateKey = r'C:\Users\Administrator\.ssh\id_rsa'  # 客户端SSH公钥
    upload_path = '/server/ceshi/'  # 要上传的文件目录

    file_upload(host, port, username, privateKey, upload_path)

报错解决方法

请输入图片描述

这个是鉴权失败，先检查客户端的公钥文件有没有写入服务端的authorized_keys文件里面

如果确定写入了，就检查一下服务端ssh的文件权限，可以按照下面的语句跟着输出一遍

chmod 700 /home/lyj/.ssh
chmod 600 /home/lyj/.ssh/authorized_keys
chown -R lyj:lyj /home/lyj/.ssh

lyj改成自己的用户名

请输入图片描述

这个是你在生成客户端密钥对的时候，设置了密码，但是在连接的时候，又没有配置密码导致，把里面的代码换成下面这个就可以了

private_key = paramiko.RSAKey(filename=privateKey, password='你的密码')  # 加载RSA私钥

请输入图片描述

或者

请输入图片描述

这两个报错是在操作nginx命令时出现，是因为当你使用sudo时，默认需要一个终端来输入密码，但在自动化脚本中没有 TTY（终端），所以报错

但是好像可以这样子，但是密码暴露在代码里面很危险

command = "echo 你的密码 | sudo /usr/local/bin/nginx -s reload"

我是这样子解决的，如果有不同见解，欢迎评论纠正

首先，我的服务端是centos服务器，如果你们的不是这个，可以ai一下

在服务器使用这个查找 nginx 可执行文件的路径

which nginx

请输入图片描述

然后打开/etc/sudoers文件，输入这两行语句，保存退出（lyj改成你们的用户名）

lyj        ALL=(ALL)     NOPASSWD: /usr/local/bin/nginx -t
lyj        ALL=(ALL)     NOPASSWD: /usr/local/bin/nginx -s reload

然后再把脚本的这两个改成你们配置的

command = "sudo /usr/local/bin/nginx -t"

command = "sudo /usr/local/bin/nginx -s reload"

就这样，大功告成啦，初版脚本，不足的地方还有好多，但能用就用嘛~

最后修改：2025 年 08 月 27 日
© 允许规范转载
如果觉得我的文章对你有用，请随意赞赏