python3 FTP文件的上传下载
内容导读
互联网集市收集整理的这篇技术教程文章主要介绍了python3 FTP文件的上传下载，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含11419字，纯文字阅读大概需要17分钟。
内容图文
FTP/Readme.md
### 执行顺序
- 先执行FTP_Server/ftp_server.py
- 再执行FTP_Server/ftp_client.py
### 测试账号
- 用户名 lily 密码 123456
- 注册或登录后，可执行上传文件和下载命令
- put a.mp4
- put 1.jpg
- get a.mp4
- get 1.jpg
### 流程
- 注册流程
- 登录流程
###FTP项目开发
- 功能:
- 多用户同时登陆
- 上传/下载文件
- 传输过程中现实进度条
- 不同用户使用自己的目录
- 每个用户拥有1G的磁盘空间
- 充分使用面向对象知识
- 思路
- 先完成客户端用户注册功能
- 注册时检查用户名s是否存在请求
```python
{'action': '_user_isexists', 'username': 'aaa', 'result': 0}
```
- 注册时检查用户名是否存在响应（不存在）
```python
{'action': '_user_isexists', 'username': 'aaa', 'result': 0}
```
- 注册时检查用户名是否存在响应（存在）
```python
{'action': '_user_isexists', 'username': 'aaa', 'result': 1}
```
- 注册提交数据
```python
{'action': '_register', 'result': 0, 'data': {'username': 'aaa', 'password': '123456', 'quotaSize': '50M', 'userHome': 'home/aaa'}}
```
- 注册成功返回响应
```python
{'action': '_register', 'result': 1, 'data': {'username': 'aaa', 'quotaSize': '50M', 'userHome': 'home/aaa'}}
```
- 再完成用户登录功能
- 请求
```python
{'action': '_login', 'username': 'lily', 'password': 'aaa', 'result': 0}
```
- 响应
```python
{'action': '_login', 'username': 'lily', 'password': '123456', 'result': 1, 'userHome': 'home/lily', 'quota_size': '50M'}
```
- 用户注册和登录时有自己的家目录和磁盘空间大小
- 注册时，用户名 密码 默认磁盘空间50M 注册或登录成功后默认进入家目录
- 用户信息保存在FTP_Server/db/user.json
- 格式
```python
{"username": "lily", "password": "1822219a532beacb9615e36dcaba3a6c", "quotaSize": "50M", "userHome": "home/lily"}
```
- 实现客户端上传和下载文件的功能
- 注册或登录成功后，自动进入用户的家目录
- 上传文件
- put 文件名
- 客户端判断文件是否存在
- 客户端获取文件名和文件大小
- 服务端判断用户磁盘是否够用
```python
{'action': '_file_upload', 'file_name': 'dydt.png', 'file_size': 39542, 'quota_size': '50M', 'user_home': 'home/lily'}
```
- 一切检查就绪，可以上传文件
```python
{"result": 1} # 上传完成
```
- 将FTP_Client下的文件上传到用户的家目录
- 下载文件
- get 文件名
- 客户端发送要下载的文件消息
- 服务端收到消息，拿到文件名，判断文件是否存在，存在则下载，不存在返回
- 将用户家目录里的文件下载到FTP_Client/download/目录下
- 上传文件的时候，需要判断磁盘空间是否满了
- 上传和下载文件的时候，有进度条
FTP/reg.png
FTP/login.png
FTP/FTP_Server/ftp_server.py
import socketserver
class MyServer(socketserver.BaseRequestHandler):
def handle(self): # 必须有的方法
while 1:
dic = self.recv_msg()
print(dic)
if hasattr(self, dic.get('action', "not_find")):
fn = getattr(self, dic.get('action', "not_find"))
if callable(fn):
fn(dic)
else:
print(fn)
else:
self.printC("方法没找到！", "red")
def _user_isexists(self, dic):
username = dic.get("username")
with open("db/user.json", "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
user_dic = json.loads(line)
if user_dic.get("username") == username:
dic["result"] = 1
print(dic)
self.send_msg(dic)
def _register(self, dic):
"""
用户注册
:param dic:
:return:
"""
user_dic = dic.get("data")
user_home = user_dic.get("userHome")
if not os.path.exists(user_home):
os.makedirs(user_home)
password = user_dic.get("password")
user_dic["password"] = MyServer.get_md5(password)
user_dic_json = json.dumps(user_dic)
with open("db/user.json", "a", encoding="utf-8") as f:
f.write(f"{user_dic_json}\n")
dic["result"] = 1
os.chdir(user_home)
del dic["data"]["password"]
print(dic)
self.send_msg(dic)
def _login(self, dic):
username = dic.get("username")
password = dic.get("password")
password = MyServer.get_md5(password)
with open("db/user.json", "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
user_dic = json.loads(line)
if user_dic.get("username") == username and user_dic.get("password") == password:
dic["result"] = 1
dic["userHome"] = user_dic.get("userHome")
dic["quota_size"] = user_dic.get("quotaSize")
os.chdir(dic["userHome"])
print(dic)
del dic["password"]
self.send_msg(dic)
def send_msg(self, dic):
"""
发送信息
:param dic:
:return:
"""
dic_json = json.dumps(dic)
dic_json_bs = dic_json.encode("utf-8")
self.request.send(struct.pack("i", len(dic_json_bs)))
self.request.send(dic_json_bs)
def recv_msg(self):
"""
接收消息
:return:
"""
recv_msg_head = self.request.recv(4)
recv_msg_len = struct.unpack("i", recv_msg_head)[0]
print(recv_msg_len)
recv_msg_content = self.request.recv(recv_msg_len).decode("utf-8")
return json.loads(recv_msg_content)
def _file_upload(self, dic):
"""
文件上传
:param dic:
:return:
"""
file_name = dic.get("file_name")
file_size = dic.get("file_size")
if dic.get("quota_size").endswith("M"):
quota_size = int(dic.get("quota_size")[:-1]) * 1024 * 1024
else:
quota_size = int(dic.get("quota_size")) * 1024 * 1024
used_quota_size = MyServer.getPathSize("./")
if quota_size > used_quota_size + file_size:
self.send_msg({"space": 1})
f = open(file_name, "wb")
while file_size > 0:
recv_msg = self.request.recv(1024)
f.write(recv_msg)
file_size -= len(recv_msg)
else:
self.send_msg({"result": 1}) # 上传完成
else:
self.send_msg({"space": 0}) # 空间不足
def _file_download(self, dic):
file_name = dic.get("filename")
if os.path.exists(file_name) and os.path.isfile(file_name):
file_size = os.path.getsize(file_name)
self.send_msg({"exists": 1, "file_size": file_size})
with open(file_name, "rb") as f:
for line in f:
self.request.send(line)
print("文件下载完成！")
else:
self.send_msg({"exists": 0})
@staticmethod
def getPathSize(path):
"""
统计目录的总大小
"""
total_size = 0
gen = os.walk(path) # 拿到一个生成器
for root, dirs, files in gen:
for d in dirs:
dir_path = os.path.join(root, d)
total_size += os.path.getsize(dir_path)
print(dir_path)
for f in files:
file_path = os.path.join(root, f)
total_size += os.path.getsize(file_path)
print(file_path)
return total_size
@staticmethod
def get_md5(str1):
md5_obj = hashlib.md5("我是盐，你想咋滴".encode("utf-8"))
md5_obj.update(str1.encode("utf-8"))
return md5_obj.hexdigest()
def printC(self, str1, color="black"):
'''添加颜色输出
color: red:红色 gre:绿色 yel:黄色
'''
col_type = 30
if color == "red":
col_type = 31
elif color == "gre":
col_type = 32
elif color == "yel":
col_type = 33
print("\033[0;%sm%s\033[0m" % (col_type, str1))
if __name__ == '__main__':
import json
import struct
import hashlib
import os
sk = socketserver.ThreadingTCPServer(("127.0.0.1", 8889), MyServer)
sk.serve_forever()
FTP/FTP_Client.py
class Client:
def __init__(self):
self.sk = None
self.username = None
self.user_home = None
self.quota_size = None
def run(self):
"""
启动入口
:return:
"""
dic = {"1": "login", "2": "register"}
while 1:
for k, v in dic.items():
self.printC("%-3s%-10s" % (k, v), "yel")
inp = input("请选择操作序号：").strip()
if hasattr(self, dic.get(inp, "not_find")):
fn = getattr(self, dic.get(inp, "not_find"))
if callable(fn):
self.connect_socket()
fn()
break
else:
self.printC("输入错误，请重新输入！", "red")
def connect_socket(self):
"""
创建socket连接
:return:
"""
self.sk = socket.socket()
self.sk.connect(("127.0.0.1", 8889))
def send_msg(self, dic):
"""
发送信息
:param dic:
:return:
"""
dic_json = json.dumps(dic)
dic_json_bs = dic_json.encode("utf-8")
self.sk.send(struct.pack("i", len(dic_json_bs)))
self.sk.send(dic_json_bs)
def recv_msg(self):
"""
接收消息
:return:
"""
recv_msg_head = self.sk.recv(4)
recv_msg_len = struct.unpack("i", recv_msg_head)[0]
recv_msg_content = self.sk.recv(recv_msg_len).decode("utf-8")
return json.loads(recv_msg_content)
def login(self):
"""
用户登录
:return: 登录成功进入welcome方法
"""
for i in range(2, -1, -1):
self.printC("用户登录".center(20, "*"), "yel")
username = input("Username: ").strip()
password = input("Password: ").strip()
user_dic = {"action": "_login", "username": username, "password": password, "result": 0}
self.send_msg(user_dic)
recv_dic = self.recv_msg()
if recv_dic.get("result"):
self.printC(f"恭喜用户{username}登录成功！", "gre")
self.username = username
self.user_home = recv_dic.get("userHome")
self.quota_size = recv_dic.get("quota_size")
self.welcome()
break
else:
self.printC(f"用户名或密码错误，你还有{i}次机会！", "red")
def register(self):
"""
用户注册
:return: 注册成功进入welcome方法
"""
while 1:
self.printC("新用户注册".center(20, "*"), "yel")
username = input("Username: ").strip()
if not username:
self.printC("用户名不能为空！", "red")
continue
if self._user_isexists(username): # 判断用户是否已存在
self.printC(f"用户{username}已存在！", "red")
continue
password = input("Password: ").strip()
if not password:
self.printC("密码不能为空！", "red")
continue
quota_size = input("Quota size: (Default 50M)").strip()
if not quota_size:
quota_size = "50M"
user_home = f"home/{username}"
user_dic = {"action": "_register", "result": 0,
"data": {"username": username, "password": password, "quotaSize": quota_size,
"userHome": user_home}}
self.send_msg(user_dic)
res_dic = self.recv_msg()
if res_dic.get("result"):
self.printC(f"用户{username}注册成功！", "gre")
self.username = username
self.user_home = user_home
self.quota_size = quota_size
self.welcome()
break
else:
self.printC(f"用户{username}注册失败！", "red")
def _user_isexists(self, username):
send_dic = {"action": "_user_isexists", "username": username, "result": 0}
self.send_msg(send_dic)
return self.recv_msg().get("result")
def welcome(self):
self.printC("欢迎来到FTP系统".center(20, "*"), "yel")
while 1:
now_path = self.user_home
inp = input(f"{now_path} # ").strip()
if inp.startswith("put "):
self.file_upload(inp)
elif inp.startswith("get "):
self.file_download(inp)
else:
self.printC("无效的命令！", "red")
def file_upload(self, command):
"""
上传文件
:param command:
:return:
"""
filename = command[4:].strip()
if os.path.exists(filename) and os.path.isfile(filename): # 判断文件是否存在
file_size = os.path.getsize(filename) # 文件大小
if int(self.quota_size[:-1]) * 1024 * 1024 > file_size:
send_dic = {"action": "_file_upload", "file_name": filename, "file_size": file_size,
"quota_size": self.quota_size, "user_home": self.user_home}
self.send_msg(send_dic)
space_dic = self.recv_msg()
if space_dic.get("space") == 0:
self.printC("上传的文件太大了，空间不足！", "red")
return
now_size = 0
with open(filename, "rb") as f:
for line in f:
self.sk.send(line)
now_size += len(line)
self.process_bar(file_size, now_size) # 进度条
recv_dic = self.recv_msg()
if recv_dic.get("result"):
self.printC(f"文件{filename}上传完成！", "gre")
else:
self.printC(f"文件{filename}太大，你的空间不足！", "red")
else:
self.printC(f"文件{filename}不存在", "red")
def file_download(self, command):
print(command)
filename = command[4:].strip()
send_dic = {"action": "_file_download", "filename": filename, "result": 0}
self.send_msg(send_dic)
recv_dic = self.recv_msg()
if recv_dic.get("exists"):
file_size = recv_dic.get("file_size")
f = open(f"download/{filename}", "wb")
now_size = 0
while file_size > now_size:
content = self.sk.recv(1024)
f.write(content)
now_size += len(content)
self.process_bar(file_size, now_size)
else:
self.printC(f"文件{filename}下载完成！", "gre")
else:
self.printC(f"文件{filename}不存在！", "red")
def process_bar(self, total, now):
"""
进度条
:param total: 总大小
:param now: 当前大小
:return:
"""
process = int(now * 100 / total)
sys.stdout.write(f"\r[{process}%] " + process * ">")
sys.stdout.flush()
def printC(self, str1, color="black"):
'''添加颜色输出
color: red:红色 gre:绿色 yel:黄色
'''
col_type = 30
if color == "red":
col_type = 31
elif color == "gre":
col_type = 32
elif color == "yel":
col_type = 33
print("\033[0;%sm%s\033[0m" % (col_type, str1))
if __name__ == '__main__':
import socket
import json
import struct
import os
import sys
Client().run()
FTP/FTP_Server/db/user.json
{"username": "lily", "password": "1822219a532beacb9615e36dcaba3a6c", "quotaSize": "50M", "userHome": "home/lily"}
{"username": "lucy", "password": "1822219a532beacb9615e36dcaba3a6c", "quotaSize": "100M", "userHome": "home/lucy"}
{"username": "user1", "password": "1822219a532beacb9615e36dcaba3a6c", "quotaSize": "50M", "userHome": "home/user1"}
内容总结
以上是互联网集市为您收集整理的python3 FTP文件的上传下载全部内容，希望文章能够帮你解决python3 FTP文件的上传下载所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。
内容备注
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。
内容手机端
扫描二维码推送至手机访问。
来源：【匿名】