基于 PyQt6 的局域网聊天室
·
基于 PyQt6 的局域网聊天室（集成星火 AI 机器人）
一、项目简介
本项目是一个基于 Python + PyQt6 实现的局域网聊天室系统，采用 C/S（客户端/服务器）架构，支持以下核心功能：
- 群聊与私聊（基于 TCP Socket 通信）
- 在线用户列表实时刷新（上线/下线广播）
- 用户名唯一性校验（重复注册会被拒绝）
- 集成讯飞星火大模型 AI 机器人（群聊 @ 触发 + 私聊直接对话）
- @ 用户选择弹窗（输入
@自动弹出在线用户列表） - 服务器 GUI 控制面板（一键启停、IP 自动获取、复制 IP）
- 自定义消息分隔符
__EOF__解决 TCP 粘包问题 - 多标签页私聊窗口（可关闭、可切换）
项目共包含三个核心源文件：
| 文件名 | 作用 |
|---|---|
chat.py |
客户端 UI 界面定义（PyQt6 设计） |
client.py |
客户端业务逻辑（连接、收发消息、私聊、@功能） |
server.py |
服务器业务逻辑（含星火 AI 集成、GUI 控制台） |
二、技术栈
- Python 3.x
- PyQt6（GUI 界面）
- socket / threading（TCP 网络通信与多线程）
- websocket-client（星火 AI WebSocket 接口）
- json（消息序列化）
- hmac / hashlib / base64（星火 API 鉴权签名）
依赖安装：
pip install PyQt6 websocket-client
三、通信协议设计
客户端与服务器之间使用 TCP 长连接通信，所有消息以 JSON 格式编码，并以 __EOF__ 作为消息分隔符，从而解决 TCP 粘包/半包问题。
消息结构示例：
{
"type": "chat",
"sender": "张三",
"time": "14:23:05",
"msg": "你好"
}
消息类型（type 字段）约定如下：
| type 值 | 含义 |
|---|---|
| 1 | 客户端上线注册 |
| 2 | 用户上线广播 |
| 3 | 私聊消息 |
| 4 | 用户下线广播 |
| 5 | 用户名注册失败（重复或保留名） |
chat |
群聊消息 |
四、功能详解
4.1 服务器端
服务器端核心类 ChatServer 继承自 QtCore.QObject，通过信号机制与 GUI 交互，主要职责：
- 监听客户端连接：在独立守护线程中循环
accept()，为每个客户端分配一个ServerThread处理消息。 - 用户名校验：注册时检查是否与已有用户重名，或使用了系统保留名
SparkAI，若冲突则返回type=5错误消息并踢出连接。 - 消息转发：
- 群聊消息广播给除 AI 虚拟客户端外的所有用户。
- 私聊消息仅转发给发送者和目标用户。
- 星火 AI 集成：
- 群聊中消息以
@SparkAI、AI:、机器人：等前缀触发，AI 在独立线程中调用星火接口生成回复并广播。 - 私聊目标为
SparkAI时，AI 使用独立的私聊上下文历史进行回复。
- 群聊中消息以
- AI 上下文管理：群聊与每个私聊用户分别维护对话历史，当历史内容超过 8000 字符时自动截断早期记录。
4.2 客户端
客户端 ClientWindow 继承自 QMainWindow，主要功能：
- 连接管理：输入 IP、端口、用户名后连接服务器，连接成功后用户名锁定不可修改。
- 消息收发：通过
ClientThread（QThread子类）异步接收消息，使用信号incoming_message投递到主线程处理。 - 多标签页私聊：双击在线用户列表即可打开私聊标签页，支持关闭。私聊消息自动路由到对应标签页。
- @ 功能：在群聊输入框输入
@时，自动弹出在线用户选择对话框，选择后自动补全用户名。 - sendTo 指令：支持在输入框直接输入
sendTo 用户名 消息内容快速发起私聊。 - 自定义对话框：使用
FixedMessageDialog替代QMessageBox，固定窗口大小，避免在不同系统下布局错乱。
4.3 UI 界面
chat.py 定义了客户端主窗口布局，采用 QGridLayout 网格布局：
- 第一行：IP、端口、用户名输入框 + 连接/关闭按钮
- 第二行：聊天区域（
QTabWidget，默认群聊页）+ 在线用户列表 - 第三行：状态栏
- 第四行：消息输入框
- 第五行：发送文件、清空、发送、退出按钮
五、完整代码
5.1 客户端 UI 界面 — chat.py
# -*- coding: utf-8 -*-
from PyQt6 import QtCore, QtGui, QtWidgets
class Ui_MainWindow(object):
def setupUi(self, MainWindow):
MainWindow.setObjectName("MainWindow")
MainWindow.resize(750, 750)
self.centralwidget = QtWidgets.QWidget(MainWindow)
self.centralwidget.setObjectName("centralwidget")
self.layoutWidget = QtWidgets.QWidget(self.centralwidget)
self.layoutWidget.setGeometry(QtCore.QRect(20, 20, 710, 680))
self.layoutWidget.setObjectName("layoutWidget")
self.gridLayout = QtWidgets.QGridLayout(self.layoutWidget)
self.gridLayout.setContentsMargins(0, 0, 0, 0)
self.gridLayout.setObjectName("gridLayout")
# --- 第一行: 基础连接设置 ---
self.label = QtWidgets.QLabel("IP:", self.layoutWidget)
self.gridLayout.addWidget(self.label, 0, 0, 1, 1)
self.ip = QtWidgets.QLineEdit(self.layoutWidget)
self.gridLayout.addWidget(self.ip, 0, 1, 1, 1)
self.label_3 = QtWidgets.QLabel("Port:", self.layoutWidget)
self.gridLayout.addWidget(self.label_3, 0, 2, 1, 1)
self.port = QtWidgets.QLineEdit(self.layoutWidget)
self.gridLayout.addWidget(self.port, 0, 3, 1, 1)
self.label_user = QtWidgets.QLabel("Name:", self.layoutWidget)
self.gridLayout.addWidget(self.label_user, 0, 4, 1, 1)
self.username = QtWidgets.QLineEdit(self.layoutWidget)
self.gridLayout.addWidget(self.username, 0, 5, 1, 1)
self.start_button = QtWidgets.QPushButton("Connect", self.layoutWidget)
self.gridLayout.addWidget(self.start_button, 0, 6, 1, 1)
self.terminal_button = QtWidgets.QPushButton("Close", self.layoutWidget)
self.gridLayout.addWidget(self.terminal_button, 0, 7, 1, 1)
# --- 聊天区域与连接列表 ---
# 标签页 (开启可关闭属性 tabsClosable)
self.chat_tabs = QtWidgets.QTabWidget(self.layoutWidget)
self.chat_tabs.setTabsClosable(True) # 允许关闭私聊标签页
self.group_chat_window = QtWidgets.QListWidget()
# 核心修改：开启自动换行 + 自适应宽度
self.group_chat_window.setWordWrap(True)
self.group_chat_window.setResizeMode(QtWidgets.QListWidget.ResizeMode.Adjust)
self.group_chat_window.setHorizontalScrollBarPolicy(QtCore.Qt.ScrollBarPolicy.ScrollBarAsNeeded)
self.group_chat_window.setSizePolicy(
QtWidgets.QSizePolicy.Policy.Expanding,
QtWidgets.QSizePolicy.Policy.Expanding
)
# 默认只保留群聊页面
self.chat_tabs.addTab(self.group_chat_window, "Group Chat (群聊)")
self.gridLayout.addWidget(self.chat_tabs, 1, 0, 1, 6)
# 在线用户列表 (提示双击私聊)
self.list_label = QtWidgets.QLabel("双击用户发起私聊:", self.layoutWidget)
self.gridLayout.addWidget(self.list_label, 1, 6, 1, 2, QtCore.Qt.AlignmentFlag.AlignBottom)
self.connection_list = QtWidgets.QListWidget(self.layoutWidget)
self.connection_list.setMaximumWidth(200)
self.gridLayout.addWidget(self.connection_list, 1, 6, 1, 2)
# --- 状态栏 ---
self.label_5 = QtWidgets.QLabel("Status:", self.layoutWidget)
self.gridLayout.addWidget(self.label_5, 2, 0, 1, 1)
self.running_status = QtWidgets.QLabel("No Connection", self.layoutWidget)
self.gridLayout.addWidget(self.running_status, 2, 1, 1, 7)
# --- 输入框 ---
self.input_window = QtWidgets.QTextEdit(self.layoutWidget)
self.input_window.setMaximumSize(QtCore.QSize(16777215, 100))
self.input_window.setPlaceholderText("此输入消息...")
self.gridLayout.addWidget(self.input_window, 3, 0, 1, 8)
# --- 底部控制栏 ---
# 调整了布局，取消了之前的下拉选择框
self.file_button = QtWidgets.QPushButton("Send File", self.layoutWidget)
self.gridLayout.addWidget(self.file_button, 4, 4, 1, 1)
self.clear_button = QtWidgets.QPushButton("Clear", self.layoutWidget)
self.gridLayout.addWidget(self.clear_button, 4, 5, 1, 1)
self.send_button = QtWidgets.QPushButton("Send", self.layoutWidget)
self.gridLayout.addWidget(self.send_button, 4, 6, 1, 1)
self.quit_button = QtWidgets.QPushButton("Quit", self.layoutWidget)
self.gridLayout.addWidget(self.quit_button, 4, 7, 1, 1)
MainWindow.setCentralWidget(self.centralwidget)
MainWindow.setWindowTitle("局域网聊天室")
QtCore.QMetaObject.connectSlotsByName(MainWindow)
5.2 客户端业务逻辑 — client.py
# -*- coding: utf-8 -*-
from PyQt6 import QtCore, QtWidgets, QtGui
import chat_ui_3 as chat_ui
import socket
import json
import sys
import base64
import os
class ClientThread(QtCore.QThread):
incoming_message = QtCore.pyqtSignal(dict)
def __init__(self, socket_conn):
super().__init__()
self.socket = socket_conn
self.buffer = b""
def run(self):
while True:
try:
raw_data = self.socket.recv(1024 * 1024)
if not raw_data:
break
self.buffer += raw_data
while b"__EOF__" in self.buffer:
msg_bytes, self.buffer = self.buffer.split(b"__EOF__", 1)
if msg_bytes.strip():
msg_str = msg_bytes.decode('utf-8')
msg_dict = json.loads(msg_str)
self.incoming_message.emit(msg_dict)
except Exception as e:
print(f"Thread connection lost: {e}")
break
class ClientWindow(QtWidgets.QMainWindow, chat_ui.Ui_MainWindow):
def __init__(self):
super().__init__()
self.setupUi(self)
self.socket = None
self.thread = None
self.ip.setText('127.0.0.1')
self.port.setText('12345')
self.username.setText('User_' + str(os.getpid())[-3:])
# 绑定按钮事件
self.quit_button.clicked.connect(self.quit)
self.send_button.clicked.connect(self.send_chat)
self.start_button.clicked.connect(self.connect_server)
self.terminal_button.clicked.connect(self.terminate)
self.clear_button.clicked.connect(self.clear_window)
self.file_button.clicked.connect(self.send_file)
# 绑定双击用户列表发起私聊
self.connection_list.itemDoubleClicked.connect(self.on_user_double_clicked)
# 绑定关闭标签页事件
self.chat_tabs.tabCloseRequested.connect(self.close_private_tab)
# ========== 初始化在线用户列表 + 监听输入框文本变化 ==========
self.online_users = [] # 维护在线用户列表（用于@选择）
self.is_inserting_at = False # 防止重复触发@选择弹窗
self.input_window.textChanged.connect(self.on_text_changed)
self.my_username = ""
# 维护一个字典，保存当前已打开的私聊窗口：{ '目标用户名': QListWidget对象 }
self.private_tabs = {}
class FixedMessageDialog(QtWidgets.QDialog):
"""固定大小的信息提示对话框（用于替换QMessageBox）"""
def __init__(self, parent=None, title="提示", message="", is_critical=False):
super().__init__(parent)
self.setWindowTitle(title)
# 固定对话框大小
self.setFixedSize(400, 150)
layout = QtWidgets.QVBoxLayout(self)
# 根据是否为错误类型设置图标
icon_label = QtWidgets.QLabel()
if is_critical:
icon_label.setPixmap(QtWidgets.QApplication.style().standardIcon(QtWidgets.QStyle.StandardPixmap.SP_MessageBoxCritical).pixmap(32, 32))
else:
icon_label.setPixmap(QtWidgets.QApplication.style().standardIcon(QtWidgets.QStyle.StandardPixmap.SP_MessageBoxInformation).pixmap(32, 32))
h_layout = QtWidgets.QHBoxLayout()
h_layout.addWidget(icon_label)
h_layout.addWidget(QtWidgets.QLabel(message), 1) # 参数1表示拉伸因子
layout.addLayout(h_layout)
btn_ok = QtWidgets.QPushButton("确定")
btn_ok.clicked.connect(self.accept)
layout.addWidget(btn_ok, 0, QtCore.Qt.AlignmentFlag.AlignCenter)
class UserSelectDialog(QtWidgets.QDialog):
"""用于@用户选择的自定义对话框，大小固定"""
def __init__(self, parent=None, user_list=None):
super().__init__(parent)
if user_list is None:
user_list = []
self.selected_user = None
self.setWindowTitle("选择@的用户")
# 固定对话框大小
self.setFixedSize(400, 200)
layout = QtWidgets.QVBoxLayout(self)
label = QtWidgets.QLabel("请选择要@的群聊用户：")
layout.addWidget(label)
self.user_combo = QtWidgets.QComboBox()
self.user_combo.addItems(user_list)
layout.addWidget(self.user_combo)
button_box = QtWidgets.QDialogButtonBox(QtWidgets.QDialogButtonBox.StandardButton.Ok | QtWidgets.QDialogButtonBox.StandardButton.Cancel)
button_box.accepted.connect(self.accept)
button_box.rejected.connect(self.reject)
layout.addWidget(button_box)
def accept(self):
"""点击OK时，保存选中的用户名"""
self.selected_user = self.user_combo.currentText()
super().accept()
# ========== 输入框文本变化监听（检测@触发选择） ==========
def on_text_changed(self):
# 1. 仅在群聊标签页（index=0）且非弹窗状态时检测
if self.chat_tabs.currentIndex() != 0 or self.is_inserting_at:
return
# 2. 获取光标位置和输入框文本
cursor = self.input_window.textCursor()
text = self.input_window.toPlainText()
pos = cursor.position()
# 3. 检测光标前是否是单独的@（避免@后已有内容）
if pos > 0 and text[pos - 1] == '@':
# 过滤：@前是空格/换行/开头，确保是新的@指令
if pos == 1 or text[pos - 2] in [' ', '\n', '\t', '']:
self.is_inserting_at = True # 标记弹窗状态，防止重复触发
# 4. 过滤掉自己，仅显示其他在线用户
selectable_users = [user for user in self.online_users if user != self.my_username]
if not selectable_users:
self.is_inserting_at = False
QtWidgets.QMessageBox.information(self, "提示", "暂无其他在线用户可@")
return
# 5. 弹出用户选择对话框
dialog = self.UserSelectDialog(self, selectable_users)
if dialog.exec() == QtWidgets.QDialog.DialogCode.Accepted and dialog.selected_user:
target_user = dialog.selected_user
# 6. 选择后自动填充@用户名 + 空格
cursor.insertText(f"{target_user} ") # 插入用户名+空格，方便继续输入
self.input_window.setTextCursor(cursor)
self.is_inserting_at = False # 重置弹窗状态
# ==============================================================
def on_user_double_clicked(self, item):
"""双击列表中的用户，打开或跳转到私聊标签页"""
target_username = item.text().replace("[Online] ", "")
if target_username == self.my_username:
# 使用自定义的固定大小消息对话框
dialog = self.FixedMessageDialog(self, "提示", "不能跟自己私聊哦！")
dialog.exec()
return
self.open_or_focus_private_tab(target_username)
def open_or_focus_private_tab(self, target_username):
if target_username in self.private_tabs:
tab_widget = self.private_tabs[target_username]
index = self.chat_tabs.indexOf(tab_widget)
self.chat_tabs.setCurrentIndex(index)
return tab_widget
else:
new_tab = QtWidgets.QListWidget()
# 新增：适配Linux的布局配置
new_tab.setWordWrap(True)
new_tab.setResizeMode(QtWidgets.QListWidget.ResizeMode.Adjust)
# 允许水平滚动，避免文本截断
new_tab.setHorizontalScrollBarPolicy(QtCore.Qt.ScrollBarPolicy.ScrollBarAsNeeded)
# 让控件自适应窗口大小
new_tab.setSizePolicy(
QtWidgets.QSizePolicy.Policy.Expanding,
QtWidgets.QSizePolicy.Policy.Expanding
)
self.private_tabs[target_username] = new_tab
index = self.chat_tabs.addTab(new_tab, f"私聊: {target_username}")
self.chat_tabs.setCurrentIndex(index)
return new_tab
def close_private_tab(self, index):
"""关闭某个标签页"""
if index == 0:
# 群聊标签页不允许关闭
return
widget = self.chat_tabs.widget(index)
# 从字典中移除记录
for user, w in list(self.private_tabs.items()):
if w == widget:
del self.private_tabs[user]
break
self.chat_tabs.removeTab(index)
widget.deleteLater()
def get_current_target(self):
"""根据当前处于前台的标签页，判断发送的目标是谁"""
index = self.chat_tabs.currentIndex()
if index == 0:
return 'All'
else:
# 标签页名称格式为 "私聊: xxx"
title = self.chat_tabs.tabText(index)
return title.replace("私聊: ", "")
def send_chat(self):
if self.thread is not None and self.thread.isRunning():
text = self.input_window.toPlainText().strip()
# 新增：打印原始输入文本和编码
print(f"【客户端发送】原始文本: {text}, UTF-8编码: {text.encode('utf-8')}")
if not text:
return
# sendTo 指令支持
if text.startswith('sendTo '):
parts = text.split(' ', 2)
if len(parts) >= 3:
target = parts[1]
msg_text = parts[2]
# 打开该用户的私聊标签页，确保发出的消息自己能看到对应窗口
self.open_or_focus_private_tab(target)
self._send_payload({'type': 3, 'target': target, 'msg': msg_text})
self.input_window.clear()
return
target = self.get_current_target()
if target == 'All':
payload = {'type': 'chat', 'msg': text} # 群发
else:
payload = {'type': 3, 'target': target, 'msg': text} # 头标签3代表私聊
self._send_payload(payload)
self.input_window.clear()
else:
self.running_status.setText("No Connection")
def send_file(self):
# 使用自定义的固定大小消息对话框
dialog = self.FixedMessageDialog(self, "功能提示", "此功能为会员专属，请联系管理员后使用")
dialog.exec()
def _send_payload(self, payload):
try:
send_bytes = json.dumps(payload).encode('utf-8') + b"__EOF__"
self.socket.sendall(send_bytes)
except Exception as e:
self.running_status.setText(f"Send Error: {e}")
def handle_incoming(self, msg):
print(f"【客户端接收】完整msg: {msg}, 消息内容: {msg.get('msg', '无')}")
msg_type = msg.get('type')
# ===== 处理用户名重复错误 =====
if msg_type == 5:
error_info = msg.get('msg', '用户名注册失败！')
# 使用自定义的固定大小消息对话框，并标记为错误类型
dialog = self.FixedMessageDialog(self, "注册失败", error_info, is_critical=True)
dialog.exec()
# 重置连接状态
self.terminate()
self.username.setEnabled(True)
return
# 1 & 2：更新列表和上线提示
if msg_type in [1, 2]:
users = msg.get('users', [])
joined_user = msg.get('username')
self.online_users = users.copy()
self.connection_list.clear()
for user in users:
item = QtWidgets.QListWidgetItem(f"[Online] {user}")
item.setForeground(QtGui.QColor("green"))
self.connection_list.addItem(item)
if joined_user:
self.print_sys_msg(f"用户 '{joined_user}' 上线了。", self.group_chat_window)
# 4：下线提示
elif msg_type == 4:
left_user = msg.get('username')
# 刷新列表
for i in range(self.connection_list.count()):
if left_user in self.connection_list.item(i).text():
self.connection_list.takeItem(i)
break
# ========== 更新在线用户列表（用户下线） ==========
if left_user in self.online_users:
self.online_users.remove(left_user)
# =======================================================
# 提示群聊和其他相关页面
self.print_sys_msg(f"用户 '{left_user}' 下线了。", self.group_chat_window)
if left_user in self.private_tabs:
self.print_sys_msg(f"对方已下线，无法继续发送消息。", self.private_tabs[left_user])
# 3 或 无头标签：处理文字聊天
else:
sender = msg.get('sender')
time_str = msg.get('time')
target = msg.get('target', 'All')
# 判断消息归属的标签页
if msg_type == 3 or (target and target != 'All'):
# 私聊消息
# 如果发送者是我自己，说明这是服务器回弹给我的记录，对方是 target
counterpart = target if sender == self.my_username else sender
target_tab = self.open_or_focus_private_tab(counterpart)
prefix = f"[私聊] {sender}"
else:
# 群聊消息
target_tab = self.group_chat_window
prefix = f"[群聊] {sender}"
if msg_type == 'chat' or msg_type == 3:
text = msg.get('msg')
# ① 去掉换行符，避免Linux下渲染截断
prefix = f"[私聊] {sender}" if (msg_type == 3 or (target and target != 'All')) else f"[群聊] {sender}"
display_msg = f"{prefix} ({time_str}): {text}"
# ② 创建Item并设置自适应大小（关键：解决...截断问题）
item = QtWidgets.QListWidgetItem(display_msg)
# 让Item宽度适配列表宽度，高度适配文本
item.setSizeHint(QtCore.QSize(target_tab.width(), item.sizeHint().height()))
# ③ 添加Item到列表
target_tab.addItem(item)
target_tab.scrollToBottom()
def print_sys_msg(self, text, target_tab):
"""向指定的标签页打印系统灰色提示"""
item = QtWidgets.QListWidgetItem(f">> System: {text}")
item.setForeground(QtGui.QColor("gray"))
target_tab.addItem(item)
target_tab.scrollToBottom()
def connect_server(self):
if self.thread is not None and self.thread.isRunning():
self.running_status.setText("Already Connected.")
return
ip = self.ip.text().strip()
port_str = self.port.text().strip()
self.my_username = self.username.text().strip()
if not self.my_username:
QtWidgets.QMessageBox.warning(self, "错误", "请输入用户名！")
return
try:
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.socket.connect((ip, int(port_str)))
self.thread = ClientThread(self.socket)
self.thread.incoming_message.connect(self.handle_incoming)
self.thread.start()
# 发送上线注册
self._send_payload({'type': 1, 'username': self.my_username})
self.running_status.setText("Connected successfully")
self.username.setEnabled(False)
except Exception as e:
self.running_status.setText(f"Connection Failed: {e}")
def terminate(self):
if self.thread is not None and self.thread.isRunning():
try:
self.socket.shutdown(socket.SHUT_RDWR)
self.socket.close()
except:
pass
self.thread.terminate()
self.thread.wait()
self.thread = None
# ===== 清空在线用户列表 =====
self.connection_list.clear()
# ===== 关闭所有私聊标签页，只保留群聊 =====
# 先删除字典中所有私聊标签页记录
self.private_tabs.clear()
# 关闭除了第一个（群聊）之外的所有标签页
while self.chat_tabs.count() > 1:
self.chat_tabs.removeTab(1)
self.clear_window()
self.running_status.setText("Connection Closed")
self.username.setEnabled(True)
def clear_window(self):
# 清除当前活跃标签页的聊天记录
index = self.chat_tabs.currentIndex()
if index >= 0:
self.chat_tabs.widget(index).clear()
self.input_window.clear()
def quit(self):
self.terminate()
sys.exit(0)
if __name__ == '__main__':
app = QtWidgets.QApplication(sys.argv)
client = ClientWindow()
client.show()
sys.exit(app.exec())
说明：客户端代码中
import chat_ui_3 as chat_ui，实际使用时请将chat.py重命名为chat_ui_3.py，或修改此处的导入名以保持一致。
5.3 服务器端业务逻辑 — server.py
# -*- coding: utf-8 -*-
import socket
import threading
import time
import json
import datetime
import _thread as thread
import base64
import hashlib
import hmac
from urllib.parse import urlparse
import ssl
from time import mktime
from urllib.parse import urlencode
from wsgiref.handlers import format_date_time
import websocket
from PyQt6 import QtCore, QtGui, QtWidgets
# ====================== 星火AI核心逻辑（复用原代码） ======================
class SparkAI:
def __init__(self, appid, api_key, api_secret, spark_url, domain):
self.APPID = appid
self.APIKey = api_key
self.APISecret = api_secret
self.Spark_url = spark_url
self.domain = domain
self.group_chat_history = [] # 群聊对话上下文
self.private_chat_history = {} # 私聊上下文：{用户名: 历史列表}
self.answer = ""
self.is_first_content = False
def create_url(self):
now = datetime.datetime.now()
date = format_date_time(mktime(now.timetuple()))
signature_origin = f"host: {urlparse(self.Spark_url).netloc}\ndate: {date}\nGET {urlparse(self.Spark_url).path} HTTP/1.1"
signature_sha = hmac.new(self.APISecret.encode('utf-8'), signature_origin.encode('utf-8'),
digestmod=hashlib.sha256).digest()
signature_sha_base64 = base64.b64encode(signature_sha).decode(encoding='utf-8')
authorization_origin = f'api_key="{self.APIKey}", algorithm="hmac-sha256", headers="host date request-line", signature="{signature_sha_base64}"'
authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')
v = {
"authorization": authorization,
"date": date,
"host": urlparse(self.Spark_url).netloc
}
url = self.Spark_url + '?' + urlencode(v)
return url
def gen_params(self, is_private=False, private_user=None):
if is_private and private_user:
history = self.private_chat_history.get(private_user, [])
else:
history = self.group_chat_history
return {
"header": {"app_id": self.APPID, "uid": private_user if is_private else "chatroom_ai"},
"parameter": {"chat": {"domain": self.domain, "temperature": 1.2, "max_tokens": 32768}},
"payload": {"message": {"text": history}}
}
def get_content_length(self, is_private=False, private_user=None):
if is_private and private_user:
history = self.private_chat_history.get(private_user, [])
else:
history = self.group_chat_history
return sum(len(item["content"]) for item in history)
def check_history_length(self, is_private=False, private_user=None):
if is_private and private_user:
history = self.private_chat_history.get(private_user, [])
while self.get_content_length(is_private=True, private_user=private_user) > 8000:
del history[0]
self.private_chat_history[private_user] = history
else:
while self.get_content_length() > 8000:
del self.group_chat_history[0]
def get_ai_response(self, question, is_private=False, private_user=None):
self.answer = ""
self.is_first_content = False
if is_private and private_user:
if private_user not in self.private_chat_history:
self.private_chat_history[private_user] = []
self.private_chat_history[private_user].append({"role": "user", "content": question})
self.check_history_length(is_private=True, private_user=private_user)
else:
self.group_chat_history.append({"role": "user", "content": question})
self.check_history_length()
def on_message(ws, message):
data = json.loads(message)
code = data['header']['code']
if code != 0:
print(f'AI请求错误: {code}, {data}')
ws.close()
return
choices = data["payload"]["choices"]
status = choices["status"]
text = choices['text'][0]
if 'content' in text and text['content']:
content = text["content"]
self.answer += content
if status == 2:
ws.close()
def on_error(ws, error):
print(f'AI连接错误: {error}')
def on_close(ws, *args):
pass
def on_open(ws):
thread.start_new_thread(lambda: ws.send(json.dumps(self.gen_params(is_private, private_user))), ())
ws_url = self.create_url()
ws = websocket.WebSocketApp(ws_url, on_message=on_message, on_error=on_error, on_close=on_close,
on_open=on_open)
ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
if is_private and private_user:
self.private_chat_history[private_user].append({"role": "assistant", "content": self.answer})
else:
self.group_chat_history.append({"role": "assistant", "content": self.answer})
return self.answer
# ====================== 客户端处理线程（复用原代码） ======================
class ServerThread(threading.Thread):
def __init__(self, server_instance, client_socket, address):
threading.Thread.__init__(self)
self.server = server_instance
self.client = client_socket
self.address = address
self.buffer = b""
def run(self):
# 仅在服务器运行中时处理消息
while self.server.is_running:
try:
raw_data = self.client.recv(1024 * 1024)
if not raw_data:
break
self.buffer += raw_data
while b"__EOF__" in self.buffer and self.server.is_running:
msg_bytes, self.buffer = self.buffer.split(b"__EOF__", 1)
if msg_bytes.strip():
msg_str = msg_bytes.decode('utf-8')
self.process_message(msg_str)
except Exception as e:
print(f"Client Exception {self.address}: {e}")
break
self.server.remove_client(self.client)
def process_message(self, msg_str):
data = json.loads(msg_str)
print(f"【服务器接收】原始消息: {data}") # 确认客户端消息是否完整
msg_type = data.get('type')
target = data.get('target')
sender_name = self.server.clients.get(self.client, {}).get('username', f"{self.address[0]}:{self.address[1]}")
data['sender'] = sender_name
data['time'] = datetime.datetime.fromtimestamp(time.time()).strftime('%H:%M:%S')
print(f"【服务器转发】处理后消息: {data}") # 确认转发前消息未被篡改
if msg_type in [1, 2]:
new_username = data.get('username')
if new_username == 'SparkAI':
error_msg = {
'type': 5,
'msg': '用户名"SparkAI"为系统保留名，无法使用！',
'time': datetime.datetime.fromtimestamp(time.time()).strftime('%H:%M:%S')
}
error_bytes = json.dumps(error_msg).encode('utf-8') + b"__EOF__"
self.client.sendall(error_bytes)
self.server.remove_client(self.client)
return
for c, info in self.server.clients.items():
if c != self.server.ai_virtual_client and info.get('username') == new_username:
error_msg = {
'type': 5,
'msg': f'用户名"{new_username}"已被占用，请更换！',
'time': datetime.datetime.fromtimestamp(time.time()).strftime('%H:%M:%S')
}
error_bytes = json.dumps(error_msg).encode('utf-8') + b"__EOF__"
self.client.sendall(error_bytes)
self.server.remove_client(self.client)
return
self.server.clients[self.client]['username'] = new_username
self.server.broadcast_users(action_type=2, changed_user=new_username)
elif msg_type == 3 or (target and target != 'All'):
send_bytes = json.dumps(data).encode('utf-8') + b"__EOF__"
for c, info in self.server.clients.items():
if info['username'] == target or info['username'] == sender_name:
try:
c.sendall(send_bytes)
except:
pass
if target == 'SparkAI':
question = data.get('msg', '').strip()
if not question:
return
def ai_private_reply_task():
try:
ai_response = self.server.spark_ai.get_ai_response(
question,
is_private=True,
private_user=sender_name
)
ai_msg = {
'type': 3,
'target': sender_name,
'sender': 'SparkAI',
'msg': ai_response,
'time': datetime.datetime.fromtimestamp(time.time()).strftime('%H:%M:%S')
}
ai_send_bytes = json.dumps(ai_msg).encode('utf-8') + b"__EOF__"
for c, info in self.server.clients.items():
if info['username'] == sender_name:
try:
c.sendall(ai_send_bytes)
except:
pass
except Exception as e:
print(f'AI私聊回复失败: {e}')
threading.Thread(target=ai_private_reply_task).start()
else:
send_bytes = json.dumps(data).encode('utf-8') + b"__EOF__"
for c in self.server.clients.keys():
if c != self.server.ai_virtual_client:
try:
c.sendall(send_bytes)
except:
pass
msg_content = data.get('msg', '').strip()
ai_call_prefixes = ['@SparkAI', 'AI：', 'AI:', '机器人：', '机器人:']
is_call_ai = any(msg_content.startswith(prefix) for prefix in ai_call_prefixes)
if is_call_ai and msg_type == 'chat':
for prefix in ai_call_prefixes:
if msg_content.startswith(prefix):
question = msg_content[len(prefix):].strip()
break
if not question:
return
def ai_reply_task():
try:
ai_response = self.server.spark_ai.get_ai_response(
question,
is_private=False
)
ai_msg = {
'type': 'chat',
'target': 'All',
'sender': 'SparkAI',
'msg': ai_response,
'time': datetime.datetime.fromtimestamp(time.time()).strftime('%H:%M:%S')
}
ai_send_bytes = json.dumps(ai_msg).encode('utf-8') + b"__EOF__"
for c in self.server.clients.keys():
if c != self.server.ai_virtual_client:
try:
c.sendall(ai_send_bytes)
except:
pass
except Exception as e:
print(f'AI群聊回复失败: {e}')
threading.Thread(target=ai_reply_task).start()
# ====================== 可控制启停的服务器核心类 ======================
class ChatServer(QtCore.QObject):
# 定义信号，用于更新GUI状态
status_signal = QtCore.pyqtSignal(str)
ip_port_signal = QtCore.pyqtSignal(str, str)
def __init__(self):
super().__init__()
self.server_socket = None
self.host = ""
self.port = 12345
self.clients = {}
self.is_running = False # 运行状态标志
self.listen_thread = None # 监听线程
# 初始化星火AI（替换为你的密钥）
self.spark_ai = SparkAI(
appid="ea155baa", # 你的APPID
api_key="7b210bf6d6a650f250200f97f6c8f112", # 你的APIKey
api_secret="ZTI5YmEzMTU0MzVjNGMzM2Y4YWRjYmJh", # 你的APISecret
spark_url="wss://spark-api.xf-yun.com/x2",
domain="spark-x"
)
# 添加SparkAI虚拟用户
self.ai_virtual_client = object()
self.clients[self.ai_virtual_client] = {'address': ('AI', 0), 'username': 'SparkAI'}
# 自动获取本机IP
self.get_local_ip()
def get_local_ip(self):
"""获取本机局域网IP"""
try:
# 方法2（备选）：通过连接外部地址获取真实出口IP
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(("8.8.8.8", 80))
self.host = s.getsockname()[0]
s.close()
except Exception as e:
self.host = "127.0.0.1"
print(f"获取IP失败: {e}")
# 发送IP和端口到GUI
self.ip_port_signal.emit(self.host, str(self.port))
def start_server(self):
"""启动服务器"""
if self.is_running:
self.status_signal.emit("服务器已在运行")
return
try:
# 创建socket并绑定
self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 设置端口复用，避免重启时端口占用
self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.server_socket.bind((self.host, self.port))
self.server_socket.listen(10)
self.is_running = True
# 启动监听线程
self.listen_thread = threading.Thread(target=self.listen_clients)
self.listen_thread.daemon = True # 守护线程，退出时自动关闭
self.listen_thread.start()
# 广播SparkAI上线
self.broadcast_users(action_type=2, changed_user='SparkAI')
self.status_signal.emit(f"服务器已启动 | IP: {self.host} | 端口: {self.port}")
except Exception as e:
self.status_signal.emit(f"启动失败: {str(e)}")
self.is_running = False
def listen_clients(self):
"""监听客户端连接（运行在独立线程）"""
while self.is_running:
try:
# 设置超时，避免阻塞无法退出
self.server_socket.settimeout(1.0)
con, addr = self.server_socket.accept()
print(f"新连接: {addr}")
self.clients[con] = {'address': addr, 'username': None}
# 启动客户端处理线程
client_thread = ServerThread(self, con, addr)
client_thread.daemon = True
client_thread.start()
except socket.timeout:
# 超时只是为了检查is_running，继续循环
continue
except Exception as e:
if self.is_running:
self.status_signal.emit(f"监听错误: {str(e)}")
break
def stop_server(self):
"""停止服务器"""
if not self.is_running:
self.status_signal.emit("服务器未运行")
return
self.is_running = False
try:
# 关闭监听socket
if self.server_socket:
self.server_socket.close()
self.server_socket = None
# 关闭所有客户端连接
for client in list(self.clients.keys()):
if client != self.ai_virtual_client:
try:
client.shutdown(socket.SHUT_RDWR)
client.close()
except:
pass
self.clients.clear()
# 重新添加AI虚拟用户
self.clients[self.ai_virtual_client] = {'address': ('AI', 0), 'username': 'SparkAI'}
self.status_signal.emit("服务器已停止")
except Exception as e:
self.status_signal.emit(f"停止失败: {str(e)}")
def broadcast_users(self, action_type, changed_user):
"""广播在线用户列表"""
if not self.is_running:
return
users = [info['username'] for info in self.clients.values() if info['username'] is not None]
msg = {'type': action_type, 'users': users, 'username': changed_user}
send_bytes = json.dumps(msg).encode('utf-8') + b"__EOF__"
for c in self.clients.keys():
if c != self.ai_virtual_client:
try:
c.sendall(send_bytes)
except:
pass
def remove_client(self, client):
"""移除客户端"""
if client == self.ai_virtual_client or not self.is_running:
return
if client in self.clients:
offline_user = self.clients[client].get('username')
print(f"客户端断开: {offline_user}")
del self.clients[client]
if offline_user:
self.broadcast_users(action_type=4, changed_user=offline_user)
try:
client.close()
except:
pass
# ====================== 服务器GUI界面（新增复制IP按钮） ======================
class ServerGUI(QtWidgets.QMainWindow):
def __init__(self):
super().__init__()
self.setWindowTitle("局域网聊天服务器")
self.resize(500, 250)
self.central_widget = QtWidgets.QWidget()
self.setCentralWidget(self.central_widget)
# 创建布局
self.layout = QtWidgets.QVBoxLayout(self.central_widget)
self.layout.setContentsMargins(30, 30, 30, 30)
self.layout.setSpacing(20)
# 1. IP和端口显示区域
self.info_group = QtWidgets.QGroupBox("服务器信息")
self.info_layout = QtWidgets.QGridLayout(self.info_group)
self.info_layout.setContentsMargins(20, 15, 20, 15)
# IP标签
self.ip_label = QtWidgets.QLabel("服务器IP:")
self.ip_value = QtWidgets.QLabel("")
self.ip_value.setStyleSheet("color: #2E8B57; font-weight: bold;")
# 新增：复制IP按钮
self.copy_ip_btn = QtWidgets.QPushButton("复制IP")
self.copy_ip_btn.setStyleSheet("""
padding: 4px 12px;
font-size: 12px;
background-color: #2196F3;
color: white;
border: none;
border-radius: 4px;
""")
self.copy_ip_btn.clicked.connect(self.copy_ip_to_clipboard)
self.info_layout.addWidget(self.ip_label, 0, 0, 1, 1)
self.info_layout.addWidget(self.ip_value, 0, 1, 1, 3)
self.info_layout.addWidget(self.copy_ip_btn, 0, 4, 1, 1) # 把复制按钮加到IP行第5列
# 端口标签
self.port_label = QtWidgets.QLabel("端口号:")
self.port_value = QtWidgets.QLabel("12345")
self.port_value.setStyleSheet("color: #2E8B57; font-weight: bold;")
self.info_layout.addWidget(self.port_label, 1, 0, 1, 1)
self.info_layout.addWidget(self.port_value, 1, 1, 1, 3)
self.layout.addWidget(self.info_group)
# 2. 控制按钮区域
self.btn_layout = QtWidgets.QHBoxLayout()
self.start_btn = QtWidgets.QPushButton("启动服务器")
self.start_btn.setStyleSheet("background-color: #4CAF50; color: white; padding: 8px 20px; font-size: 14px;")
self.start_btn.clicked.connect(self.on_start_click)
self.stop_btn = QtWidgets.QPushButton("停止服务器")
self.stop_btn.setStyleSheet("background-color: #f44336; color: white; padding: 8px 20px; font-size: 14px;")
self.stop_btn.clicked.connect(self.on_stop_click)
self.stop_btn.setEnabled(False) # 初始禁用
self.btn_layout.addStretch()
self.btn_layout.addWidget(self.start_btn)
self.btn_layout.addWidget(self.stop_btn)
self.btn_layout.addStretch()
self.layout.addLayout(self.btn_layout)
# 3. 状态显示区域
self.status_label = QtWidgets.QLabel("状态: 未运行")
self.status_label.setStyleSheet("color: #666; font-size: 14px;")
self.layout.addWidget(self.status_label, alignment=QtCore.Qt.AlignmentFlag.AlignCenter)
# 初始化服务器核心
self.chat_server = ChatServer()
# 绑定信号
self.chat_server.status_signal.connect(self.update_status)
self.chat_server.ip_port_signal.connect(self.update_ip_port)
# 自动获取IP
self.chat_server.get_local_ip()
def copy_ip_to_clipboard(self):
"""新增：复制IP地址到剪贴板"""
ip_text = self.ip_value.text()
if ip_text:
# 获取系统剪贴板并设置内容
clipboard = QtWidgets.QApplication.clipboard()
clipboard.setText(ip_text)
# 显示复制成功提示（2秒后恢复原状态）
original_status = self.status_label.text()
self.status_label.setText(f"状态: 已复制IP {ip_text} 到剪贴板 ")
self.status_label.setStyleSheet("color: #2196F3; font-size: 14px; font-weight: bold;")
# 2秒后恢复原状态
QtCore.QTimer.singleShot(2000, lambda: self.update_status(original_status))
def update_ip_port(self, ip, port):
"""更新IP和端口显示"""
self.ip_value.setText(ip)
self.port_value.setText(port)
def update_status(self, status):
"""更新状态显示"""
self.status_label.setText(f"状态: {status}")
# 更新按钮状态
if "已启动" in status:
self.start_btn.setEnabled(False)
self.stop_btn.setEnabled(True)
self.status_label.setStyleSheet("color: #4CAF50; font-size: 14px; font-weight: bold;")
elif "已停止" in status:
self.start_btn.setEnabled(True)
self.stop_btn.setEnabled(False)
self.status_label.setStyleSheet("color: #f44336; font-size: 14px; font-weight: bold;")
elif "已复制IP" not in status: # 排除复制提示的情况
self.status_label.setStyleSheet("color: #ff9800; font-size: 14px; font-weight: bold;")
def on_start_click(self):
"""启动服务器按钮点击"""
self.chat_server.start_server()
def on_stop_click(self):
"""停止服务器按钮点击"""
self.chat_server.stop_server()
def closeEvent(self, event):
"""窗口关闭时停止服务器"""
if self.chat_server.is_running:
self.chat_server.stop_server()
event.accept()
# ====================== 主函数 ======================
if __name__ == '__main__':
import sys
app = QtWidgets.QApplication(sys.argv)
server_gui = ServerGUI()
server_gui.show()
sys.exit(app.exec())
六、运行方式
6.1 启动服务器
python server.py
点击界面上的「启动服务器」按钮，界面会显示本机局域网 IP 和端口（默认 12345），可点击「复制IP」将 IP 复制到剪贴板，方便分享给局域网内的其他用户。
6.2 启动客户端
python client.py
在客户端界面填写：
- IP：服务器所在主机的局域网 IP
- Port：12345
- Name：自定义用户名（不能与已有用户重复，也不能使用保留名
SparkAI）
点击「Connect」即可连接。
6.3 使用 AI 机器人
- 群聊触发：在群聊输入框输入
@SparkAI 你好或AI: 今天天气如何等前缀，AI 会自动回复到群聊。 - 私聊触发：双击在线用户列表中的
SparkAI，在私聊标签页直接发送消息即可与 AI 一对一对话。
七、关键设计点说明
7.1 粘包处理
TCP 是流式协议，没有消息边界。本项目在每条 JSON 消息末尾追加 __EOF__ 作为分隔符，接收端在缓冲区中按 __EOF__ 切分，确保完整解析每条消息。
7.2 线程模型
- 服务器端：主线程运行 GUI，监听线程
accept连接，每个客户端分配一个ServerThread处理消息，AI 调用在独立线程中执行避免阻塞。 - 客户端：主线程运行 GUI，
ClientThread（QThread）负责接收消息，通过 PyQt 信号将消息投递回主线程处理，保证 UI 操作线程安全。
7.3 AI 上下文管理
SparkAI 类为群聊和每个私聊用户分别维护对话历史列表，当历史内容总字符数超过 8000 时，从头部删除最早的消息，避免超出模型上下文窗口限制。
7.4 用户名唯一性校验
服务器在收到 type=1/2 注册消息时，遍历所有已注册用户检查重名，同时禁止使用保留名 SparkAI。冲突时返回 type=5 错误消息并断开该连接。
7.5 AI 虚拟客户端
服务器在 clients 字典中预先注册一个 ai_virtual_client（使用 object() 作为 key），用户名为 SparkAI。这样 AI 在用户列表中显示为在线用户，但不会真正建立 socket 连接，广播消息时也会跳过它。
八、总结
本项目完整实现了一个具备群聊、私聊、AI 机器人集成、用户管理等功能的局域网聊天室，代码结构清晰，适合作为 PyQt6 网络编程、多线程编程以及大模型 API 集成的学习参考。后续可扩展的方向包括：
- 文件传输功能（目前为占位）
- 消息持久化存储
- 用户头像与富文本消息
- 群组管理（创建/加入/退出群组）
- AI 多轮对话上下文持久化
如需使用星火 AI，请前往讯飞开放平台申请自己的 APPID、APIKey、APISecret，并替换 server.py 中 SparkAI 初始化的对应参数。
中国智能体开发者社区，聚焦智能体与大模型开发，提供前沿资讯、实用工具链、开源项目及行业案例。通过技术沙龙、开发者大赛等活动，促进经验交流与协作，助力开发者快速构建创新智能应用。
更多推荐
5
0- 0
已为社区贡献1条内容
所有评论(0)