Python3的FTP文件上传下载

来源:maozexijr 发布时间:2018-11-03 15:21:43 阅读量:1265

注:代码中的变量命名规则采用的是“驼峰式”命名法,这可能带来代码检查提示,在PyCharm中可以Alt + Enter忽略此类检查




1、配置文件 ftp_config.py



# FTP服务器的IP、端口、账户、密码

host = "172.16.53.113"

port = 21  # not str

username = "admin"

password = "123456"

 

# remote和local是相对客户端的

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: 服务端文件或文件夹的绝对或相对路径

    :param localAbsDir: 客户端文件夹的绝对路径,如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)  # 服务端文件或文件夹的绝对路径

 

        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: 服务端文件的相对路径,含文件后缀,如/srcDir/

    :param localAbsDir: 客户端文件夹的绝对路径,如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: 服务端文件的相对路径,含文件后缀,如/srcDir/file.txt

    :param localAbsDir: 客户端文件夹的绝对路径,如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: 服务端文件夹相对路径,可以为None、"",此时文件上传到homeDir

    :param localPath: 客户端文件或文件夹路径,当路径以localDir开始,文件保存到homeDir的相对路径下

    :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):  # 绝对路径

                localRelDir = localPath.replace(ftp_config.localDir, "/")

                localRelDir = formatPath(localRelDir)

            else:  # 相对(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: 服务端文件夹相对路径,可以为None、"",此时文件上传到homeDir

    :param localAbsDir: 客户端文件夹路径,当路径以localDir开始,文件保存到homeDir的相对路径下

    :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: 服务端文件夹相对路径,可以为None、"",此时文件上传到homeDir

    :param localAbsPath: 客户端文件路径,当路径以localDir开始,文件保存到homeDir的相对路径下

    :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、测试 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("输入 服务端文件或文件夹的相对路径:")

    localPath = input("输入 客户端文件夹的绝对路径:")

    runFTP(remotePath, localPath)



控制台:


输入 服务端文件或文件夹的相对路径:/dir1/

输入 客户端文件夹的绝对路径: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



执行前:





执行后:



--------------------- 

作者:maozexijr 

来源:CSDN 

原文:https://blog.csdn.net/MAOZEXIJR/article/details/78967806 

版权声明:本文为博主原创文章,转载请附上博文链接!


标签: 服务器搭建
分享:
评论:
你还没有登录,请先