python 实现简单的FTP程序

class MyFtpServer(socketserver.BaseRequestHandler):
 
 def handle(self): # 重写handle方法,处理socket请求
 print(f"连接来自{self.client_address}的客户端")
 commom_obj = Commom()
 data_analy = DataAnalysis()
 login_obj = Login()
 while 1:
 # 执行用户选项：1、登陆系统 2、注册账号。并返回一个结果
 status_id = login_obj.run_client_choice(self.request, commom_obj)
 if status_id == "01": # 登陆成功
 if not self.run_ftp_server(data_analy,commom_obj): # 执行ftpserver主功能
  break
 elif int(status_id) == -1: # client断开连接了
 break
 print(f"客户端{self.client_address}断开了连接")

 def run_ftp_server(self,data_analy,commom_obj):
 """"
 登陆成功后，接收客户端发来的命令，并进行处理
 :param data_analy:负责解析，执行客户端命令的对象
 :param commom_obj:程序执行时所需的数据对象
 :return 返回false代表客户端断开连接了
 """
 while True:
 try:
 cmd_len_pack = self.request.recv(4) 
 cmd_len = struct.unpack('i',cmd_len_pack)[0] # 获取命令长度，防止粘包 
 except Exception:
 break
 recv_data = self.request.recv(cmd_len).decode('utf-8') # 接收客户端数据
 if recv_data.upper() == "Q": # 客户端提出断开连接了
 break
 # 解析，处理客户端的命令
 data_analy.syntax_analysis(recv_data, self.request, commom_obj)
 return False
if __name__ == '__main__':
 print('运行FTP服务')
 ip_port = ('192.168.10.10',9000)
 # 创建并发服务端对象
 server = socketserver.ThreadingTCPServer(ip_port, MyFtpServer)
 # 开启服务
 server.serve_forever()

class Login(FileOperation):
 """
 登陆注册类。主要负责用户的登陆认证，和用户注册。
 """
 def run_client_choice(self,socket_obj,commom):
 """
 获取客户端的请求，1是登陆，2是注册用户
 :param socket_obj: socket对象
 :param commom: ftpserver运行时所需要的数据对象
 :return:
 """
 recv_choice = socket_obj.recv(1).decode("utf-8") # 获取用户选项：1是登陆，2是注册用户
 if recv_choice == "1": # client请求登陆
 return self.login_authen(socket_obj,commom)
 elif recv_choice == "2": # client请求注册账号
 return self.register_user(socket_obj,commom)
 else:
 return -1 # client断开连接了
 # 用户登陆认证
 def login_authen(self,socket_obj,commom):
 """
 客户端登陆认证
 :param socket_obj: socket对象
 :param commom: ftpserver运行时需要的数据对象
 :return:返回1代表登陆成功
 """
 # 接收client发来的用户名，密码
 recv_userPwd = self.recv_data(socket_obj).decode("utf-8").split("|") 
 # 效验用户名密码
 check_ret = self.check_user_pwd(recv_userPwd, socket_obj,commom)
 if check_ret: # 用户名密码正确
 self.check_user_home_dir(commom,recv_userPwd[0]) # 检测用户家目录
 return commom.status_info["login_success"]
 else:
 return commom.status_info["login_fail"]
 ...
 # 注册用户
 def register_user(self,socket_obj,commom):
 """
 :param socket_obj:
 :param commom:
 :return: 返回是否允许注册的结果，1允许客户端注册，2拒绝客户端注册
 """
 while True:
 choice_id = input("请输入回应码：1是允许注册，2是不允许注册:")
 if choice_id.isdigit() and 3 > int(choice_id) > 0:
 socket_obj.send(choice_id.encode("utf-8")) # 发通知告知客户端，处理结果
 if choice_id == "1": # 注册用户
  return self.client_register(socket_obj, commom)
 return choice_id
 else:
 print("您输入的信息有误，请重新输入。")
 ...

class DataAnalysis(FileOperation):
 """
 数据分析处理类，主要负责解析client发送过来的指令。
 """
 def syntax_analysis(self,recv_data, socket_obj, commom):
 """
 负责解析客户端传来的数据。
 :param recv_data:接收到的客户端用户数据
 :param socket_obj:socket对象
 :param commom:数据对象
 :return:
 """
 clientData = recv_data.split(" ")
 if hasattr(self,clientData[0]): # 判断对象方法是否存在
 get_fun = getattr(self,clientData[0])#获取对象方法
 get_fun(clientData,socket_obj,commom) # 运行对象方法
 else:
 pass
 ...

socket_obj = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
socket_obj.connect(("192.168.10.10",9000))

client_obj = Client()
client_obj.run_ftp_client(socket_obj) # 接收用户输入的选项，执行对应的功能

class Client(Login,DataAnalysis):
 def run_ftp_client(self,socket_obj):
 """
 运行用户输入的选项：1、是登陆 2、是注册账号
 :return:
 """
 while True:
 self.login_menu() # 打印系统菜单
 choice_id = self.get_user_choice() # 获取用户输入的选项
 if choice_id:
 if self.run_user_choice(choice_id,socket_obj):
  break
 else:
 print("您输入的有误")
 def get_user_choice(self):
 """
 获取用户输入的选项
 :return:
 """
 choice_id = input("请输入选项：")
 if choice_id.isdigit() and 4 > int(choice_id) > 0 or choice_id.upper() == "Q":
 return choice_id
 return False
 def run_user_choice(self,choice_id,socket_obj):
 if choice_id == "1": # 登陆系统
 socket_obj.send(choice_id.encode("utf-8")) # 发通知告知服务器准备登陆
 if self.run_login(socket_obj) == True: # 执行登陆
 return True
 elif choice_id == "2": # 注册用户
 socket_obj.send(choice_id.encode("utf-8")) # 请求服务器，注册用户
 self.register_user(socket_obj) # 执行注册
 elif choice_id.upper() == "Q": # 退出程序
 socket_obj.send(choice_id.encode("utf-8")) # 通知服务器，准备退出程序
 socket_obj.close()
 print("程序正常退出")
 return True
 def run_login(self,socket_obj,):
 """
 运行登陆认证模块，如果登陆成功执行程序主逻辑，否则重新登陆。
 :param socket_obj:
 :return:
 """
 if self.login_authention(socket_obj):
 while True:
 send_data = input(">").strip(" ") # 获取发送数据（用户执行的命令）
 if send_data.upper() == "Q": # 正常退出程序
  socket_obj.send(send_data.encode("utf-8")) # 通知服务区断开连接
  socket_obj.close()
  print("程序正常退出")
  return True
 if self.syntax_analysis(send_data, socket_obj): # 解析用户数据并处理数据
  print("异常退出")
  return True
 return False
 def login_menu(self):
 print("-"*41)
 print(" 欢迎登陆迷你FTPv1.0")
 print("-"*41)
 print("1、登陆系统")
 print("2、用户注册")
 print("Q、退出程序")

class Login(Commom):
 def login_authention(self,socket_obj):
 """
 登陆认证
 :param socket_obj:socket 对象
 :return:
 """
 user_pwd = self.get_user_pwd() # 获取用户名密码
 self.send_data(socket_obj,user_pwd) # 将用户名和密码发给服务器
 recv_status = socket_obj.recv(2).decode("utf-8") # 等待接收状态码
 print(self.status_info[recv_status]) # 打印状态码对应的结果
 if self.status_info[recv_status] == '登录成功':
 return True
 return False
 ...
 def register_user(self,socket_obj):
 """
 等待服务端反馈是否允许注册用户。
 :param socket_obj:
 :return:
 """
 print("请等待服务端回应.....")
 recv_status = socket_obj.recv(1).decode("utf-8")
 if recv_status == "1": # 服务端同意申请账号
 user_pwd = self.get_regist_user_pwd() # 获取注册用户名和密码
 if user_pwd:
 self.send_data(socket_obj,user_pwd)
 result = socket_obj.recv(2).decode("utf-8")
 print(self.status_info[result])
 else:
 print("用户名密码有误")
 else: # 客户端拒绝申请账号的请求
 print("服务端拒绝了您申请账号的请求，请与管理员取得联系。")
 return False
 ...

class DataAnalysis(FileOperation):
 def syntax_analysis(self,cmd,socket_obj):
 """
 解析用户输入的命令。
 :param cmd:用户执行的命令，如：put 上传的文件
 :param socket_obj:socket对象发送和接收数据
 :return:
 """
 cmd_split = cmd.split(" ") # 将字符串命令分割成列表，用于验证命令是否存在
 if hasattr(self,cmd_split[0]):
 run_fun = getattr(self,cmd_split[0])
 run_fun(cmd_split,socket_obj)
 else:
 print("无效的命令")
 ...