关键参数：
sftp_route：远程路径local_route：本地路径（windows需要r解析）header：txt表头字段（列表方式）data：txt内容信息（列表嵌套列表方式）filename_success：是否创建success文件默认为falsefilename：文件名称（无需带后缀默认为txt文件）
注意：
上传文件需要带文件名称
下载只需要到文件路径即可
import os
import time
import paramiko
class SftpLocal:
"""
SFTP操作
支持配置协议，主机，账号，密码
支持生成文件
支持下载
支持删除
支持上传
"""
def __init__(self,
sftp_route: str,
local_route: str):
"""
:param sftp_route: SFTP路径
:param local_route: 本地路径
"""
self.sftp_route = sftp_route
self.local_route = local_route
# SFTP配置协议登录信息
self.protocol = 'sftp' # SFTP协议
self.host = '10.100.***.**' # 服务器主机名或IP地址
self.port = 9999 # SFTP服务器端口号，默认为22
self.username = '******' # 服务器用户名
self.password = ''******'' # 服务器密码
# 创建SSH客户端并连接到SFTP服务器
self.ssh = paramiko.SSHClient()
self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
self.ssh.connect(self.host, self.port, self.username, self.password)
# 创建SFTP对象
self.sftp = self.ssh.open_sftp()
def upload(self):
"""
样例：
local_route = '/path/to/local/file.txt'
sftp_route = '/path/to/remote/file.txt'
:param sftp_route: sftp路径
:param local_route: 本地路径
:return:
"""
# 上传本地文件到指定路径
self.sftp.put(self.local_route, self.sftp_route)
return True
def download(self):
"""
:param sftp_route: sftp路径
:param local_route: 本地路径
:return:
"""
files = self.sftp.listdir(self.sftp_route)
for filename in files:
remote_file = os.path.join(self.sftp_route, filename)
local_file = os.path.join(self.local_route, filename)
self.sftp.get(remote_file, local_file)
return True
def delete(self):
"""
删除sftp路径下的文件
:param sftp_route:
:return:
"""
self.sftp.remove(self.sftp_route)
return True
def write_txt(self, route: str, filename: str, header: list, data: list, filename_success: bool = False):
"""
# 定义表头和数据
header = ['Name', 'Age', 'Gender']
data = [
['Alice', 25, 'Female'],
['Bob', 30, 'Male']
]
:param route: 文件路径
:param filename: 文件名称
:param filename_success: 是否要生成一个success文件
:param args: 随机表头参数和内容以及其他随机参数
:return:
"""
if not filename_success:
# 打开文件并写入表头和数据
with open(route + filename + ".txt", 'w') as f:
# 写入表头
f.write(','.join(header) + '\n')
# 写入数据
for row in data:
f.write(','.join(row) + '\n')
else:
# 打开文件并写入表头和数据
with open(route + filename + ".txt", 'w') as f:
# 写入表头
f.write(','.join(header) + '\n')
# 写入数据
for row in data:
f.write(','.join(row) + '\n')
self.create_txt_success(route + filename)
return True
def create_txt_success(self, filename: str):
open(filename + ".txt" + ".success", "a")
return True
# 调试
# 上传
# SftpLocals = SftpLocal(sftp_route='/****/marketing/7433800/20231129/123.txt',
# local_route=r'G:\pycharmDl\job\MarkeScriptTest\sftp_local_txt\123.txt')
# print(SftpLocals.upload())
# 下载
# SftpLocals = SftpLocal(sftp_route='/****/marketing/7433800/20231129/',
# local_route = r'G:\pycharmDl\job\MarkeScriptTest\sftp_local_txt')
# print(SftpLocals.download())
# 创建
# SftpLocals = SftpLocal(sftp_route='/****/marketing/7433800/20231129/',
# local_route=r'G:\pycharmDl\job\MarkeScriptTest\sftp_local_txt')
# local_route = r'G:\pycharmDl\job\MarkeScriptTest\sftp_local_txt'
# today = time.strftime('%Y%m%d', time.localtime(time.time()))
# header = ['custNum', 'cell']
# data = [['asdfqwert', '13112341234'], ['asdfqwert', '13112341234']]
# print(SftpLocals.write_txt(route=local_route,
# filename=today,
# header=header,
# data=data,
# filename_success=True))