376 lines
14 KiB
Python
376 lines
14 KiB
Python
import tkinter as tk
|
|
from tkinter import messagebox, scrolledtext
|
|
import json
|
|
import threading
|
|
import websockets
|
|
import asyncio
|
|
import requests
|
|
from urllib.parse import urlparse
|
|
from datetime import datetime
|
|
|
|
class ChatClient:
|
|
def __init__(self):
|
|
# 客户端配置
|
|
self.client_version = "1.3"
|
|
self.min_server_version = "1.3"
|
|
self.language = "Python"
|
|
|
|
# 网络组件
|
|
self.websocket = None
|
|
self.loop = None
|
|
self.ws_thread = None
|
|
self.current_user = None
|
|
self.server_url = ""
|
|
self.connected = False
|
|
self.ws_url = "" # 新增的初始化
|
|
|
|
self.ws_url = input('请选择连接IP')
|
|
self.server_url = self.ws_url
|
|
|
|
# 初始化UI
|
|
self.root = tk.Tk()
|
|
self.root.title(f"聊天客户端 v{self.client_version}")
|
|
self.setup_ui()
|
|
self.root.protocol("WM_DELETE_WINDOW", self.cleanup)
|
|
|
|
def setup_ui(self):
|
|
# 登录框架
|
|
self.login_frame = tk.Frame(self.root, padx=20, pady=20)
|
|
self.login_frame.pack()
|
|
|
|
# 服务器配置
|
|
tk.Label(self.login_frame, text="服务器地址:").grid(row=0, column=0, sticky="w")
|
|
self.server_entry = tk.Entry(self.login_frame, width=30)
|
|
self.server_entry.grid(row=0, column=1, sticky="ew")
|
|
self.server_entry.insert(0, "http://127.0.0.1:3000")
|
|
|
|
# 认证信息
|
|
tk.Label(self.login_frame, text="用户名:").grid(row=1, column=0, sticky="w")
|
|
self.username_entry = tk.Entry(self.login_frame)
|
|
self.username_entry.grid(row=1, column=1, sticky="ew")
|
|
|
|
tk.Label(self.login_frame, text="密码:").grid(row=2, column=0, sticky="w")
|
|
self.password_entry = tk.Entry(self.login_frame, show="*")
|
|
self.password_entry.grid(row=2, column=1, sticky="ew")
|
|
|
|
# 操作按钮
|
|
btn_frame = tk.Frame(self.login_frame)
|
|
btn_frame.grid(row=3, columnspan=2, pady=10)
|
|
|
|
tk.Button(btn_frame, text="登录", command=self.on_login).pack(side="left", padx=5)
|
|
tk.Button(btn_frame, text="注册", command=self.on_register).pack(side="left", padx=5)
|
|
|
|
# 状态显示
|
|
self.status_label = tk.Label(self.login_frame, text="等待连接...", fg="gray")
|
|
self.status_label.grid(row=4, columnspan=2)
|
|
|
|
# 聊天主界面（初始隐藏）
|
|
self.chat_frame = tk.Frame(self.root)
|
|
|
|
# 消息显示区域
|
|
self.chat_display = scrolledtext.ScrolledText(
|
|
self.chat_frame,
|
|
wrap=tk.WORD,
|
|
width=60,
|
|
height=20,
|
|
state='disabled'
|
|
)
|
|
self.chat_display.pack(pady=10, fill=tk.BOTH, expand=True)
|
|
|
|
# 消息输入区域
|
|
input_frame = tk.Frame(self.chat_frame)
|
|
input_frame.pack(fill=tk.X, pady=5)
|
|
|
|
self.message_entry = tk.Entry(input_frame)
|
|
self.message_entry.pack(side=tk.LEFT, fill=tk.X, expand=True)
|
|
self.message_entry.bind("<Return>", lambda e: self.send_message())
|
|
|
|
tk.Button(input_frame, text="发送", command=self.send_message).pack(side=tk.LEFT)
|
|
|
|
# 底部状态栏
|
|
self.connection_status = tk.Label(self.chat_frame, text="未连接", fg="red")
|
|
self.connection_status.pack(side=tk.BOTTOM, fill=tk.X)
|
|
|
|
def authenticate(self, action, username, password):
|
|
"""处理认证逻辑（登录/注册）"""
|
|
try:
|
|
# 规范化URL
|
|
self.server_url = self.server_entry.get().strip()
|
|
if not self.server_url.startswith(("http://", "https://")):
|
|
self.server_url = "http://" + self.server_url
|
|
|
|
# 准备请求
|
|
url = f"{self.server_url.rstrip('/')}/api/{action}"
|
|
headers = {"Content-Type": "application/json"}
|
|
payload = json.dumps({"username": username, "password": password})
|
|
|
|
# 发送请求
|
|
self.update_status(f"{action}中...", "blue")
|
|
response = requests.post(url, headers=headers, data=payload, timeout=10)
|
|
|
|
# 检查响应
|
|
if response.status_code != 200:
|
|
error_msg = f"HTTP {response.status_code}"
|
|
if response.text:
|
|
error_msg += f" - {response.text[:100]}{'...' if len(response.text)>100 else ''}"
|
|
raise ConnectionError(error_msg)
|
|
|
|
# 解析JSON
|
|
try:
|
|
data = response.json()
|
|
except json.JSONDecodeError:
|
|
raise ValueError(f"无效的JSON响应: {response.text[:100]}...")
|
|
|
|
# 处理结果
|
|
if not data.get("success"):
|
|
raise ValueError(data.get("message", f"{action}失败"))
|
|
|
|
# 认证成功
|
|
self.current_user = {
|
|
"username": username,
|
|
"is_admin": data.get("is_admin", False)
|
|
}
|
|
self.start_chat_session()
|
|
|
|
except Exception as e:
|
|
error_msg = f"{action}错误: {str(e)}"
|
|
self.update_status(error_msg, "red")
|
|
print(f"[DEBUG] {error_msg}\nURL: {url}\nResponse: {getattr(response, 'text', '')}")
|
|
|
|
def start_chat_session(self):
|
|
"""启动聊天会话"""
|
|
self.login_frame.pack_forget()
|
|
self.chat_frame.pack(fill=tk.BOTH, expand=True)
|
|
self.update_connection_status("连接中...", "orange")
|
|
|
|
# 启动WebSocket连接
|
|
self.loop = asyncio.new_event_loop()
|
|
self.ws_thread = threading.Thread(target=self.run_websocket, daemon=True)
|
|
self.ws_thread.start()
|
|
|
|
def run_websocket(self):
|
|
"""运行WebSocket客户端"""
|
|
asyncio.set_event_loop(self.loop)
|
|
self.loop.run_until_complete(self.websocket_handler())
|
|
|
|
async def websocket_handler(self):
|
|
"""处理WebSocket连接"""
|
|
try:
|
|
# 建立连接前设置ws_url
|
|
self.ws_url = self.server_url.replace("http", "ws") + "/chat"
|
|
|
|
async with websockets.connect(self.ws_url) as websocket:
|
|
self.websocket = websocket
|
|
self.connected = True
|
|
self.update_connection_status("已连接", "green")
|
|
|
|
# 1. 发送客户端信息
|
|
await websocket.send(json.dumps({
|
|
"type": "client-info",
|
|
"version": self.client_version,
|
|
"language": self.language
|
|
}))
|
|
|
|
# 2. 发送认证信息
|
|
await websocket.send(json.dumps({
|
|
"type": "login",
|
|
"username": self.current_user["username"],
|
|
"password": self.password_entry.get()
|
|
}))
|
|
|
|
# 3. 消息处理循环
|
|
while self.connected:
|
|
message = await websocket.recv()
|
|
self.handle_server_message(json.loads(message))
|
|
|
|
except Exception as e:
|
|
error_msg = f"连接错误: {str(e)}"
|
|
self.display_message(error_msg, "system")
|
|
self.update_connection_status("连接断开", "red")
|
|
finally:
|
|
self.connected = False
|
|
|
|
def handle_server_message(self, message):
|
|
"""处理服务端消息"""
|
|
msg_type = message.get("type")
|
|
|
|
if msg_type == "error":
|
|
self.display_message(message.get("message", "未知错误"), "error")
|
|
|
|
elif msg_type == "server-info":
|
|
server_version = message.get("version")
|
|
if server_version != self.client_version:
|
|
warn_msg = f"版本警告: 服务端({server_version})≠客户端({self.client_version})"
|
|
self.display_message(warn_msg, "warning")
|
|
|
|
elif msg_type == "history":
|
|
for msg in message.get("data", []):
|
|
self.display_history_message(msg)
|
|
|
|
elif msg_type == "message":
|
|
self.display_message(
|
|
message.get("content"),
|
|
"admin" if message.get("is_admin") else "user",
|
|
message.get("sender")
|
|
)
|
|
|
|
elif msg_type == "system":
|
|
self.display_message(message.get("message"), "system")
|
|
|
|
def send_message(self):
|
|
"""发送消息到服务器"""
|
|
message = self.message_entry.get().strip()
|
|
if not message or not self.connected:
|
|
return
|
|
|
|
try:
|
|
# 处理命令
|
|
if message.startswith("///"):
|
|
self.handle_command(message)
|
|
else:
|
|
# 发送普通消息
|
|
asyncio.run_coroutine_threadsafe(
|
|
self.send_websocket_message({
|
|
"type": "message",
|
|
"content": message
|
|
}),
|
|
self.loop
|
|
)
|
|
|
|
self.message_entry.delete(0, tk.END)
|
|
|
|
|
|
except Exception as e:
|
|
self.display_message(f"发送失败: {str(e)}", "error")
|
|
|
|
async def send_websocket_message(self, message):
|
|
"""通过WebSocket发送消息"""
|
|
if self.websocket:
|
|
await self.websocket.send(json.dumps(message))
|
|
|
|
def handle_command(self, command):
|
|
"""处理管理员命令"""
|
|
if not self.current_user.get("is_admin"):
|
|
self.display_message("需要管理员权限", "error")
|
|
return
|
|
|
|
# 示例命令处理
|
|
if command == "///help":
|
|
help_text = [
|
|
"可用命令:",
|
|
"///help - 显示帮助",
|
|
"///me - 显示我的信息",
|
|
"///list - 列出在线用户",
|
|
"///op <用户> - 授予管理员权限",
|
|
"///deop <用户> - 取消管理员权限",
|
|
"///kick <用户> - 踢出用户"
|
|
]
|
|
self.display_message("\n".join(help_text), "system")
|
|
else:
|
|
asyncio.run_coroutine_threadsafe(
|
|
self.send_websocket_message({
|
|
"type": "message",
|
|
"content": command
|
|
}),
|
|
self.loop
|
|
)
|
|
|
|
def display_message(self, message, msg_type="normal", sender=None):
|
|
"""显示消息到聊天区域"""
|
|
colors = {
|
|
"system": "blue",
|
|
"error": "red",
|
|
"warning": "orange",
|
|
"admin": "purple",
|
|
"user": "black",
|
|
"normal": "green"
|
|
}
|
|
|
|
self.root.after(0, lambda: self._append_message(
|
|
text=message,
|
|
color=colors.get(msg_type, "black"),
|
|
sender=sender
|
|
))
|
|
|
|
def display_history_message(self, message):
|
|
"""显示历史消息"""
|
|
self._append_message(
|
|
text=message.get("content", ""),
|
|
color="gray",
|
|
sender=message.get("sender"),
|
|
timestamp=message.get("timestamp")
|
|
)
|
|
|
|
def _append_message(self, text, color, sender=None, timestamp=None):
|
|
"""内部方法：添加消息到文本框"""
|
|
self.chat_display.config(state='normal')
|
|
|
|
if timestamp:
|
|
time_str = datetime.fromisoformat(timestamp).strftime("%H:%M:%S")
|
|
self.chat_display.insert(tk.END, f"[{time_str}] ", "time")
|
|
|
|
if sender:
|
|
self.chat_display.insert(tk.END, f"{sender}: ", "sender")
|
|
|
|
self.chat_display.insert(tk.END, text + "\n", color)
|
|
self.chat_display.config(state='disabled')
|
|
self.chat_display.see(tk.END)
|
|
|
|
# 配置标签样式
|
|
for tag in ["time", "sender", color]:
|
|
self.chat_display.tag_config(tag, foreground=color)
|
|
|
|
def update_status(self, text, color="black"):
|
|
"""更新登录界面状态"""
|
|
self.root.after(0, lambda: self.status_label.config(text=text, fg=color))
|
|
|
|
def update_connection_status(self, text, color):
|
|
"""更新连接状态"""
|
|
self.root.after(0, lambda: self.connection_status.config(text=text, fg=color))
|
|
|
|
def on_login(self):
|
|
"""登录按钮事件"""
|
|
username = self.username_entry.get().strip()
|
|
password = self.password_entry.get().strip()
|
|
|
|
if not all([self.server_entry.get(), username, password]):
|
|
self.update_status("所有字段不能为空", "red")
|
|
return
|
|
|
|
threading.Thread(
|
|
target=self.authenticate,
|
|
args=("login", username, password),
|
|
daemon=True
|
|
).start()
|
|
|
|
def on_register(self):
|
|
"""注册按钮事件"""
|
|
username = self.username_entry.get().strip()
|
|
password = self.password_entry.get().strip()
|
|
|
|
if not all([self.server_entry.get(), username, password]):
|
|
self.update_status("所有字段不能为空", "red")
|
|
return
|
|
|
|
threading.Thread(
|
|
target=self.authenticate,
|
|
args=("register", username, password),
|
|
daemon=True
|
|
).start()
|
|
|
|
def cleanup(self):
|
|
"""清理资源"""
|
|
self.connected = False
|
|
try:
|
|
if self.websocket:
|
|
asyncio.run_coroutine_threadsafe(self.websocket.close(), self.loop)
|
|
if self.loop:
|
|
self.loop.call_soon_threadsafe(self.loop.stop)
|
|
except:
|
|
pass
|
|
self.root.destroy()
|
|
|
|
if __name__ == "__main__":
|
|
# 运行客户端
|
|
client = ChatClient()
|
|
client.root.mainloop() |