【Python學習 】Python實現的FTP上傳和下載功能
阿新 • • 發佈：2019-02-03
一、背景
最近公司的一些自動化操作需要使用Python來實現FTP的上傳和下載功能。因此參考網上的例子，擼了一段程式碼來實現了該功能，下面做個記錄。
二、ftplib介紹
Python中預設安裝的ftplib模組定義了FTP類，其中函式有限，可用來實現簡單的ftp客戶端，用於上傳或下載檔案。
ftplib中的FTP 主要有以下這些方法
這些方法，可以參考原始碼，也可以參考上面貼的官方文件來進行學習。
我們主要需要使用到的api有以下幾個：
- 1、連線伺服器：
ftp=FTP() #設定變數
ftp.connect(“IP”,”port”) #連線的ftp sever和埠
ftp.login(“user”,”password”)#連線的使用者名稱，密碼如果匿名登入則用空串代替即可
- 2、上傳檔案：
fp=open(‘E:/test.xlsx’,’rb’)
cmd=’STOR filepath/test.xlsx’
ftp.storbinary(cmd, fp)
storbinary是以二進位制形式上傳檔案。
cmd: STOR命令，是FTP的一個命令，後面需要加上儲存檔案的路徑及檔名
fp: 一個開啟的檔案物件，‘rb’,以二進位制形式開啟檔案
- 3、下載檔案：
ftp.nlst(path) #獲取目錄下的檔案
file_handle=open(filename,”wb”).write #以寫模式在本地開啟檔案
ftp.retrbinaly(“RETR filename.txt”,file_handle) #接收伺服器上檔案並寫入本地檔案
- 4、其他方法：
ftp.set_debuglevel(2) #開啟除錯級別2，顯示詳細資訊
ftp.set_pasv(0) #0主動模式 1 #被動模式
print ftp.getwelcome() #打印出歡迎資訊
ftp.cmd(“xxx/xxx”) #更改遠端目錄
ftp.set_debuglevel(0) #關閉除錯模式
ftp.quit() #退出ftp
ftp.dir() #顯示目錄下檔案資訊
ftp.mkd(pathname) #新建遠端目錄
ftp.pwd() #返回當前所在位置
ftp.rmd(dirname) #刪除遠端目錄
ftp.delete(filename) #刪除遠端檔案
ftp.rename(fromname, toname) #將fromname修改名稱為toname。
三、使用ftplib來實現FTP上傳下載功能
參考官網以及網際網路其他資源，擼了下面這份程式碼，作為一個FTP工具類
#!/usr/bin/python
# -*- coding: UTF-8 -*-
from ftplib import FTP
import os
import sys
import time
import socket
class MyFTP:
"""
ftp自動下載、自動上傳指令碼，可以遞迴目錄操作
作者：歐陽鵬
部落格地址：http://blog.csdn.net/ouyang_peng/article/details/79271113
"""
def __init__(self, host, port=21):
""" 初始化 FTP 客戶端
引數:
host:ip地址
port:埠號
"""
# print("__init__()---> host = %s ,port = %s" % (host, port))
self.host = host
self.port = port
self.ftp = FTP()
# 重新設定下編碼方式
self.ftp.encoding = 'gbk'
self.log_file = open("log.txt", "a")
self.file_list = []
def login(self, username, password):
""" 初始化 FTP 客戶端
引數:
username: 使用者名稱
password: 密碼
"""
try:
timeout = 60
socket.setdefaulttimeout(timeout)
# 0主動模式 1 #被動模式
self.ftp.set_pasv(True)
# 開啟除錯級別2，顯示詳細資訊
# self.ftp.set_debuglevel(2)
self.debug_print('開始嘗試連線到 %s' % self.host)
self.ftp.connect(self.host, self.port)
self.debug_print('成功連線到 %s' % self.host)
self.debug_print('開始嘗試登入到 %s' % self.host)
self.ftp.login(username, password)
self.debug_print('成功登入到 %s' % self.host)
self.debug_print(self.ftp.welcome)
except Exception as err:
self.deal_error("FTP 連線或登入失敗 ，錯誤描述為：%s" % err)
pass
def is_same_size(self, local_file, remote_file):
"""判斷遠端檔案和本地檔案大小是否一致
引數:
local_file: 本地檔案
remote_file: 遠端檔案
"""
try:
remote_file_size = self.ftp.size(remote_file)
except Exception as err:
# self.debug_print("is_same_size() 錯誤描述為：%s" % err)
remote_file_size = -1
try:
local_file_size = os.path.getsize(local_file)
except Exception as err:
# self.debug_print("is_same_size() 錯誤描述為：%s" % err)
local_file_size = -1
self.debug_print('local_file_size:%d , remote_file_size:%d' % (local_file_size, remote_file_size))
if remote_file_size == local_file_size:
return 1
else:
return 0
def download_file(self, local_file, remote_file):
"""從ftp下載檔案
引數:
local_file: 本地檔案
remote_file: 遠端檔案
"""
self.debug_print("download_file()---> local_path = %s ,remote_path = %s" % (local_file, remote_file))
if self.is_same_size(local_file, remote_file):
self.debug_print('%s 檔案大小相同，無需下載' % local_file)
return
else:
try:
self.debug_print('>>>>>>>>>>>>下載檔案 %s ... ...' % local_file)
buf_size = 1024
file_handler = open(local_file, 'wb')
self.ftp.retrbinary('RETR %s' % remote_file, file_handler.write, buf_size)
file_handler.close()
except Exception as err:
self.debug_print('下載檔案出錯，出現異常：%s ' % err)
return
def download_file_tree(self, local_path, remote_path):
"""從遠端目錄下載多個檔案到本地目錄
引數:
local_path: 本地路徑
remote_path: 遠端路徑
"""
print("download_file_tree()---> local_path = %s ,remote_path = %s" % (local_path, remote_path))
try:
self.ftp.cwd(remote_path)
except Exception as err:
self.debug_print('遠端目錄%s不存在，繼續...' % remote_path + " ,具體錯誤描述為：%s" % err)
return
if not os.path.isdir(local_path):
self.debug_print('本地目錄%s不存在，先建立本地目錄' % local_path)
os.makedirs(local_path)
self.debug_print('切換至目錄: %s' % self.ftp.pwd())
self.file_list = []
# 方法回撥
self.ftp.dir(self.get_file_list)
remote_names = self.file_list
self.debug_print('遠端目錄 列表: %s' % remote_names)
for item in remote_names:
file_type = item[0]
file_name = item[1]
local = os.path.join(local_path, file_name)
if file_type == 'd':
print("download_file_tree()---> 下載目錄： %s" % file_name)
self.download_file_tree(local, file_name)
elif file_type == '-':
print("download_file()---> 下載檔案： %s" % file_name)
self.download_file(local, file_name)
self.ftp.cwd("..")
self.debug_print('返回上層目錄 %s' % self.ftp.pwd())
return True
def upload_file(self, local_file, remote_file):
"""從本地上傳檔案到ftp
引數:
local_path: 本地檔案
remote_path: 遠端檔案
"""
if not os.path.isfile(local_file):
self.debug_print('%s 不存在' % local_file)
return
if self.is_same_size(local_file, remote_file):
self.debug_print('跳過相等的檔案: %s' % local_file)
return
buf_size = 1024
file_handler = open(local_file, 'rb')
self.ftp.storbinary('STOR %s' % remote_file, file_handler, buf_size)
file_handler.close()
self.debug_print('上傳: %s' % local_file + "成功!")
def upload_file_tree(self, local_path, remote_path):
"""從本地上傳目錄下多個檔案到ftp
引數:
local_path: 本地路徑
remote_path: 遠端路徑
"""
if not os.path.isdir(local_path):
self.debug_print('本地目錄 %s 不存在' % local_path)
return
self.ftp.cwd(remote_path)
self.debug_print('切換至遠端目錄: %s' % self.ftp.pwd())
local_name_list = os.listdir(local_path)
for local_name in local_name_list:
src = os.path.join(local_path, local_name)
if os.path.isdir(src):
try:
self.ftp.mkd(local_name)
except Exception as err:
self.debug_print("目錄已存在 %s ,具體錯誤描述為：%s" % (local_name, err))
self.debug_print("upload_file_tree()---> 上傳目錄： %s" % local_name)
self.upload_file_tree(src, local_name)
else:
self.debug_print("upload_file_tree()---> 上傳檔案： %s" % local_name)
self.upload_file(src, local_name)
self.ftp.cwd("..")
def close(self):
""" 退出ftp
"""
self.debug_print("close()---> FTP退出")
self.ftp.quit()
self.log_file.close()
def debug_print(self, s):
""" 列印日誌
"""
self.write_log(s)
def deal_error(self, e):
""" 處理錯誤異常
引數：
e：異常
"""
log_str = '發生錯誤: %s' % e
self.write_log(log_str)
sys.exit()
def write_log(self, log_str):
""" 記錄日誌
引數：
log_str：日誌
"""
time_now = time.localtime()
date_now = time.strftime('%Y-%m-%d', time_now)
format_log_str = "%s ---> %s \n " % (date_now, log_str)
print(format_log_str)
self.log_file.write(format_log_str)
def get_file_list(self, line):
""" 獲取檔案列表
引數：
line：
"""
file_arr = self.get_file_name(line)
# 去除 . 和 ..
if file_arr[1] not in ['.', '..']:
self.file_list.append(file_arr)
def get_file_name(self, line):
""" 獲取檔名
引數：
line：
"""
pos = line.rfind(':')
while (line[pos] != ' '):
pos += 1
while (line[pos] == ' '):
pos += 1
file_arr = [line[0], line[pos:]]
return file_arr
if __name__ == "__main__":
my_ftp = MyFTP("172.28.180.117")
my_ftp.login("ouyangpeng", "ouyangpeng")
# 下載單個檔案
my_ftp.download_file("G:/ftp_test/XTCLauncher.apk", "/App/AutoUpload/ouyangpeng/I12/Release/XTCLauncher.apk")
# 下載目錄
# my_ftp.download_file_tree("G:/ftp_test/", "App/AutoUpload/ouyangpeng/I12/")
# 上傳單個檔案
# my_ftp.upload_file("G:/ftp_test/Release/XTCLauncher.apk", "/App/AutoUpload/ouyangpeng/I12/Release/XTCLauncher.apk")
# 上傳目錄
# my_ftp.upload_file_tree("G:/ftp_test/", "/App/AutoUpload/ouyangpeng/I12/")
my_ftp.close()3.1 測試下載遠端單個檔案
我們將 程式入口改為如下所示的程式碼，測試下下載單個檔案
測試前的測試環境如下圖所示：
if __name__ == "__main__":
my_ftp = MyFTP("172.28.180.117")
my_ftp.login("ouyangpeng", "ouyangpeng")
# 下載單個檔案
my_ftp.download_file("G:/ftp_test/XTCLauncher.apk", "/App/AutoUpload/ouyangpeng/I12/Release/XTCLauncher.apk")
my_ftp.close()執行結果為：
"C:\Code Python\JenkinsAPI\venv\Scripts\python.exe" "C:/Code Python/JenkinsAPI/ftptest.py"
2018-02-06 ---> 開始嘗試連線到 172.28.180.117
2018-02-06 ---> 成功連線到 172.28.180.117
2018-02-06 ---> 開始嘗試登入到 172.28.180.117
2018-02-06 ---> 成功登入到 172.28.180.117
2018-02-06 ---> 220 (vsFTPd 2.3.5)
2018-02-06 ---> download_file()---> local_path = G:/ftp_test/XTCLauncher.apk ,remote_path = /App/AutoUpload/ouyangpeng/I12/Release/XTCLauncher.apk
2018-02-06 ---> local_file_size:-1 , remote_file_size:16749148
2018-02-06 ---> >>>>>>>>>>>>下載檔案 G:/ftp_test/XTCLauncher.apk ... ...
2018-02-06 ---> close()---> FTP退出
Process finished with exit code 0
重新整理下目錄，檢視檔案已經下載成功！
3.2 測試下載遠端目錄
我們將 程式入口改為如下所示的程式碼，測試下載整個資料夾，執行之前的環境如下所示：
if __name__ == "__main__":
my_ftp = MyFTP("172.28.180.117")
my_ftp.login("ouyangpeng", "ouyangpeng")
# 下載目錄
my_ftp.download_file_tree("G:/ftp_test/", "App/AutoUpload/ouyangpeng/I12/")
my_ftp.close()執行結果為：
"C:\Code Python\JenkinsAPI\venv\Scripts\python.exe" "C:/Code Python/JenkinsAPI/ftptest.py"
2018-02-06 ---> 開始嘗試連線到 172.28.180.117
2018-02-06 ---> 成功連線到 172.28.180.117
2018-02-06 ---> 開始嘗試登入到 172.28.180.117
2018-02-06 ---> 成功登入到 172.28.180.117
2018-02-06 ---> 220 (vsFTPd 2.3.5)
download_file_tree()---> local_path = G:/ftp_test/ ,remote_path = App/AutoUpload/ouyangpeng/I12/
2018-02-06 ---> 切換至目錄: /App/AutoUpload/ouyangpeng/I12
2018-02-06 ---> 遠端目錄 列表: [['d', 'Debug'], ['d', 'Release']]
download_file_tree()---> 下載目錄： Debug
download_file_tree()---> local_path = G:/ftp_test/Debug ,remote_path = Debug
2018-02-06 ---> 本地目錄G:/ftp_test/Debug不存在，先建立本地目錄
2018-02-06 ---> 切換至目錄: /App/AutoUpload/ouyangpeng/I12/Debug
2018-02-06 ---> 遠端目錄 列表: []
2018-02-06 ---> 返回上層目錄 /App/AutoUpload/ouyangpeng/I12
download_file_tree()---> 下載目錄： Release
download_file_tree()---> local_path = G:/ftp_test/Release ,remote_path = Release
2018-02-06 ---> 本地目錄G:/ftp_test/Release不存在，先建立本地目錄
2018-02-06 ---> 切換至目錄: /App/AutoUpload/ouyangpeng/I12/Release
2018-02-06 ---> 遠端目錄 列表: [['-', 'XTCLauncher.apk']]
download_file()---> 下載檔案： XTCLauncher.apk
2018-02-06 ---> download_file()---> local_path = G:/ftp_test/Release\XTCLauncher.apk ,remote_path = XTCLauncher.apk
2018-02-06 ---> local_file_size:-1 , remote_file_size:16749148
2018-02-06 ---> >>>>>>>>>>>>下載檔案 G:/ftp_test/Release\XTCLauncher.apk ... ...
2018-02-06 ---> 返回上層目錄 /App/AutoUpload/ouyangpeng/I12
2018-02-06 ---> 返回上層目錄 /App/AutoUpload/ouyangpeng
2018-02-06 ---> close()---> FTP退出
Process finished with exit code 0
重新整理下目錄，檢視整個目錄已經下載成功！
目錄下的檔案也成功下載下來
3.3 測試上傳單個檔案到遠端FTP伺服器
我們將 程式入口改為如下所示的程式碼，測試下載整個資料夾，執行之前的環境如下所示：
if __name__ == "__main__":
my_ftp = MyFTP("172.28.180.117")
my_ftp.login("ouyangpeng", "ouyangpeng")
# 上傳單個檔案
my_ftp.upload_file("G:/ftp_test/Release/XTCLauncher.apk", "/App/AutoUpload/ouyangpeng/I12/Release/XTCLauncher.apk")
my_ftp.close()執行結果為：
"C:\Code Python\JenkinsAPI\venv\Scripts\python.exe" "C:/Code Python/JenkinsAPI/ftptest.py"
2018-02-06 ---> 開始嘗試連線到 172.28.180.117
2018-02-06 ---> 成功連線到 172.28.180.117
2018-02-06 ---> 開始嘗試登入到 172.28.180.117
2018-02-06 ---> 成功登入到 172.28.180.117
2018-02-06 ---> 220 (vsFTPd 2.3.5)
2018-02-06 ---> local_file_size:16749148 , remote_file_size:-1
2018-02-06 ---> 上傳: G:/ftp_test/Release/XTCLauncher.apk成功!
2018-02-06 ---> close()---> FTP退出
Process finished with exit code 0
重新整理下介面，可以看到檔案上傳成功了！
3.4 測試上傳資料夾到遠端FTP伺服器
我們將 程式入口改為如下所示的程式碼，測試下載整個資料夾，執行之前的環境如下所示：
if __name__ == "__main__":
my_ftp = MyFTP("172.28.180.117")
my_ftp.login("ouyangpeng", "ouyangpeng")
# 上傳目錄
my_ftp.upload_file_tree("G:/ftp_test/", "/App/AutoUpload/ouyangpeng/I12/")
my_ftp.close()執行結果為：
"C:\Code Python\JenkinsAPI\venv\Scripts\python.exe" "C:/Code Python/JenkinsAPI/ftptest.py"
2018-02-06 ---> 開始嘗試連線到 172.28.180.117
2018-02-06 ---> 成功連線到 172.28.180.117
2018-02-06 ---> 開始嘗試登入到 172.28.180.117
2018-02-06 ---> 成功登入到 172.28.180.117
2018-02-06 ---> 220 (vsFTPd 2.3.5)
2018-02-06 ---> 切換至遠端目錄: /App/AutoUpload/ouyangpeng/I12
2018-02-06 ---> 目錄已存在 Debug ,具體錯誤描述為：550 Create directory operation failed.
2018-02-06 ---> upload_file_tree()---> 上傳目錄： Debug
2018-02-06 ---> 切換至遠端目錄: /App/AutoUpload/ouyangpeng/I12/Debug
2018-02-06 ---> upload_file_tree()---> 上傳目錄： Release
2018-02-06 ---> 切換至遠端目錄: /App/AutoUpload/ouyangpeng/I12/Release
2018-02-06 ---> upload_file_tree()---> 上傳檔案： XTCLauncher.apk
2018-02-06 ---> local_file_size:16749148 , remote_file_size:-1
2018-02-06 ---> 上傳: G:/ftp_test/Release\XTCLauncher.apk成功!
2018-02-06 ---> upload_file_tree()---> 上傳檔案： v1.jpg
2018-02-06 ---> local_file_size:203677 , remote_file_size:-1
2018-02-06 ---> 上傳: G:/ftp_test/v1.jpg成功!
2018-02-06 ---> close()---> FTP退出
Process finished with exit code 0
重新整理下介面，可以看到檔案上傳成功了！
3.5 解決中文編碼奔潰的問題
準備上傳含有中文的檔案到FTP伺服器的時候，會奔潰，奔潰如下所示：
2018-02-06 ---> 切換至遠端目錄: /App/AutoUpload/ouyangpeng/I12
2018-02-06 ---> upload_file_tree()---> 上傳目錄： Debug
2018-02-06 ---> 切換至遠端目錄: /App/AutoUpload/ouyangpeng/I12/Debug
2018-02-06 ---> upload_file_tree()---> 上傳檔案： Python程式設計快速上手__讓繁瑣工作自動化.pdf
2018-02-06 ---> local_file_size:14773136 , remote_file_size:-1
Traceback (most recent call last):
File "C:/Code Python/JenkinsAPI/ftptest.py", line 277, in <module>
my_ftp.upload_file_tree("G:/ftp_test/", "/App/AutoUpload/ouyangpeng/I12/")
File "C:/Code Python/JenkinsAPI/ftptest.py", line 203, in upload_file_tree
self.upload_file(src, local_name)
File "C:/Code Python/JenkinsAPI/ftptest.py", line 172, in upload_file
self.ftp.storbinary('STOR %s' % remote_file, file_handler, buf_size)
File "D:\Python\lib\ftplib.py", line 502, in storbinary
with self.transfercmd(cmd, rest) as conn:
File "D:\Python\lib\ftplib.py", line 397, in transfercmd
return self.ntransfercmd(cmd, rest)[0]
File "D:\Python\lib\ftplib.py", line 363, in ntransfercmd
resp = self.sendcmd(cmd)
File "D:\Python\lib\ftplib.py", line 270, in sendcmd
self.putcmd(cmd)
File "D:\Python\lib\ftplib.py", line 197, in putcmd
self.putline(line)
File "D:\Python\lib\ftplib.py", line 192, in putline
self.sock.sendall(line.encode(self.encoding))
UnicodeEncodeError: 'latin-1' codec can't encode characters in position 11-16: ordinal not in range(256)
Process finished with exit code 1
提示我們
UnicodeEncodeError: 'latin-1' codec can't encode characters in position 11-16: ordinal not in range(256)檢視 Python ftplib模組，預設的編碼就是 latin-1
我們將這個ftp的編碼修改下即可上傳成功。將MyFTP的 _init_ 方法修改下，設定下ftp的encoding為gbk即可
def __init__(self, host, port=21):
""" 初始化 FTP 客戶端
引數:
host:ip地址
port:埠號
"""
# print("__init__()---> host = %s ,port = %s" % (host, port))
self.host = host
self.port = port
self.ftp = FTP()
# 重新設定下編碼方式
self.ftp.encoding = 'gbk'
self.log_file = open("log.txt", "a")
self.file_list = []增加一句 self.ftp.encoding = 'gbk',即可重新設定encoding了。
再次執行即可成功上傳，如下圖所示：
執行結果為：
"C:\Code Python\JenkinsAPI\venv\Scripts\python.exe" "C:/Code Python/JenkinsAPI/ftptest.py"
2018-02-06 ---> 開始嘗試連線到 172.28.180.117
2018-02-06 ---> 成功連線到 172.28.180.117
2018-02-06 ---> 開始嘗試登入到 172.28.180.117
2018-02-06 ---> 成功登入到 172.28.180.117
2018-02-06 ---> 220 (vsFTPd 2.3.5)
2018-02-06 ---> 切換至遠端目錄: /App/AutoUpload/ouyangpeng/I12
2018-02-06 ---> upload_file_tree()---> 上傳目錄： Debug
2018-02-06 ---> 切換至遠端目錄: /App/AutoUpload/ouyangpeng/I12/Debug
2018-02-06 ---> upload_file_tree()---> 上傳檔案： Python程式設計快速上手__讓繁瑣工作自動化.pdf
2018-02-06 ---> local_file_size:14773136 , remote_file_size:-1
2018-02-06 ---> 上傳: G:/ftp_test/Python程式設計快速上手__讓繁瑣工作自動化.pdf成功!
2018-02-06 ---> upload_file_tree()---> 上傳目錄： Release
2018-02-06 ---> 切換至遠端目錄: /App/AutoUpload/ouyangpeng/I12/Release
2018-02-06 ---> upload_file_tree()---> 上傳檔案： XTCLauncher.apk
2018-02-06 ---> local_file_size:16749148 , remote_file_size:-1
2018-02-06 ---> 上傳: G:/ftp_test/Release\XTCLauncher.apk成功!
2018-02-06 ---> upload_file_tree()---> 上傳檔案： v1.jpg
2018-02-06 ---> local_file_size:203677 , remote_file_size:-1
2018-02-06 ---> 上傳: G:/ftp_test/v1.jpg成功!
2018-02-06 ---> upload_file_tree()---> 上傳檔案： XTCLauncher.apk
2018-02-06 ---> local_file_size:16749148 , remote_file_size:-1
2018-02-06 ---> 上傳: G:/ftp_test/XTCLauncher.apk成功!
2018-02-06 ---> close()---> FTP退出
Process finished with exit code 0
四、參考連結
如果覺得本文對您有所幫助，歡迎您掃碼下圖所示的支付寶和微信支付二維碼對本文進行隨意打賞。您的支援將鼓勵我繼續創作！
