python实现FTP程序

川长思鸟来 2021-12-15 02:19 190阅读 0赞

# python实现FTP程序 #

程序源码

上传功能

查看文件

cd功能

创建目录

# #

# 程序源码 #

##  目录结构 ##

![1477786-20190413205001190-1101653024.png][]

## 服务端 ##

## 主程序 ##

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import optparse
    import socketserver
    import server
    import configs
    
    
    
    class ArgvHandler():
        def __init__(self):
            #命令行解析
            self.op=optparse.OptionParser()
            self.op.add_option('-S', '--server', dest='server')
            self.op.add_option('-P', '--port', dest='port')
            options,args=self.op.parse_args() #options的值为添加对应的值，args为输入的无关值
    
            #进行命令分发
            self.verify_args(options,args)
        #命令分发
        def verify_args(self,options,args):
            print(options,args)
            cmd=args[0]
            print(cmd)
            if hasattr(self,str(cmd)):
                func=getattr(self,str(cmd))
                func()
        #启动服务端
        def start(self):
    
            #启动服务端
            s=socketserver.ThreadingTCPServer((configs.IP,int(configs.PORT)),server.ServerHandler)
            #永久启动服务端
            s.serve_forever()
    
        #帮助信息
        def help(self):
            pass
    if __name__ =='__main__':
        ArgvHandler()

main.py

## 核心代码 ##

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import socketserver
    import json
    import configparser
    import configs
    import os
    
    '''
    
    
    
    '''
    
    STATUS_CODE={
        250:"Invalid cmd format,e.g:{'action':'get','filename':'test.py','size':344}",
        251:"Invalid cmd",
        252:"Invalid auth data",
        253:"Wrong username or password",
        254:"Passed authentication",
        255:"Filename doesn't provided",
        256:"File doesn't exist on server",
        257:"ready to send file",
        258:"md5 verification",
        800:"the file exist,but not enough,is continue",
        801:"the file exist",
        802:"ready to receive datas",
        900:"md5 valdate success",
    
    }
    
    
    
    #服务端处理代码
    class ServerHandler(socketserver.BaseRequestHandler):
        def handle(self):
            #接收客户端信息，循环接收处理
            while 1:
                try:
                    data=self.request.recv(1024).strip().decode('utf8')
                except Exception as e:
                    continue
                #将数据转成一个json字典的形式
                if data :
                    data=json.loads(data)
                    print(data)
    
                    #传过来的数据格式
                    # {'action':'auth',
                    # 'username':'admin',
                    # 'password':'123456'}
    
                    #外层判断命令是否为空
                    #内层循环判断是否有这个命令
                    if data.get('action'):
                        if hasattr(self,data.get('action')):
                            func=getattr(self,data.get('action'))
                            func(**data)
                        else:
                            print()
                    else:
                         print('')
    
    
    
    #认证
        def auth(self,**data):
            username=data['username']
            passwd=data['password']
            print(username,passwd)
            print(4)
            username=self.authenticate(username,passwd)
    
            if username:
                self.send_reponse(254)
            else:
                self.send_reponse(253)
    
    
        def authenticate(self,user,passwd):
            print(5)
            cfg=configparser.ConfigParser()
            # cfg.read(configs.auth_path)
            cfg.read('auth.cfg')
            if user in cfg.sections():
    
    
                if cfg[user]['passwd']==passwd:
                    #
                    BASE_DIR = os.path.dirname(os.path.abspath(__file__))
                    self.user=user
                    #定义用户的本地目录，在home目录下
                    self.mainPath=os.path.join(BASE_DIR,'home',self.user)
                    print("登陆成功")
                    return user
    
        def send_reponse(self,state_code):
            print(6)
            response={
         "status_code":state_code,"status_mes":STATUS_CODE[state_code]}
            self.request.sendall(json.dumps(response).encode("utf-8"))
    
    
    #上传
        def put(self,**data):
            print("data:",data)
            file_name=data.get('file_name')
            file_size=data.get('file_size')
            target_path=data.get('target_path')
            #这个路径对应的上传的完整路径:self.mainPath,target_path+file_name路径
            #self.mainPath：home+username
            #self.target:文件类型
            #file_name:文件名
    
            #完整的文件名
            has_receive=0
            abs_path=os.path.join(self.mainPath,target_path,file_name)
            print(abs_path)
    
    
            #上传文件：
            # 1.已经存在这个文件：如果文件完整，则返回文件存在，如果文件不完整，则返回选择续传还是不续传
            # 2.没有这个文件：直接上传文件
            #对应文件打开模式：不存在：wb  续传：ab 存在完整：不打开
    
            # 文件存在
            if os.path.exists(abs_path):
    
                file_has_size=os.stat(abs_path).st_size
                print(file_size,file_has_size)
                # 文件存在,且完整
                if file_size==file_has_size:
                    print(1)
                    self.request.sendall('801'.encode("utf8"))
                    return
                #断点续传
                elif file_has_size<file_size:
                    self.request.sendall('800'.encode("utf8"))
                    choice=self.request.recv(1024).decode('utf8')
                    #续传
                    print(2)
                    if choice=="Y":
                        self.request.sendall(str(file_has_size).encode("utf8"))
                        has_receive+=file_has_size
                        #这里注意光标的位置，以追加方式打开，光标在文件里的位置在最后
                        f=open(abs_path,"ab")
                    #不续传，重传
                    else:
                        f = open(abs_path, "wb")
    
            else:
                #文件不存在
                print(3)
                self.request.sendall('802'.encode("utf8"))
                f=open(abs_path,"wb")
            #has_receive:已经接收到数据大小
            while has_receive<file_size:
                try:
                    data=self.request.recv(1024)
                    f.write(data)
                    has_receive+=len(data)
                except Exception as e:
                    break
            f.close()
    #cd与ls都要依靠用户的self.mainPath去做处理
    #而self.mainPath一直都是用户当前所在的目录
    #查看文件
        def ls(self,**data):
            file_list=os.listdir(self.mainPath)
            if len(file_list)==0:
                file_str="<empty dir>"
            else:
                file_str="\n".join(file_list)
            self.request.sendall(file_str.encode("utf8"))
    #cd命令
        def cd(self,**data):
            #cd image
            dirname=data.get("dirname")
            if dirname=="..":
                # 返回目录名
                self.mainPath=os.path.dirname(self.mainPath)
            else:
    
                self.mainPath=os.path.join(self.mainPath,dirname)
            self.request.sendall(self.mainPath.encode('utf8'))
    #创建目录
        def mkdir(self,**data):
            dirname=data.get("dirname")
            path=os.path.join(self.mainPath,dirname)
            if not os.path.exists(path):
                if "/" in dirname:
                    os.makedirs(path)
                else:
                    os.mkdir(path)
                self.request.sendall("dirname create success".encode("utf8"))
            else:
                self.request.sendall("dirname exist".encode("utf8"))

server.py

## 日志文件 ##

logger.py

##  配置文件 ##

config.py

import os
    IP="127.0.0.1"
    PORT="4444"
    
    # BASE_DIR=os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    # auth_path=os.path.join(BASE_DIR,"conf","auth.cfg")

##  数据文件 ##

auth.cfg

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

[DEFAULT]
    
    [yuan]
    passwd=123098
    
    [root]
    passwd=wenli

保存用户名与密码

# 客户端 #

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import optparse
    import socket
    import json
    import re
    import os
    import sys
    
    
    
    STATUS_CODE={
        250:"Invalid cmd format,e.g:{'action':'get','filename':'test.py','size':344}",
        251:"Invalid cmd",
        252:"Invalid auth data",
        253:"Wrong username or password",
        254:"Passed authentication",
        255:"Filename doesn't provided",
        256:"File doesn't exist on server",
        257:"ready to send file",
        258:"md5 verification",
        800:"the file exist,but not enough,is continue",
        801:"the file exist",
        802:"ready to receive datas",
        900:"md5 valdate success",
    
    }
    class ClientHandler():
        def __init__(self):
            self.op=optparse.OptionParser()
            self.op.add_option('-S','--server',dest='server')
            self.op.add_option('-P', '--port', dest='port')
            self.op.add_option('-u', '--username', dest='username')
            self.op.add_option('-p', '--password', dest='password')
            self.options, self.args = self.op.parse_args()
            #上传文件的绝对路径
            #os.path.abspath(__file__):获得当前文件的绝对路径
            # os.path.dirname(os.path.abspath(__file__))：获得上一级目录
            self.mainPath=os.path.dirname(os.path.abspath(__file__))
            self.last = 0
    
            #下面两个函数是实例化类的时候就执行
            #对参数进行处理
            self.verify_args(self.options)
            #连接服务端，进行用户操作
            self.make_connection()
    
        # 对参数进行处理
        def verify_args(self,options):
            server=options.server
            port=options.port
            username=options.username
            password=options.password
            #对参数进行简单判断
            if re.match(r"^(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$",str(server)):
                return True
            else:
                exit("IP vaild")
            if int(port) and int(port) < 65535:
                return True
            else:
                exit("the port is in 0-65535")
    
            if username is None:
                return True
            else:
                exit("用户名为空")
    
            if password is None:
                return True
            else:
                exit("密码为空")
    
    
    
        #连接服务端
        def make_connection(self):
            #创建socket
            self.sock=socket.socket()
            self.sock.connect((self.options.server,int(self.options.port)))
    
        # 用户操作
        def interactive(self):
            #先进行身份认证
            if self.authenticate():
                print('''请选择你要进行的操作:
                1.上传文件示例：put filename file 
                2.查看文件：ls
                3.进入目录：cd dirname
                4.创建目录：mkdir dirname
                5.退出：quit
                ''')
                while 1:
                    cmd_info=input("[%s]："%self.user).strip()#put test.py file
    
                    cmd_list=cmd_info.split()#默认以空格分隔
                    if cmd_list[0]=="quit":break
                    if hasattr(self,cmd_list[0]):
                        func=getattr(self,cmd_list[0])
                        func(*cmd_list)
    
    
    
    
    
        #传输文件的路径必须和客户端路径同级
        def put(self,*cmd_list):
            action,local_path,target_path=cmd_list
            print( action,local_path,target_path)
            # os.path.join(self.mainPath, local_path):self.mainPath+local_path得到本地完全的图片的路径
            #这里首先对格式进行判断
    
    
            #获取文件完整路径
            local_path=os.path.join(self.mainPath,local_path)
            #获取文件的名字
            file_name=os.path.basename(local_path)
            #获取文件大小
            file_size=os.stat(local_path).st_size
            data = {
                'action': 'put',
                'file_name':file_name,
                'file_size':file_size,
                'target_path':target_path#类型
            }
            has_sent=0
            self.sock.send(json.dumps(data).encode('utf-8'))
            is_exist=int(self.sock.recv(1024).decode("utf-8"))
            if is_exist==800:
                print(STATUS_CODE[is_exist])
                choice=input('请输入你的选择是否选择续传：[Y/N]')
                if choice.upper()=="Y":
                    self.sock.sendall("Y".encode("utf8"))
                    continue_position=self.sock.recv(1024).decode("utf8")
                    has_sent+=int(continue_position)
                else:
                    self.sock.sendall("N".encode("utf8"))
            elif is_exist==801:
                print(STATUS_CODE[is_exist])
                return
            else:
                print(STATUS_CODE[is_exist])
    
            f=open(local_path,"rb")
            #将光标的位置调整到has_sent
            f.seek(has_sent)
            while has_sent<file_size:
                data=f.read(1024)
                self.sock.sendall(data)
                has_sent+=len(data)
                self.show_progress(has_sent,file_size)
            f.close()
    
        def show_progress(self,has,total):
           rate=float(has)/float(total)
           rate_num=int(rate*100)
           sys.stdout.write("%s%% %s\r"%(rate_num,"#"* rate_num))
    
        def ls(self,*cmd_list):
            data= {
                'action': 'ls',
            }
            self.sock.sendall((json.dumps(data).encode("utf8")))
            reponse=self.sock.recv(1024).decode("utf8")
            print(reponse)
    
        def cd(self,*cmd_list):
            data={
                'action': 'cd',
                'dirname':cmd_list[1]
            }
            self.sock.sendall((json.dumps(data).encode("utf8")))
            reponse = self.sock.recv(1024).decode("utf8")
            # self.current_dir=reponse
            print("当前目录："+str(reponse))
    
        def mkdir(self,*cmd_list):
    
            data = {
                'action': 'mkdir',
                'dirname': cmd_list[1]
            }
            self.sock.sendall((json.dumps(data).encode("utf8")))
            reponse = self.sock.recv(1024).decode("utf8")
            print(reponse)
    
    
    
        #身份认证
        def authenticate(self):
            #如果用户名或者密码有一个为空，则要求再次输入，反之进入下一步
            if self.options.username is None or self.options.password is None:
                username=input('username:')
                password=input('password')
                return self.get_auth_result(username,password)
    
            return self.get_auth_result(self.options.username,self.options.password)
    
        #发送认证信息
        def get_auth_result(self,username,password):
            # 构建数据
            dic={
                'action':'auth',
                'username':username,
                'password':password
            }
            #发送认证信息
            self.sock.send(json.dumps(dic).encode('utf-8'))
            #判断服务端回复信息，进行认证
            response=self.response()
            if response['status_code']==254:
                self.user=username
                self.current_dir=username
                print(STATUS_CODE[254])
            else:
                print(response['status_code']+response['status_mes'])
            return True
    
        #接收信息
        def response(self):
            data = self.sock.recv(1024).decode('utf-8')
            data = json.loads(data)
            return data
    
    
    
    ch=ClientHandler()
    # 用户操作，在用户操作里面会进行身份认证
    
    ch.interactive()

FTP\_Client.py

# 上传功能 #

# 查看文件 #

# cd功能 #

# 创建目录 #

转载于:https://www.cnblogs.com/-wenli/p/10702919.html

[1477786-20190413205001190-1101653024.png]: /images/20211214/bd44d6f5503d42218f34cbed1c9d6df5.png
[ContractedBlock.gif]: https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif
[ExpandedBlockStart.gif]: /images/20211214/b857300deb9a44819eb62d3a338b6056.png