Python3的FTP文件上传下载

关于 FTP 和 SFTP 服务搭建，参考链接

另关于 Python3 的 SFTP 文件上传下载，参见我的另外一篇博客，链接

注：代码中的变量命名规则采用的是“驼峰式”命名法，这可能带来代码检查提示，在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

执行前：

执行后：