82 lines
2.5 KiB
Python
82 lines
2.5 KiB
Python
|
# -*- coding: UTF-8 -*-
|
|||
|
|
"""
|
|||
|
|
@Project -> File ：IoD_data_analysis_tool -> file_util
|
|||
|
|
@IDE ：PyCharm
|
|||
|
|
@Author ：rengengchen
|
|||
|
|
@Date ：2022/5/10 17:21
|
|||
|
|
@Desc ：
|
|||
|
|
"""
|
|||
|
|
import os
|
|||
|
|
import queue
|
|||
|
|
import shutil
|
|||
|
|
|
|||
|
|
import paramiko
|
|||
|
|
|
|||
|
|
|
|||
|
|
def list_files(dir_paths):
|
|||
|
|
files = []
|
|||
|
|
for root, dir_path, filepath in walk(dir_paths):
|
|||
|
|
if filepath:
|
|||
|
|
files.append(os.path.join(root, filepath))
|
|||
|
|
return files
|
|||
|
|
|
|||
|
|
|
|||
|
|
def walk(dir_paths):
|
|||
|
|
dir_queue = queue.Queue()
|
|||
|
|
if isinstance(dir_paths, str):
|
|||
|
|
dir_paths = [dir_paths]
|
|||
|
|
for dir_path in dir_paths:
|
|||
|
|
dir_queue.put(dir_path)
|
|||
|
|
while not dir_queue.empty():
|
|||
|
|
dirname = dir_queue.get()
|
|||
|
|
for root, dirs, files in os.walk(dirname):
|
|||
|
|
for dirname in dirs:
|
|||
|
|
dir_queue.put(os.path.join(root, dirname))
|
|||
|
|
yield root, dirname, None
|
|||
|
|
for filename in files:
|
|||
|
|
yield root, None, filename
|
|||
|
|
|
|||
|
|
|
|||
|
|
def copy(s, t):
|
|||
|
|
if os.path.isfile(s):
|
|||
|
|
shutil.copy(s, t)
|
|||
|
|
else:
|
|||
|
|
if not os.path.exists(t):
|
|||
|
|
os.mkdir(t)
|
|||
|
|
s = os.path.abspath(s)
|
|||
|
|
t = os.path.abspath(t)
|
|||
|
|
for root, dirname, filename in walk(s):
|
|||
|
|
if dirname:
|
|||
|
|
os.mkdir(os.path.join(t, dirname))
|
|||
|
|
else:
|
|||
|
|
shutil.copy(os.path.join(root, filename), os.path.join(root.replace(s, t), filename))
|
|||
|
|
|
|||
|
|
|
|||
|
|
class RemoteFileUtil:
|
|||
|
|
|
|||
|
|
def __init__(self, ip, username, password, port=22, local_dir=None, remote_dir=None):
|
|||
|
|
tran = paramiko.Transport((ip, port))
|
|||
|
|
tran.connect(username=username, password=password)
|
|||
|
|
self.sftp = paramiko.SFTPClient.from_transport(tran).getfo()
|
|||
|
|
self.local_dir = local_dir
|
|||
|
|
self.remote_dir = remote_dir
|
|||
|
|
|
|||
|
|
def ls(self, remote_dir=None):
|
|||
|
|
if remote_dir is None:
|
|||
|
|
remote_dir = self.remote_dir
|
|||
|
|
return self.sftp.listdir_attr(remote_dir)
|
|||
|
|
|
|||
|
|
def upload_file(self, local_filepath=None, remote_filepath=None, filename=None):
|
|||
|
|
if local_filepath is None:
|
|||
|
|
local_filepath = os.path.join(self.local_dir, filename)
|
|||
|
|
if remote_filepath is None:
|
|||
|
|
remote_filepath = os.path.join(self.remote_dir, filename)
|
|||
|
|
self.sftp.put(local_filepath, remote_filepath)
|
|||
|
|
|
|||
|
|
def download_file(self, local_filepath=None, remote_filepath=None, filename=None):
|
|||
|
|
if local_filepath is None:
|
|||
|
|
local_filepath = os.path.join(self.local_dir, filename)
|
|||
|
|
if remote_filepath is None:
|
|||
|
|
remote_filepath = os.path.join(self.remote_dir, filename)
|
|||
|
|
self.sftp.get(remote_filepath, local_filepath)
|