【软件开发之聊天软件】

这个代码是开发聊天软件使用，大家可参照代码使用。
注：

这个代码仅供参考,大家可以参照我的代码来修改

废话少说，上代码：

# ----数据协议相关配置----
# ----数据协议相关配置----
from threading import Thread
from typing import Anyimport self as selfREQUEST_LOGIN = '0001'  # 登陆请求
REQUEST_CHAT = '0002'  # 聊天请求
RESPONSE_LOGIN_RESULT = '1001'  # 登陆结果响应
RESPONSE_CHAT = '1002'  # 聊天响应
DELIMITER = '|'  # 自定义协议数据分割符
SERVER_IP = '127.0.0.1'  # 服务器地址
SERVER_PORT = 8090  # 服务器端口# ----数据库相关配置----   #具体参数自己设置
DB_HOST = '127.0.0.1'  # 这里是你数据库IP地址
DB_PORT = 3306  # 这里是你数据库端口
DB_NAME = 'py_chat'  # 这里是你数据库名
DB_USER = 'root'  # 这里是你数据库登陆账号
DB_PASSWD = '123456'  # 这里是你数据库密码
CHARSET = 'utf-8'# ----数据协议相关配置----
REQUEST_LOGIN = '0001'  # 登陆请求
REQUEST_CHAT = '0002'  # 聊天请求
RESPONSE_LOGIN_RESULT = '1001'  # 登陆结果响应
RESPONSE_CHAT = '1002'  # 聊天响应
DELIMITER = '|'  # 自定义协议数据分割符
SERVER_IP = '127.0.0.1'  # 服务器地址
SERVER_PORT: int = 8090  # 服务器端口class ResponseProtocol(object):@staticmethoddef response_chat(nickname, messages, DELIMITER=None, RESPONSE_CHAT=None):"""拼接聊天相应，数据格式为：“相应协议编号|聊天发送者昵称|聊天信息”:param RESPONSE_CHAT: :param DELIMITER: :param nickname: 聊天内容发送者昵称:param messages: 聊天内容:return: 聊天相应协议格式字符串"""return DELIMITER.join([RESPONSE_CHAT, nickname, messages])'import *'class Server(object):"""自定义套接字，负责初始化服务器套接字需要的相关参数"""def __init__(self):# 初始化套结字# 创建请求的ID和方法关联字典self.server_socket = Noneself.requset_handle_function = {}self.register(REQUEST_LOGIN, self.request_login_handle)self.register(REQUEST_CHAT, self.request_chat_handle)# 创建保存当前登录用户字典self.clients = ()'''注册消息类型和处理函数到字典'''def register(self, requeset_id, handle_function):self.requset_handle_function[requeset_id] = handle_function'''启动程序'''def startup(self):"""启动器"""while True:print('正在等待客户端连接')soc, addr = self.server_socket.accept()# print ('获取到客户端连接')client_soc = SocketWrapper(soc)# 启动线程处理该用户请求Thread(target=lambda: self.request_handle(client_soc)).start()'''处理客户端数据'''def request_handle(self, client_soc):while True:# 接收客户端数据recv_data = client_soc.recv_data()if not recv_data:# 没有接收到数据客户端应该已经关闭self.remve_offline_user(client_soc)client_soc.close()break'''解析数据'''parse_data = self.parse_request_text(recv_data)'''分析请求类型，并依据请求类型调用相应的分类处理''''''# 获得使用的方法名 方法名 = 字典[value]  注： 如 字典[key]可以互相找到字典[value]# 此处 字典[key]=0001 对应得字典[value] = REQUEST_LOGIN#例子：parse_data = '0001|XXX|XXX'parse_data['requset_id'] = ‘0001’requset_handle_function['0001'] = self.request_login_handlehandle_funtion = self.request_login_handle'''handle_funtion = self.requset_handle_function.get(parse_data['requset_id'])if handle_funtion:# 按照方法名调用方法handle_funtion(client_soc, parse_data)else:# 如果传输内容不匹配，返回错误请求self.request_err_handle(client_soc)'''用户离线操作'''def remve_offline_user(self, client_soc):for username, info in self.clients.items():if info['sock'] == client_soc:print(self.clients[username]['nickname'] + '已经离开')del self.clients[username]break'''解析客户端发送来的数据''''''解析数据内容'''def parse_request_text(self, recv_data):'''登录信息登录信息：0001|username|password聊天信息：0002|username|messages错误信息：err'''print('解析客户端数据:' + recv_data)requset_list = recv_data.split(DELIMITER)requset_data = {}requset_data['requset_id'] = requset_list[0]if requset_data['requset_id'] == REQUEST_LOGIN:requset_data['username'] = requset_list[1]requset_data['password'] = requset_list[2]elif requset_data['requset_id'] == REQUEST_CHAT:requset_data['username'] = requset_list[1]requset_data['messages'] = requset_list[2]return requset_data'''登录处理'''def request_login_handle(self, client_sock, requet_data):# print('收到登录请求')username = requet_data['username']password = requet_data['password']# 查询用户是否合法ret, nickname, username = self.check_user_login(username, password)# 如果登录成功，则保存用户连接套接字if ret == '1':self.clients[username] = {'sock': client_sock, 'nickname': nickname}# 组装响应结果response_text = response_login_result(ret, nickname, username)# 发送响应结果client_sock.send_data(response_text)'''聊天处理'''def request_chat_handle(self, client_sock, requet_data):# 获取消息内容username = requet_data['username']messages = requet_data['messages']try:nickname = self.clients[username]['nickname']except:client_sock.send_data('您未登录，请登录后再发消息')return# 拼接发送给客户的消息文本msg = ResponseProtocol.response_chat(nickname, messages)# 转发消息给在线用户for u_name, info in self.clients.items():if username == u_name:continueinfo['sock'].send_data(msg)print(msg)'''错误信息处理'''def request_err_handle(self, client_sock):print("传输数据出错------")client_sock.send_data('数据无效，请重新确认')'''检查用户是否登录成功，返回检查结果（0/失败，1/成功，昵称，用户账号'''def check_user_login(self, username, password):# print("正在检测是否成功")# 从数据库查询用户信息sql = "select * from users where username = '%s' " % usernameresult = self.db.get_one(sql)# 如果没有查询结果，用户不存在，登录失败if not result:# print("用户不存在，登录失败")return '-1', ' ', username# 密码不匹配，说明密码错误，登录失败if password != result["password"]:# print("密码错误，登录失败")return '0', ' ', username# 登录成功# print("验证正确，登录成功")print(result['nickname'] + "进入聊天室")return '1', result['nickname'], usernamedef Server():passdef Server() -> None:if __name__ == '__main__':Server().startup()def response_login_result(result: str, nickname: str, username: str, DELIMITER=None,RESPONSE_LOGIN_RESULT=None) -> str:"""拼接登陆响应:param RESPONSE_LOGIN_RESULT::param result:登陆结果，0或则1，0标识登陆失败，1标识登陆成功:param nickname:登陆名，登陆失败，该值为空字符串:param username:登陆ID，登陆失败，该值为空字符串:return 登陆结果相应格式字符串"""return DELIMITER.join([RESPONSE_LOGIN_RESULT, result, nickname, username])import socket
import configclass ServerSoket(socket.socket):"""自定义套接字，负责初始化服务器套接字需要的相关参数"""def __init__(self):# 设置TCP类型# 初始化套结字super(ServerSoket, self).__init__(socket.AF_INET, socket.SOCK_STREAM)self.bind((config.SERVER_IP, config.SERVER_PORT))# 设置为监听模式self.listen(128)class Server(object):"""服务器的核心类"""def __init__(self):# 初始化套结字self.server_socket = ServerSoket()# 创建请求的ID和方法关联字典self.requset_handle_function = {}self.register(REQUEST_LOGIN, self.request_login_handle)self.register(REQUEST_CHAT, self.request_chat_handle)# 创建保存当前登录用户字典self.clients = {}self.db = DBHandle()def startup(self):"""启动器"""while True:print('正在等待客户端连接')soc, addr = self.server_socket.accept()# print ('获取到客户端连接')client_soc = SocketWrapper(soc)# 启动线程处理该用户请求Thread(target=lambda: self.request_handle(client_soc)).start()def __init__(self):# 初始化套结字self.server_socket = ServerSoket()# 创建请求的ID和方法关联字典self.requset_handle_function = {}self.register(REQUEST_LOGIN, self.request_login_handle)self.register(REQUEST_CHAT, self.request_chat_handle)# 创建保存当前登录用户字典self.clients = {}def register(self, requeset_id, handle_function):"""注册消息类型和处理函数到字典"""self.requset_handle_function = 'handle_function'def request_handle(self, client_soc):'''处理客户端数据'''def recv_data() -> object:passrecv_data()
while True:# 接收客户端数据recv_data: Any = recv_data()if not recv_data:# 没有接收到数据客户端应该已经关闭self.remve_offline_user()"""client_soc.close()"""break'''解析数据'''parse_data = self.parse_request_text(recv_data)'''分析请求类型，并依据请求类型调用相应的分类处理''''''# 获得使用的方法名 方法名 = 字典[value]  注： 如 字典[key]可以互相找到字典[value]# 此处 字典[key]=0001 对应得字典[value] = REQUEST_LOGIN#例子：parse_data = '0001|XXX|XXX'parse_data['requset_id'] = ‘0001’requset_handle_function['0001'] = self.request_login_handlehandle_funtion = self.request_login_handle'''handle_funtion = self.requset_handle_function.get(parse_data['requset_id'])if handle_funtion:# 按照方法名调用方法handle_funtion(parse_data)else:# 如果传输内容不匹配，返回错误请求self.request_err_handle()'''登录处理'''def request_login_handle(self, client_sock, requet_data):# print('收到登录请求')username = requet_data['username']password = requet_data['password']# 查询用户是否合法ret, nickname, username = self.check_user_login(username, password)# 如果登录成功，则保存用户连接套接字if ret == '1':self.clients[username] = {'sock': client_sock, 'nickname': nickname}# 组装响应结果response_text = response_login_result(ret, nickname, username)# 发送响应结果client_sock.send_data(response_text)'''聊天处理'''def request_chat_handle(self, client_sock, requet_data):# 获取消息内容username = requet_data['username']messages = requet_data['messages']try:nickname = self.clients[username]['nickname']except:client_sock.send_data('您未登录，请登录后再发消息')return# 拼接发送给客户的消息文本msg = ResponseProtocol.response_chat(nickname, messages)# 转发消息给在线用户for u_name, info in self.clients.items():if username == u_name:continueinfo['sock'].send_data(msg)print(msg)'''检查用户是否登录成功，返回检查结果（0/失败，1/成功，昵称，用户账号'''def check_user_login(self, username, password):# print("正在检测是否成功")# 从数据库查询用户信息sql = "select * from users where username = '%s' " % usernameresult = self.db.get_one(sql)# 如果没有查询结果，用户不存在，登录失败if not result:# print("用户不存在，登录失败")return '-1', ' ', username# 密码不匹配，说明密码错误，登录失败if password != result["password"]:# print("密码错误，登录失败")return '0', ' ', username# 登录成功# print("验证正确，登录成功")print(result['nickname'] + "进入聊天室")return '1', result['nickname'], usernamefrom pymysql import connect
from config import *class DBHandle(object):'''mysql管理器'''def __init__(self):'''初始化数据库'''self.conn = connect(host=DB_HOST,port=DB_PORT,database=DB_NAME,user=DB_USER,password=DB_PASSWD)self.cursor = self.conn.cursor()# 释放数据库资源def close_db(self):self.cursor.close()self.conn.close()def get_one(self, sql):# 执行SQL结果self.cursor.execute(sql)# 获取查询结果query_result = self.cursor.fetchone()# 判断是否有结果if not query_result:return None# 获得字段名称列表fileds = [filed[0] for filed in self.cursor.description]# 保存返回结果return_data = {}for filed, value in zip(fileds, query_result):return_data[filed] = value# 查询结果return return_data'''用户离线操作'''def remve_offline_user(self, client_soc):for username, info in self.clients.items():if info['sock'] == client_soc:print(self.clients[username]['nickname'] + '已经离开')del self.clients[username]breakdef request_chat_handle(self, client_sock, requet_data):# 获取消息内容username = requet_data['username']messages = requet_data['messages']try:nickname = self.clients[username]['nickname']except:client_sock.send_data('您未登录，请登录后再发消息')return# 拼接发送给客户的消息文本msg = ResponseProtocol.response_chat(nickname, messages)# 转发消息给在线用户for u_name, info in self.clients.items():if username == u_name:continueinfo['sock'].send_data(msg)print(msg)from tkinter import Tk
from tkinter import Label, Entry, Frame, Button, LEFT, ENDclass WindowLogin(Tk):"""登陆窗口"""def __init__(self):super(WindowLogin, self).__init__()# 设置窗口属性self.window_init()# 填充控件self.add_widgets()# self.on_reset_button_click (lambda :print(self.get_username()))# self.on_login_button_click (lambda: print(self.get_password()))"""初始化窗口属性"""def window_init(self):# 设置窗口标题self.title('登陆窗口')# 设置窗口不能被拉伸self.resizable(False, False)# 获取窗口的位置变量window_width = 255window_height = 100screenWidth = self.winfo_screenwidth()screenHeight = self.winfo_screenheight()pos_x = (screenWidth - window_width) / 2pos_y = (screenHeight - window_height) / 2# 设置窗口大小和位置self.geometry('%dx%d+%d+%d' % (window_width, window_height, pos_x, pos_y))"""添加控件到窗口"""def add_widgets(self):"""添加控件到窗口"""# 用户名提示标签username_label = Label(self)username_label['text'] = '用户名:'username_label.grid(row=0, column=0, padx=10, pady=5)# 用户名输入文本框username_entry = Entry(self, name='username_entry')username_entry['width'] = 20username_entry.grid(row=0, column=1, padx=10, pady=5)# 密码提示标签password_label = Label(self)password_label['text'] = '密 码:'password_label.grid(row=1, column=0)# 密码输入文本框password_entry = Entry(self, name='password_entry')password_entry['show'] = '*'username_entry['width'] = 20password_entry.grid(row=1, column=1)# 按钮区button_frame = Frame(self, name='button_frame')# 重置按钮reset_button = Button(button_frame, name='reset_button')reset_button['text'] = '重置'reset_button.pack(side=LEFT, padx=40)# 登录按钮login_button = Button(button_frame, name='login_button')login_button['text'] = '登录'login_button.pack(side=LEFT)button_frame.grid(row=2, columnspan=2, pady=5)def get_username(self):"""获取用户名"""return self.children['username_entry'].get()def get_password(self):"""获取密码"""return self.children['password_entry'].get()def clear_username(self):""" 清空用户名"""return self.children['username_entry'].delete(0, END)def clear_password(self):""" 清空用户名"""return self.children['password_entry'].delete(0, END)def on_reset_button_click(self, command):"""重置按钮的响应注册"""reset_button = self.children['button_frame'].children['reset_button']reset_button['command'] = commanddef on_login_button_click(self, command):"""登录按钮的响应注册"""login_button = self.children['button_frame'].children['login_button']login_button['command'] = command  # 把command函数赋值给登录按钮的command，点击时调用commanddef on_window_close(self, command):"""关闭窗口的响应注册"""self.protocol('WM_DELETE_WINDOW', command)from config import *class RequestProtocol(object):"""服务器响应协议的格式字符串处理"""@staticmethoddef request_login_result(username, password):"""拼接登陆响应:param username:登陆用户名，登陆失败，该值为空字符串:param password:登陆密码:return 登陆结果相应格式字符串"""return DELIMITER.join([REQUEST_LOGIN, username, password])@staticmethoddef request_chat(username, messages):"""拼接聊天相应，数据格式为：“相应协议编号|聊天发送者账号|聊天信息”:param REQUEST_CHAT::param username: 聊天内容发送者账号:param messages: 聊天内容:return: 聊天相应协议格式字符串"""return DELIMITER.join([REQUEST_CHAT, username, messages])from threading import Thread
from tkinter.messagebox import showinfoclass Client(object):def __init__(self):"""初始化客户端资源"""# 初始化登陆窗口self.window = WindowLogin()self.window.on_login_button_click(self.send_login_data)self.window.on_reset_button_click(self.clear_inputs)# 创建客户端套接字self.conn = 'ClientSocket()'# 初始化消息处理函数self.response_handle_funtion = {}self.regist(RESPONSE_LOGIN_RESULT, self.response_login_handle)self.regist(RESPONSE_CHAT, self.response_chat_handle)def regist(self, requeset_id, handle_function):"""注册消息和消息对应的方法到字典里"""self.response_handle_funtion[requeset_id] = handle_functiondef startup(self):'''开启窗口'''self.conn.connect()Thread(target=self.response_handle).start()self.window.mainloop()def clear_inputs(self):"""清空窗口内容"""self.window.clear_password()self.window.clear_username()def send_login_data(self):username = self.window.get_username()password = self.window.get_password()request_text = RequestProtocol.request_login_result(username, password)self.conn.send_data(request_text)def response_handle(self):"不断收发服务器消息"while True:recv_data = self.conn.recv_data()print('收到服务器消息：' + recv_data)response_data = self.parse_response_data(recv_data)# 根据事件类型，调用指定方法名handle_funtionn = self.response_handle_funtion[response_data['response_id']]if handle_funtionn:handle_funtionn(response_data)@staticmethoddef parse_response_data(recv_data):'''登陆响应消息：1001|成功/失败|昵称|账号聊天响应消息：1002|发送者昵称|消息内容'''# 使用协议约定的符号来切割消息response_data_list = recv_data.split(DELIMITER)# 解析消息的各个组成部分response_data = {}response_data['response_id'] = response_data_list[0]if response_data['response_id'] == RESPONSE_LOGIN_RESULT:response_data['result'] = response_data_list[1]response_data['nickname'] = response_data_list[2]response_data['username'] = response_data_list[3]elif response_data['response_id'] == RESPONSE_CHAT:response_data['nickname'] = response_data_list[1]response_data['message'] = response_data_list[2]return response_datadef response_chat_handle(self, response_data):print('接收到聊天消息~', response_data)def response_login_handle(self, response_data):result = response_data['result']if result != '1':showinfo('提示', '账号或密码错误')returnnickname = response_data['nickname']username = response_data['username']print('%s 的昵称为 %s ,已经登录聊天室' % (username, nickname))

本文为优化的代码，此文基础为借鉴，改进是由我改的，所以此文选择原创，本文的基础原创我不记得了，如是你的原创，大家可以私信找我

好了，今天就讲到这里了，下课

【软件开发之聊天软件】相关推荐

仿微信界面开发的聊天软件和聊天客户端功能介绍
系统模式:独立系统版本,独立服务器部署,服务器配置:建议 4 核 8G:10M 带宽既可以下是聊天系统客户端的功能介绍,服务端的功能我会再写一篇文章接着仿微信聊天开发的聊天软件以及聊天客户端功能介 ...
windowsphone开发_APP软件开发用哪些软件比较好
说到APP软件开发,大家能否联想到APP使用情况呢.我们在生活上离不开手机,现在时代发展越来越快,2020年处于短视频流量时代,像抖音等这些软件APP想必大家都听说过,这样的一款APP开发需要多长时间 ...
软件工程：浅谈人工智能软件开发与传统软件开发的区别
题目:浅谈人工智能软件开发与传统软件开发的区别摘要:人工智能的飞速发展带动着软件工程的发展,最终使得软件工程产生新的变革.因为人工智能特有的性质,因而导致了人工智能软件与传统软件的差异性.本文对比了 ...
恩施软件开发人员每月多少钱_恩施软件开发学习,恩施软件开发学习哪家好,恩施软件开发学习一般能拿多少工资...
恩施软件开发学习,恩施软件开发学习哪家好,恩施软件开发学习一般能拿多少工资首页 > 软件 > 恩施软件开发学习作者:镀金池发布时间:2017-11-22 18:54 因为cent ...
浅谈移动软件开发与电脑软件开发发展前景
首先,先说下当下的情况,我国经济的高速发展对计算机软件行业需求越来越高,因此国家根据实际需求加大了对软件行业的各项投入,如资金.人力.物力.政策. 软件行业是国家重点鼓励和支持的行业,他的发展前景肯定 ...
用Delphi开发视频聊天软件
一.引言我们知道视频聊天软件的关键技术在于采集视频,并实时传输给聊天软件在线的人.对于视频的采集,这里采用微软公司的关于数字视频的一个软件包VFW(Video for Windows).相信很多人对 ...
软件开发丨关于软件重构的灵魂四问
在软件工程学中重构就是在不改变软件现有功能的基础上,通过调整程序代码改善软件的质量.性能,使其程序的设计模式和架构更趋合理,提高软件的扩展性和维护性. 摘要在本文中,您会了解到如下的内容: 先添加新 ...
在线教育软件开发教育APP软件开发分析
随着互联网的快速发展,教育机构也纷纷改变了发展模式,教育APP软件的开发大受欢迎,司科认为教育软件的开发不仅能让教学资源共享,更能促进教育行业的发展. 为什么教育类APP软件值得开发 1.使用教育AP ...
抢单软件开发原理_软件开发原理
抢单软件开发原理 Rubbish software is produced when we try to do everything at once. 当我们尝试一次做所有事情时,就会产生垃圾软件. ...

【软件开发之聊天软件】

【软件开发之聊天软件】相关推荐

最新文章

热门文章