前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >python ftp常用操作

python ftp常用操作

作者头像
py3study
发布2020-01-10 11:20:03
6190
发布2020-01-10 11:20:03
举报
文章被收录于专栏:python3

需求

快速进行ftp上传 ,下载,查询文件

原来直接在shell下操作:需要【连接,输用户名,输密码,单文件操作,存在超时限制】

太过于繁琐,容易操作失败

改进

一句命令,搞定多文件上传,下载,查询,列表等操作

后期可以加入更强大的功能

源代码

代码语言:javascript
复制
#!/usr/bin/python# -*- coding:utf-8 -*-#ftp.py#    wklken@yeah.net#this script is used to do some operations more convenient via ftp
  #1.[p]upload many files in the same time,show md5s
  #2.[g]download many files in the same time,show md5s
  #3.[l]list all the files on ftp site
  #4.[f]search a file on ftp site,return True or Flase
  #5.[h]show help info
  
  #add upload and download operations  20111210 version0.1
  #add md5sum after ops 20120308 version0.2
  import sys,os,ftplib,socket
  CONST_HOST = "ip"
  CONST_USERNAME = "username"
  CONST_PWD = "pwd"
  CONST_BUFFER_SIZE = 8192
  COLOR_NONE = "\033[m"
  COLOR_GREEN = "\033[01;32m"
  COLOR_RED = "\033[01;31m"
  COLOR_YELLOW = "\033[01;33m"
  
  def connect():
    try:
        ftp = ftplib.FTP(CONST_HOST)
        ftp.login(CONST_USERNAME,CONST_PWD)
        return ftp
    except socket.error,socket.gaierror:
        print("FTP is unavailable,please check the host,username and password!")
        sys.exit(0)def disconnect(ftp):
    ftp.quit()
    
    def upload(ftp, filepath):
        f = open(filepath, "rb")
        file_name = os.path.split(filepath)[-1]
        try:
            ftp.storbinary('STOR %s'%file_name, f, CONST_BUFFER_SIZE)
        except ftplib.error_perm:
            return False
            return True
            
    def download(ftp, filename):
        f = open(filename,"wb").write
        try:
            ftp.retrbinary("RETR %s"%filename, f, CONST_BUFFER_SIZE)
        except ftplib.error_perm:
            return False
        return True
        
    def list(ftp):
        ftp.dir()def find(ftp,filename):
        ftp_f_list = ftp.nlst()
        if filename in ftp_f_list:
            return True
        else:
            return Falsedef help():
    print("help info:")
    print("[./ftp.py l]\t show the file list of the ftp site ")
    print("[./ftp.py f filenamA filenameB]\t check if the file is in the ftp site")
    print("[./ftp.py p filenameA filenameB]\t upload file into ftp site")
    print("[./ftp.py g filenameA filenameB]\t get file from ftp site")
    print("[./ftp.py h]\t show help info")
    print("other params are invalid")def main():
    args = sys.argv[1:]
    if len(args) == 0:
        print("Params needed!")
        sys.exit(0)

    ftp = connect()

    success_list = []
    failed_list = []

    if args[0] == "p":
        f_list = args[1:]

        for up_file in f_list:
        if not os.path.exists(up_file):
            print(("UPLOAD: %s "+COLOR_RED+"FAILED"+COLOR_NONE+"  :file not exist")%up_file)
            continue
        elif not os.path.isfile(up_file):
            print(("UPLOAD: %s "+COLOR_RED+"FAILED"+COLOR_NONE+"  :%s is not a file")%(up_file,up_file))
            continue

        if upload(ftp, up_file):
            success_list.append(up_file)
        else:
            failed_list.append(up_file)
        if len(success_list) > 0 :
        print((COLOR_GREEN + "UPLOAD SUCCESS: %s" + COLOR_NONE)%(" ".join(success_list)))
        print("md5sum:")
        for f in success_list:
            print( os.popen("md5sum " + f).read()[:-1])
        if len(failed_list) > 0:
        print((COLOR_RED + "UPLOAD FAILED: %s" + COLOR_NONE)%(" ".join(failed_list)))

    elif args[0] == "g":
        f_list = args[1:]

        for down_file in f_list:
        if not find(ftp,down_file):
            print(("DOWNLOAD: %s "+COLOR_RED+"FAILED"+COLOR_NONE+"  :%s is not in the ftp site")%(down_file,down_file))
            continue

        if download(ftp, down_file):
            success_list.append(down_file)
        else:
            failed_list.append(down_file)
        if len(success_list) > 0 :
        print((COLOR_GREEN + "DOWNLOAD SUCCESS: %s" + COLOR_NONE)%(" ".join(success_list)))
        print("md5sum:")
        for f in success_list:
            print( os.popen("md5sum " + f).read()[:-1])
        if len(failed_list) > 0:
        print((COLOR_RED + "DOWNLOAD FAILED: %s" + COLOR_NONE)%(" ".join(failed_list)))

    elif args[0] == "l":
        list(ftp)
    elif args[0] == "f":
        f_list = args[1:]
        for f_file in f_list:
        if find(ftp,f_file):
            print(("SEARCH: %s "+COLOR_GREEN+"EXIST"+COLOR_NONE)%f_file)
        else:
            print(("SEARCH: %s "+COLOR_RED+"NOT EXIST"+COLOR_NONE)%f_file)
            if len(f_file) > 3:
            print("Similar File List:")
            s = ftp.nlst()
            print  ", ".join([k for k in s if f_file in k])

    elif args[0] == "h":
        help()
    else:
        print("args are invalid!")
        help()

    disconnect(ftp)if __name__ == "__main__":
  main()

常用函数

用手册查看,以下只是简略,因为没用用到,[待整理]:

代码语言:javascript
复制
login(user='',passwd='', acct='')     登录到FTP 服务器,所有的参数都是可选的
pwd()                       当前工作目录
cwd(path)                   把当前工作目录设置为path
dir([path[,...[,cb]])       显示path 目录里的内容,可选的参数cb 是一个回调函数,会被传给retrlines()方法
nlst([path[,...])           与dir()类似,但返回一个文件名的列表,而不是显示这些文件名
retrlines(cmd [, cb])       给定FTP 命令(如“RETR filename”),用于下载文本文件。可选的回调函数cb 用于处理文件的每一行
retrbinary(cmd, cb[,bs=8192[, ra]])     与retrlines()类似,只是这个指令处理二进制文件。回调函数cb 用于处理每一块(块大小默认为8K)下载的数据。
storlines(cmd, f)   给定FTP 命令(如“STOR filename”),以上传文本文件。要给定一个文件对象f
storbinary(cmd, f[,bs=8192])    与storlines()类似,只是这个指令处理二进制文件。要给定一个文件对象f,上传块大小bs 默认为8Kbs=8192])
rename(old, new)    把远程文件old 改名为new
delete(path)     删除位于path 的远程文件
mkd(directory)  创建远程目录
本文参与 腾讯云自媒体同步曝光计划,分享自作者个人站点/博客。
原始发表:2019/07/26 ,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 需求
  • 改进
  • 源代码
  • 常用函数
相关产品与服务
云服务器
云服务器(Cloud Virtual Machine,CVM)提供安全可靠的弹性计算服务。 您可以实时扩展或缩减计算资源,适应变化的业务需求,并只需按实际使用的资源计费。使用 CVM 可以极大降低您的软硬件采购成本,简化 IT 运维工作。
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档