reorganize file structure

common/api_worker.py

@@ -0,0 +1,65 @@
+import requests
+from PySide6.QtCore import QObject, Signal
+from common.utils import APIException, BASE_URL
+
+
+# --- 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))

common/utils.py

@@ -0,0 +1,46 @@
+from PySide6.QtWidgets import QWidget
+from qfluentwidgets import (
+    InfoBarIcon, MessageBox,
+)
+
+# --- 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