效果预览
代码
代码采用AI辅助编写
import sys
import json
import threading
import asyncio
from PyQt5.QtWidgets import *
from PyQt5.QtCore import *
from PyQt5.QtGui import *
import requests
import websockets# ================= 1. 报文显示控件 =================
class PacketDisplay(QTextEdit):def __init__(self, parent=None):super().__init__(parent)self.setReadOnly(True)self.setLineWrapMode(QTextEdit.NoWrap)font = QFont("Consolas", 10)self.setFont(font)self.is_hex_mode = Falseself.history: list[bytes] = []self._pending: list[bytes] = []self._max_history = 2000self._timer = QTimer(self)self._timer.setSingleShot(True)self._timer.timeout.connect(self._flush)self.fmt_normal = QTextCharFormat()self.fmt_normal.setFont(font)self.fmt_hidden = QTextCharFormat()self.fmt_hidden.setBackground(QBrush(QColor(128, 128, 128)))self.fmt_hidden.setForeground(QBrush(QColor(255, 255, 255)))self.fmt_hidden.setFont(font)self.fmt_hidden.setFontWeight(QFont.Bold)def append_raw_data(self, data: bytes):if not data: returnself._pending.append(data)if not self._timer.isActive():self._timer.start(20)def toggle_view(self):self.is_hex_mode = not self.is_hex_modeself.clear()cursor = self.textCursor()cursor.movePosition(QTextCursor.Start)for pkt in self.history:self._render_packet(cursor, pkt)self.setTextCursor(cursor)self.ensureCursorVisible()def _flush(self):if not self._pending: returncursor = self.textCursor()cursor.movePosition(QTextCursor.End)for data in self._pending:self.history.append(data)if len(self.history) > self._max_history:self.history.pop(0)self._render_packet(cursor, data)self._pending.clear()self.setTextCursor(cursor)self.ensureCursorVisible()def _render_packet(self, cursor: QTextCursor, data: bytes):if self.is_hex_mode:for i in range(0, len(data), 16):chunk = data[i:i+16]hex_part = ' '.join(f'{b:02x}' for b in chunk)ascii_part = ''.join(chr(b) if 32 <= b < 127 else '.' for b in chunk)cursor.insertText(f"{i:04x} {hex_part:<48} {ascii_part}\n", self.fmt_normal)cursor.insertText("\n", self.fmt_normal)else:text = data.decode('utf-8', errors='replace')i, n = 0, len(text)while i < n:ch = text[i]if ch == '\r' and i + 1 < n and text[i+1] == '\n':cursor.insertText('\\r\\n', self.fmt_hidden)cursor.insertText('\n', self.fmt_normal)i += 2elif ch == '\n':cursor.insertText('\\n', self.fmt_hidden)cursor.insertText('\n', self.fmt_normal)i += 1elif ch == '\r':cursor.insertText('\\r', self.fmt_hidden)cursor.insertText('\n', self.fmt_normal)i += 1elif ch == '\t':cursor.insertText('\\t', self.fmt_hidden)i += 1elif ch == ' ':cursor.insertText('·', self.fmt_hidden)i += 1elif ord(ch) < 32 or ord(ch) == 127:cursor.insertText(f'\\x{ord(ch):02x}', self.fmt_hidden)i += 1else:j = iwhile j < n and not (text[j] in '\r\n\t ' or ord(text[j]) < 32 or ord(text[j]) == 127):j += 1cursor.insertText(text[i:j], self.fmt_normal)i = j# cursor.insertText('\n', self.fmt_normal)# ================= 2. 网络工作线程 =================
class NetWorker(QObject):sig_data = pyqtSignal(bytes) # 实时报文（SSE 逐行 / WS 帧 / HTTP Body）sig_resp_headers = pyqtSignal(str) # 响应头（含状态行）sig_resp_body = pyqtSignal(bytes) # 响应体（非 SSE 时为完整内容；SSE 时为占位提示）sig_status = pyqtSignal(str)sig_error = pyqtSignal(str)sig_finished = pyqtSignal()def __init__(self):super().__init__()self._stop_flag = threading.Event()self._resp = None # HTTP/SSE 响应用于中断self._ws_instance = None # WS 连接实例用于中断self._ws_loop = Noneself.is_running = Falsedef start_task(self, protocol, method, url, headers_json, body):if self.is_running: returnt = threading.Thread(target=self._run,args=(protocol, method, url, headers_json, body),daemon=True)t.start()def stop(self):self._stop_flag.set()if self._resp:try: self._resp.close()except: passif self._ws_instance and self._ws_loop:try: asyncio.run_coroutine_threadsafe(self._ws_instance.close(), self._ws_loop)except: passdef _run(self, protocol, method, url, headers_json, body):self.is_running = Trueself._stop_flag.clear()self._resp = Noneself._ws_instance = Noneself._ws_loop = Nonetry:headers = json.loads(headers_json) if headers_json.strip() else {}self.sig_status.emit(f"🚀 正在发起 {protocol} 请求: {url}")if protocol == "HTTP":self._do_http(method, url, headers, body)elif protocol == "WebSocket":self._do_ws(url, headers)else:self.sig_error.emit("未知协议类型")except requests.exceptions.RequestException as e:if not self._stop_flag.is_set(): self.sig_error.emit(f"❌ 网络异常: {str(e)}")except Exception as e:if not self._stop_flag.is_set(): self.sig_error.emit(f"❌ 异常: {str(e)}")finally:self.is_running = Falseself.sig_finished.emit()def _do_http(self, method, url, headers, body):# stream=True 确保可以中断 + 支持 SSE 流式读取self._resp = requests.request(method, url, headers=headers, data=body, timeout=600, stream=True)# 1. 先发送响应头（状态行 + 头）status_line = f"HTTP/1.1 {self._resp.status_code} {self._resp.reason or ''}"hdr_lines = "\n".join(f"{k}: {v}" for k, v in self._resp.headers.items())self.sig_resp_headers.emit(f"{status_line}\n{hdr_lines}")self.sig_status.emit(f"✅ 响应状态: {self._resp.status_code}")# 2. 判断是否 SSEcontent_type = self._resp.headers.get('Content-Type', '')is_sse = 'text/event-stream' in content_type.lower()if is_sse:# SSE 模式：逐行推送，Body 面板显示占位self.sig_status.emit("🔗 检测到 SSE 流 (text/event-stream)，开始接收事件...")for line in self._resp.iter_content(chunk_size=256, decode_unicode=False):if self._stop_flag.is_set(): breakif line:self.sig_resp_body.emit(line)self.sig_data.emit(line)self.sig_status.emit(f"📦 接收 SSE 行: {len(line)} 字节")else:# 普通 HTTP：一次性读取 Bodycontent = self._resp.contentself.sig_resp_body.emit(content)self.sig_data.emit(content)def _do_ws(self, url, headers):async def ws_loop():self._ws_loop = asyncio.get_running_loop()async with websockets.connect(url, extra_headers=headers) as ws:self._ws_instance = wsself.sig_status.emit("🔗 WebSocket 已连接，等待服务端推送...")# WebSocket 无标准 HTTP 响应体/头，发送提示占位self.sig_resp_headers.emit("WebSocket 握手成功（无标准 HTTP 响应头）")self.sig_resp_body.emit(b"")async for msg in ws:if self._stop_flag.is_set(): breakdata = msg if isinstance(msg, bytes) else msg.encode('utf-8')self.sig_data.emit(data)self.sig_status.emit(f"📦 接收 WS 帧: {len(data)} 字节")asyncio.run(ws_loop())# ================= 3. 主界面 =================
class MainWindow(QMainWindow):def __init__(self):super().__init__()self.setWindowTitle("HTTP/WS/SSE 测试客户端")self.resize(1250, 850)self.worker = NetWorker()self._setup_ui()self._bind_signals()def _setup_ui(self):central = QWidget()self.setCentralWidget(central)main_layout = QHBoxLayout(central)# ---- 左侧 ----left_panel = QGroupBox("📤 请求配置")left_layout = QFormLayout(left_panel)left_layout.setSpacing(10)self.cb_protocol = QComboBox()# 🆕 仅保留 HTTP / WebSocket；SSE 由 HTTP 响应头自动判定self.cb_protocol.addItems(["HTTP", "WebSocket"])self.cb_protocol.currentTextChanged.connect(self._on_protocol_change)self.cb_method = QComboBox()self.cb_method.addItems(["GET", "POST", "PUT", "DELETE", "PATCH"])self.le_url = QLineEdit("http://localhost:8000/test")self.le_url.setPlaceholderText("http(s):// 或 ws(s)://")self.te_headers = QTextEdit()self.te_headers.setMaximumHeight(100)self.te_headers.setPlaceholderText('{\n "Content-Type": "application/json",\n "Accept": "text/event-stream"\n}')self.te_body = QTextEdit()self.te_body.setPlaceholderText("请求体 (仅 HTTP 有效)")self.btn_action = QPushButton("发送请求 / 连接")self.btn_action.setStyleSheet("QPushButton { padding: 10px; font-weight: bold; background-color: #2196F3; color: white; border-radius: 4px; }")self.btn_action.clicked.connect(self._handle_action)left_layout.addRow("协议:", self.cb_protocol)left_layout.addRow("方法:", self.cb_method)left_layout.addRow("URL:", self.le_url)left_layout.addRow("Headers:", self.te_headers)left_layout.addRow("Body:", self.te_body)left_layout.addRow(self.btn_action)# ---- 右侧 ----right_panel = QTabWidget()# Tab 1: 完整响应 (Headers + Body)self.te_resp_headers = QTextEdit()self.te_resp_headers.setReadOnly(True)self.te_resp_headers.setFont(QFont("Consolas", 9))self.te_resp_headers.setPlaceholderText("HTTP 响应头将显示在这里...")self.te_resp_body = QTextEdit()self.te_resp_body.setReadOnly(True)self.te_resp_body.setFont(QFont("Consolas", 10))resp_splitter = QSplitter(Qt.Vertical)resp_splitter.addWidget(self.te_resp_headers)resp_splitter.addWidget(self.te_resp_body)resp_splitter.setSizes([160, 400])resp_tab = QWidget()resp_layout = QVBoxLayout(resp_tab)resp_layout.setContentsMargins(0, 0, 0, 0)resp_layout.addWidget(resp_splitter)right_panel.addTab(resp_tab, "📄 完整响应")# Tab 2: 实时报文self.pkt_display = PacketDisplay()toolbar = QHBoxLayout()self.btn_toggle = QPushButton("🔄 切换 Hex / Text")self.btn_toggle.clicked.connect(self.pkt_display.toggle_view)toolbar.addWidget(self.btn_toggle)toolbar.addStretch()pkt_tab = QWidget()pkt_layout = QVBoxLayout(pkt_tab)pkt_layout.setContentsMargins(4, 4, 4, 4)pkt_layout.addLayout(toolbar)pkt_layout.addWidget(self.pkt_display)right_panel.addTab(pkt_tab, "⚡ 实时报文")splitter = QSplitter(Qt.Horizontal)splitter.addWidget(left_panel)splitter.addWidget(right_panel)splitter.setSizes([380, 870])main_layout.addWidget(splitter)self.statusBar().showMessage("就绪")def _bind_signals(self):self.worker.sig_status.connect(lambda msg: self.statusBar().showMessage(msg, 5000))self.worker.sig_resp_headers.connect(self.te_resp_headers.setPlainText)self.worker.sig_resp_body.connect(self._show_response)self.worker.sig_data.connect(self.pkt_display.append_raw_data)self.worker.sig_error.connect(lambda msg: QMessageBox.critical(self, "错误", msg))self.worker.sig_finished.connect(self._on_task_finished)def _on_protocol_change(self, proto):is_http = proto == "HTTP"self.cb_method.setEnabled(is_http)self.te_body.setEnabled(is_http)if not self.worker.is_running:self.btn_action.setText("发送请求 / 连接")def _handle_action(self):if self.btn_action.text() == "断开连接":self.worker.stop()returnif self.worker.is_running:QMessageBox.warning(self, "提示", "当前有任务正在运行，请先断开")returnproto = self.cb_protocol.currentText()method = self.cb_method.currentText()url = self.le_url.text().strip()headers = self.te_headers.toPlainText().strip()body = self.te_body.toPlainText().strip()if not url:QMessageBox.warning(self, "提示", "URL 不能为空")return# 清空旧数据self.te_resp_headers.clear()self.te_resp_body.clear()self.pkt_display.history.clear()self.pkt_display.clear()self.btn_action.setText("断开连接")self.btn_action.setStyleSheet("QPushButton { padding: 10px; font-weight: bold; background-color: #d32f2f; color: white; border-radius: 4px; }")self.worker.start_task(proto, method, url, headers, body)def _on_task_finished(self):self.btn_action.setText("发送请求 / 连接")self.btn_action.setStyleSheet("QPushButton { padding: 10px; font-weight: bold; background-color: #2196F3; color: white; border-radius: 4px; }")def _show_response(self, data: bytes):if not data: returnold_text = self.te_resp_body.toPlainText()try:text = data.decode('utf-8')formatted = json.dumps(json.loads(text), indent=4, ensure_ascii=False)self.te_resp_body.setPlainText(old_text + formatted)except Exception:self.te_resp_body.setPlainText(old_text + data.decode('utf-8', errors='replace'))# 移动光标到最底下self.te_resp_body.moveCursor(QTextCursor.End)def closeEvent(self, event):self.worker.stop()event.accept()if __name__ == '__main__':app = QApplication(sys.argv)app.setStyle("Fusion")# 设置窗口图标icon_provider = QFileIconProvider()app.setWindowIcon(icon_provider.icon(QFileInfo(app.applicationFilePath())))win = MainWindow()win.show()sys.exit(app.exec_())