python-文件服务器测试

0x01 思路

使用tcp协议传输文件

json解析二进制的时候,有问题, 不建议使用,推荐 pickle

0x02 代码

服务器端

# 测试文件服务器
# 实现文件的上传和下载
import socket
import os
import json
import pickle
sk = socket.socket()
sk.bind(("127.0.0.1",10051))
sk.listen()

#等待连接
conn,addr = sk.accept()
while True:
    if getattr(conn,'_closed'):
        conn, addr = sk.accept()
    msg_get = conn.recv(109600)
    #msg_get_str = msg_get.decode('utf-8')
    #msg_info = json.loads(msg_get_str)
    #msg_choice = msg_info["choice"]
    msg_info = pickle.loads(msg_get)
    msg_choice = msg_info["choice"]
    # 1 列文件
    # 2 下载文件
    # 3 上传文件
    spec_file_path = os.path.dirname(__file__) + "/" + "files"
    file_lists = os.listdir(spec_file_path)
    if msg_choice == '1':
        #列目录
        file_list_str = pickle.dumps(file_lists)
        conn.send(file_list_str)
    elif msg_choice == '2':
        # 下载文件
        # 需要输入下载的文件名
        file_name = msg_info["file_name"]
        if file_name in file_lists:
            # 下载操作
            file_name_path = spec_file_path+"/"+file_name
            with open(file_name_path,"rb") as f:
                filebytes = f.read()
                #conn.send(filebytes.encode("utf-8"))
                conn.send(filebytes)
        else:
            conn.send("没有可以下载的文件".encode("utf-8"))
    elif msg_choice == '3':
        # 上传文件,禁止覆盖
        file_name = msg_info["file_name"]
        file_content = msg_info['file_content']
        #file_content = str.encode(file_content_str)
        if file_name in file_lists:
            conn.send("已存在该文件,禁止覆盖".encode("utf-8"))
        else:
            file_name_path = spec_file_path + "/" + file_name
            with open(file_name_path,"wb") as f:
                # 获取字节流
                f.write(file_content)

    continue_str = input("是否继续 Y or N?")
    if continue_str=='N':
        break

sk.close()

客户端:

# 上传文件客户端
import socket
import os
import json
import pickle


sk = socket.socket()
sk.connect(("127.0.0.1",10051))
user_choice={"1":"查看服务器的文件","2":"下载文件","3":"上传文件"}

for k,v in user_choice.items():
    print(k,v)
spec_file_path = os.path.dirname(__file__) + "/" + "upload_files"
file_lists = os.listdir(spec_file_path)

while 1:
    choice_str = input("请输入你的选择:")
    if choice_str=='1':
        # 列目录
        dict_str = {"choice":"1"}
        send_str = pickle.dumps(dict_str);
        sk.send(send_str)
        msg_str = sk.recv(1024)
        #pickle.loads(msg_str)
        print(pickle.loads(msg_str))
    elif choice_str=='2':
        # 下载文件
        file_name = input("请输入文件名:")
        dict_str = {"choice": "2","file_name":file_name}
        send_str = pickle.dumps(dict_str);
        sk.send(send_str)
        # 接收文件
        file_content = sk.recv(109600)
        file_name_path = spec_file_path + "/" + file_name
        with open(file_name_path, "wb") as f:
            # 获取字节流
            f.write(file_content)

    elif choice_str=='3':
        # 上传文件
        file_name = input("请输入文件名:")
        file_name_path = spec_file_path + "/" + file_name
        if file_name in file_lists:
            with open(file_name_path, "rb") as f:
                # 获取字节流
                file_content = f.read()
                dict_str = {"choice": "3", "file_name": file_name,"file_content":file_content}
                #send_str = json.dumps(dict_str);
                send_str = pickle.dumps(dict_str)
                # 上传文件
                sk.send(send_str)
        else:
            print("选择的文件不存在")

    else:
        print("对不起,选择失败!")


sk.close()