chakcy_code_repository/pyside_queue_sqlite
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

modify: 封装webview通用模块

Browse Source
This commit is contained in:
chakcy
2025-09-30 11:01:00 +08:00
parent 7abb876100
commit 01bd8e0f5e
14 changed files with 186 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/config/__init__.py
+6 -1
View File
@@ -5,4 +5,9 @@ try:
except:
config = {}
__all__ = ["config"]
def get_info() -> dict:
return config
__all__ = ["config", "get_info"]
app/scheduler_manager/__init__.py
+3 -7
View File
@@ -1,12 +1,8 @@
from queue_sqlite.scheduler import QueueScheduler
from .students.add_student import add_student
from .students.add_student import add_student
from .students.students_listen import students
scheduler = QueueScheduler()
scheduler = QueueScheduler(scheduler_type="async")
__all__ = [
"scheduler",
"add_student",
"students"
]
__all__ = ["scheduler", "add_student", "students"]
app/scheduler_manager/students/add_student.py
+3 -4
View File
@@ -1,9 +1,8 @@
from queue_sqlite.mounter.task_mounter import TaskMounter
from queue_sqlite.mounter import task
from queue_sqlite.model import MessageItem
import time
@TaskMounter.task(meta={"task_name": "add_student"})
@task(meta={"task_name": "add_student"})
def add_student(message_item: MessageItem):
# create a new instance of the AddStudentDilalog dialog
# time.sleep(3)
@@ -11,7 +10,7 @@ def add_student(message_item: MessageItem):
return {"message": "学生添加成功"}
@TaskMounter.task(meta={"task_name": "test"})
@task(meta={"task_name": "test"})
def test(message_item: MessageItem):
# print(f"测试任务: {message_item}")
return {"message": "测试成功"}
app/utils/__init__.py
+4
View File
@@ -0,0 +1,4 @@
from .webview_queue_sqlite_operations import WebViewQueueSqliteOperations
__all__ = ["WebViewQueueSqliteOperations"]
app/utils/base_webview_bridge.py
+8
View File
@@ -0,0 +1,8 @@
from PySide6.QtCore import QObject
class BaseWebViewBridge(QObject):
def __init__(self, webview, parent=None):
self.webview = webview
super().__init__(parent)
pass
app/utils/webview_controller.py
+44
View File
@@ -0,0 +1,44 @@
from PySide6.QtCore import QObject
from .webview_queue_sqlite_operations import WebViewQueueSqliteOperations
from .base_webview_bridge import BaseWebViewBridge
from typing import List
from PySide6.QtWebChannel import QWebChannel
class WebviewController(QObject):
def __init__(
self,
web_engine_page,
webview_bridge_list: List[type[BaseWebViewBridge]],
web_webview,
scheduler,
parent=None,
):
super().__init__(parent)
self.page = web_engine_page
self.bridge_instances = {}
# 创建bridge实例并保存
for bridge_class in webview_bridge_list:
instance = bridge_class(web_webview)
bridge_name = bridge_class.__name__.lower()
setattr(self, bridge_name, instance)
self.bridge_instances[bridge_name] = instance
self.queue_sqlite_operations = WebViewQueueSqliteOperations(
web_webview, scheduler
)
self.channel = QWebChannel()
# 使用已创建的实例注册到channel
for bridge_name, instance in self.bridge_instances.items():
self.channel.registerObject(bridge_name, instance)
# 为了向前兼容，确保注册名为 "entrance" 的对象
if "webviewbridge" in self.bridge_instances:
self.channel.registerObject(
"entrance", self.bridge_instances["webviewbridge"]
)
self.channel.registerObject("queueSqlite", self.queue_sqlite_operations)
self.page.setWebChannel(self.channel)
app/utils/webview_queue_sqlite_operations.py
+54
View File
@@ -0,0 +1,54 @@
import json
from PySide6.QtCore import QObject, Slot, Signal
from qframelesswindow.webengine import FramelessWebEngineView
from queue_sqlite.model import MessageItem
from queue_sqlite.scheduler import QueueScheduler
class WebViewQueueSqliteOperations(QObject):
"""
用于前端调用QueueSqlite的队列操作
"""
update_queue_state_signal = Signal(dict, str)
def __init__(
self,
webview: FramelessWebEngineView,
queue_scheduler: QueueScheduler,
parent=None,
):
super().__init__(parent)
self.webview = webview
self.page = self.webview.page()
self.update_queue_state_signal.connect(self.update_queue_state)
self.queue_scheduler = queue_scheduler
@Slot(str, str, result=str)
def send_message(self, message: str, store_key: str):
def callback(result: MessageItem):
print(f"send_message callback: {result}")
self.update_queue_state_signal.emit(result.to_dict(), store_key)
self.queue_scheduler.send_message(
MessageItem.from_dict(json.loads(message)), callback
)
def update_queue_state(self, result: dict, store_key: str):
json_str = json.dumps(result)
print(f"result: {result}")
print(f"json_str: {json_str}")
js_code = f"""
try {{
if (typeof window.queueStore.updateQueueData === 'function') {{
window.queueStore.updateQueueData('{store_key}', {json_str});
}} else {{
console.log('updateQueueData function not found');
}}
}} catch (error) {{
console.error('Error in updateQueueData:', error);
}}
"""
print(f"执行的JavaScript代码: {js_code}")
self.page.runJavaScript(js_code)
app/view/web_view/entrance.py
+4 -60
View File
@@ -7,59 +7,17 @@ import base64
import tempfile
from pathlib import Path
from PySide6.QtCore import QObject, Slot, QUrl
from PySide6.QtWebChannel import QWebChannel
from PySide6.QtCore import Slot, QUrl
from PySide6.QtWebEngineCore import QWebEngineSettings
from qframelesswindow.webengine import FramelessWebEngineView
from queue_sqlite.model import MessageItem
from ...scheduler_manager import scheduler
from ...signal import global_signals
from ...utils.base_webview_bridge import BaseWebViewBridge
class WebViewQueueSqliteOperations(QObject):
"""
用于前端调用QueueSqlite的队列操作
"""
def __init__(self, webview: FramelessWebEngineView, parent=None):
super().__init__(parent)
self.webview = webview
self.page = self.webview.page()
global_signals.update_queue_state.connect(self.update_queue_state)
@Slot(str, str, result=str)
def send_message(self, message: str, store_key: str):
def callback(result: MessageItem):
print(f"send_message callback: {result}")
global_signals.update_queue_state.emit(result.to_dict(), store_key)
scheduler.send_message(MessageItem.from_dict(json.loads(message)), callback)
def update_queue_state(self, result: dict, store_key: str):
json_str = json.dumps(result)
print(f"result: {result}")
print(f"json_str: {json_str}")
js_code = f"""
try {{
if (typeof window.queueStore.updateQueueData === 'function') {{
window.queueStore.updateQueueData('{store_key}', {json_str});
}} else {{
console.log('updateQueueData function not found');
}}
}} catch (error) {{
console.error('Error in updateQueueData:', error);
}}
"""
print(f"执行的JavaScript代码: {js_code}")
self.page.runJavaScript(js_code)
class WebviewBridge(QObject):
class Entrance(BaseWebViewBridge):
"""Webview桥接类，用于处理前端和Python后端之间的通信"""
def __init__(self, webview, parent=None):
super().__init__(parent)
super().__init__(webview, parent)
self.webview = webview
# 启用本地文件访问
@@ -362,17 +320,3 @@ class WebviewBridge(QObject):
except Exception as e:
# 确保任何异常都返回有效的JSON
return json.dumps({"success": False, "error": str(e)})
class WebviewController(QObject):
"""Webview控制器类，负责管理Webview的逻辑"""
def __init__(self, web_engine_page, web_webview, parent=None):
super().__init__(parent)
self.page = web_engine_page
self.bridge = WebviewBridge(web_webview)
self.queue_sqlite_operations = WebViewQueueSqliteOperations(web_webview)
self.channel = QWebChannel()
self.channel.registerObject("entrance", self.bridge)
self.channel.registerObject("queueSqlite", self.queue_sqlite_operations)
self.page.setWebChannel(self.channel)
app/view/web_view/web_view_interface.py
+6 -2
View File
@@ -15,8 +15,10 @@ from PySide6.QtWidgets import (
from PySide6.QtCore import QPoint, Qt, QUrl
from qframelesswindow.webengine import FramelessWebEngineView
from .entrance import WebviewController
from ...config import config
from ...utils.webview_controller import WebviewController
from .entrance import Entrance
from ...scheduler_manager import scheduler
class DevToolsWindow(FramelessWebEngineView):
@@ -60,7 +62,9 @@ class WebViewInterface(QFrame):
# 创建WebviewController实例
self.page = self.webView.page()
self.controller = WebviewController(self.page, self.webView)
self.controller = WebviewController(
self.page, [Entrance], self.webView, scheduler
)
# 加载初始页面
self.load_current_page()