party-admin/bin/distribute.py

import paramiko
import os


class SSHConnect:
    # 定义一个私有变量，用来保存ssh连接通道，初始化为None
    __transport = None

    # 初始化构造函数（主机，用户名，密码，端口，默认22）
    def __init__(self, hostname, username, password, port=22):
        self.hostname = hostname
        self.port = port
        self.username = username
        self.password = password
        # 创建 ssh 连接通道
        self.connect()

    # 建立ssh 连接通道，并绑定在 __transport 上
    def connect(self):
        try:
            # 设置SSH 连接的远程主机地址和端口
            self.__transport = paramiko.Transport((self.hostname, self.port))
            # 通过用户名和密码连接SSH服务端
            self.__transport.connect(username=self.username, password=self.password)
        except Exception as e:
            # 连接出错
            print(e)

    # 执行linux命令
    def exec(self, command):

        # 创建 ssh 客户端
        ssh = paramiko.SSHClient()
        # 指定连接的通道
        ssh._transport = self.__transport

        # 调用 exec_command 方法执行命令
        stdin, stdout, stderr = ssh.exec_command(command)

        # 获取命令结果，返回是二进制，需要编码一下
        res = stdout.read().decode('utf-8')
        # 获取错误信息
        error = stderr.read().decode('utf-8')

        # 如果没出错
        if error.strip():
            # 返回错误信息
            return error
        else:
            # 返回结果
            return res

    # 前端打包（入参work_path为项目目录）
    def build(self, work_path):
        # 开始打包
        print('########### run build ############')
        # 打包命令
        cmd = 'npm run build:prod'
        # 切换到需要项目目录
        os.chdir(work_path)
        # 当前项目目录下执行打包命令
        if os.system(cmd) == 0:
            # 打包完成
            print('########### build complete ############')

    # 文件上传
    def upload(self, local_path, target_path):
        # 判断路径问题
        if not os.path.exists(local_path):
            return print('local path is not exist')

        print('upload...')

        # 实例化一个 sftp 对象,指定连接的通道
        sftp = paramiko.SFTPClient.from_transport(self.__transport)
        # 打包后的文件路径
        local_path = os.path.join(local_path, 'dist')
        # 本地路径转换，将windows下的 \ 转成 /
        local_path = '/'.join(local_path.split('\\'))

        # 递归上传文件
        self.upload_file(sftp, local_path, target_path)

        print('finish upload...')
        # 关闭连接
        self.close()

    # 递归上传文件
    def upload_file(self, sftp, local_path, target_path):
        # 判断当前路径是否是文件夹
        if not os.path.isdir(local_path):
            # 如果是文件，获取文件名
            file_name = os.path.basename(local_path)
            # 检查服务器文件夹是否存在
            self.check_remote_dir(sftp, target_path)
            # 服务器创建文件
            target_file_path = os.path.join(target_path, file_name).replace('\\', '/')
            # 上传到服务器
            sftp.put(local_path, target_file_path)
        else:
            # 查看当前文件夹下的子文件
            file_list = os.listdir(local_path)
            # 遍历子文件
            for p in file_list:
                # 拼接当前文件路径
                current_local_path = os.path.join(local_path, p).replace('\\', '/')
                # 拼接服务器文件路径
                current_target_path = os.path.join(target_path, p).replace('\\', '/')
                # 如果已经是文件，服务器就不需要创建文件夹了
                if os.path.isfile(current_local_path):
                    # 提取当前文件所在的文件夹
                    current_target_path = os.path.split(current_target_path)[0]
                # 递归判断
                self.upload_file(sftp, current_local_path, current_target_path)

    # 创建服务器文件夹
    def check_remote_dir(self, sftp, target_path):
        try:
            # 判断文件夹是否存在
            sftp.stat(target_path)
        except IOError:
            # 创建文件夹
            self.exec(r'mkdir -p %s ' % target_path)

    # 自动化打包部署
    def auto_deploy(self, local_path, target_path):
        # 打包构建
        self.build(local_path)
        # 清空文件
        # self.clear_remote_dir(target_path)
        # 文件上传
        self.upload(local_path, target_path)

    # 清空文件夹
    def clear_remote_dir(self, target_path):
        if target_path[-1] == '/':
            cmd = f'rm -rf {target_path}*'
        else:
            cmd = f'rm -rf {target_path}/*'
        self.exec(cmd)

    # 关闭连接
    def close(self):
        self.__transport.close()

    # 销毁实例
    def __del__(self):
        self.__transport.close()


if __name__ == '__main__':
    # 项目目录
    project_path = r'd:\Project\P2205-Platform\inspur-ui'
    # d:\Project\P2205-Platform\inspur-ui
    # 服务器目录
    # remote_path = r'/usr/local/tomcat/webapps/'
    remote_path = r'/usr/local/tomcat/webapps/inspur-web/'
    
    # 实例化
    ssh = SSHConnect(hostname='10.201.10.200', username='root', password='inspur')
    # 自动打包部署
    ssh.auto_deploy(project_path, remote_path)