FTP（File Transfer Protocol）雖然古老，但在自動化部署、備份同步、嵌入式設備檔案傳輸等場景依然廣泛使用。Python 標準庫 ftplib 提供了基礎 FTP 客戶端功能，但不支援目錄遞歸上傳下載。本文提供一個完整、可生產環境使用的 FTPSync 類，支援檔案與目錄的雙向遞歸傳輸，相容 Windows 與 Linux 平台。
set_pasv(False)），相容多種防火牆環境#!/usr/bin/env python
# -*- coding: utf-8 -*-
import ftplib
import os
import sys
class FTPSync(object):
"""支援目錄遞歸的 FTP 同步客戶端"""
def __init__(self, host, port=21):
self.conn = ftplib.FTP()
self.conn.connect(host, port)
def login(self, username, password):
self.conn.login(username, password)
self.conn.set_pasv(False) # 使用主動模式，相容性更好
print(self.conn.welcome)
# ==================== 內部工具方法 ====================
def _is_ftp_file(self, ftp_path):
"""判斷遠端路徑是否為檔案"""
try:
ftp_path = ftp_path.rstrip('/')
parent = os.path.dirname(ftp_path)
name = os.path.basename(ftp_path)
return name in self.conn.nlst(parent)
except ftplib.error_perm:
return False
def _ftp_list_callback(self, line):
"""LIST 指令回調，用於判斷目錄"""
parts = line.split()
if len(parts) >= 1 and parts[0].startswith('d') and parts[-1] == self._ftp_target_name:
self._is_dir_flag = True
def _is_ftp_dir(self, ftp_path):
"""判斷遠端路徑是否為目錄"""
ftp_path = ftp_path.rstrip('/')
if ftp_path in ('.', './', ''):
return True
self._ftp_target_name = os.path.basename(ftp_path)
parent = os.path.dirname(ftp_path)
self._is_dir_flag = False
try:
self.conn.retrlines(f'LIST {parent}', self._ftp_list_callback)
except ftplib.error_perm:
pass
return self._is_dir_flag
# ==================== 檔案操作 ====================
def get_file(self, ftp_path, local_path='.'):
"""下載單一檔案"""
ftp_path = ftp_path.rstrip('/')
if not self._is_ftp_file(ftp_path):
print(f'ERROR: 遠端檔案不存在: {ftp_path}')
return
filename = os.path.basename(ftp_path)
# 確定本地儲存路徑
if os.path.isdir(local_path):
local_filepath = os.path.join(local_path, filename)
elif os.path.isdir(os.path.dirname(local_path)):
local_filepath = local_path
else:
print(f'ERROR: 本地目錄不存在: {os.path.dirname(local_path)}')
return
with open(local_filepath, 'wb') as f:
self.conn.retrbinary(f'RETR {ftp_path}', f.write)
print(f'下載完成: {ftp_path} -> {local_filepath}')
def put_file(self, local_path, ftp_path='.'):
"""上傳單一檔案"""
if not os.path.isfile(local_path):
print(f'ERROR: 本地檔案不存在: {local_path}')
return
ftp_path = ftp_path.rstrip('/')
filename = os.path.basename(local_path)
with open(local_path, 'rb') as f:
# 遠端是目錄 -> 上傳到該目錄下，檔名不變
if self._is_ftp_dir(ftp_path):
remote_path = os.path.join(ftp_path, filename)
self.conn.storbinary(f'STOR {remote_path}', f)
# 遠端父目錄存在 -> 指定檔名上傳
elif self._is_ftp_dir(os.path.dirname(ftp_path)):
self.conn.storbinary(f'STOR {ftp_path}', f)
else:
print(f'ERROR: 遠端路徑錯誤: {ftp_path}')
return
print(f'上傳完成: {local_path} -> {ftp_path}')
# ==================== 目錄遞歸操作 ====================
def get_dir(self, ftp_path, local_path='.', is_root=True):
"""遞歸下載整個目錄"""
ftp_path = ftp_path.rstrip('/')
if not self._is_ftp_dir(ftp_path):
print(f'ERROR: 遠端目錄不存在: {ftp_path}')
return
# 根呼叫時初始化本地目標目錄
if is_root:
if not os.path.isdir(local_path):
os.makedirs(local_path)
local_path = os.path.join(local_path, os.path.basename(ftp_path))
if not os.path.isdir(local_path):
os.makedirs(local_path)
# 進入遠端目錄，遞歸處理
self.conn.cwd(ftp_path)
try:
items = self.conn.nlst()
except ftplib.error_perm:
items = []
for item in items:
local_item = os.path.join(local_path, item)
if self._is_ftp_dir(item):
self.get_dir(item, local_item, is_root=False)
else:
self.get_file(item, local_item)
# 返回上層目錄
self.conn.cwd('..')
def put_dir(self, local_path, ftp_path='.', is_root=True):
"""遞歸上傳整個目錄"""
if not os.path.isdir(local_path):
print(f'ERROR: 本地目錄不存在: {local_path}')
return
ftp_path = ftp_path.rstrip('/')
# 根呼叫時初始化遠端目標目錄
if is_root:
if not self._is_ftp_dir(ftp_path):
self.conn.mkd(ftp_path)
ftp_path = os.path.join(ftp_path, os.path.basename(local_path))
if not self._is_ftp_dir(ftp_path):
self.conn.mkd(ftp_path)
# 進入本地目錄，遞歸處理
old_cwd = os.getcwd()
os.chdir(local_path)
try:
for item in os.listdir('.'):
if os.path.isdir(item):
self.put_dir(item, os.path.join(ftp_path, item), is_root=False)
else:
self.put_file(item, ftp_path)
finally:
os.chdir(old_cwd)
# ==================== 使用範例 ====================
if __name__ == '__main__':
ftp = FTPSync('192.168.1.110')
ftp.login('test', 'test')
# 檔案操作範例
# ftp.put_file('111.txt', 'a/b') # 上傳到目錄，不重命名
# ftp.put_file('111.txt', 'a/112.txt') # 上傳並重命名
# ftp.get_file('/a/111.txt', r'D:') # 下載到目錄
# ftp.get_file('/a/111.txt', r'D:112.txt') # 下載並重命名
# 目錄操作範例
# ftp.get_dir('a/b/c', r'D:a') # 下載到現有目錄
# ftp.get_dir('a/b/c', r'D:aa') # 下載到新目錄
ftp.put_dir('b', 'a') # 上傳到現有目錄
ftp.put_dir('b', 'aa/B/') # 上傳到新目錄
FTP 協定本身不提供直接判斷路徑類型的指令。本實現採用 NLST 列出父目錄內容，檢查名稱是否存在來判斷檔案；對於目錄，使用 LIST -l 解析權限位（以 d 開頭）確認。這種方式雖需兩次網路往返，但相容性最好，不依賴伺服器擴展指令。
get_dir 和 put_dir 使用 is_root 旗標區分首次呼叫與遞歸呼叫：
這確保了 put_dir('src', 'dst') 會在遠端建立 dst/src/... 結構，而非扁平化合併。
本地遞歸上傳時，使用 os.chdir() 進入來源目錄，配合 try...finally 確保無論發生何種異常都能恢復原工作目錄，避免影響後續操作或主程式。
預設 set_pasv(False) 使用主動模式。許多企業防火牆對主動模式支援較好（只需開放 20/21 埠），而被動模式需要開放大範圍高位埠。若遇到連線成功但列表/傳輸卡住，可嘗試切換至 set_pasv(True)。
retrbinary/storbinary 的回調函數中累計位元組數，顯示進度條REST 指令配合 APPE 實現斷點續傳threading 或 asyncio 並行傳輸多個檔案ftplib.FTP_TLS 實現 FTPS 加密傳輸這個 FTPSync 類解決了 ftplib 不支援目錄遞歸的痛點，提供了生產級的錯誤處理與路徑判斷邏輯。無論是自動化部署腳本、定時備份任務，還是嵌入式設備日誌收集，都能直接拿來使用或輕鬆擴展。掌握其核心原理——FTP 指令組合與遞歸狀態管理——也能幫助你在其他網路協定（如 SFTP、WebDAV）上實現類似功能。
