目录
第三方库wxPython
-
wsPython是第三方库，代码实现基于C++的wxWidgets库封装，呈现的界面风格和系统本地风格一致 -
多人聊天室，可以拥有多个客户端，一个服务端
使用wxPython库实现聊天室界面
客户端界面
-
客户端界面示意图
-
示例代码：
pythonimport threading from socket import socket, AF_INET, SOCK_STREAM import wx class TomClient(wx.Frame): def __init__(self, client_name): # 调用父类的初始化方法 # None: 没有父级窗口 # id表示当前窗口的一个编号 # pos 窗体的打开位置 # size 窗体的大小，单位是像素，400宽，450高度 wx.Frame.__init__(self, None, id=1001, title=client_name+'客户端界面', pos=wx.DefaultPosition, size=(400,450)) # 创建面板对象 pl = wx.Panel(self) # 在面板中放上盒子 box = wx.BoxSizer(wx.VERTICAL) # 垂直方向布局 # 可伸缩的网格布局 fgz1 = wx.FlexGridSizer(wx.HSCROLL) # 水平方向布局 # 创建2个按钮 conn_btn = wx.Button(pl, size=(200, 40), label='连接') dis_conn_btn = wx.Button(pl, size=(200, 40), label='断开') # 把两个按钮放到可伸缩的网格布局 fgz1.Add(conn_btn, 1, wx.TOP|wx.LEFT) fgz1.Add(dis_conn_btn, 1, wx.TOP|wx.RIGHT) # （可伸缩的网格布局）添加到box中 box.Add(fgz1, 1, wx.ALIGN_CENTRE) # 只读文本框，显示聊天内容 self.show_text = wx.TextCtrl(pl, size=(400, 210), style=wx.TE_MULTILINE|wx.TE_READONLY) box.Add(self.show_text, 1, wx.ALIGN_CENTRE) # 创建聊天内容的文本框 self.chat_text = wx.TextCtrl(pl, size=(400, 120), style=wx.TE_MULTILINE) box.Add(self.chat_text, 1, wx.ALIGN_CENTRE) # 可伸缩的网格布局 fgz2 = wx.FlexGridSizer(wx.HSCROLL) # 水平方向布局 # 创建2个按钮 reset_btn = wx.Button(pl, size=(200, 40), label='重置') send_btn = wx.Button(pl, size=(200, 40), label='发送') fgz2.Add(reset_btn, 1, wx.TOP|wx.LEFT) fgz2.Add(send_btn, 1, wx.TOP|wx.RIGHT) # （可伸缩的网格布局）添加到box中 box.Add(fgz2, 1, wx.ALIGN_CENTRE) # 将盒子放到面板中 pl.SetSizer(box) '''------------------------------以上代码都是界面的绘制代码------------------------------''' self.Bind(wx.EVT_BUTTON, self.connect_to_server, conn_btn) # 实例属性的设置 self.client_name = client_name self.isConnected = False # 存储客户端连接服务器的状态，默认False，没连接 self.client_socket = None # 设置客户端的socket对象为空 # 给“发送”按钮绑定一个事件 self.Bind(wx.EVT_BUTTON, self.send_to_server, send_btn) # 给“断开”按钮绑定一个事件 self.Bind(wx.EVT_BUTTON, self.dis_conn_server, dis_conn_btn) # 给“重置”按钮绑定一个事件 self.Bind(wx.EVT_BUTTON, self.reset, reset_btn) def reset(self, event): self.chat_text.Clear() # 文本框中内容就没有了 def dis_conn_server(self, event): # 发送断开的信息 self.client_socket.send('bye'.encode('utf-8')) # 改变连接状态 self.isConnected = False def send_to_server(self,event): # 判断连接状态 if self.isConnected: # 从可写文本框获取 input_data = self.chat_text.GetValue() if input_data != '': # 向服务器端发送数据 self.client_socket.send(input_data.encode('utf-8')) # 发送完数据之后，清空文本框 self.chat_text.SetValue('') def connect_to_server(self, event): print(f'客户端{self.client_name}连接服务器成功') # 如果客户端没有连接服务器，则开始连接 if not self.isConnected: # 等价于 isConnected == False # TCP编程步骤 server_host_port = ('127.0.0.1', 8888) # 创建socket对象 self.client_socket = socket(AF_INET, SOCK_STREAM) # 发送连接请求 self.client_socket.connect(server_host_port) # 只要连接成功，发送一条数据 self.client_socket.send(self.client_name.encode('utf-8')) # 启动一个线程，客户端的线程与服务器的会话线程进行会话 client_thread = threading.Thread(target=self.recv_data) # 设置成守护进程，窗体关闭，子线程也结束了 client_thread.daemon = True # 修改一下连接状态 self.isConnected = True # 启动线程 client_thread.start() def recv_data(self): # 如果是连接状态 while self.isConnected: # 接收来自服务器端数据 data = self.client_socket.recv(1024).decode('utf-8') # 显示到只读文本框中 self.show_text.AppendText('-'*40+'\n'+data+'\n') if __name__ == '__main__': # 初始化App app = wx.App() # 创建自己的客户端界面对象 name = input('请输入客户端名称') # client = TomClient('tom客户端') client = TomClient(name) client.Show() # 可以改成 TomClient('tom客户端').Show() # 循环刷新显示 app.MainLoop()
服务端界面
-
服务器端界面示意图
-
示例代码
pythonimport threading import time import wx from socket import socket, AF_INET, SOCK_STREAM class TomServer(wx.Frame): def __init__(self): wx.Frame.__init__(self, None, id=1002, title='服务器端界面', pos=wx.DefaultPosition, size=(400, 450)) # 窗口放一个面板 pl = wx.Panel(self) # 面板上放一个盒子 box = wx.BoxSizer(wx.VERTICAL) # 垂直方向自动排版 # 可伸缩的网格布局 fgz1 = wx.FlexGridSizer(wx.HSCROLL) # 水平方向布局 start_server_btn = wx.Button(pl, size=(133, 40), label='启动服务') record_btn = wx.Button(pl, size=(133, 40), label='保存聊天记录') stop_server_btn = wx.Button(pl, size=(133, 40), label='停止服务') # 放到可伸缩的网格布局中 fgz1.Add(start_server_btn, 1, wx.TOP) fgz1.Add(record_btn, 1, wx.TOP) fgz1.Add(stop_server_btn, 1, wx.TOP) # 将可伸缩的网格放到box中 box.Add(fgz1, 1, wx.ALIGN_CENTRE) # 只读文本框，显示聊天内容 self.show_text = wx.TextCtrl(pl, size=(400, 410), style=wx.TE_MULTILINE|wx.TE_READONLY) box.Add(self.show_text, 1, wx.ALIGN_CENTRE) # 把盒子放到面板中 pl.SetSizer(box) '''------------------------------以上代码都是界面的绘制代码------------------------------''' '''以下代码是服务器功能实现的必要属性''' self.isOn = False # 存储服务器启动状态，默认值False，默认没有启动 # 服务器端绑定的IP地址和端口号 self.host_port = ('', 8888) # 空的字符串代表的是本机的所有IP # 创建socket对象 self.server_socket = socket(AF_INET, SOCK_STREAM) # 绑定IP地址和端口号 self.server_socket.bind(self.host_port) # 监听 self.server_socket.listen(5) # 创建一个字典，存储与客户端对话的线程 self.session_thread_dict = {} # key-value {客户端的名称key：会话线程value} # 当鼠标点击“启动服务”按钮时，要执行的操作 self.Bind(wx.EVT_BUTTON, self.start_server, start_server_btn) # 为保存聊天记录按钮绑定事件 self.Bind(wx.EVT_BUTTON, self.save_record, record_btn) # 为“断开”按钮绑定事件 self.Bind(wx.EVT_BUTTON, self.stop_server, stop_server_btn) def stop_server(self, event): print('服务器已经停止服务') self.isOn = False def save_record(self, event): # 获取只读文本框的内容 record_data = self.show_text.GetValue() with open('record.log', 'w',encoding='utf-8') as file: file.write(record_data) def start_server(self, event): # 判断服务器是否已经启动，只有服务器没有启动的时候才启动 if not self.isOn: # 等价于 self.isOn==False # 启动服务器 self.isOn = True # 创建主线程对象，函数式创建主线程 main_thread = threading.Thread(target=self.do_work) # 设置为守护线程，父线程执行结束（窗体界面）子线程也自动关闭 main_thread.daemon = True # 启动主线程 main_thread.start() def do_work(self): # 判断isOn的值 while self.isOn: # 接收客户端的连接请求 session_socket, client_addr = self.server_socket.accept() # 客户端发送连接请求之后，发送过来的第一条数据为客户端的名称，客户端的名称去作为字典中的键 user_name = session_socket.recv(1024).decode('utf-8') # 创建一个会话线程对象 session_thread = SessionThread(session_socket, user_name, self) # 存储到字典中 self.session_thread_dict[user_name] = session_thread # 启动会话线程 session_thread.start() # 输出服务器的提示信息 self.show_info_and_send_client('服务器通知', f'欢迎{user_name}进入聊天室！', time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())) # 当self.isOn的值为False时，执行关闭socket对象 self.server_socket.close() def show_info_and_send_client(self, data_source, data, data_time): # 字符串操作 send_data = f'{data_source}:{data}\n时间：{data_time}' # 只读文本框 self.show_text.AppendText('-'*40+'\n'+send_data+'\n') # 每个客户端都发送一次 for client in self.session_thread_dict.values(): # 判断当前会话状态是否开启 if client.isOn: client.client_socket.send(send_data.encode('utf-8')) # 服务器端会话线程的类 class SessionThread(threading.Thread): def __init__(self, client_socket, user_name, server): # 调用父类的初始化方法 threading.Thread.__init__(self) self.client_socket = client_socket self.user_name = user_name self.server = server self.isOn = True # 会话线程是否启动 def run(self) -> None: print(f'客户端：{self.user_name}已经和服务器连接成功，服务器启动一个会话进程') while self.isOn: # 从客户端接收数据，存到data中 data = self.client_socket.recv(1024).decode('utf-8') # 如果客户端点击断开按钮，先给服务器发送一句话，消息自定义 if data == 'bye': self.isOn = False # 发送一条服务器通知 self.server.show_info_and_send_client('服务器通知', f'{self.user_name}离开了聊天室', time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())) else: # 其他聊天信息显示给所有的客户端，包括服务器端也显示 self.server.show_info_and_send_client(self.user_name, data, time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())) # 关闭socket self.client_socket.close() if __name__ == '__main__': # 初始化App app = wx.App() # 创建自己的服务器端界面对象 server = TomServer() server.Show() # 循环刷新显示 app.MainLoop()
本文作者:柯南
本文链接:
版权声明:©2024 柯南 All rights reserved.
目录