python之路——作业:高级FTP(仅供参考)
Posted
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了python之路——作业:高级FTP(仅供参考)相关的知识,希望对你有一定的参考价值。
一、作业需求
1. 用户加密认证
2. 多用户同时登陆
3. 每个用户有自己的家目录且只能访问自己的家目录
4. 对用户进行磁盘配额、不同用户配额可不同
5. 用户可以登陆server后,可切换目录
6. 查看当前目录下文件
7. 上传下载文件,保证文件一致性
8. 传输过程中现实进度条
9.支持断点续传
二、实现功能
1、多用户同时登录注册(已有用户:japhi、alex;密码都是123)
2、上传/下载文件(已有示例文件)
3、查看不同用户自己家得目录下文件,且只能访问自己的家目录
4、对用户进行磁盘配额,不同用户配额不同(使用random函数,随机给用户一个内存大小(10m-20m))
5、用户登录后,可对用户目录下文件目录进行操作,包含:ls(查看当前操作目录下文件)、cd(切换当前操作目录)、rm(删除文件)、mkdir(创建目录)
6、上传下载文件,保证文件一致性,且在传输过程中实现进度条
7、支持断点续传(*******暂未实现*******)
三、目录说明
FTP/
|-- FTPClient/ #客户端文件夹
| |-- 示例文件夹/ #客户端上传/下载示例文件夹
| |-- Client_start.py #客户端启动程序
|
|-- FTPServer/ #服务端文件夹
| |-- bin/
| | |-- __init__.py
| | |-- Server_start.py #程序启动的主入口
| |
| |-- conf/
| | |-- setting.py #配置文件
| |
| |-- db/ #用户数据
| | |-- alex #用户名alex的数据文件夹
| | |-- japhi #用户名japhi的数据文件夹
| |
| |-- home/
| | |-- alex/ #用户alex用户家目录
| | |-- japhi/ #用户japhi用户家目录
| |
|-- |-- log/
| |-- log_sys.log #日志文件(未启用)
|
|-- |-- src/
| | |-- __init__.py
| | |-- common.py #公共功能
| | |-- Server_start.py #程序启动的主入口
| | |-- user.py #用户类及方法
|-- db/ #用户数据
| | |-- alex #用户名alex的数据文件夹
| | |-- japhi #用户名japhi的数据文件夹
|-- FTP.png #流程图
|-- README.txt
四、流程图
五、代码说明
1、FTPClient/Client_start.py
from __future__ import division import socket,os,sys,time,hashlib,math updir = os.path.join(os.path.dirname(os.path.abspath(__file__)),"示例文件夹") HOST = "localhost" PORT = 9998 def upload(client,user_info,name): ‘‘‘ 客户端上传文件的函数 :param client:scoket客户端标志 :param user_info:客户端登陆用户的信息 :param name:客户端登陆用户的名字 :return:none ‘‘‘ print("\\033[1;37m当前可选上传\\033[0m".center(40,"*")) dic = {} for root, dirs, files in os.walk(updir): for i,j in enumerate(files): k = i+1 dic[k] = j print("\\033[1;37m%s:%s\\033[0m"%(k,j)) choice = input("请输入要上传的文件序号:>>>").strip() if choice.isdigit() and 0 < int(choice) <= len(dic): command = "upload+"+user_info+"+"+dic[int(choice)] client.sendall(bytes(command,encoding="utf-8")) res = client.recv(1024) if str(res,encoding="utf-8") == "True": dir = os.path.join(updir,dic[int(choice)]) f = open(dir,"rb") md5 = hashlib.md5() length = os.stat(dir).st_size client.send(str(length).encode()) sign = client.recv(1024).decode() if sign == "ok": data = f.read() md5.update(data) client.sendall(data) f.close() client.send(md5.hexdigest().encode()) res_sign = client.recv(1024) if res_sign == b‘True‘: print("\\033[1;37m文件上传成功\\033[0m") elif res_sign == b‘False‘: print("\\033[1;37m文件上传失败\\033[0m") exit() elif sign == "no": print("\\033[1;37m磁盘空间不足\\033[0m") exit() else: print("\\033[1;37m输入有误\\033[0m") def download(client,user_info,name): ‘‘‘ 客户端下载文件的函数 :param client: scoket客户端标志 :param user_info: 客户端登陆的用户信息 :param name:客户端登陆的用户名字 :return: none ‘‘‘ dic = {} command = "download+"+user_info client.sendall(bytes(command, encoding="utf-8")) data = client.recv(4069) res = eval(str(data, encoding="utf-8")) if len(res) == 0: print("\\033[1;31m当前目录下暂无文件\\033[0m".center(40, "-")) else: for i,j in enumerate(res): k = i + 1 dic[k] = j print("\\033[1;37m%s:%s\\033[0m" % (k, j)) choice = input("请选择要下载的文件序号:>>>") cm = dic[int(choice)] client.sendall(bytes(cm, encoding="utf-8")) print("\\033[1;37m准备开始下载文件\\033[0m") dir = os.path.join(updir, dic[int(choice)]) res = str(client.recv(1024).decode()).split("+") res_length = res[0] or_md5 = res[1] # print(or_md5) length = 0 f = open(dir, "wb") m = hashlib.md5() while length < int(res_length): if int(res_length) - length > 1024: # 要收不止一次 size = 1024 else: # 最后一次了,剩多少收多少 size = int(res_length) - length # print("最后一次剩余的:", size) data = client.recv(size) length += len(data) m.update(data) f.write(data) progressbar(length, int(res_length)) else: new_md5 = m.hexdigest() # print(new_md5) f.close() if new_md5 == or_md5: print("\\033[1;37m文件下载成功\\033[0m") return True else: print("\\033[1;37m文件下载失败\\033[0m") return False def switch(client,user_info,name): ‘‘‘ 切换目录操作函数,包括“ls”,“cd”,“rm”,“mkdir” :param client: 客户端 :param user_info: 用户信息 :param name: 用户名 :return: none ‘‘‘ command = ‘‘‘ ls cd rm mkdir 目录名 ‘‘‘ # while True: print("\\033[1;33m%s\\033[0m" % command) c = input("请输入您要操作的命令:>>>").strip() if c == "ls": view_file(client, user_info, name) elif c == "cd": cm = "cd+" + user_info client.sendall(cm.encode("utf-8")) dirs = eval(client.recv(1024).decode()) if len(dirs) != 0: for j in dirs: print("\\033[1;37m目录:%s\\033[0m" % (j)) choice = input("请输入“cd”的目录名称:>>").strip() if choice in dirs: client.sendall(choice.encode("utf-8")) sign = client.recv(1024).decode() # print(len(res[0]),len(res[1])) if sign == "True": print("\\033[1;31m%s目录切换成功\\033[0m" % (choice)) else: print("\\033[1;31m%s目录切换失败\\033[0m" % (choice)) else: print("\\033[1;31m输入有误\\033[0m") client.sendall("error".encode("utf-8")) exit() else: print("\\033[1;31m无其它目录\\033[0m") elif c.split(" ")[0] == "mkdir": cm = "mkdir+" + user_info + "+" + c.split(" ")[1] # print(cm) client.sendall(cm.encode("utf-8")) res = client.recv(1024).decode() if res == "True": print("\\033[1;37m目录创建成功\\033[0m") else: print("\\033[1;37m目录创建失败\\033[0m") elif c == "rm": cm = "rm+" + user_info # print(cm) client.sendall(cm.encode("utf-8")) res_list = eval(client.recv(1024)) if len(res_list) == 0: print("\\033[1;37m无其它文件\\033[0m") client.sendall("error".encode("utf-8")) else: for i in res_list: print("\\033[1;37m文件:%s\\033[0m" % i) choice = input("请输入“rm”的文件名称:>>").strip() if choice in res_list: client.sendall(choice.encode("utf-8")) if client.recv(1024).decode() == "True": print("\\033[1;37m文件删除成功\\033[0m") else: print("\\033[1;37m文件删除失败\\033[0m") else: print("\\033[1;37m输入有误\\033[0m") client.sendall("error".encode("utf-8")) exit() else: print("\\033[1;37m输入有误\\033[0m") exit() def view_file(client,user_info,name): ‘‘‘ 客户端查看当前目录下文件的函数 :param client: scoket客户端标志 :param user_info: 客户端登陆的用户信息 :param name: 客户端登陆的用户名字 :return: none ‘‘‘ command = "view+"+user_info client.sendall(bytes(command,encoding="utf-8")) dirs = client.recv(1024) # print(dirs,"1111111111111111") if dirs.decode() == "False": dir = [] else: dir = eval(str(dirs,encoding="utf-8")) files = client.recv(1024) file = eval(str(files, encoding="utf-8")) client.sendall("true".encode("utf-8")) storage = str(client.recv(1024).decode()) # print(storage) if len(file) == 0 and len(dir) == 0: print("\\033[1;31m当前目录下暂无文件\\033[0m".center(40, "-")) else: print("\\033[1;33m当前目录包含以下文件内容\\033[0m".center(30,"*")) print("\\033[1;35m磁盘大小:%skb\\033[0m" % storage) for j in dir: print("\\033[1;35m目录:%s\\033[0m"%j) for i in file: print("\\033[1;35m文件:%s\\033[0m"%i) print("".center(33,"*")) def operate(client,user_info,name): ‘‘‘ 客户端操作主函数 :param client: scoket客户端标志 :param user_info: 客户端登陆的用户信息 :param name: 客户端登陆的用户名字 :return: none ‘‘‘ dic = {"1":upload,"2":download,"4":view_file,"3":switch} info = ‘‘‘------操作指令------ 1、上传文件 2、下载文件 3、切换目录操作 4、查看目录下文件 5、退出 ‘‘‘ while True: print("\\033[1;33m%s\\033[0m" % info) choice = input("请输入你要操作的命令:>>>").strip() if choice.isdigit() and 0 < int(choice) <= len(dic): dic.get(choice)(client,user_info,name) elif choice.isdigit() and int(choice) == 5: break else: print("\\033[1;31m输出错误\\033[0m".center(40, "-")) def com_parse(client,com): ‘‘‘ 客户端用户登陆注册命中解析函数 :param client: 客户端scoket标志 :param com: 命令 :return: 登陆成功返回True,否则False ‘‘‘ # print(com) client.sendall(bytes(com,encoding="utf-8")) re = client.recv(4096) if str(re,encoding="utf-8") == "Success": return True elif str(re, encoding="utf-8") == "Success": return False def login(client,data): ‘‘‘ 客户端用户登陆函数 :param client: 客户端scoket标志 :param data: 数据 :return: none ‘‘‘ name = input("请输入您的名字:>>>").strip() psd = input("请输入密码:>>>").strip() user_info = name+"+"+psd com = "login+"+user_info # com_parse(client, com) if com_parse(client,com): print("\\033[1;31m登陆成功\\033[0m") operate(client,user_info,name) else: print("\\033[1;31m登陆出现异常\\033[0m") def register(client,data): ‘‘‘ 客户端用户注册函数 :param client: 客户端scoket标志 :param data: 数据 :return: none ‘‘‘ name = input("请输入您的名字:>>>").strip() psd = input("请输入密码:>>>").strip() com = "register+" + name + "+" + psd if com_parse(client,com): print("\\033[1;31m注册成功\\033[0m") user_info = name + "+" + psd operate(client, user_info, name) else: print("\\033[1;31m注册出现异常\\033[0m") def quit(client,data): ‘‘‘ 程序退出函数 :param client: 客户端scoket标志 :param data: 用户数据 :return: none ‘‘‘ exit() def main_func(client,data): ‘‘‘ 客户端主菜单函数 :param client: 客户端scoket标志 :param data: 数据 :return: none ‘‘‘ dic = {"1":login,"2":register,"3":quit} info = ‘‘‘------用户登录界面------*{0}* 1、登陆 2、注册 3、退出 ‘‘‘.format(data) print("\\033[1;33m%s\\033[0m"%info) what = input("你要干嘛?>>>").strip() if what.isdigit() and 0 < int(what) <= len(dic): dic.get(what)(client,data) else: print("\\033[1;31m输出错误\\033[0m".center(40,"-")) def progressbar(cur, total): ‘‘‘ 进度条处理函数 :param cur: 当前文件内容长度 :param total: 总长度 :return: none ‘‘‘ percent = ‘{:.2%}‘.format(cur / total) sys.stdout.write(‘\\r‘) sys.stdout.write(‘[%-50s] %s‘ % (‘=‘ * int(math.floor(cur * 50 / total)), percent)) # print(‘[%-50s] %s\\r‘ % (‘=‘ * int(math.floor(cur * 50 / total)), percent),) sys.stdout.flush() time.sleep(0.01) if cur == total: sys.stdout.write(‘\\n‘) if __name__ == ‘__main__‘: client = socket.socket(socket.AF_INET,socket.SOCK_STREAM) client.connect(("localhost",PORT)) client.send("True".encode("utf-8")) # print("True 发送成功") # main_func(client,client.recv(1024)) main_func(client,"connect") client.close()
2、FTPServer/conf/settings.py
import os basedir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) user_home = "%s/FTPServer/home"%basedir user_info = "%s/db"%basedir # print(user_home) # print(user_info) HOST = "0.0.0.0" PORT = 9998
3、FTPServer/src/common.py
from __future__ import division import logging,os,pickle,sys,uuid,math,time frame = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.append(frame) # from conf import setting # # def sys_logging(content,levelname): # ‘‘‘ # 程序记录日志函数 # :param content: 日志的内容 # :param levelname: 日志的等级 # :return: none # ‘‘‘ # _filename = os.path.join(setting.log_dir,"log_sys.log") # log = logging.getLogger(_filename) # logging.basicConfig(filename=_filename,level=logging.INFO,format=‘%(asctime)s-%(levelname)s-%(message)s‘, datefmt=‘%m/%d/%Y %I:%M:%S %p‘) # if levelname == ‘debug‘: # logging.debug(content) # elif levelname == ‘info‘: # logging.info(content) # elif levelname == ‘warning‘: # logging.warning(content) # elif levelname == ‘error‘: # logging.error(content) # elif levelname == ‘critical‘: # logging.critical(content) def show(msg,msg_type): ‘‘‘ 程序不同信息打印的字体颜色 :param msg: 打印信息 :param msg_type: 打印信息的类型 :return: none ‘‘‘ if msg_type == "info": show_msg = "\\033[1;35m%s\\033[0m"%msg elif msg_type == "error": show_msg = "\\033[1;31m%s\\033[0m"%msg elif msg_type == "msg": show_msg = "\\033[1;37m%s\\033[0m"%msg else: show_msg = "\\033[1;32m%s\\033[0m"%msg print(show_msg) def progressbar(cur, total): ‘‘‘ 进度条输出函数 :param cur: 目前的长度 :param total: 总共的长度 :return: none ‘‘‘ percent = ‘{:.2%}‘.format(cur / total) sys.stdout.write(‘\\r‘) sys.stdout.write(‘[%-50s] %s‘ % (‘=‘ * int(math.floor(cur * 50 / total)), percent)) # print(‘[%-50s] %s\\r‘ % (‘=‘ * int(math.floor(cur * 50 / total)), percent),) sys.stdout.flush() time.sleep(0.01) if cur == total: sys.stdout.write(‘\\n‘)
4、FTPServer/src/Server_start.py
import socketserver,os,sys Base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.append(Base_dir) from conf import settings from common import show from user import User class FTPserver(socketserver.BaseRequestHandler): ‘‘‘ 服务端类 ‘‘‘ def handle(self): ‘‘‘ 重构handle :return: none ‘‘‘ if self.request.recv(1024) == b‘True‘: show("收到{0}的连接请求,正在通信中。。。".format(self.client_address),"info") # try: while True: self.cmd = self.request.recv(4069) # print(self.cmd) if not self.cmd: break elif self.cmd == b‘‘: break else: data = str(self.cmd.decode(encoding="utf-8")) res = data.split("+") if hasattr(self,res[0]): func = getattr(self,res[0]) func(res) else: show("wrong action","error") # except Exception as e: # print(e) # show("客户端发生错误","erroe") def login(self,res): ‘‘‘ 登陆函数 :param res: 命令结果 :return: none ‘‘‘ show("收到客户端登陆的请求,正在登陆。。。", "msg") name = res[1] psd = res[2] user = User(name, psd) sign = user.login() if sign: self.request.sendall(bytes("Success", encoding="utf-8")) else: self.request.sendall(bytes("Failure", encoding="utf-8")) def register(self,res): ‘‘‘ 注册 :param res: 命令结果 :return: none ‘‘‘ show("收到客户端注册的请求,正在注册。。。", "msg") name = res[1] psd = res[2] user = User(name, psd) if user.register(): self.request.sendall(bytes("Success", encoding="utf-8")) else: self.request.sendall(bytes("Failure", encoding="utf-8")) def view(self,res): ‘‘‘ 查看当前目录下文件函数 :param res: 命令结果 :return: none ‘‘‘ show("收到客户端查看当前目录文件的请求。。。", "msg") name = res[1] psd = res[2] user = User(name, psd) dirs,files = user.view_file() # print(dirs,files) dir = str(dirs) file = str(files) if len(dirs) == 0: self.request.sendall("False".encode("utf-8")) else: self.request.sendall(bytes(dir, encoding="utf-8")) self.request.sendall(bytes(file, encoding="utf-8")) self.request.recv(1024) dic = User.info_read(name) storage = str(dic["storage"]) # print(storage,type(storage)) self.request.sendall(bytes(storage, encoding="utf-8")) show("当前目录文件查看或创建成功", "info") def upload(self,res): ‘‘‘ 上传文件函数 :param res: 命令结果 :return: none ‘‘‘ show("收到客户端上传文件的请求。。。", "msg") name = res[1] filename = res[3] self.request.sendall(bytes("True", encoding="utf-8")) res = int(self.request.recv(1024).decode()) # print(res,"nice") if User.receive(filename, name, res,self.request): self.request.sendall(bytes("True", encoding="utf-8")) else: self.request.sendall(bytes("False", encoding="utf-8")) def download(self,res): ‘‘‘ 下载文件函数 :param res: 命令结果 :return: none ‘‘‘ show("收到客户端下载文件的请求。。。", "msg") name = res[1] psd = res[2] user = User(name, psd) dirs,files = user.view_file() file = str(files) self.request.sendall(bytes(file, encoding="utf-8")) res = self.request.recv(1024).decode() # print(str(res)) if User.download_file(res,name,self.request): show("文件下载成功", "info") else: show("文件下载失败", "error") def cd(self,res): ‘‘‘ “cd”函数 :param res: 命令结果 :return: none ‘‘‘ show("收到客户端“cd”的请求。。。", "msg") name = res[1] psd = res[2] user = User(name,psd) res = user.cd_command(self.request) dirs = str(res) self.request.sendall(str(dirs).encode("utf-8")) dir1 = self.request.recv(1024).decode() if dir1 == "error": show("客户端输入错误","error") self.request.close() else: sign = user.cd_dir(self.request,dir1,name) # print(sign) self.request.sendall(str(sign).encode("utf-8")) def mkdir(self,res): show("收到客户端“mkdir”的请求。。。", "msg") name = res[1] psd = res[2] user = User(name,psd) dir_name = res[3] # print(dir_name) sign = user.mkdir(self.request,dir_name) # print(sign) if sign: self.request.sendall("True".encode("utf-8")) else: self.request.sendall("False".encode("utf-8")) def rm(self,res): show("收到客户端“rm”的请求。。。", "msg") name = res[1] psd = res[2] user = User(name,psd) files = user.rm(self.request) # print(files) self.request.sendall(str(files).encode("utf-8")) file_name = self.request.recv(1024).decode() if file_name == "error": show("客户端不稳定","error") else: sign = user.rm_file(self.request,file_name) if sign: show("文件删除成功", "info") self.request.sendall("True".encode("utf-8")) else: self.request.sendall("False".encode("utf-8")) if __name__ == ‘__main__‘: show("等待客户端连接。。。", "info") server = socketserver.ThreadingTCPServer(("localhost",settings.PORT),FTPserver) server.serve_forever()
5、FTPServer/src/user.py
import os,sys,pickle,socket,time,random,hashlib Base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.append(Base_dir) from conf import settings from common import show,progressbar class User(object): ‘‘‘ 用户类 ‘‘‘ def __init__(self,username,psd): self.name = username self.password = psd self.home_path = settings.user_home + "/" +self.name def login(self): ‘‘‘ 用户登陆方法 :return: ‘‘‘ user_dic = User.info_read(self.name) print(user_dic,"qqqq") if user_dic.get(self.name) == self.password: show("登陆成功","info") user_dic["dir"] = self.home_path User.info_write(self.name,user_dic) return True else: show("登陆失败,用户名或密码错误","error") return False def register(self): ‘‘‘ 用户注册方法 :return: ‘‘‘ dic = {} dic[self.name] = self.password dic["storage"] = random.randint(10240,20480) dic["dir"] = self.home_path if User.info_write(self.name,dic): show("注册成功","info") os.mkdir(self.home_path) os.mkdir("%s/others" % self.home_path) with open("%s\\空白文件" % self.home_path, "w") as f: f.write("空白文件") return True else: show("注册失败","error") return False def view_file(self): ‘‘‘ 查看当前目录下文件 :return: 目录下文件名组成的列表 ‘‘‘ if not os.path.exists(self.home_path): os.mkdir(self.home_path) os.mkdir("%s/others"%self.home_path) with open("%s\\空白文件"%self.home_path,"w") as f: f.write("空白文件") user_dic = User.info_read(self.name) if user_dic["dir"] == os.path.join(os.path.join(Base_dir, "home"), self.name): dir = os.path.join(os.path.join(Base_dir, "home"), self.name) else: dir = user_dic["dir"] for root, dirs, files in os.walk(dir): return dirs,files def cd_command(self,con): for root,dirs,files in os.walk(self.home_path): return dirs def cd_dir(self,con,dir,name): next_dir = self.home_path+"/" +dir user_dic = User.info_read(name) # print(user_dic) user_dic["dir"] = next_dir User.info_write(name,user_dic) return True def mkdir(self,con,res_dir): user_dic = User.info_read(self.name) if user_dic["dir"] == os.path.join(os.path.join(Base_dir, "home"), self.name): dir = os.path.join(os.path.join(Base_dir, "home"), self.name) else: dir = user_dic["dir"] next_dir = dir+"/" +res_dir if os.path.exists(next_dir): show("该目录已存在", "error") return False else: os.mkdir(next_dir) show("目录创建成功","info") return True def rm(self,con): user_dic = User.info_read(self.name) if user_dic["dir"] == os.path.join(os.path.join(Base_dir, "home"), self.name): dir = os.path.join(os.path.join(Base_dir, "home"), self.name) else: dir = user_dic["dir"] for root,dirs,files in os.walk(dir): return files def rm_file(self,con,file_name): user_dic = User.info_read(self.name) if user_dic["dir"] == os.path.join(os.path.join(Base_dir, "home"), self.name): dir = os.path.join(os.path.join(Base_dir, "home"), self.name) else: dir = user_dic["dir"] os.remove(dir+"/" +file_name) return True @staticmethod def download_file(filename,name,con): ‘‘‘ 下载文件静态方法 :param filename: 文件名 :param name: 用户名 :param con: 标志 :return: none ‘‘‘ user_dic = User.info_read(name) if user_dic["dir"] == os.path.join(os.path.join(Base_dir, "home"), name): user_dir = os.path.join(os.path.join(Base_dir, "home"), name) else: user_dir = user_dic["dir"] dir = os.path.join(user_dir, filename) f = open(dir,"rb") m = hashlib.md5() data = f.read() m.update(data) a = str(len(data))+"+"+ m.hexdigest() # print(a) con.sendall(bytes(a, encoding="utf-8")) con.sendall(data) f.close() return True @staticmethod def receive(filename,name,res,con): ‘‘‘ 接收文件静态方法 :param filename: 文件名 :param name: 用户名 :param con: 标志 :return: none ‘‘‘ user_dic = User.info_read(name) if user_dic["dir"] == os.path.join(os.path.join(Base_dir, "home"), name): dir = os.path.join(os.path.join(os.path.join(Base_dir, "home"), name), filename) else: dir_name = user_dic["dir"] dir = os.path.join(dir_name, filename) # print(res,user_dic["storage"]) if res/1024 < user_dic["storage"]: con.sendall("ok".encode("utf-8")) length = 0 f = open(dir, "wb") md5 = hashlib.md5() while length < res: if res - length > 1024: # 要收不止一次 size = 1024 else: # 最后一次了,剩多少收多少 size = res - length # print("最后一次剩余的:", size) data = con.recv(size) length += len(data) md5.update(data) f.write(data) progressbar(length, res) else: new_md5 = md5.hexdigest() f.close() # Num = User.download_Progress(res, length, Num) or_md5 = con.recv(1024) # print(new_md5) # print(or_md5) if new_md5 == or_md5.decode(): show("文件下载成功", "info") return True else: show("文件不一致", "error") return False elif res/1024 > user_dic["storage"]: con.sendall("no".encode("utf-8")) show("磁盘空间不足", "error") return False @staticmethod def info_read(name): ‘‘‘ 读取用户数据的静态方法 :param name: 用户名 :return: 字典 ‘‘‘ user_dir = os.path.join(settings.user_info,name) if os.path.exists(user_dir): with open(user_dir,"rb") as f: dic = pickle.load(f) return dic else: print("用户数据为空") @staticmethod def info_write(name,dic): ‘‘‘ 写入用户数据的静态方法 :param name:用户名 :param dic:用户信息字典 :return:True ‘‘‘ user_dir = os.path.join(settings.user_info, name) with open(user_dir,"wb") as f: pickle.dump(dic,f) return True
六、界面显示
以上是关于python之路——作业:高级FTP(仅供参考)的主要内容,如果未能解决你的问题,请参考以下文章