🚀 Python 操作 SFTP 和 FTP 完全指南

1

pip install paramiko

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

import paramiko
from paramiko import SSHClient, AutoAddPolicy

def sftp_demo():
host = "sftp.example.com"
port = 22
username = "your_username"
password = "your_password"# 或使用密钥

# 创建SSH客户端
ssh = SSHClient()
ssh.set_missing_host_key_policy(AutoAddPolicy())

try:
# 建立连接
ssh.connect(host, port, username, password)

# 创建SFTP客户端
sftp = ssh.open_sftp()

# 1. 上传文件
sftp.put("local_file.txt", "/remote/path/file.txt")

# 2. 下载文件
sftp.get("/remote/path/file.txt", "local_copy.txt")

# 3. 创建目录
sftp.mkdir("/remote/new_folder")

# 4. 列出目录内容
file_list = sftp.listdir("/remote/path")
print("目录内容:", file_list)

# 5. 删除文件
sftp.remove("/remote/path/old_file.txt")

# 6. 重命名文件
sftp.rename("/remote/path/old_name.txt", "/remote/path/new_name.txt")

# 7. 获取文件信息
file_attr = sftp.stat("/remote/path/file.txt")
print(f"文件大小: {file_attr.st_size} bytes")

except Exception as e:
print(f"SFTP操作失败: {str(e)}")
finally:
# 关闭连接
sftp.close()
ssh.close()

sftp_demo()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

def sftp_upload_dir(sftp, local_dir, remote_dir):
"""递归上传整个目录"""
import os

# 确保远程目录存在
try:
sftp.mkdir(remote_dir)
except IOError:
pass# 目录已存在

# 遍历本地目录
for item in os.listdir(local_dir):
local_path = os.path.join(local_dir, item)
remote_path = remote_dir + "/" + item

if os.path.isfile(local_path):
sftp.put(local_path, remote_path)
print(f"上传文件: {local_path} -> {remote_path}")
elif os.path.isdir(local_path):
sftp_upload_dir(sftp, local_path, remote_path)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

def sftp_with_key():
host = "sftp.example.com"
username = "your_username"
private_key_path = "/path/to/private_key.pem"

# 加载私钥
key = paramiko.RSAKey.from_private_key_file(private_key_path)

ssh = SSHClient()
ssh.set_missing_host_key_policy(AutoAddPolicy())
ssh.connect(host, username=username, pkey=key)

sftp = ssh.open_sftp()
# ...后续操作同上

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

from ftplib import FTP, FTP_TLS
import os

def ftp_demo():
host = "ftp.example.com"
username = "your_username"
password = "your_password"

try:
# 创建FTP连接
ftp = FTP(host)
ftp.login(username, password)
print(ftp.getwelcome())# 打印欢迎信息

# 1. 上传文件
with open("local_file.txt", "rb") as f:
ftp.storbinary("STOR remote_file.txt", f)

# 2. 下载文件
with open("local_copy.txt", "wb") as f:
ftp.retrbinary("RETR remote_file.txt", f.write)

# 3. 列出目录内容
files = []
ftp.dir(files.append)# 列出当前目录
print("目录内容:", files)

# 4. 创建目录
ftp.mkd("new_folder")

# 5. 切换目录
ftp.cwd("subdirectory")

# 6. 删除文件
ftp.delete("old_file.txt")

# 7. 重命名文件
ftp.rename("old_name.txt", "new_name.txt")

# 8. 获取文件大小
size = ftp.size("remote_file.txt")
print(f"文件大小: {size} bytes")

except Exception as e:
print(f"FTP操作失败: {str(e)}")
finally:
ftp.quit()

ftp_demo()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

def ftps_demo():
host = "ftps.example.com"
username = "your_username"
password = "your_password"

try:
# 创建加密连接
ftp = FTP_TLS(host)
ftp.login(username, password)
ftp.prot_p()# 切换到加密数据连接

# 后续操作与普通FTP相同...
with open("local_file.txt", "rb") as f:
ftp.storbinary("STOR secure_file.txt", f)

finally:
ftp.quit()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

def ftp_download_dir(ftp, remote_dir, local_dir):
"""递归下载整个目录"""
import os

# 创建本地目录
if not os.path.exists(local_dir):
os.makedirs(local_dir)

# 切换到远程目录
ftp.cwd(remote_dir)

# 获取文件列表
files = []
ftp.retrlines('LIST', files.append)

for line in files:
# 解析FTP列表行
parts = line.split()
filename = parts[-1]

# 跳过特殊目录
if filename in ['.', '..']:
continue

# 处理目录
if line.startswith('d'):
ftp_download_dir(ftp, filename, os.path.join(local_dir, filename))
# 处理文件
else:
local_path = os.path.join(local_dir, filename)
with open(local_path, 'wb') as f:
ftp.retrbinary(f'RETR {filename}', f.write)
print(f"下载文件: {filename}")

# 返回上级目录
ftp.cwd('..')

1
2
3
4
5
6
7

# 避免密码硬编码
import getpass
password = getpass.getpass("Enter SFTP password: ")

# 或使用环境变量
import os
password = os.environ.get("SFTP_PASSWORD")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

from paramiko import SSHException

try:
# SFTP操作代码
except SSHException as ssh_err:
print(f"SSH错误: {ssh_err}")
except IOError as io_err:
print(f"文件操作错误: {io_err}")
except Exception as e:
print(f"未知错误: {e}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

# SFTP测试
from unittest.mock import patch, MagicMock

def test_sftp_upload():
with patch('paramiko.SSHClient') as mock_ssh:
mock_sftp = MagicMock()
mock_ssh.return_value.open_sftp.return_value = mock_sftp

# 调用上传函数
sftp_upload_file("test.txt", "/remote/test.txt")

# 验证是否调用上传
mock_sftp.put.assert_called_with("test.txt", "/remote/test.txt")

1
2
3
4
5
6
7

# 使用缓冲区优化大文件传输
BUFFER_SIZE = 1024 * 1024# 1MB

def sftp_upload_large_file(local_path, remote_path):
with open(local_path, 'rb') as f:
sftp.putfo(f, remote_path, file_size=os.path.getsize(local_path),
callback=None, confirm=True)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

from concurrent.futures import ThreadPoolExecutor

def parallel_upload(files):
with ThreadPoolExecutor(max_workers=5) as executor:
futures = []
for local, remote in files:
futures.append(executor.submit(sftp.put, local, remote))

for future in futures:
future.result()# 等待所有任务完成