简介
本示例主要是用Python的socket，使用UDP协议实现一个FTP服务端、FTP客户端，用来实现文件的传输。在公司内网下，可以不适用U盘的情况下，纯粹使用网络，来实现文件服务器的搭建，进而实现文件的网络传输。同时用来理解Python的socket使用。
服务端运行起来后，会把服务器上面的指定目录作为根目录提供给客户端使用，即客户端可以访问、下载服务端设置的根目录里面的文件内容
客户端和服务端之间支持的一些简单命令如下：
- “ll”或者“ls” 查看当前目录下的所有文件或者目录
- “pwd” 查看当前所在的目录（根目录是服务端设置的“D:\var”目录）
- “get 文件名” 下载指定的文件到客户端配置的目录（客户端指定的根目录，在运行时配置）
- “get 目录” 下载指定的目录到客户端配置的目录
- “get all” 把当前所在的目录的所有文件、目录下载到客户端配置的目录
- “cd” 把客户端的目录切换到根目录
- “cd 目录” 把客户端的目录切换到指定的目录
- “cd ..” 把客户端的目录切换到上一级目录
客户端和服务端之间的通信，是把dict格式使用pickle.dumps()和pickle.loads()转成对应的bytes类型进行传输的。dict格式参考代码。
代码
file_handler.py
该文件就是把对文件的一些操作进行提取出来，供UDP服务端使用
import os
import logging
import traceback
LOG_FORMAT = "%(asctime)s - %(levelname)s [%(filename)s-%(funcName)s] Line: %(lineno)s] - %(message)s"
logging.basicConfig(level=logging.DEBUG, format=LOG_FORMAT)
_logger = logging.getLogger()
class FileHandler:
def __init__(self, logger=None):
"""
由于设置成可以支持自定义的logger，因此就没有设置成工具类形式，使用的时候，还是要实例化该类的
:param logger:
"""
self.__logger = logger if logger else _logger
def list_dir(self, dir_path: str, return_absolute_path=False) -> list:
"""
获取指定目录下面的文件或者目录的列表
返回列表，里面中的每一个元素都是元组，元组的第一个值是文件或者目录名，第二个值是"d"(代表目录)或者"f"(代表文件)
:param dir_path:
:param return_absolute_path: True: 返回值是返回文件或者目录的绝对路径； False: 只返回文件名或者目录名
:return: [('download', 'd'), ('mylog.txt', 'f')]
"""
ret_list = []
try:
if os.path.exists(dir_path):
ret = os.listdir(dir_path)
for item in ret:
if os.path.isdir(os.path.join(dir_path, item)):
if return_absolute_path:
ret_list.append((os.path.join(dir_path, item), 0, "d"))
else:
ret_list.append((item, 0, "d"))
if os.path.isfile(os.path.join(dir_path, item)):
size = os.path.getsize(os.path.join(dir_path, item))
if return_absolute_path:
ret_list.append((os.path.join(dir_path, item), size, "f"))
else:
ret_list.append((item, size, "f"))
except Exception:
self.__logger.error("Can not list dir: [%s]" % dir_path + traceback.format_exc())
finally:
return ret_list
def seek_file(self, file_path: str, start_index: int, length=1024) -> tuple:
"""
通过二进制格式读取指定一个文件指定范围的内容
:param file_path:
:param start_index:
:param length: 读取的字节数
:return:
"""
# 下一次访问的时候的起始start_index值。 -1代表已经访问到文件结尾了，不用再访问该文件了。
content_bytes = b''
next_index = -1
if not os.path.exists(file_path):
message = "File[%s] not exists !!!" % file_path
self.__logger.error(message)
raise Exception(message)
file_size = os.path.getsize(file_path) # 文件大小
if start_index >= file_size:
return content_bytes, next_index
try:
# print("### file_size: ", file_size)
with open(file_path, "rb") as fh:
fh.seek(start_index) # 游标跳到指定位置
content_bytes = fh.read(length) # 读取文件内容
# print("content_bytes: ", content_bytes)
# print("type(content_bytes): ", type(content_bytes))
if start_index + length < file_size:
next_index = start_index + length
except Exception:
self.__logger.error("Seek file exception !!! " + traceback.format_exc())
finally:
return content_bytes, next_index
if __name__ == '__main__':
file = r"D:\var\download\system.log"
file_target = r"D:\var\download\system.txt"
file = r"D:\软件安装包\NetAssist.exe"
file_target = r"D:\软件安装包\NetAssist_copy.exe"
file_obj = FileHandler()
# ret = file_obj.seek_file(file, start_index=17, length=30)
# print("ret: ", ret)
# file_obj.copy_file(file, file_target, 1024 * 1000)
服务端代码
""" 使用socket的udp协议实现的一个ftp服务端。 服务器和客户端之间传递数据格式： 1、服务端和客户端统一使用Python的字典格式（也就是本例中自定义的"通信协议"），格式形如： { "type": "cmd", # 支持的值有： "cmd"、"download" "body": "ll", # 在cmd模式下，常用的命令有: ll、ls、cd 指定目录、pwd "pwd": ["folder1", "folder2"], "status": 1, "uuid": "b93e21e659f711ee9285a46bb6f59f55" # uuid.uuid1().hex，用来保证客户端和服务端 } 2、客户端使用pickle模块的pickle.dumps(dict类型数据)把要发送的数据转成bytes类型 3、客户端使用socket的udp传输转成的bytes数据给服务端 4、服务端接收到从客户端发送过来的bytes类型的数据，再使用pickle.loads(bytes类型数据)把数据转成原始的dict类型。 使用socket的udp，既能接收数据，又能发送数据，因此服务端和客户端都是相对的。 """ import os import pickle import sys import socket import logging import time import traceback from file_handler import FileHandler LOG_FORMAT = "%(asctime)s - %(levelname)s [%(filename)s-%(funcName)s] Line: %(lineno)s] - %(message)s" logging.basicConfig(level=logging.DEBUG, format=LOG_FORMAT) _logger = logging.getLogger() _HOST = "127.0.0.1" _PORT = 8090 # 服务器端口号 _ROOT_DIR = r"D:\var" # 服务端给客户端展示的可以进行下载的文件路径 __all__ = ["FTPServerUDP"] class FTPServerUDP: def __init__(self, host="", port=None, root_dir="", logger=None): self.__host = host if host else _HOST self.__port = port if port else _PORT self.__root_dir = root_dir if root_dir else _ROOT_DIR self.__logger = logger if logger else _logger self.__file_handler = FileHandler() self.__socket_obj = self.__get_socket_obj() self.__message_type_unsupported = "Unsupported message type" self.__message_type_server_inner_error = "Server internal error" self.__message_type_path_not_exists = "Target path not exists" self.__message_type_ok = "ok" def __get_socket_obj(self) -> socket.socket: socket_obj = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # UDP socket_obj.bind((self.__host, self.__port)) # 服务端必须绑定IP和端口 return socket_obj def __message_handler(self, message_tuple: tuple): # 给客户端返回的数据类型，最终还需要使用pickle.dumps()把字典类型转成bytes类型进行发送 response_message_dict = { "type": "cmd", "body": self.__message_type_unsupported, "pwd": [], # 客户端的相对于服务端self.__root_dir的相对路径 "status": 0 # 0:代表后端处理结果异常; 1:代表后端处理结果正常 } recv_message_bytes = message_tuple[0] client_tuple = message_tuple[1] # 客户端元组信息，形如 ('127.0.0.1', 59699) # 确保接收到的消息是 pickle.dumps() 转成的bytes类型，并把数据转成dict类型 # recv_message_dict 的格式形如上面的response_message_dict格式 recv_message_dict = self.__check_recv_message_type(recv_message_bytes) if not recv_message_dict: response_message_dict["body"] = self.__message_type_unsupported response_message_dict["status"] = 0 self.__send_bytes_message(response_message_dict, client_tuple) return # 把客户端的进入的目录，赋值给返回值的目录。 response_message_dict["pwd"] = recv_message_dict.get("pwd", []) # 把客户端传递进来的uuid，赋值给返回值的目录 response_message_dict["uuid"] = recv_message_dict.get("uuid", "") # 接收到的消息符合规范，就需要根据"type"类型进行分类。 try: print("recv_message_dict: ", recv_message_dict) if recv_message_dict.get("type", "") == "cmd": self.__cmd_handler(recv_message_dict, response_message_dict, client_tuple) elif recv_message_dict.get("type", "") == "download": self.__download_handler(recv_message_dict, response_message_dict, client_tuple) pass except Exception: self.__logger.error("Server message handler exception !!!" + traceback.format_exc()) response_message_dict["status"] = 0 response_message_dict["body"] = self.__message_type_server_inner_error self.__send_bytes_message(response_message_dict, client_tuple) def __check_recv_message_type(self, recv_message_bytes: bytes) -> dict: """ 确保接收到的消息是 pickle.dumps() 转成的bytes类型，并且通过pickle.loads()把bytes类型的数据转成dict后， dict类型数据中要有： "type"、"body"、"pwd"、"uuid" 等字段 :param recv_message_bytes: :return: """ ret_dict = {} try: message_dict = pickle.loads(recv_message_bytes) except Exception: return ret_dict if not isinstance(message_dict, dict): return ret_dict # 接收到的dict类型的消息中，必须要有如下的字段 if not {"type", "body", "pwd", "uuid"}.issubset(set(message_dict.keys())): return ret_dict return message_dict def __send_bytes_message(self, message_dict: dict, client_tuple: tuple): """ 使用pickle.dumps()把字典格式的消息(message_dict)，转成bytes发送给客户端(client_tuple) :param message_dict: :param client_tuple: :return: """ try: message_bytes = pickle.dumps(message_dict) self.__socket_obj.sendto(message_bytes, client_tuple) except Exception: self.__logger.error("Send message to client exception !!! " + traceback.format_exc()) message_dict["status"] = 0 message_dict["body"] = self.__message_type_server_inner_error message_bytes = pickle.dumps(message_dict) self.__socket_obj.sendto(message_bytes, client_tuple) def __cmd_handler(self, recv_message_dict: dict, response_message_dict: dict, client_tuple: tuple): """ 处理消息体的type参数值是"cmd"，这种命令行的消息 :param recv_message_dict: :param response_message_dict: :param client_tuple: :return: """ cmd_str = recv_message_dict.get("body", "").lower().strip() if not cmd_str or (not isinstance(cmd_str, str)): # 返回值形如: {'type': 'cmd', 'body': 'Unsupported message type', 'pwd': ['download']} response_message_dict["body"] = self.__message_type_unsupported response_message_dict["status"] = 0 self.__send_bytes_message(response_message_dict, client_tuple) return if cmd_str in ("ls", "ll"): # 查看当前目录的文件目录列表 # 返回值形如: {'type': 'cmd', 'body': [('folder1', 'd'), ('file1.txt', 'f')]} customer_dir = self.__root_dir if recv_message_dict.get("pwd", []): customer_dir = os.path.join(self.__root_dir, *recv_message_dict.get("pwd", [])) response_message_dict["body"] = self.__file_handler.list_dir(customer_dir) response_message_dict["status"] = 1 # 处理结果符合预期，状态设置成1 self.__send_bytes_message(response_message_dict, client_tuple) return if cmd_str.startswith("cd"): # 切换到下一个目录 if cmd_str.strip() == "cd": response_message_dict["pwd"] = [] response_message_dict["body"] = self.__message_type_ok response_message_dict["status"] = 1 # 处理结果符合预期，状态设置成1 self.__send_bytes_message(response_message_dict, client_tuple) return target_dir = cmd_str.split(" ")[-1] if target_dir == "..": response_message_dict["pwd"] = response_message_dict["pwd"][0:-1] response_message_dict["body"] = self.__message_type_ok response_message_dict["status"] = 1 # 处理结果符合预期，状态设置成1 self.__send_bytes_message(response_message_dict, client_tuple) return if target_dir == "/": response_message_dict["body"] = self.__message_type_ok response_message_dict["status"] = 1 # 处理结果符合预期，状态设置成1 self.__send_bytes_message(response_message_dict, client_tuple) return if not os.path.exists(os.path.join(self.__root_dir, *recv_message_dict.get("pwd", []), target_dir)): # 返回值形如: {'type': 'cmd', 'body': 'Target path not exists', 'pwd': ['folder1']} response_message_dict["body"] = self.__message_type_path_not_exists response_message_dict["status"] = 0 self.__send_bytes_message(response_message_dict, client_tuple) return elif not (os.path.join(self.__root_dir, *recv_message_dict.get("pwd", []), target_dir).startswith(self.__root_dir)): # 客户进入的目录，必须是以 self.__root_dir 开头，不能进入到其他目录 # 返回值形如: {'type': 'cmd', 'body': 'Target path not exists', 'pwd': ['folder1']} # response_message_dict["pwd"] = [] response_message_dict["body"] = self.__message_type_path_not_exists response_message_dict["status"] = 0 self.__send_bytes_message(response_message_dict, client_tuple) return elif os.path.isfile(os.path.join(self.__root_dir, *recv_message_dict.get("pwd", []), target_dir)): # 文件 # 返回值形如: {'type': 'cmd', 'body': 'Target path not exists', 'pwd': ['folder1']} response_message_dict["body"] = self.__message_type_path_not_exists response_message_dict["status"] = 0 self.__send_bytes_message(response_message_dict, client_tuple) return else: # 返回值形如： {'type': 'cmd', 'body': 'ok', 'pwd': ['folder1', 'folder2']} response_message_dict["pwd"].append(target_dir) response_message_dict["body"] = self.__message_type_ok response_message_dict["status"] = 1 # 处理结果符合预期，状态设置成1 self.__send_bytes_message(response_message_dict, client_tuple) return if cmd_str == "pwd": # 返回值形如 {'type': 'cmd', 'body': ['folder1'], 'pwd': ['folder1']} response_message_dict["body"] = response_message_dict.get("pwd", []) response_message_dict["status"] = 1 # 处理结果符合预期，状态设置成1 self.__send_bytes_message(response_message_dict, client_tuple) return self.__send_bytes_message(response_message_dict, client_tuple) return def __download_handler(self, recv_message_dict: dict, response_message_dict: dict, client_tuple: tuple): """ 处理消息体的type参数值是"download"，这种下载单个文件的命令行的消息 :param recv_message_dict: :param response_message_dict: :param client_tuple: :return: """ response_message_dict["type"] = "download" try: file_name = recv_message_dict["body"]["file_name"] # 要下载的文件的名字 pwd_list = recv_message_dict["pwd"] # 要下载的文件所在的路径 start_index = recv_message_dict["body"]["start_index"] # 要下载的文件的起始位置索引 length = recv_message_dict["body"]["length"] # 单次下载的字节数 if length > 1024 * 10: # 单次传输的数据太大的话，会报错。通过测试，单次发送10KB数据比较合适。 length = 1024 * 10 file_name_absolute = os.path.join(self.__root_dir, *pwd_list, file_name) # 要下载的文件的绝对路径 content_bytes, next_index = self.__file_handler.seek_file(file_path=file_name_absolute, start_index=start_index, length=length) response_message_dict["body"] = { "file_name": file_name, "pwd": pwd_list, "content_bytes": content_bytes, "start_index": start_index, "next_index": next_index } response_message_dict["status"] = 1 self.__send_bytes_message(response_message_dict, client_tuple) except Exception: response_message_dict["body"] = self.__message_type_server_inner_error response_message_dict["status"] = 0 self.__logger.error("Download file exception !!!" + traceback.format_exc()) self.__send_bytes_message(response_message_dict, client_tuple) def run(self): """ 调用该方法，启动服务端 :return: """ self.__logger.info("Server is running at [%s@%s], wating for client ..." % (self.__host, self.__port)) while True: try: # message_tuple 是一个元组类型: (b'消息体', ('127.0.0.1', 61040))，第一个值是bytes类型的消息，第二个值是客户端信息 message_tuple = self.__socket_obj.recvfrom(1024 * 1024) # 从客户端接收到的消息 self.__message_handler(message_tuple) except Exception: self.__logger.error("FTP Server Error: " + traceback.format_exc()) time.sleep(1) if __name__ == '__main__': ftp_server_obj = FTPServerUDP() ftp_server_obj.run()
客户端代码
""" 使用socket的udp协议实现的一个ftp客户端。 服务器和客户端之间传递数据格式： 1、服务端和客户端统一使用Python的字典格式（也就是本例中自定义的"通信协议"），格式形如： { "type": "cmd", # 支持的值有： "cmd"、"download" "body": "ll", # 在cmd模式下，常用的命令有: ll、ls、cd 指定目录、pwd "pwd": ["folder1", "folder2"], "status": 1, "uuid": "b93e21e659f711ee9285a46bb6f59f55" # uuid.uuid1().hex，用来保证客户端和服务端 } 2、客户端使用pickle模块的pickle.dumps(dict类型数据)把要发送的数据转成bytes类型 3、客户端使用socket的udp传输转成的bytes数据给服务端 4、服务端接收到从客户端发送过来的bytes类型的数据，再使用pickle.loads(bytes类型数据)把数据转成原始的dict类型。 使用socket的udp，既能接收数据，又能发送数据，因此服务端和客户端都是相对的。 """ import socket import sys import time import pickle import os import logging import traceback import uuid LOG_FORMAT = "%(asctime)s - %(levelname)s [%(filename)s-%(funcName)s] Line: %(lineno)s] - %(message)s" logging.basicConfig(level=logging.DEBUG, format=LOG_FORMAT) _logger = logging.getLogger() class FTPClientUDP: def __init__(self, host="127.0.0.1", port=8090, download_path="D:\my_download", logger=None): self.__host = host # 服务器的IP地址 self.__port = port # 服务器的端口号 self.__download_root_path = download_path # 要下载的文件路径 self.__logger = logger if logger else _logger if not os.path.exists(download_path): os.makedirs(download_path, True) self.__socket_obj = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # UDP连接对象 self.__is_downloading = False # 用来标识是否在下载文件，当在下载文件的时候，就不会再接收输入参数 self.__pwd_list = [] # 用来存储用户所在的当前目录 self.__server_tuple = (self.__host, self.__port) # 服务端的地址和端口号组成的元组 def __cmd_handler(self, cmd_dict: dict): """ :param cmd_dict: 形如：{'type': 'cmd', 'body': 'll', 'pwd': [], 'status': 1, 'uuid': 'f5c2f4d263f111ee877aa46bb6f59f55'} :return: """ # print("cmd_dict: ", cmd_dict) cmd_bytes = pickle.dumps(cmd_dict) if cmd_dict["body"] in ["ll", "ls"]: self.__cmd_handler_ls(cmd_dict, cmd_bytes) elif cmd_dict["body"].startswith("cd "): self.__cmd_handler_cd(cmd_dict, cmd_bytes) elif cmd_dict["body"] == "pwd": self.__cmd_handler_pwd(cmd_dict, cmd_bytes) elif cmd_dict["body"].startswith("get "): self.__cmd_handler_get(cmd_dict, cmd_bytes) else: print("Command [%s] not supported !!!" % cmd_dict["body"]) print() def __cmd_handler_ls(self, cmd_dict: dict, cmd_bytes: bytes): """ 发送类似于"ls -la" 命令给服务端，并把返回值打印到终端 :param cmd_dict: :param cmd_bytes: :return: """ recv_message_dict = self.__recv_normal_cmd_handler(cmd_dict, cmd_bytes) self.__print_normal_info(recv_message_dict) def __cmd_handler_cd(self, cmd_dict: dict, cmd_bytes: bytes): """ 发送类似于"cd 指定目录"命令给服务端，并把返回值打印到终端 :param cmd_dict: :param cmd_bytes: :return: """ recv_message_dict = self.__recv_normal_cmd_handler(cmd_dict, cmd_bytes) self.__print_normal_info(recv_message_dict) if recv_message_dict.get("status", 0): self.__pwd_list = recv_message_dict["pwd"] # 保存客户端进入的目录 def __cmd_handler_pwd(self, cmd_dict: dict, cmd_bytes: bytes): """ 发送 "pwd" 命令给服务端，并把返回值打印到终端 :param cmd_dict: :param cmd_bytes: :return: """ recv_message_dict = self.__recv_normal_cmd_handler(cmd_dict, cmd_bytes) self.__print_pwd_info(recv_message_dict) if recv_message_dict.get("status", 0): self.__pwd_list = recv_message_dict["pwd"] # 保存客户端进入的目录 def __cmd_handler_get(self, cmd_dict: dict, cmd_bytes: bytes): # 获取当前目录下面的所有的文件，包括文件夹 if cmd_dict["body"].lower().strip() == "get all": self.__download_handler_all_file() return # 获取当前目录下面"get "命令后面的一个文件 command = cmd_dict["body"] file_name = cmd_dict["body"].split(" ")[-1] # 字符串类型的要下载的文件名 if not file_name: return # 首先获取当前目录"ll"的返回值 cmd_dict_ll = {'type': 'cmd', 'body': 'll', 'pwd': cmd_dict["pwd"], 'status': 1, 'uuid': uuid.uuid1().hex } ret_dict = self.__recv_normal_cmd_handler(cmd_dict_ll) can_download = False file_size = 0 # 文件大小 pwd = [] # 文件所处路径 file_type = "f" # f:文件 d:目录 for item_tuple in ret_dict["body"]: # item_tuple形如 ('aa.txt', 34, 'f') if item_tuple[0] == file_name: can_download = True file_size = item_tuple[1] pwd = ret_dict["pwd"] file_type = item_tuple[-1] if can_download and file_type == "f": self.__download_handler_one_file(file_name, file_size, pwd) print() if can_download and file_type == "d": raw_pwd = self.__pwd_list # 存储用户最原始的pwd目录。 self.__download_handler_one_folder(file_name, pwd) self.__pwd_list = raw_pwd # 把用户最原始的pwd目录再重新赋给self.__pwd_list变量中。 else: print() def __download_handler_one_file(self, file_name: str, file_size=0, pwd=[]): # 发送下载单个文件的命令格式，下载的时候，要来回更新 download_cmd_dict['body']['start_index']的值 download_cmd_dict = { 'type': 'download', 'body': { "file_name": file_name, "pwd": pwd, "start_index": 0, # "length": 1024 * 1024 # 使用UDP传输的数据，单次数据太大的话会报错 }, 'pwd': pwd, 'status': 1, 'uuid': uuid.uuid1().hex } try: file_absolute = os.path.join(self.__download_root_path, *pwd, file_name) if not os.path.exists(os.path.dirname(file_absolute)): os.mkdir(os.path.dirname(file_absolute)) with open(file_absolute, "wb") as fh: while True: download_cmd_bytes = pickle.dumps(download_cmd_dict) ret_dict = self.__recv_normal_cmd_handler(download_cmd_dict, download_cmd_bytes) if not ret_dict["status"]: # 下载途中失败了 self.__logger.error("Download exception: " + str(ret_dict["body"])) break fh.write(ret_dict["body"]["content_bytes"]) next_index = ret_dict["body"]["next_index"] if ret_dict["body"]["next_index"] == -1: # 说明已经把文件下载完了 break download_cmd_dict["body"]["start_index"] = ret_dict["body"]["next_index"] download_cmd_dict["uuid"] = uuid.uuid1().hex self.__print_download_info(file_absolute, file_size, next_index) self.__print_download_successfully_info(file_absolute) except Exception: self.__logger.error("Download file exception !!!" + traceback.format_exc()) def __download_handler_one_folder(self, file_name: str, pwd=[]): """ 下载一个文件夹。 :param file_name: 即文件夹名字。进入到该方法，默认文件名是存在的 :param pwd: :return: """ if not os.path.exists(os.path.join(self.__download_root_path, *pwd, file_name)): os.makedirs(os.path.join(self.__download_root_path, *pwd, file_name)) cmd_dict_ll = {'type': 'cmd', 'body': 'll', 'pwd': pwd, # 该参数会在后面的代码中进行替换更新 'status': 1, 'uuid': uuid.uuid1().hex } cmd_dict_cd = {'type': 'cmd', 'body': 'cd ' + file_name, # 进入到的目录 'pwd': pwd, 'status': 1, 'uuid': uuid.uuid1().hex } # 先进入指定目录 recv_message_dict_cd = self.__recv_normal_cmd_handler(cmd_dict_cd, None) pwd = recv_message_dict_cd["pwd"] # 保存进入的目录 # 获取指定目录的文件列表 cmd_dict_ll["pwd"] = pwd recv_message_dict_ll = self.__recv_normal_cmd_handler(cmd_dict_ll, None) # print("&&&&&&&v recv_message_dict_ll: ", recv_message_dict_ll) if recv_message_dict_ll["status"] and recv_message_dict_ll["body"]: for item_tuple in recv_message_dict_ll["body"]: # print("@@@@ item_tuple: ", item_tuple) if item_tuple[-1] == "f": self.__download_handler_one_file(file_name=item_tuple[0], file_size=item_tuple[1], pwd=pwd ) elif item_tuple[-1] == "d": self.__download_handler_one_folder(file_name=item_tuple[0], pwd=pwd) def __download_handler_all_file(self): """ 下载当前目录下面的所有文件及文件夹 :return: """ if not os.path.exists(os.path.join(self.__download_root_path, *self.__pwd_list)): os.makedirs(os.path.join(self.__download_root_path, *self.__pwd_list)) cmd_dict_ll = {'type': 'cmd', 'body': 'll', 'pwd': self.__pwd_list, # 该参数会在后面的代码中进行替换更新 'status': 1, 'uuid': uuid.uuid1().hex } recv_message_dict_ll = self.__recv_normal_cmd_handler(cmd_dict_ll, None) if recv_message_dict_ll["status"] and recv_message_dict_ll["body"]: for item_tuple in recv_message_dict_ll["body"]: if item_tuple[-1] == "f": self.__download_handler_one_file(file_name=item_tuple[0], file_size=item_tuple[1], pwd=recv_message_dict_ll["pwd"] ) elif item_tuple[-1] == "d": self.__download_handler_one_folder(file_name=item_tuple[0], pwd=recv_message_dict_ll["pwd"] ) def __recv_normal_cmd_handler(self, cmd_dict: dict, cmd_bytes: bytes=None, try_times=3) -> dict: """ 持续发送一条命令到客户端，直到正常接收到数据后结束 :param cmd_dict: 形如 {'type': 'cmd', 'body': 'get aa.txt', 'pwd': [], 'status': 1, 'uuid': '464899225dcb11ee9a91a46bb6f59f55'} :param cmd_bytes: :param try_times: 重试的次数，即 :return: 形如 {'type': 'cmd', 'body': [('log', 0, 'd'), ('mylog.txt', 8, 'f')], 'pwd': [], 'status': 1, 'uuid': '464899235dcb11eebf99a46bb6f59f55'} 或者 {'type': 'cmd', 'body': {'file_name': 'logger.log', 'pwd': ['log'], 'content_bytes': b'2023-', 'start_index': 0, 'next_index': 5}, 'pwd': [], 'status': 1, 'uuid': '464899235dcb11eebf99a46bb6f59f55'} """ ret_dict = {} cmd_bytes = cmd_bytes if cmd_bytes else pickle.dumps(cmd_dict) try: for x in range(try_times): # print("cmd_dict_ll: ", cmd_dict) self.__socket_obj.sendto(cmd_bytes, self.__server_tuple) # 服务器端的地址 recv_message_tuple = self.__socket_obj.recvfrom(1024 * 1024) recv_message_bytes = recv_message_tuple[0] recv_message_dict = pickle.loads(recv_message_bytes) if cmd_dict.get("uuid", "False") == recv_message_dict.get("uuid", "True"): ret_dict = recv_message_dict break time.sleep(0.1) except Exception: self.__logger.error("Recv normal cmd info exception !!!" + traceback.format_exc()) finally: return ret_dict def __print_normal_info(self, recv_message_dict: dict): message_body = recv_message_dict.get("body", None) if isinstance(message_body, list): for item in message_body: print("%-20s %-20s %-20s" % (item[0], item[1], item[2])) print() return True if isinstance(message_body, str): print("%-20s" % message_body) print() return True return True def __print_pwd_info(self, recv_message_dict: dict): pwd_list = recv_message_dict.get("pwd", []) if not pwd_list: print("%-20s" % "/") else: pwd_str = '/' + '/'.join(pwd_list) print("%-20s" % pwd_str) print() return True def __print_download_info(self, file_name, file_size, next_index): sys.stdout.write("\r[%s] -->: %s" % (file_name, ("%.3f" % (next_index / file_size* 100) + "%"))) def __print_download_successfully_info(self, file_name): """ 最终打印100% :param file_name: :return: """ sys.stdout.write("\r[%s] -->: %s" % (file_name, "100%")) print() def run(self): while True: try: # 没有下载任务的话，则接收输入新的命令 if not self.__is_downloading: cmd = input("Input: ") cmd_dict = { "type": "cmd", "body": cmd, "pwd": self.__pwd_list, "status": 1, "uuid": uuid.uuid1().hex } self.__cmd_handler(cmd_dict) print("================================================") else: time.sleep(1) except Exception: self.__logger.error("Client exception: " + traceback.format_exc()) time.sleep(1) if __name__ == '__main__': ftp_client_obj = FTPClientUDP() ftp_client_obj.run()
使用
先运行服务端代码，再运行客户端代码。然后再在客户端输入响应的命令即可
