中山市原創(chuàng)動(dòng)力網(wǎng)絡(luò)科技有限公司
創(chuàng)您所想與您同在
免費(fèi)互聯(lián)網(wǎng)品牌營銷咨詢服務(wù)
立刻咨詢
13450931319來源：maozexijr 發(fā)布時(shí)間：2018-11-03 15:21:43 閱讀量：1580
注：代碼中的變量命名規(guī)則采用的是“駝峰式”命名法，這可能帶來代碼檢查提示，在PyCharm中可以Alt + Enter忽略此類檢查
1、配置文件 ftp_config.py
# FTP服務(wù)器的IP、端口、賬戶、密碼
host = "172.16.53.113"
port = 21 # not str
username = "admin"
password = "123456"
# remote和local是相對(duì)客戶端的
homeDir = "D:/FTP/"
localDir = "E:/FTP/"
2、主要代碼 ftp_client.py
# !/usr/bin/python3
# coding: utf-8
from config import ftp_config
import ftplib
import os
# 獲取連接
def getConnect(host, port, username, password):
"""
:param host: FTP ip
:param port: FTP port
:param username: FTP userName
:param password: FTP password
:return: ftp
"""
print("FTP connection...")
result = [1, ""]
try:
ftp = ftplib.FTP()
# ftp.set_debuglevel(2)
ftp.connect(host, port)
ftp.login(username, password)
result = [1, "connection success", ftp]
except Exception as e:
result = [-1, "connection fail, reason:{0}".format(e)]
return result
# 下載
def download(ftp, remotePath, localAbsDir):
"""
:param ftp:
:param remotePath: 服務(wù)端文件或文件夾的絕對(duì)或相對(duì)路徑
:param localAbsDir: 客戶端文件夾的絕對(duì)路徑，如E:/FTP/downDir/
:return:
"""
result = [1, ""]
try:
remotePath = formatPath(remotePath)
localAbsDir = formatPath(localAbsDir)
remoteRel = ""
if remotePath == "":
remotePath = ftp_config.homeDir
else:
if remotePath.startswith(ftp_config.homeDir):
remoteRel = remotePath.replace(ftp_config.homeDir, "/")
remoteRel = formatPath(remoteRel)
else:
remoteRel = remotePath
if localAbsDir == "":
localAbsDir = ftp_config.localDir
localAbsDir = formatPath(localAbsDir)
remoteAbs = formatPath(ftp_config.homeDir, remoteRel) # 服務(wù)端文件或文件夾的絕對(duì)路徑
if os.path.isdir(remoteAbs):
rs = downloadDir(ftp, remoteRel, localAbsDir)
else:
rs = downloadFile(ftp, remoteRel, localAbsDir)
if rs[0] == -1:
result[0] = -1
result[1] = result[1] + "\n" + rs[1]
except Exception as e:
result = [-1, "download fail, reason:{0}".format(e)]
return result
# 下載指定文件夾下的所有
def downloadDir(ftp, remoteRelDir, localAbsDir):
"""
:param ftp:
:param remoteRelDir: 服務(wù)端文件的相對(duì)路徑,含文件后綴，如/srcDir/
:param localAbsDir: 客戶端文件夾的絕對(duì)路徑，如E:/FTP/downDir/
:return:
"""
print("start download dir by use FTP...")
result = [1, ""]
try:
remoteRelDir = formatPath(remoteRelDir)
localAbsDir = formatPath(localAbsDir)
files = [] # 文件
dirs = [] # 文件夾
remotePaths = ftp.nlst(remoteRelDir)
if len(remotePaths) > 0:
for remotePath in remotePaths:
remotePath = formatPath(remotePath)
if isDir(ftp, remotePath):
dirs.append(remotePath)
else:
files.append(remotePath)
ftp.cwd("") # 切回homeDir
if len(files) > 0:
for rrp in files: # rrp is relPath
rs = downloadFile(ftp, rrp, localAbsDir)
if rs[0] == -1:
result[0] = -1
result[1] = result[1] + "\n" + rs[1]
if len(dirs) > 0:
for rrd in dirs: # rrd is relDir
dirName = lastDir(rrd)
localAbsDir = formatPath(localAbsDir, dirName)
rs = downloadDir(ftp, rrd, localAbsDir)
if rs[0] == -1:
result[0] = -1
result[1] = result[1] + "\n" + rs[1]
except Exception as e:
result = [-1, "download fail, reason:{0}".format(e)]
return result
# 下載指定文件
def downloadFile(ftp, remoteRelPath, localAbsDir):
"""
:param ftp:
:param remoteRelPath: 服務(wù)端文件的相對(duì)路徑,含文件后綴，如/srcDir/file.txt
:param localAbsDir: 客戶端文件夾的絕對(duì)路徑，如E:/FTP/downDir/
:return:
"""
print("start download file by use FTP...")
result = [1, ""]
try:
fileName = os.path.basename(remoteRelPath) # 文件名
localAbsPath = formatPath(localAbsDir, fileName)
splitPaths = os.path.split(localAbsPath)
lad = splitPaths[0]
lad = formatPath(lad)
if not os.path.exists(lad):
os.makedirs(lad)
handle = open(localAbsPath, "wb")
ftp.retrbinary("RETR %s" % remoteRelPath, handle.write, 1024)
handle.close()
result = [1, "download " + splitPaths[1] + " success"]
except Exception as e:
result = [-1, "download fail, reason:{0}".format(e)]
return result
# 上傳
def upload(ftp, remoteRelDir, localPath):
"""
:param ftp:
:param remoteRelDir: 服務(wù)端文件夾相對(duì)路徑，可以為None、""，此時(shí)文件上傳到homeDir
:param localPath: 客戶端文件或文件夾路徑，當(dāng)路徑以localDir開始，文件保存到homeDir的相對(duì)路徑下
:return:
"""
result = [1, ""]
try:
remoteRelDir = formatPath(remoteRelDir)
localPath = formatPath(localPath)
localRelDir = ""
if localPath == "":
localPath = ftp_config.localDir
localPath = formatPath(localPath)
else:
if localPath.startswith(ftp_config.localDir): # 絕對(duì)路徑
localRelDir = localPath.replace(ftp_config.localDir, "/")
localRelDir = formatPath(localRelDir)
else: # 相對(duì)(localDir)路徑
localPath = formatPath(ftp_config.localDir, localPath)
if remoteRelDir == "":
remoteRelDir = formatPath("/uploadFiles/", localRelDir)
else:
if remoteRelDir.startswith(ftp_config.homeDir):
remoteRelDir = remoteRelDir.replace(ftp_config.homeDir, "/")
remoteRelDir = formatPath(remoteRelDir)
if os.path.isdir(localPath): # isDir
rs = uploadDir(ftp, remoteRelDir, localPath)
else: # isFile
rs = uploadFile(ftp, remoteRelDir, localPath)
if rs[0] == -1:
result[0] = -1
result[1] = result[1] + "\n" + rs[1]
except Exception as e:
result = [-1, "upload fail, reason:{0}".format(e)]
return result
# 上傳指定文件夾下的所有
def uploadDir(ftp, remoteRelDir, localAbsDir):
"""
:param ftp:
:param remoteRelDir: 服務(wù)端文件夾相對(duì)路徑，可以為None、""，此時(shí)文件上傳到homeDir
:param localAbsDir: 客戶端文件夾路徑，當(dāng)路徑以localDir開始，文件保存到homeDir的相對(duì)路徑下
:return:
"""
print("start upload dir by use FTP...")
result = [1, ""]
try:
for root, dirs, files in os.walk(localAbsDir):
if len(files) > 0:
for fileName in files:
localAbsPath = localAbsDir + fileName
rs = uploadFile(ftp, remoteRelDir, localAbsPath)
if rs[0] == -1:
result[0] = -1
result[1] = result[1] + "\n" + rs[1]
if len(dirs) > 0:
for dirName in dirs:
rrd = formatPath(remoteRelDir, dirName)
lad = formatPath(localAbsDir, dirName)
rs = uploadDir(ftp, rrd, lad)
if rs[0] == -1:
result[0] = -1
result[1] = result[1] + "\n" + rs[1]
break
except Exception as e:
result = [-1, "upload fail, reason:{0}".format(e)]
return result
# 上傳指定文件
def uploadFile(ftp, remoteRelDir, localAbsPath):
"""
:param ftp:
:param remoteRelDir: 服務(wù)端文件夾相對(duì)路徑，可以為None、""，此時(shí)文件上傳到homeDir
:param localAbsPath: 客戶端文件路徑，當(dāng)路徑以localDir開始，文件保存到homeDir的相對(duì)路徑下
:return:
"""
print("start upload file by use FTP...")
result = [1, ""]
try:
try:
ftp.cwd(remoteRelDir)
except ftplib.error_perm:
try:
ftp.mkd(remoteRelDir)
except ftplib.error_perm:
print("U have no authority to make dir")
fileName = os.path.basename(localAbsPath)
remoteRelPath = formatPath(remoteRelDir, fileName)
handle = open(localAbsPath, "rb")
ftp.storbinary("STOR %s" % remoteRelPath, handle, 1024)
handle.close()
result = [1, "upload " + fileName + " success"]
except Exception as e:
result = [-1, "upload fail, reason:{0}".format(e)]
return result
# 判斷remote path isDir or isFile
def isDir(ftp, path):
try:
ftp.cwd(path)
ftp.cwd("..")
return True
except:
return False
# return last dir'name in the path, like os.path.basename
def lastDir(path):
path = formatPath(path)
paths = path.split("/")
if len(paths) >= 2:
return paths[-2]
else:
return ""
# 格式化路徑或拼接路徑并格式化
def formatPath(path, *paths):
"""
:param path: 路徑1
:param paths: 路徑2-n
:return:
"""
if path is None or path == "." or path == "/" or path == "//":
path = ""
if len(paths) > 0:
for pi in paths:
if pi == "" or pi == ".":
continue
path = path + "/" + pi
if path == "":
return path
while path.find("\\") >= 0:
path = path.replace("\\", "/")
while path.find("//") >= 0:
path = path.replace("//", "/")
if path.find(":/") > 0: # 含磁盤符 NOT EQ ZERO, OS.PATH.ISABS NOT WORK
if path.startswith("/"):
path = path[1:]
else:
if not path.startswith("/"):
path = "/" + path
if os.path.isdir(path): # remote path is not work
if not path.endswith("/"):
path = path + "/"
elif os.path.isfile(path): # remote path is not work
if path.endswith("/"):
path = path[:-1]
elif path.find(".") < 0: # maybe it is a dir
if not path.endswith("/"):
path = path + "/"
else: # maybe it is a file
if path.endswith("/"):
path = path[:-1]
# print("new path is " + path)
return path
3、測(cè)試 main.py
# !/usr/bin/python3
# coding: utf-8
from config import ftp_config
from clients import ftp_client
import sys
def runFTP(remotePath, localPath):
result = ftp_client.getConnect(
host=ftp_config.host,
port=ftp_config.port,
username=ftp_config.username,
password=ftp_config.password
)
if result[0] != 1:
print(result[1])
sys.exit()
else:
print("connection success")
ftp = result[2]
result = ftp_client.download(
ftp=ftp,
remotePath=remotePath,
localAbsDir=localPath
)
# result = ftp_client.upload(
# ftp=ftp,
# remoteRelDir=remotePath,
# localPath=localPath
# )
# ftp.quit()
print("全部成功" if result[0] == 1 else "部分失敗")
print(result[1])
sys.exit()
def main():
remotePath = input("輸入 服務(wù)端文件或文件夾的相對(duì)路徑：")
localPath = input("輸入 客戶端文件夾的絕對(duì)路徑：")
runFTP(remotePath, localPath)
控制臺(tái)：
輸入 服務(wù)端文件或文件夾的相對(duì)路徑：/dir1/
輸入 客戶端文件夾的絕對(duì)路徑：E:/FTP/downDir/
FTP connection...
connection success
start download dir by use FTP...
start download file by use FTP...
start download file by use FTP...
start download file by use FTP...
start download dir by use FTP...
start download file by use FTP...
start download file by use FTP...
start download file by use FTP...
全部成功
download dir1_file1.txt success
download dir1_file2.txt success
download dir1_file3.txt success
download dir2_file1.txt success
download dir2_file2.txt success
download dir2_file3.txt success
執(zhí)行前：
執(zhí)行后：
---------------------
作者：maozexijr
來源：CSDN
原文：https://blog.csdn.net/MAOZEXIJR/article/details/78967806
版權(quán)聲明：本文為博主原創(chuàng)文章，轉(zhuǎn)載請(qǐng)附上博文鏈接！
在線
客服
客服
熱線
7*24小時(shí)客服服務(wù)熱線
關(guān)注
微信
關(guān)注官方微信
微信登錄
QQ登錄
微博登錄
售前咨詢