import sys import requests from PySide6.QtWidgets import ( QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QTableWidget, QTableWidgetItem, QPushButton, QMessageBox, QDialog, QFormLayout, QLineEdit, QComboBox, QTextEdit, QDialogButtonBox, QAbstractItemView ) from PySide6.QtCore import Qt, QThread, Signal, QObject from typing import Dict, Any, Optional # --- API 客户端配置 --- BASE_URL = "http://127.0.0.1:17701/api/v1/ext" # 安全状态与中文的映射 (根据新要求) SAFE_MAP = { 1: "安全 (safe)", 0: "未知 (unsure)", -1: "不安全 (unsafe)", -2: "不详 (unknown)" } # 反向映射，用于 QComboBox SAFE_MAP_INV = {v: k for k, v in SAFE_MAP.items()} # --- API 工作线程 --- class ApiWorker(QObject): """在单独的线程中执行网络请求，避免 GUI 冻结""" # 定义信号 queryAllFinished = Signal(list) addOneFinished = Signal(dict) updateOneFinished = Signal(dict) deleteOneFinished = Signal(dict) error = Signal(str) def __init__(self): super().__init__() self.session = requests.Session() @staticmethod def _handle_response(response): """辅助函数：检查 HTTP 响应""" if not response.ok: try: detail = response.json().get("detail", response.text) raise Exception(f"API 错误 (状态 {response.status_code}): {detail}") except requests.JSONDecodeError: raise Exception(f"API 错误 (状态 {response.status_code}): {response.text}") return response.json() def do_query_all(self): """执行 query_all 操作""" try: response = self.session.get(f"{BASE_URL}/query_all") data = self._handle_response(response) self.queryAllFinished.emit(data) except Exception as e: self.error.emit(str(e)) def do_add_one(self, extension: Dict[str, Any]): """执行 add_one 操作 (dict)""" try: response = self.session.post(f"{BASE_URL}/add_one", json=extension) data = self._handle_response(response) self.addOneFinished.emit(data) except Exception as e: self.error.emit(str(e)) def do_update_one(self, item_id: str, payload: Dict[str, Any]): """执行 update_one 操作 (dict)""" try: response = self.session.put(f"{BASE_URL}/update_one/{item_id}", json=payload) data = self._handle_response(response) self.updateOneFinished.emit(data) except Exception as e: self.error.emit(str(e)) def do_delete_one(self, item_id: str): """执行 delete_one 操作 (str ID)""" try: response = self.session.delete(f"{BASE_URL}/delete_one/{item_id}") data = self._handle_response(response) self.deleteOneFinished.emit(data) except Exception as e: self.error.emit(str(e)) # --- 添加/编辑 对话框 --- class ExtensionDialog(QDialog): """用于添加或编辑插件条目的对话框。""" def __init__(self, parent=None, extension_data: Optional[Dict] = None): super().__init__(parent) self.is_edit_mode = extension_data is not None self.setWindowTitle("编辑插件" if self.is_edit_mode else "添加插件") self.setMinimumWidth(400) # 创建控件 self.id_edit = QLineEdit() self.name_edit = QLineEdit() self.safe_combo = QComboBox() self.notes_edit = QTextEdit() # 填充 SAFE 下拉框 (使用新的 MAP) for text in SAFE_MAP_INV.keys(): self.safe_combo.addItem(text) # 布局 form_layout = QFormLayout() form_layout.addRow("ID:", self.id_edit) form_layout.addRow("名称 (NAME):", self.name_edit) form_layout.addRow("安全状态 (SAFE):", self.safe_combo) form_layout.addRow("备注 (NOTES):", self.notes_edit) self.button_box = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel) self.button_box.accepted.connect(self.accept) self.button_box.rejected.connect(self.reject) main_layout = QVBoxLayout() main_layout.addLayout(form_layout) main_layout.addWidget(self.button_box) self.setLayout(main_layout) if self.is_edit_mode: self.id_edit.setText(extension_data.get("ID", "")) self.id_edit.setReadOnly(True) # ID 不允许修改 self.name_edit.setText(extension_data.get("NAME", "")) safe_status = extension_data.get("SAFE", 0) self.safe_combo.setCurrentText(SAFE_MAP.get(safe_status, "未知 (unsure)")) self.notes_edit.setText(extension_data.get("NOTES", "")) def get_data(self) -> Dict[str, Any]: """获取对话框中用于 'add_one' 的数据""" return { "ID": self.id_edit.text(), "NAME": self.name_edit.text(), "SAFE": SAFE_MAP_INV.get(self.safe_combo.currentText(), 0), "NOTES": self.notes_edit.toPlainText() } def get_update_payload(self) -> Dict[str, Any]: """获取对话框中用于 'update_one' 的数据 (不含ID)""" return { "NAME": self.name_edit.text(), "SAFE": SAFE_MAP_INV.get(self.safe_combo.currentText(), 0), "NOTES": self.notes_edit.toPlainText() } # --- 主窗口 --- class MainWindow(QMainWindow): def __init__(self): super().__init__() self.setWindowTitle("插件管理器") self.setGeometry(100, 100, 800, 600) self.table_widget = QTableWidget() self.setup_table() self.refresh_button = QPushButton("刷新 (Query All)") self.add_button = QPushButton("添加 (Add One)") self.delete_button = QPushButton("删除选定 (Delete One)") button_layout = QHBoxLayout() button_layout.addWidget(self.refresh_button) button_layout.addWidget(self.add_button) button_layout.addWidget(self.delete_button) button_layout.addStretch() main_layout = QVBoxLayout() main_layout.addLayout(button_layout) main_layout.addWidget(self.table_widget) central_widget = QWidget() central_widget.setLayout(main_layout) self.setCentralWidget(central_widget) self.setup_worker_thread() # 连接信号和槽 self.refresh_button.clicked.connect(self.worker.do_query_all) self.add_button.clicked.connect(self.on_add) self.delete_button.clicked.connect(self.on_delete) self.table_widget.doubleClicked.connect(self.on_edit) # 初始加载数据 self.worker.do_query_all() def setup_table(self): """初始化表格""" self.table_widget.setColumnCount(5) self.table_widget.setHorizontalHeaderLabels( ["ID", "名称 (NAME)", "安全 (SAFE)", "更新日期 (UPDATE_DATE)", "备注 (NOTES)"]) self.table_widget.setEditTriggers(QAbstractItemView.NoEditTriggers) self.table_widget.setSelectionBehavior(QAbstractItemView.SelectRows) # 整行选中 self.table_widget.setSelectionMode(QAbstractItemView.SingleSelection) # 只允许单选 self.table_widget.horizontalHeader().setStretchLastSection(True) self.table_widget.resizeColumnsToContents() def setup_worker_thread(self): """初始化后台工作线程和 QObject""" self.thread = QThread() self.worker = ApiWorker() self.worker.moveToThread(self.thread) # 连接 Worker 的新信号 self.worker.queryAllFinished.connect(self.populate_table) self.worker.addOneFinished.connect(self.handle_operation_finished) self.worker.updateOneFinished.connect(self.handle_operation_finished) self.worker.deleteOneFinished.connect(self.handle_operation_finished) self.worker.error.connect(self.handle_error) self.thread.start() def closeEvent(self, event): """关闭窗口时，确保线程也退出""" self.thread.quit() self.thread.wait() super().closeEvent(event) # --- GUI 槽函数 (在主线程中运行) --- def populate_table(self, data: list): """使用从 API 获取的数据填充表格""" self.table_widget.setRowCount(0) for row, item in enumerate(data): self.table_widget.insertRow(row) id_item = QTableWidgetItem(item.get("ID")) name_item = QTableWidgetItem(item.get("NAME")) safe_val = item.get("SAFE", 0) safe_item = QTableWidgetItem(SAFE_MAP.get(safe_val, "未知 (unsure)")) # 使用新 MAP date_item = QTableWidgetItem(item.get("UPDATE_DATE")) notes_item = QTableWidgetItem(item.get("NOTES")) # 存储原始数据到 ID 项中，以便编辑时检索 id_item.setData(Qt.UserRole, item) self.table_widget.setItem(row, 0, id_item) self.table_widget.setItem(row, 1, name_item) self.table_widget.setItem(row, 2, safe_item) self.table_widget.setItem(row, 3, date_item) self.table_widget.setItem(row, 4, notes_item) self.table_widget.resizeColumnsToContents() def handle_operation_finished(self, result: dict): """添加/更新/删除成功后的通用处理器""" QMessageBox.information(self, "操作成功", f"操作已成功完成。\n{result}") self.worker.do_query_all() # 重新加载数据 def handle_error(self, error_message: str): """显示来自工作线程的错误消息""" QMessageBox.critical(self, "API 错误", f"与服务器通信时发生错误:\n{error_message}") # --- 按钮点击处理 --- def on_add(self): """打开“添加”对话框 (add_one)""" dialog = ExtensionDialog(self) if dialog.exec(): data = dialog.get_data() if not data.get("ID") or not data.get("NAME"): self.handle_error("ID 和 名称 (NAME) 不能为空。") return self.worker.do_add_one(data) def on_edit(self, index): """打开“编辑”对话框（通过双击）(update_one)""" row = index.row() item_data = self.table_widget.item(row, 0).data(Qt.UserRole) item_id = item_data.get("ID") dialog = ExtensionDialog(self, extension_data=item_data) if dialog.exec(): update_payload = dialog.get_update_payload() self.worker.do_update_one(item_id, update_payload) def on_delete(self): """删除当前选定的行 (delete_one)""" current_row = self.table_widget.currentRow() if current_row < 0: QMessageBox.warning(self, "未选择", "请选择一行进行删除。") return id_item = self.table_widget.item(current_row, 0) item_id = id_item.text() reply = QMessageBox.question( self, "确认删除", f"你确定要删除插件 (ID: {item_id}) 吗？", QMessageBox.Yes | QMessageBox.No, QMessageBox.No ) if reply == QMessageBox.Yes: self.worker.do_delete_one(item_id) # --- 运行 GUI --- if __name__ == "__main__": app = QApplication(sys.argv) window = MainWindow() window.show() sys.exit(app.exec())