Python 操作 SFTP 和 FTP 完全指南

本文提供 Python 操作 SFTP 和 FTP 的完整解决方案，涵盖连接管理、文件传输、目录操作等核心功能。

点赞标记，明天就能用上这几个技巧！
linux 运维、shell、python、网络爬虫、数据采集等定定做，请私信。。。

核心库对比

SFTP 操作指南 (使用 Paramiko)

安装依赖

pip install paramiko

基础连接与文件操作

import paramikofrom paramiko import SSHClient, AutoAddPolicydef sftp_demo():host = "sftp.example.com"port = 22username = "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()

高级功能 - 递归上传目录

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 + "/" + itemif 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)

使用 SSH 密钥认证

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()# ...后续操作同上

FTP 操作指南 (使用 ftplib)

基础连接与文件操作

from ftplib import FTP, FTP_TLSimport osdef 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()

使用 TLS/SSL 加密 (FTPS)

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()

递归下载目录 (高级)

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. 连接安全

• SFTP优先于FTP

• 使用FTPS取代普通FTP

• 禁用FTP明文传输

2. 认证安全

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

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

3. 错误处理

from paramiko import SSHExceptiontry:# 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. 本地测试服务器

• SFTP: sudo apt-get install openssh-server

• FTP: sudo apt-get install vsftpd

2. Python测试库

# SFTP测试from unittest.mock import patch, MagicMockdef 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. 大文件传输

# 使用缓冲区优化大文件传输BUFFER_SIZE = 1024 * 1024# 1MBdef 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)

2. 并行传输

from concurrent.futures import ThreadPoolExecutordef 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()# 等待所有任务完成

扩展学习资源

1. 官方文档

• Paramiko 文档

• Python ftplib 文档

2. 替代库

• pysftp：更简单的SFTP封装

• ftputil：增强版FTP客户端

• fabric：自动化部署工具（包含SFTP）

3. 相关协议

• SCP：scp 模块 (基于paramiko)

• WebDAV：webdavclient3 库

• Cloud Storage：boto3 (AWS S3), google-cloud-storage

总结

根据您的安全需求选择协议：

• 需要加密传输：优先使用 SFTP

• 仅内部网络：可使用普通 FTP

• 需要合规性：使用 SFTP 或 FTPS

提示：实际应用中提议将连接管理封装为类，支持连接池和自动重连机制。

linux 运维、shell、python、网络爬虫、数据采集等定定做，请私信。。。
都看到这里了， 记得帮忙一键三连， 点赞、评论、收藏、关注