Python_ssh_sftp

# -*- coding: utf-8 -*-
import paramiko
import threading
import logging as logger

logger.basicConfig(level=logger.INFO)

def remote_command(host_ip, host_port, username, password, command):
    """
    远程执行Linux命令
    :param host_ip: 主机IP地址
    :type host_ip: str
    :param host_port: 主机SSH端口号
    :type host_port: int
    :param username: SSH用户名
    :type username: str
    :param password: SSH密码
    :type password: str
    :param command: 要执行的命令
    :type command: str
    """
    client = None
    try:
        # 创建SSH客户端
        client = paramiko.SSHClient()
        # 第一次SSH远程连接时会提示输入yes或者no
        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        client.connect(host_ip, host_port, username=username,
                       password=password, timeout=10)
        logger.info(f"开始在远程服务器 {host_ip} 上执行命令: {command}")
        # 执行命令
        stdin, stdout, stderr = client.exec_command(command, get_pty=True)
        # 获取命令执行结果，返回的数据是一个字符串
        result = stdout.read().decode('utf-8')
        logger.info(f"{host_ip} 执行结果: {result}")

        error = stderr.read().decode('utf-8')
        if error != "":
            logger.info(f"{host_ip} 错误信息: {error}")
        else:
            pass
    except Exception as e:
        logger.error(e)
    finally:
        client.close()

def batch_remote_command(host_list, host_port, username, password, command):
    """
    批量远程执行Linux命令
    :param host_list: 主机IP地址列表
    :type host_list: list
    :param host_port: 主机SSH端口号
    :type host_port: int
    :param username: SSH用户名
    :type username: str
    :param password: SSH密码
    :type password: str
    :param command: 要执行的命令
    :type command: str
    """
    thread_list = []
    for ip in host_list:
        thread = threading.Thread(target=remote_command, args=(
            ip, host_port, username, password, command))
        # 将生成的线程添加到列表中
        thread_list.append(thread)
    for t in thread_list:
        # 开始执行线程
        t.start()
    for t in thread_list:
        # 挂起线程，等待所有线程结束
        t.join()

def upload_file_to_remote_server(host, port, username, password, local_path, remote_path):
    """
    上传文件从本地到远程服务器
    Args:
        host (str): 远程服务器的IP地址或主机名
        port (int): 远程服务器的端口
        username (str): 远程服务器的用户名
        password (str): 远程服务器的密码
        local_path (str): 本地文件路径，要上传的文件
        remote_path (str): 远程文件路径，上传的目标位置
    """
    # 连接远程服务器
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect(hostname=host, port=port, username=username, password=password)
    # 打开SFTP客户端
    sftp = ssh.open_sftp()
    # 上传本地文件至远程服务器
    sftp.put(local_path, remote_path)
    # 关闭SFTP客户端和SSH连接
    sftp.close()
    ssh.close()

def download_file_from_remote_server(host, port, username, password, remote_path, local_path):
    """
    从远程服务器下载文件到本地
    Args:
        host (str): 远程服务器的IP地址或主机名
        port (int): 远程服务器的端口
        username (str): 远程服务器的用户名
        password (str): 远程服务器的密码
        remote_path (str): 远程文件路径，要下载的文件
        local_path (str): 本地文件路径，下载的目标位置
    """
    # 连接远程服务器
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect(hostname=host, port=port, username=username, password=password)
    # 打开SFTP客户端
    sftp = ssh.open_sftp()
    # 下载远程服务器文件至本地
    sftp.get(remote_path, local_path)
    # 关闭SFTP客户端和SSH连接
    sftp.close()
    ssh.close()

if __name__ == '__main__':
    upload_file_to_remote_server(host='h.vimll.com', port=xxx, username='xxx',
                                 password='xxxx', local_path='D:\\git-python\\常用工具包\\MTU.py', remote_path='MTU.py')
    download_file_from_remote_server(host='h.vimll.com', port=xxx,
                                     username='xxxx', password='xxxx', remote_path='MTU.py', local_path='D:\\test.py')