Python 连接Linux服务器完成上传下载和执行命令及查询目录下的文件
记录一些用于连接linux获取远端文件或者上传文件的小工具，另外还有执行shell命令和查找linux目录下文件是否存在
"""
@Time : 2021/7/22 10:31
@Author : chenhj
@File : sshclient.py
@Software: PyCharm
@Comment :
"""
import osfrom stat
import S_ISDIR
import re
import paramikofrom paramiko.ssh_exception
import NoValidConnectionsError, AuthenticationException, SSHException
class SshRemoteConnection:
def __init__(self, hostname, port, user, password):
self.hostname = hostname
self.port = port
self.user = user
self.password = password
def do_connect(self, conn):
"""创建连接"""
if conn == "SSHClient":
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
client.connect(hostname=self.hostname,port=self.port,username=self.user,password=self.password)
print("正在连接 {}".format(self.hostname).center(100, '*'))
except NoValidConnectionsError as e:
print(f"连接失败:{e}".center(100, '*'))
except AuthenticationException as e:
print(f"密码错误:{e}".center(100, '*'))
return client
elif conn == "Transport":
tran = paramiko.Transport(self.hostname, int(self.port))
try:
tran.connect(username=self.user, password=self.password)
except SSHException as e:
print(f'连接失败:{e}'.center(100, '*'))
return tran
def do_cmd(self, cmd):
"""执行shell命令，多条命令以分号‘；’隔开"""
conn = self.do_connect('SSHClient')
stdin, stdout, stderr = conn.exec_command(cmd, get_pty=True)
res, err = stdout.read().decode('utf-8'), stderr.read().decode('utf-8')
result = res if res else err
print('执行成功'.center(100, '*'))
print(result)
return result
def do_put_file(self, local_path, remote_path):
"""上传单个文件"""
print('正在上传文件...'.center(100, '*'))
conn = self.do_connect('Transport')
sftp = paramiko.SFTPClient.from_transport(conn)
try:
sftp.put(local_path, remotepath=remote_path)
print(f'{local_path}文件上传到{self.hostname}主机的{remote_path}')
except SSHException as e:
print(f'上传失败:{e}'.center(100, '*'))
conn.close()
def __get_all_local_files(self, local_dir):
"""获取需要批量上传目录下的所有文件"""
all_files = list()
files = os.listdir(local_dir)
for i in files:
local_filename = os.path.join(local_dir, i)
if os.path.isdir(i):
all_files.extend(self.__get_all_local_files(local_filename))
else:
all_files.append(local_filename)
return all_files
def do_put_all_files(self, local_path, remote_path):
"""批量上传所有文件"""
print('正在上传文件...'.center(100, '*'))
conn = self.do_connect('Transport')
sftp = paramiko.SFTPClient.from_transport(conn)
if remote_path[-1] == '/':
remote_path = remote_path[0:-1]
all_files = self.__get_all_local_files(local_path)
try:
for file in all_files:
local_filename = os.path.split(file)[-1]
remote_filename = remote_path + '/' + local_filename
print('{} 文件正在传输中...'.center(100, '*').format(remote_filename))
sftp.put(file, remote_filename)
print('文件传输成功...'.center(100, '*').format(remote_filename))
except SSHException as e:
print(f'上传失败:{e}'.center(100, '*'))
conn.close()
def do_get_file(self, remote_path, local_path):
"""下载文件"""
print('正在下载文件...'.center(100, '*'))
conn = self.do_connect('Transport')
sftp = paramiko.SFTPClient.from_transport(conn)
try:
sftp.get(remotepath=remote_path, localpath=local_path)
print(f'{remote_path}文件从{self.hostname}主机下载到{local_path}成功')
except SSHException as e:
print(f'下载失败:{e}'.center(100, '*'))
conn.close()
def __get_all_remote_files(self, sftp, remote_dir):
"""获取需要批量下载目录下的所有文件"""
all_files = list()
if remote_dir[-1] == '/':
remote_path = remote_dir[0:-1]
files = sftp.listdir_attr(remote_dir)
for i in files:
remote_filename = remote_dir + '/' + i.filename
if S_ISDIR(i.st_mode):
all_files.extend(self.__get_all_remote_files(sftp, remote_filename))
else:
all_files.append(remote_filename)
return all_files
def do_get_all_files(self, remote_path, local_path):
"""批量下载目录下所有文件"""
print('正在批量下载文件...'.center(100, '*'))
conn = self.do_connect('Transport')
sftp = paramiko.SFTPClient.from_transport(conn)
all_files = self.__get_all_remote_files(sftp, remote_path)
for file in all_files:
filename = file.split('/')[-1]
local_filename = os.path.join(local_path, filename)
print(f'{filename} 文件正在传输中...'.center(100, '*'))
sftp.get(file, local_filename)
print('传输完成...'.center(100, '*'))
conn.close()
def do_search_file(self, check_dir, checkup=None, fuzzy=True):
"""校验目录下的文件，默认模糊查询fuzzy=True，匹配忽略大小写"""
res = self.do_cmd(f'cd {check_dir};ls')
dir_list = [i for i in res.split('\r\n') if i != '']
num = 0
all_file = []
print('正在获取检索结果...'.center(100, '*'))
for file in dir_list:
if fuzzy:
ma = re.compile(f'{checkup}', flags=re.IGNORECASE).findall(f'{file}')
else:
ma = re.compile(f'{checkup}').findall(f'{file}')
if ma:
print(file)
all_file.append(file)
num += 1
else:
print('...')
print(f'检索完毕...找到{num}个结果'.center(100, '*'))
return all_file
def do_review_contents(self):
pass
if __name__ == "__main__":
s = SshRemoteConnection('99.99.45.123', 22, 'user', "123!@#")
s.do_cmd('pwd;ls -l')
localpath = 'D:\\00_data\\test'
filename = '111.DAT'
remotepath = '/data/20210719'
remote_filename = filename
s.do_get_all_files(remotepath, localpath)
s.do_search_file('/data/20210719', 'txt')
本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)
