import pysftp
import paramiko
import os
import unittest
import json
import warnings
warnings.filterwarnings("ignore")
class SftpUtils(object):
"""
cmd: execute shell command on remote linux server.
exist_file : if remote path exists on remote return True ,else return False.
exist_dir : if remote path exists on remote return True ,else return False.
is_dir: if remote path is dir return True ,else return False.
is_file : if remote path is file return True ,else return False.
remove: remove the remote file on remote linux .
rm: remove dir and all files/dir under remote dir and dir it self.
mkdir: mkdir dir on remote linux but it parent dir must exists ,if it already exist delete remote dir exists first and mkdir remote dir .
makedirs: mkdir the dir on remote linux , if it parent dir and dir self not exists,
it will mkdir parent dir and dir, else mkdir dir.
get: get remote linux file to local ,local filename and remote filename must absolute filepath.
put: upload local file to remote linux local filename and remote filename must absolute filepath.
rename:
rename a file or directory on the remote host
:param str remote_src: the remote file/directory to rename
:param str remote_dest: the remote file/directory to put it
open: read remote file content and return readlines list
cwd : cd curdir to newdir
getcwd: get remote linux cursor current dir path and return it
chmod : change file r,w,x grants.
chown: change file or dir owner
opens: read and write file on remote linux ,if remote file not exists it will create it first when write .
exists: if dir or file exists on remote linux return True ,else return False .
list_dir: query all files under dir ,the same with shell ls -l -a dir/ .
ger_recursive:
recursive copy remote files or dir under remote dir to localdir
put_recursive: recursive copy localdir's all file or dir under it to remote dir
"""
def __init__(self, ip, port, user, pwd):
self.ip = ip
self.port = port
self.user = user
self.pwd = pwd
self.sftp = self.sftp_client()
def sftp_client(self):
# https://pysftp.readthedocs.io/en/release_0.2.9/cookbook.html#pysftp-cnopts
# private_key="/home/app/privatekey_file"
cnopts = pysftp.CnOpts()
cnopts.hostkeys = None
sftp = pysftp.Connection(host=self.ip, username=self.user, password=self.pwd, port=self.port, cnopts=cnopts)
return sftp
def cmd(self, shell_cmd: str):
"""get stdout or err"""
std = self.sftp.execute(shell_cmd)
if std:
readlines = ''
for line in std:
readlines = readlines + line.decode("utf-8")
return readlines
def exists_file(self, filepath):
try:
self.sftp.stat(filepath)
except Exception:
return False
else:
return True
def exists_dir(self, dirpath):
try:
self.sftp.stat(dirpath)
except Exception:
return False
else:
return True
def is_dir(self, dirpath):
return True if self.sftp.isdir(remotepath=dirpath) else False
def is_file(self, filepath):
return True if self.sftp.isfile(remotepath=filepath) else False
def remove(self, filepath):
"""remove remote file"""
if self.exists_file(filepath):
self.sftp.remove(filepath)
else:
raise FileNotFoundError("%s file not find on remote ! " % filepath)
def rm(self, dirpath):
""" rmdir just delete empty dirs """
if self.exists_dir(dirpath):
files = self.sftp.listdir(dirpath)
for file in files:
filepath = os.path.join(dirpath, file).replace("\",'/')
if self.is_dir(filepath):
self.rm(filepath)
else:
self.remove(filepath)
self.sftp.rmdir(dirpath)
else:
raise FileNotFoundError("%s dir not find on remote !" % dirpath)
def mkdir(self, remote_dir):
"""parent dir must exists"""
if self.exists(remote_dir):
self.rm(remote_dir)
self.sftp.mkdir(remotepath=remote_dir)
else:
self.sftp.mkdir(remotepath=remote_dir)
def mkdirs(self, remote_dir):
"""mkdir -p /usr/local/mkdir,it can create dir that parent dirs not exists,if exists,it can also create"""
self.sftp.makedirs(remotedir=remote_dir)
def get(self, remote_path, local_path):
"""sftp get,download """
self.sftp.get(remotepath=remote_path, localpath=local_path, callback=None)
def put(self, local_path, remote_path):
"""sftp put,upload """
self.sftp.put(localpath=local_path, remotepath=remote_path)
def rename(self, remote_path, new_name_path):
"""rename file or dir on remote """
self.sftp.rename(remote_src=remote_path, remote_dest=new_name_path)
def open(self, remote_filepath, mode):
"""open 'r+' or 'r' file ,return str"""
readlines = self.sftp.open(remote_file=remote_filepath, mode=mode)
return list(readlines)
def getcwd(self):
return self.sftp.getcwd()
def cwd(self, remote_path):
"""change dir to given remote_path"""
return self.sftp.cwd(remotepath=remote_path)
def chmod(self, remote_path, mode: int):
""" change file grants for w,r,x"""
self.sftp.chmod(remotepath=remote_path, mode=mode)
def chown(self, remote_path, uid: int, gid: int):
"""change owner of user and group """
self.sftp.chown(remote_path, uid, gid)
def chdir(self, remote_path):
"""cwd()== chdir()"""
self.sftp.chdir(remote_path)
def touch(self, filepath):
"""if exists ,pass it or raise exception """
if self.exists_file(filepath):
self.remove(filepath)
self.sftp_client().execute("touch %s" % filepath)
else:
self.sftp_client().execute("touch %s" % filepath)
def close(self):
self.sftp.close()
def opens(self, filepath, mode, file_data=None):
"""file remote read and write ,return str"""
tp = paramiko.Transport(self.ip, self.port)
tp.connect(username=self.user, password=self.pwd)
pipe = paramiko.sftp_client.SFTPClient.from_transport(tp)
if mode == "w" or mode == "w+":
with pipe.open(filepath, mode) as f:
f.write(file_data)
else:
if mode == "r" or mode == "r+":
with pipe.open(filepath, mode)as f:
return f.read().decode("utf-8")
def exists(self,remotepath):
return True if self.sftp.exists(remotepath) else False
def list_dir(self,dir):
return self.sftp.listdir(dir)
def get_recursive(self,remote_dir,localpdir,):
"""
preserve modification time on files if True keep modify file or dir last operate time ,else False
localpdir : the parent dir path of local receive linux dir
remote_dir : the target of copy remote_dir ,the result dir name is same with remote
"""
remote_pdir=os.path.dirname(remote_dir)
# local_pdir = os.path.dirname(localdir)
remote_target_dir=os.path.split(remote_dir)[-1]
self.sftp.chdir(remote_pdir) #cd remote_pdir
if self.exists(localpdir):
self.sftp.get_r(remote_target_dir,localpdir,preserve_mtime=False)
else:
#local create localdir pdir that not exisit
os.makedirs(localpdir)
self.sftp.get_r(remote_target_dir, localpdir, preserve_mtime=False)
def put_recursive(self,localdir,remote_pdir,preserve_mtime=False):
"""
remote_pdir: the parent dir of receive local targetdir ,if parent exists copy target same
target dirname diredctly to remote_dir,if parent dir not exists ,first makedirs parent dir of remote_pdir,then copy same name
target dirname to remote_pdir
localdir: local target dir absolute dir path
"""
local_targetdir = os.path.split(localdir)[-1]
remote_receivedir = os.path.join(remote_pdir, local_targetdir)
local_pdir=os.path.dirname(localdir)
os.chdir(local_pdir) #cd local pdir
if self.sftp.exists(remote_receivedir):
self.sftp.put_r(local_targetdir,remote_receivedir)
else:
#remote create pdir not exists
self.sftp.makedirs(remote_receivedir)
self.sftp.put_r( local_targetdir,remote_receivedir,preserve_mtime)
def local_rm(dirpath):
if os.path.exists(dirpath):
files = os.listdir(dirpath)
for file in files:
filepath = os.path.join(dirpath, file).replace("\",'/')
if os.path.isdir(filepath):
local_rm(filepath)
else:
os.remove(filepath)
os.rmdir(dirpath)
class TestSftp(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.sftp=SftpUtils(ip="192.168.110.151",port=22,user="root",pwd="admin")
@classmethod
def tearDownClass(cls):
pass
def test_cmd(self):
shell="ip addr "
readlines=self.sftp.cmd(shell)
self.assertIn("192.168.110.151",readlines)
def test_touch(self):
path="/usr/local/test.txt"
self.sftp.touch(path)
res=self.sftp.exists(path)
self.assertEqual(res,True)
r=self.sftp.is_file(path)
self.assertEqual(r,True)
self.sftp.remove(path)
res2 = self.sftp.exists(path)
self.assertEqual(res2, False)
def test_isdir(self):
r=self.sftp.is_dir("/usr/local")
self.assertEqual(r,True)
k=self.sftp.is_dir("/fff/fcdh")
self.assertEqual(k,False)
def test_isfile(self):
r="/usr/tets.txt"
res=self.sftp.is_file(r)
self.assertEqual(res,False)
def test_mkdir(self):
self.sftp.mkdir("/usr/local/testmkdir")
r=self.sftp.exists("/usr/local/testmkdir")
self.assertEqual(r,True)
# self.sftp.rm("/usr/local/testmkdir")
def test_makedirs(self):
path="/usr/local/makedirs/mk1"
self.sftp.mkdirs(path)
r=self.sftp.exists(path)
self.assertEqual(r,True)
def test_rm(self):
path="/usr/local/testrm/rmdir/"
self.sftp.mkdirs(path)
files=self.sftp.list_dir("/usr/local/testrm")
self.assertIn('rmdir',files)
self.sftp.touch(path+"test.txt")
self.sftp.rm("/usr/local/testrm")
r=self.sftp.exists(path)
self.assertEqual(r,False)
def test_opens(self):
path="/usr/local/test.txt"
data={"k":99,"v":{"a":9990},"dd":0,"oo":6556}
self.sftp.opens(path,'w+',json.dumps(data))
lines=self.sftp.open(path,'r')
def test_get_r(self):
remote="/usr/local/listdir"
local="/usr/local/ttt"
# self.sftp.chdir("/usr/local")
# get current dir
# print(self.sftp.getcwd())
self.sftp.get_recursive(remote,local)
local_rm(local)
def test_put_r(self):
local="/usr/local/upload/listdir"
remote="/usr/local/rrre"
self.sftp.put_recursive(local,remote)
self.sftp.rm(remote)
if __name__ == '__main__':
unittest.main(verbosity=2)
测试：
python -m unittest sftp_tools.TestSftp
[root@hostuser local]# sh sftptool.sh
..........
----------------------------------------------------------------------
Ran 10 tests in 3.017s
OK