作业要求
0、实现用户登陆
1、实现上传和下载
3、每个用户都有自己的家目录，且只可以访问自己的家目录
4、对用户进行磁盘配额，每个用户的空间不同，超过配额不允许下载和上传
5、允许用户在指定的家目录随意切换目录
6、允许用户在自己的家目录切换目录
7、允许上传和下载文件，并判断文件的一致性
目前还未终稿，还在持续优化中
客户端核心代码
import socket
import os
import hashlib
import subprocess
import json
import time HOST = "127.0.0.1"
PORT = 8000
ip_bind = (HOST, PORT)
ftp_client = socket.socket()
ftp_client.connect(ip_bind)
def test_md5(path):
md5 = hashlib.md5()
f = open(path,"rb")
while True:
d = f.read(10)
if not d:
break
else:
f_md5 = md5.update(d)
file_md5 = md5.hexdigest()
return file_md5 while True:
client_login = ftp_client.recv(8192)
print(str(client_login, encoding="utf-8"))
cline_user = input("用户名:")
ftp_client.sendall(bytes(cline_user, encoding="utf-8"))
client_login = ftp_client.recv(8192)
print(str(client_login, encoding="utf-8"))
cline_password = input("密码:")
ftp_client.sendall(bytes(cline_password, encoding="utf-8"))
ftp_server = ftp_client.recv(8192)
if str(ftp_server, encoding="utf-8") == "用户名或者密码错误":
print(str(ftp_server, encoding="utf-8"))
continue
else:
print(str(ftp_server, encoding="utf-8"))
break while True:
client_func = input("你可以查看和切换目录,你可以上传和下载文件:")
ftp_client.sendall(bytes(client_func, encoding="utf-8")) if client_func.startswith("get"):
server_reply = ftp_client.recv(8192)
get_info = str(server_reply,encoding="utf-8")
print(get_info)
file_md5 = get_info.split("|")[1].split(":")[1]
file_size = int(get_info.split("|")[2].split(":")[1])
file_name = get_info.split("|")[0].split(":")[1]
sever_ack = ftp_client.recv(8192)
if str(sever_ack,encoding="utf-8") == "ok":
print(str(sever_ack,encoding="utf-8"))
get_file_path = "F:\\ftp\\ftpclient\\%s" %(file_name)
with open(get_file_path,"wb") as file:
temp_size = int(os.path.getsize(get_file_path))
while temp_size < file_size:
# print("临时大小:",temp_size,"实际大小:",file_size)
data = ftp_client.recv(8192)
file.write(data)
temp_size = int(os.path.getsize(get_file_path))
print(temp_size,file_size)
print("下载文件[%s]完成" %(file_name))
elif client_func.startswith("put"):
pass
else:
server_reply = ftp_client.recv(8192)
print(str(server_reply, encoding="utf-8"))
服务端核心代码
import socket
import hashlib
import subprocess
import json
import os
import random
import hashlib
root = "F:\\ftp\\ftpserver\\root"
bob = "F:\\ftp\\ftpserver\\bob"
user_info = {}
test_md5 = hashlib.md5() def test_md5(path):
md5 = hashlib.md5()
f = open(path,"rb")
while True:
d = f.read(10)
if not d:
break
else:
f_md5 = md5.update(d)
file_md5 = md5.hexdigest()
return file_md5 with open("E:\\python\\网络编辑_socket\\ftp作业\\user_data", "r", encoding="utf-8") as f:
for line in f:
temp = line.split("|")
temp_dict = {temp[0]: {"密码": temp[1], "配额": temp[2]}}
user_info.update(temp_dict)
temp = []
temp_dict = {} HOST = "127.0.0.1"
PORT = 8000
ip_bind = (HOST, PORT)
FTP_server = socket.socket()
FTP_server.bind(ip_bind)
FTP_server.listen(1)
conn, add = FTP_server.accept()
while True:
conn.sendall(bytes("FTP_server:请输入用户名", encoding="utf-8"))
cline_user = str(conn.recv(8192), encoding="utf-8")
conn.sendall(bytes("FTP_server:请输入密码", encoding="utf-8"))
cline_password = str(conn.recv(8192), encoding="utf-8")
if cline_user in user_info.keys():
if cline_password == user_info[cline_user]["密码"]:
welcome = "欢迎用户\033[31;1m[%s]\033[0m登陆FTP服务器" % (cline_user)
send_info = welcome.center(100, "-")
conn.sendall(bytes(send_info, encoding="utf-8"))
break
else:
conn.sendall(bytes("用户名或者密码错误", encoding="utf-8"))
continue
else:
conn.sendall(bytes("用户名或者密码错误", encoding="utf-8"))
continue client_path = "F:\\ftp\\ftpserver\\" + cline_user
os.chdir(client_path)
client_pwd = client_path
while True:
client_func = conn.recv(8192)
if str(client_func, encoding="utf-8") == "dir":
os.chdir(client_pwd)
a = os.listdir(client_pwd)
b = []
for i in a:
path = client_pwd + "\\" + i
if os.path.isdir(path):
c = "dir|" + i
b.append(c)
elif os.path.isfile(path):
c = "file|" + i
b.append(c)
else:
pass
ret = json.dumps(b)
conn.sendall(bytes(ret, encoding="utf-8"))
elif str(client_func, encoding="utf-8").startswith("cd"):
if str(client_func, encoding="utf-8").strip() == "cd":
os.chdir(client_path)
s = "切换成功"
conn.sendall(bytes(s, encoding="utf-8"))
client_pwd = client_path
# print(client_pwd) else:
cd_func = str(client_func, encoding="utf-8").split(" ")
cd_path = client_pwd + "\\" + cd_func[1]
if os.path.exists(cd_path):
if os.path.isdir(cd_path):
os.chdir(cd_path)
s = "切换成功"
conn.sendall(bytes(s, encoding="utf-8"))
client_pwd = cd_path
else:
s = "请输入正确的目录信息"
conn.sendall(bytes(s, encoding="utf-8"))
else:
s = "该目录不存在"
conn.sendall(bytes(s, encoding="utf-8"))
elif str(client_func, encoding="utf-8").startswith("get"): a = str(client_func, encoding="utf-8").split(" ")
file_path = client_pwd + "\\" + a[1]
file_size = str(os.path.getsize(file_path)) if os.path.exists(client_pwd + "\\" + a[1]):
if os.path.isdir(client_pwd + a[1]):
s = "请输入正确的文件路径信息"
conn.sendall(bytes(s, encoding="utf-8"))
elif os.path.isfile(client_pwd + "\\" + a[1]):
print(client_pwd + "\\" + a[1],type(client_pwd + "\\" + a[1]))
md5 = str(test_md5(client_pwd + "\\" + a[1]))
print(md5)
with open(client_pwd + "\\" + a[1], "rb") as file:
s = "文件名称:%(file_name)s|文件md5:%(file_md5)s|文件大小:%(size)s" %{"file_name":a[1],"file_md5":md5,"size":file_size}
print(s)
conn.sendall(bytes(s,encoding="utf-8"))
conn.sendall(bytes("ok",encoding="utf-8"))
for line in file:
# f2.write(line)
data = conn.sendall(line)
conn.sendall(bytes("",encoding="utf-8"))
else:
s = "该路径不存在"
conn.sendall(bytes(s, encoding="utf-8"))
elif str(client_func, encoding="utf-8").startswith("put"):
pass
else:
s = "错误的输入，请重新输入"
conn.sendall(bytes(s, encoding="utf-8"))