websocket小工具

websocketman不注册老是自动跳到关于页面，实在受不了了，自己写了一个。  有需要的网盘自取

源码部分


https://wwtg.lanzouo.com/b026xsrlxc
密码:8n2fimport websocket
import threading
import tkinter as tk
from tkinter import scrolledtext, ttk, font
import json

1. class WebSocketApp:
2.     def __init__(self, root):
3.         self.root = root
4.         self.root.title("WebSocket 客户端")
5.         self.ws = None
7.         # 设置窗口大小
8.         self.root.geometry("800x600")
10.         self.custom_font = font.Font(family="Courier New", size=12)
12.         # 1. WebSocket 地址输入框和连接按钮
13.         self.url_frame = ttk.LabelFrame(root, text="连接地址", padding=10)
14.         self.url_frame.pack(fill="x", padx=10, pady=10)
16.         self.url_entry = tk.Entry(self.url_frame, font=self.custom_font)
17.         self.url_entry.insert(0, "ws://127.0.0.1:8000/ws")
18.         self.url_entry.pack(side="left", fill="x", expand=True, padx=5)
20.         self.connect_btn = tk.Button(self.url_frame, text="开始连接", command=self.connect, font=self.custom_font)
21.         self.connect_btn.pack(side="left", padx=5)
23.         self.close_btn = tk.Button(self.url_frame, text="关闭连接", command=self.close, font=self.custom_font)
24.         self.close_btn.pack(side="left", padx=5)
26.         # 2. 发送区：文本域 + 按钮
27.         self.send_frame = ttk.LabelFrame(root, text="发送区", padding=10)
28.         self.send_frame.pack(fill="both", padx=10, pady=10, expand=True)
30.         self.send_text = scrolledtext.ScrolledText(
31.             self.send_frame,
32.             width=60,
33.             height=8,
34.             font=self.custom_font,
35.             wrap=tk.WORD,
36.             spacing1=2,
37.             spacing2=2,
38.             spacing3=2,
39.             undo=True  # 启用撤销功能
40.         )
41.         self.send_text.pack(side="left", fill="both", expand=True, padx=(0, 5))
43.         # 绑定 Ctrl+Z 和 Ctrl+Y 快捷键
44.         self.send_text.bind('', lambda event: self.send_text.edit_undo())
45.         self.send_text.bind('', lambda event: self.send_text.edit_redo())
46.         self.send_text.bind('', lambda event: self.send_text.tag_add(tk.SEL, "1.0", tk.END) or "break")
48.         self.send_btn_frame = tk.Frame(self.send_frame)
49.         self.send_btn_frame.pack(side="right", fill="y")
51.         self.format_btn = tk.Button(self.send_btn_frame, text="格式化", command=self.format_text, font=self.custom_font)
52.         self.format_btn.pack(pady=5)
54.         self.send_btn = tk.Button(self.send_btn_frame, text="发送", command=self.send_message, font=self.custom_font)
55.         self.send_btn.pack(pady=5)
57.         self.clear_send_btn = tk.Button(self.send_btn_frame, text="清空", command=self.clear_send, font=self.custom_font)
58.         self.clear_send_btn.pack(pady=5)
60.         # 3. 显示区：内容展示框 + 清空按钮
61.         self.display_frame = ttk.LabelFrame(root, text="显示区", padding=10)
62.         self.display_frame.pack(fill="both", padx=10, pady=10, expand=True)
64.         self.recv_text = scrolledtext.ScrolledText(
65.             self.display_frame,
66.             width=60,
67.             height=15,
68.             font=self.custom_font,
69.             wrap=tk.WORD
70.         )
71.         self.recv_text.pack(side="left", fill="both", expand=True, padx=(0, 5))
73.         self.clear_display_btn = tk.Button(self.display_frame, text="清空", command=self.clear_display, font=self.custom_font)
74.         self.clear_display_btn.pack(side="right", pady=5)
76.     def on_message(self, ws, message):
77.         self.recv_text.insert(tk.END, f"收到消息: {message}\n")
78.         self.recv_text.see(tk.END)
80.     def on_error(self, ws, error):
81.         self.recv_text.insert(tk.END, f"发生错误: {error}\n")
82.         self.recv_text.see(tk.END)
84.     def on_close(self, ws, close_status_code, close_msg):
85.         self.recv_text.insert(tk.END, "连接已关闭\n")
86.         self.recv_text.see(tk.END)
88.     def on_open(self, ws):
89.         self.recv_text.insert(tk.END, "连接已建立\n")
90.         self.recv_text.see(tk.END)
92.     def connect(self):
93.         try:
94.             ws_url = self.url_entry.get()
95.             if not ws_url:
96.                 self.recv_text.insert(tk.END, "请输入 WebSocket 地址\n")
97.                 return
98.             self.ws = websocket.WebSocketApp(ws_url,
99.                                              on_open=self.on_open,
100.                                              on_message=self.on_message,
101.                                              on_error=self.on_error,
102.                                              on_close=self.on_close)
103.             ws_thread = threading.Thread(target=self.ws.run_forever)
104.             ws_thread.daemon = True
105.             ws_thread.start()
106.         except Exception as e:
107.             self.recv_text.insert(tk.END, f"连接失败: {e}\n")
109.     def format_text(self):
110.         try:
111.             # self.send_text.edit_undo(False)  # 禁用撤销
112.             text = self.send_text.get("1.0", tk.END).strip()
113.             if not text:
114.                 return
116.             try:
117.                 json_obj = json.loads(text)
118.                 formatted_text = json.dumps(json_obj, indent=2, ensure_ascii=False)
119.                 self.send_text.delete("1.0", tk.END)
120.                 self.send_text.insert("1.0", formatted_text)
121.             except json.JSONDecodeError:
122.                 cleaned_text = " ".join(text.split())
123.                 self.send_text.delete("1.0", tk.END)
124.                 self.send_text.insert("1.0", cleaned_text)
125.         except Exception as e:
126.             self.recv_text.insert(tk.END, f"格式化失败: {e}\n")
128.     def send_message(self):
129.         if self.ws:
130.             message = self.send_text.get("1.0", tk.END).strip()
131.             if message:
132.                 self.ws.send(message)
133.                 self.recv_text.insert(tk.END, f"已发送消息: {message}\n")
134.                 self.recv_text.see(tk.END)
135.             else:
136.                 self.recv_text.insert(tk.END, "发送内容不能为空\n")
138.     def close(self):
139.         if self.ws:
140.             self.ws.close()
141.             self.ws = None
143.     def clear_send(self):
144.         self.send_text.delete("1.0", tk.END)
146.     def clear_display(self):
147.         self.recv_text.delete("1.0", tk.END)
149. if __name__ == "__main__":
150.     root = tk.Tk()
151.     app = WebSocketApp(root)
152.     root.mainloop()

复制代码

您好，想向您请教一个问题，为何我用 pyhont 3.12.10 ，用 vscode 运行后，输入正确的ip地址后，点连接，会提示以下的错误呢

连接失败: WebSocketApp.__init__() got an unexpected keyword argument 'on_open'

但如果用您的就没有问题，不晓得是不是哪里没有配置正确，您的代码我是直接复制的，并未改 。

>python --version
Python 3.12.10

谢谢

3.12.10经测试无上述问题
根据报错来看请您查看WebSocketApp的__init__方法是否传入了除TKroot实例之外的参数, 例如异常中提示的on_open参数

谢谢了，这个有什么用

哪方面的网站呢？

用于websocket连接测试

感谢分享，简单实用。不过在线工具也不少，留着备用。

应该添加一个显示连接状态的模块，可以直观看到有没有连接成功

您好，请问这个有源码么，谢谢

已加上源码