Python操作FTP服务器

Ftplib常用函数介绍

Python中官方包中自带有的ftplib模块定义了FTP类。
import ftplib # 加载ftp模块
ftp = ftplib.FTP(host) # 设置变量连接ftp server
ftp.login("user","password") # 使用账号密码登陆ftp， 若没有账号密码则以匿名anonymously登陆
ftp.cwd(pathname) # 设置ftp当前操作的路径
ftp.retrbinary("RETR filename.txt", open(filename,"wb").write, bufsize) # 下载ftp文件，并将文件保存在本地文件，bufsize缓存大小
ftp.cmd("xxx/xxx") # 更改远程目录
open(filename,"wb").write # 以写模式在本地打开文件
ftp.dir() # 显示目录下文件信息
ftp.nlst() # 获取目录下的文件
ftp.mkd(pathname) # 新建远程目录
ftp.pwd() # 返回当前所在位置
ftp.rmd(dirname) # 删除远程目录
ftp.delete(filename) # 删除远程文件
ftp.rename(fromname, toname) # 将fromname修改名称为toname。
ftp.storbinaly("STOR filename.txt",file_handel,bufsize) # 上传目标文件

文件下载Demo

import ftplib
import os
import socket

remote_host = 'hostname'
remote_dir = '/pub/data/dir'
remote_file = 'readme.txt'


def readmemain():
    try:
        ftp = ftplib.FTP(remote_host)  # 实例化FTP对象连接host
    except (socket.error, socket.gaierror):
        print('ERROR cannot reach "%s"' %remote_host)
        return
    print('..Connected to remote_host "%s"..' %remote_host)

    try:
        ftp.login()  # 匿名登陆
    except ftplib.error_perm:
        print('ERROR cannot login anonymously')
        ftp.quit()
        return
    print('...logged in as "anonymously"...')

    try:
        ftp.cwd(remote_dir)  # 设置ftp当前操作路径
    except ftplib.error_perm:
        print('ERROR cannot cd to "%s"' % remote_dir)
        ftp.quit()
    print('....Change to "%s" folder ....' % remote_dir)

    try:
        ftp.retrbinary('RETR %s' % remote_file, open(remote_file, 'wb').write)  # 下载readme.txt文件
    except ftplib.error_perm:
        print('ERROR cannot remote_file "%s"' % remote_file)
        os.unlink(remote_file)
    else:
        print('.....Download "%s" to cwd.....' % remote_file)
    ftp.quit()
    return

文件批量下载Demo

import ftplib
import os


class FtpDownloader(object):
    remote_dir = '/pub/data/dir/'
    def __init__(self, host='hostname'):
        self.conn = ftplib.FTP(host)
        self.conn.login()
        print('login success!')

    def cwd(self, year=''):
        self.dir = self.remote_dir + str(year)
        self.conn.cwd(self.dir)
        print('cwd remote_dir: {}'.format(self.dir))

    def dirInfo(self):
        self.conn.dir()

    def nlstListInfo(self):
        files_list = self.conn.nlst()
        return [file for file in files_list if file != '.' and file != '..']

    def downloadFile(self, filename='readme.txt'):
        f = open(filename, 'wb')
        self.conn.retrbinary('RETR %s' % filename, f.write)
        print('Download file success: %s' % filename)
        f.close()
        return True


if __name__ == '__main__':
    current_path = os.getcwd()  # 获取当前工作路径（运行Python时的路径）
    ftptest = FtpDownloader()  # 实例化FtpDownloader对象
    for y in range(2003, 2020):
        year_dir_path = current_path + f'\\{y}'  # 构建要存文件的文件夹路径
        if not os.path.exists(year_dir_path):
            os.mkdir(year_dir_path)  # 创建要存文件的文件夹
        os.chdir(year_dir_path)  # 改变工作路径
        ftptest.cwd(y)  # 设置ftp操作路径
        filelist = ftptest.nlstListInfo()  # 列出所有文件名
        for file in filelist:
            if file[0] == '5':  # 筛选要下载的文件
                ftptest.downloadFile(file)  # 下载文件到当前工作路径

留下评论