[Python]基于本地局域网的聊天程序

笙芝 · 2025-5-31 23:33:57

摘要：基于本地局域网的聊天程序源码

源码

使用python编写，运用 socket和tkinter等模块实现多人在线聊天，拥有基础GUI界面

import socket
import threading
import sys
import time
import tkinter as tk
from tkinter import scrolledtext, messagebox
class ChatServer:
def __init__(self, host='0.0.0.0', port=12345):
self.host = host
self.port = port
self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.server.bind((host, port))
self.server.listen(5)
self.clients = []
self.nicknames = []
self.heartbeat_interval = 30 # 心跳检测间隔(秒)
self.last_heartbeat = {}
def broadcast(self, message):
for client in self.clients:
client.send(message)
def handle(self, client):
index = self.clients.index(client)
nickname = self.nicknames[index]
self.last_heartbeat[client] = time.time()
while True:
try:
message = client.recv(1024)
if message == b'HEARTBEAT':
self.last_heartbeat[client] = time.time()
continue
self.broadcast(message)
except:
index = self.clients.index(client)
self.clients.remove(client)
client.close()
nickname = self.nicknames[index]
self.broadcast(f'{nickname} 离开了聊天室!'.encode('utf-8'))
self.nicknames.remove(nickname)
if client in self.last_heartbeat:
del self.last_heartbeat[client]
break
def receive(self):
print('服务器已启动，等待连接...')
while True:
try:
client, address = self.server.accept()
print(f'已连接: {str(address)}')
except socket.error as e:
print(f'接受连接时出错: {e}')
continue
try:
client.send('NICK'.encode('utf-8'))
nickname = client.recv(1024).decode('utf-8')
if not nickname:
raise ConnectionError('Empty nickname received')
self.nicknames.append(nickname)
self.clients.append(client)
except (ConnectionError, socket.error) as e:
print(f'客户端连接异常: {e}')
client.close()
continue
print(f'昵称是: {nickname}')
self.broadcast(f'{nickname} 加入了聊天室!'.encode('utf-8'))
client.send('已连接到服务器!'.encode('utf-8'))
thread = threading.Thread(target=self.handle, args=(client,))
thread.start()
class ChatClientGUI(tk.Tk):
def __init__(self, host='127.0.0.1', port=12345):
super().__init__()
self.title("聊天室客户端")
self.geometry("600x800")
self.configure(bg='#f0f0f0')
# 昵称输入
self.nickname_frame = tk.Frame(self, bg='#f0f0f0')
self.nickname_frame.pack(pady=10)
tk.Label(self.nickname_frame, text="输入昵称:", bg='#f0f0f0').pack(side=tk.LEFT)
self.nickname_entry = tk.Entry(self.nickname_frame, width=30)
self.nickname_entry.pack(side=tk.LEFT, padx=5)
tk.Button(self.nickname_frame, text="连接", command=self.connect_server).pack(side=tk.LEFT)
# 聊天显示区
self.chat_frame = tk.Frame(self)
self.chat_frame.pack(pady=10, padx=10, fill=tk.BOTH, expand=True)
self.chat_area = scrolledtext.ScrolledText(self.chat_frame, wrap=tk.WORD, state='disabled')
self.chat_area.pack(fill=tk.BOTH, expand=True)
# 消息输入区
self.input_frame = tk.Frame(self, bg='#f0f0f0')
self.input_frame.pack(pady=10, padx=10, fill=tk.X)
self.message_entry = tk.Entry(self.input_frame, width=50)
self.message_entry.config(font=('Microsoft YaHei', 10)) # 设置支持中文的字体
self.message_entry.pack(side=tk.LEFT, fill=tk.X, expand=True)
self.message_entry.bind('<Return>', self.send_message)
self.message_entry.bind('<Key>', lambda e: 'break' if e.keysym == 'Escape' else None) # 防止ESC键关闭输入法
tk.Button(self.input_frame, text="发送", command=self.send_message).pack(side=tk.LEFT, padx=5)
# 网络连接
self.client = None
self.host = host
self.port = port
self.nickname = ""
def connect_server(self):
self.nickname = self.nickname_entry.get().strip()
if not self.nickname:
messagebox.showerror("错误", "请输入昵称")
return
try:
self.client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.client.connect((self.host, self.port))
# 发送昵称
self.client.send(self.nickname.encode('utf-8'))
# 启动接收线程
receive_thread = threading.Thread(target=self.receive, daemon=True)
receive_thread.start()
# 禁用昵称输入
self.nickname_entry.config(state='disabled')
self.nickname_frame.children['!button'].config(state='disabled')
# 启用消息输入
self.message_entry.config(state='normal')
except ConnectionRefusedError:
messagebox.showerror("错误", "服务器未运行")
except socket.error as e:
messagebox.showerror("错误", f"连接错误: {e}")
def receive(self):
while True:
try:
message = self.client.recv(1024).decode('utf-8')
self.display_message(message)
except ConnectionResetError:
self.display_message("服务器连接已断开")
self.client.close()
break
except socket.error as e:
self.display_message(f"网络错误: {e}")
self.client.close()
break
def send_message(self, event=None):
message = self.message_entry.get()
if message and self.client:
try:
full_message = f'{self.nickname}: {message}'
self.client.send(full_message.encode('utf-8'))
self.message_entry.delete(0, tk.END)
except UnicodeEncodeError:
messagebox.showerror("编码错误", "无法发送包含特殊字符的消息")
except socket.error as e:
messagebox.showerror("网络错误", f"发送失败: {e}")
def display_message(self, message):
self.chat_area.config(state='normal')
self.chat_area.insert(tk.END, message + '\n')
self.chat_area.config(state='disabled')
self.chat_area.see(tk.END)
class ChatClient:
def __init__(self, host='127.0.0.1', port=12345):
self.nickname = input('输入你的昵称: ')
self.client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
self.client.connect((host, port))
except ConnectionRefusedError:
print('服务器未运行，请先启动服务器')
sys.exit(1)
except socket.error as e:
print(f'连接错误: {e}')
sys.exit(1)
def receive(self):
while True:
try:
message = self.client.recv(1024).decode('utf-8')
if message == 'NICK':
self.client.send(self.nickname.encode('utf-8'))
else:
print(message)
except ConnectionResetError:
print('服务器连接已断开')
self.client.close()
break
except socket.error as e:
print(f'网络错误: {e}')
self.client.close()
break
def write(self):
while True:
message = f'{self.nickname}: {input("")}'
self.client.send(message.encode('utf-8'))
def check_server_running(host='127.0.0.1', port=12345):
try:
test_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
test_socket.settimeout(1)
test_socket.connect((host, port))
test_socket.close()
return True
except:
return False
def find_available_port(start_port=12345, max_tries=100):
for port in range(start_port, start_port + max_tries):
try:
test_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
test_socket.bind(('127.0.0.1', port))
test_socket.close()
return port
except:
continue
return None
if __name__ == '__main__':
if check_server_running():
port = find_available_port()
if port is None:
print('无法找到可用端口，请稍后再试')
sys.exit(1)
# 使用GUI客户端
app = ChatClientGUI(port=port)
app.mainloop()
else:
choice = input('未检测到服务器，是否作为服务器启动？(y/n): ')
if choice.lower() == 'y':
print('[!]你当前作为服务器启动,可以再次启动程序并使用客户端启动')
print('[!]请不要关闭此窗口')
server = ChatServer()
server.receive()
else:
port = find_available_port()
if port is None:
print('无法找到可用端口，请稍后再试')
sys.exit(1)
# 使用GUI客户端
app = ChatClientGUI(port=port)
app.mainloop()

复制代码

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

账号		自动登录	找回密码
密码			立即注册