import sys import requests from datetime import datetime, timezone from zoneinfo import ZoneInfo from PySide6.QtWidgets import ( QApplication, QWidget, QVBoxLayout, QHBoxLayout, QFormLayout, QAbstractItemView ) from PySide6.QtCore import ( Qt, QThread, Signal, QObject, QAbstractTableModel, QModelIndex, QLocale, QAbstractListModel ) from PySide6.QtGui import QIcon from qfluentwidgets import ( MSFluentWindow, setTheme, Theme, TableView, PushButton, MessageBox, MessageBoxBase, LineEdit, InfoBar, TextEdit, InfoBarIcon, FluentTranslator, TeachingTip, TeachingTipTailPosition, ModelComboBox, ) from qfluentwidgets import FluentIcon as Fi import resources # dict[str, str | int] 就是 # { # "ID": "xxx", # "NAME": "zzz", # "SAFE": -2, # "UPDATE_DATA": "aaa", # "NOTES": "" # } # --- API 客户端配置 --- BASE_URL = "https://safe-marks.oranj.work/api/v1/ext" SAFE = [1, 0, -1, -2] SAFE_MAP = { 1: "安全", 0: "未知", -1: "不安全", -2: "未记录" } # 反向映射，用于 QComboBox SAFE_MAP_INV = {v: k for k, v in SAFE_MAP.items()} SAFE_MAP_ICON = { 1: InfoBarIcon.SUCCESS.icon(), 0: InfoBarIcon.WARNING.icon(), -1: InfoBarIcon.ERROR.icon(), -2: InfoBarIcon.INFORMATION.icon(), } class APIException(Exception): pass def show_quick_tip(widget: QWidget, caption: str, text: str): mb = MessageBox(caption, text, widget) mb.cancelButton.setHidden(True) mb.buttonLayout.insertStretch(0, 1) mb.buttonLayout.setStretch(1, 0) mb.yesButton.setMinimumWidth(100) mb.setClosableOnMaskClicked(True) mb.exec() def accept_warning(widget: QWidget, condition: bool, caption: str = "警告", text: str = "你确定要继续吗？") -> bool: if condition: mb = MessageBox(caption, text, widget) if not mb.exec(): return True return False # --- 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 APIException(f"API 错误 (状态 {response.status_code}): {detail}") except requests.JSONDecodeError: raise APIException(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, str | int]): """执行 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, str | int]): """执行 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 SafeStatusListModel(QAbstractListModel): def __init__(self, parent: QWidget = None): super().__init__(parent) self.safe_status = SAFE def rowCount(self, /, parent = ...): return len(self.safe_status) def columnCount(self, parent, /): return 1 def data(self, index: QModelIndex, /, role: int = ...): row = index.row() if role == Qt.ItemDataRole.EditRole: return SAFE_MAP[self.safe_status[row]] elif role == Qt.ItemDataRole.DecorationRole: return SAFE_MAP_ICON[self.safe_status[row]] elif role == Qt.ItemDataRole.UserRole: return self.safe_status[row] return None # --- 添加/编辑 对话框 --- class ExtensionDialog(MessageBoxBase): """用于添加或编辑插件条目的对话框。""" def __init__(self, extension_data: dict[str, str | int] = None, parent=None): super().__init__(parent) self.is_edit_mode = extension_data is not None self.setMinimumWidth(400) # 创建控件 self.lne_id = LineEdit(self) self.lne_name = LineEdit(self) self.safe_combo = ModelComboBox(self) self.notes_edit = TextEdit(self) # 填充 SAFE 下拉框 (使用新的 MAP) safe_list_model = SafeStatusListModel(self) self.safe_combo.setModel(safe_list_model) # 布局 form_layout = QFormLayout() form_layout.addRow("ID:", self.lne_id) form_layout.addRow("名称:", self.lne_name) form_layout.addRow("安全状态:", self.safe_combo) form_layout.addRow("备注:", self.notes_edit) self.cw = QWidget(self) main_layout = QVBoxLayout() main_layout.addLayout(form_layout) self.cw.setLayout(main_layout) self.viewLayout.addWidget(self.cw) if self.is_edit_mode: self.lne_id.setText(extension_data.get("ID", "")) self.lne_id.setReadOnly(True) # ID 不允许修改 self.lne_name.setText(extension_data.get("NAME", "")) safe_status: int = extension_data.get("SAFE", -2) self.safe_combo.setCurrentIndex(SAFE.index(safe_status)) self.notes_edit.setText(extension_data.get("NOTES", "")) else: self.safe_combo.setCurrentIndex(SAFE.index(-2)) def validate(self) -> bool: if len(self.lne_id.text()) == 0: TeachingTip.create( target=self.lne_id, title="错误", content="ID 不能为空！", icon=InfoBarIcon.ERROR, isClosable=True, duration=2000, tailPosition=TeachingTipTailPosition.BOTTOM, parent=self ) return False if len(self.lne_name.text()) == 0: TeachingTip.create( target=self.lne_name, title="错误", content="名称不能为空！", icon=InfoBarIcon.ERROR, isClosable=True, duration=2000, tailPosition=TeachingTipTailPosition.BOTTOM, parent=self, ) return False return True def get_data(self) -> dict[str, str | int]: """获取对话框中用于 'add_one' 的数据""" return { "ID": self.lne_id.text(), "NAME": self.lne_name.text(), "SAFE": self.safe_combo.currentData(), "NOTES": self.notes_edit.toPlainText() } def get_update_payload(self) -> dict[str, str | int]: """获取对话框中用于 'update_one' 的数据 (不含ID)""" return { "NAME": self.lne_name.text(), "SAFE": self.safe_combo.currentData(), "NOTES": self.notes_edit.toPlainText() } class ExtensionsDataTable(QAbstractTableModel): def __init__(self, ext_data: list[dict[str, str | int]], parent=None): super().__init__(parent) self.ext_data = ext_data or [] self.headers = ["ID", "名称", "安全性", "更新日期", "备注"] self.col_map = { 0: "ID", 1: "NAME", 2: "SAFE", 3: "UPDATE_DATE", 4: "NOTES", } def rowCount(self, /, parent = ...): return len(self.ext_data) def columnCount(self, /, parent = ...): return len(self.headers) def headerData(self, section: int, orientation: Qt.Orientation, /, role: int = ...): if orientation == Qt.Orientation.Horizontal: if role == Qt.ItemDataRole.DisplayRole: return self.headers[section] return None def data(self, index: QModelIndex, /, role: int = ...): row = index.row() col = index.column() ext = self.ext_data[row] col_val = ext[self.col_map[col]] if role == Qt.ItemDataRole.DisplayRole: if col == 2: return SAFE_MAP[col_val] elif col == 3: return self.iso2custom(col_val) else: return col_val if role == Qt.ItemDataRole.DecorationRole: if col == 2: return SAFE_MAP_ICON[col_val] # 一行的每一个列都返回所有数据，这样不管双击一行的哪一个位置，都可以获取到所有数据 if role == Qt.ItemDataRole.UserRole: return ext return None def update_data(self, ext_data: list[dict[str, str | int]]): self.beginResetModel() self.ext_data.clear() self.ext_data.extend(ext_data) self.endResetModel() @staticmethod def iso2custom(iso_time: str) -> str: dt = datetime.fromisoformat(iso_time) dt = dt.replace(tzinfo=timezone.utc) dt = dt.astimezone(ZoneInfo("America/New_York")) custom_format = "%Y-%m-%d %H:%M:%S" return dt.strftime(custom_format) class MainInterface(QWidget): def __init__(self, ext_data: list[dict[str, str | int]] = None, parent=None): super().__init__(parent) self.setObjectName("main") self.vly_m = QVBoxLayout() self.setLayout(self.vly_m) self.hly_top = QHBoxLayout() self.pbn_refresh = PushButton(Fi.SYNC, "刷新", self) self.pbn_add = PushButton(Fi.ADD, "添加", self) self.pbn_delete = PushButton(Fi.DELETE, "删除", self) self.hly_top.addWidget(self.pbn_refresh) self.hly_top.addWidget(self.pbn_add) self.hly_top.addWidget(self.pbn_delete) self.hly_top.addStretch(1) self.vly_m.addLayout(self.hly_top) self.tbv_m = TableView(self) self.ext_model = ExtensionsDataTable(ext_data, self) self.tbv_m.setModel(self.ext_model) self.tbv_m.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers) self.tbv_m.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectRows) # 整行选中 self.tbv_m.setSelectionMode(QAbstractItemView.SelectionMode.SingleSelection) # 只允许单选 self.tbv_m.verticalHeader().hide() self.tbv_m.setBorderVisible(True) self.tbv_m.setBorderRadius(4) self.tbv_m.horizontalHeader().setStretchLastSection(True) self.tbv_m.setColumnWidth(0, 250) self.tbv_m.setColumnWidth(1, 200) self.tbv_m.setColumnWidth(3, 180) self.vly_m.addWidget(self.tbv_m) # --- 主窗口 --- class MainWindow(MSFluentWindow): def __init__(self): super().__init__() self.setWindowTitle("插件安全标记") self.setWindowIcon(QIcon(":/logo.png")) # self.navigationInterface.hide() self.resize(1000, 760) desktop = QApplication.screens()[0].availableGeometry() w, h = desktop.width(), desktop.height() self.move(w // 2 - self.width() // 2, h // 2 - self.height() // 2) setTheme(Theme.LIGHT) # --- UI --- self.main_interface = MainInterface(parent=self) self.addSubInterface(self.main_interface, Fi.HOME, "主页", Fi.HOME_FILL) # --- 初始化后台工作线程和 QObject --- self.thread = QThread() self.worker = ApiWorker() self.worker.moveToThread(self.thread) # 连接 Worker 的新信号 self.worker.queryAllFinished.connect(self.main_interface.ext_model.update_data) 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() # 连接信号和槽 self.main_interface.pbn_refresh.clicked.connect(self.worker.do_query_all) self.main_interface.pbn_add.clicked.connect(self.on_add) self.main_interface.pbn_delete.clicked.connect(self.on_delete) self.main_interface.tbv_m.doubleClicked.connect(self.on_edit) # 初始加载数据 self.worker.do_query_all() def closeEvent(self, event): """关闭窗口时，确保线程也退出""" self.thread.quit() self.thread.wait() super().closeEvent(event) # --- GUI 槽函数 (在主线程中运行) --- def handle_operation_finished(self, result: dict): """添加/更新/删除成功后的通用处理器""" InfoBar.success("", "操作已成功完成。", duration=2000, parent=self.window()) self.worker.do_query_all() # 重新加载数据 def handle_error(self, error_message: str): """显示来自工作线程的错误消息""" show_quick_tip(self, "API 错误", f"与服务器通信时发生错误:\n{error_message}") # --- 按钮点击处理 --- def on_add(self): """打开“添加”对话框 (add_one)""" dialog = ExtensionDialog(parent=self) if dialog.exec(): data = dialog.get_data() self.worker.do_add_one(data) def on_edit(self, index: QModelIndex): """打开“编辑”对话框（通过双击）(update_one)""" item_data = index.data(Qt.ItemDataRole.UserRole) item_id = item_data.get("ID") dialog = ExtensionDialog(extension_data=item_data, parent=self) if dialog.exec(): update_payload = dialog.get_update_payload() self.worker.do_update_one(item_id, update_payload) def on_delete(self): """删除当前选定的行 (delete_one)""" # 前面设定了只能选中一行 ext_data: list[dict] = [index.data(Qt.ItemDataRole.UserRole) for index in self.main_interface.tbv_m.selectedIndexes() if index.column() == 0] if len(ext_data) == 0: show_quick_tip(self, "警告", "你没有选中任何插件。") return ext_d: dict = ext_data[0] if accept_warning(self, True, "警告", f"你确定要删除插件【{ext_d.get('NAME')}】吗？"): return self.worker.do_delete_one(ext_d.get("ID")) # --- 运行 GUI --- def main(): app = QApplication(sys.argv) translator = FluentTranslator(QLocale(QLocale.Language.Chinese, QLocale.Country.China)) app.installTranslator(translator) win = MainWindow() win.show() return app.exec() if __name__ == '__main__': main()