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

modify: h5 插件系统更新

Browse Source
This commit is contained in:
chakcy
2025-09-13 12:13:52 +08:00
parent 3631b383d6
commit f94c4b57e4
43 changed files with 16649 additions and 1151 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/view/web_view/entrance.py
+245 -53
View File
@@ -1,41 +1,100 @@
from PySide6.QtCore import QObject, Slot, QUrl
from PySide6.QtWebChannel import QWebChannel
from PySide6.QtWebEngineCore import QWebEngineSettings
import time
import os
import time
import zipfile
from pathlib import Path
import json
import shutil
import base64
import tempfile
from pathlib import Path
from PySide6.QtCore import QObject, Slot, QUrl
from PySide6.QtWebChannel import QWebChannel
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
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):
"""Webview桥接类，用于处理前端和Python后端之间的通信"""
def __init__(self, webview, parent=None):
super().__init__(parent)
self.webview = webview
# 启用本地文件访问
self.webview.settings().setAttribute(QWebEngineSettings.WebAttribute.LocalContentCanAccessFileUrls, True)
self.webview.settings().setAttribute(QWebEngineSettings.WebAttribute.LocalContentCanAccessRemoteUrls, True)
self.webview.settings().setAttribute(QWebEngineSettings.WebAttribute.AllowRunningInsecureContent, True)
self.webview.settings().setAttribute(QWebEngineSettings.WebAttribute.AllowGeolocationOnInsecureOrigins, True)
self.webview.settings().setAttribute(
QWebEngineSettings.WebAttribute.LocalContentCanAccessFileUrls, True
)
self.webview.settings().setAttribute(
QWebEngineSettings.WebAttribute.LocalContentCanAccessRemoteUrls, True
)
self.webview.settings().setAttribute(
QWebEngineSettings.WebAttribute.AllowRunningInsecureContent, True
)
self.webview.settings().setAttribute(
QWebEngineSettings.WebAttribute.AllowGeolocationOnInsecureOrigins, True
)
# 设置Web安全策略
self.webview.page().settings().setAttribute(QWebEngineSettings.WebAttribute.LocalContentCanAccessFileUrls, True)
self.webview.page().settings().setAttribute(QWebEngineSettings.WebAttribute.LocalContentCanAccessRemoteUrls, True)
self.webview.page().settings().setAttribute(QWebEngineSettings.WebAttribute.AllowRunningInsecureContent, True)
self.webview.page().settings().setAttribute(
QWebEngineSettings.WebAttribute.LocalContentCanAccessFileUrls, True
)
self.webview.page().settings().setAttribute(
QWebEngineSettings.WebAttribute.LocalContentCanAccessRemoteUrls, True
)
self.webview.page().settings().setAttribute(
QWebEngineSettings.WebAttribute.AllowRunningInsecureContent, True
)
# 创建插件目录
self.plugins_dir = Path(os.getcwd()) / "plugins"
self.plugins_dir.mkdir(exist_ok=True)
self.plugins_dir = Path("plugins")
os.makedirs(self.plugins_dir, exist_ok=True)
# 插件信息文件
self.plugins_info_file = self.plugins_dir / "plugins.json"
if not self.plugins_info_file.exists():
with open(self.plugins_info_file, 'w') as f:
with open(self.plugins_info_file, "w") as f:
json.dump([], f)
@Slot(str, str, str, result=str)
@@ -44,55 +103,181 @@ class WebviewBridge(QObject):
try:
# 将base64编码的文件内容解码并保存为临时文件
file_data = base64.b64decode(file_content)
# 创建临时文件来保存上传的ZIP文件
with tempfile.NamedTemporaryFile(suffix='.zip', delete=False) as tmp_file:
with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as tmp_file:
tmp_file.write(file_data)
tmp_file_path = tmp_file.name
# 判断插件是否已存在
if os.path.exists(self.plugins_dir / name):
os.unlink(tmp_file_path)
return json.dumps({"success": False, "error": "插件名称已存在"})
# 解压插件
success, error = self.extract_plugin(tmp_file_path, name)
success, msg = self.extract_plugin(tmp_file_path, name)
if not success:
# 清理临时文件
os.unlink(tmp_file_path)
return json.dumps({"success": False, "error": error})
return json.dumps({"success": False, "error": msg})
# 读取 plugin.config.json 文件
plugin_config_file = msg.replace("index.html", "plugin.config.json")
if not os.path.exists(plugin_config_file):
os.unlink(tmp_file_path)
return json.dumps({"success": False, "error": "插件配置文件不存在"})
with open(plugin_config_file, "r") as f:
plugin_config = json.load(f)
plugin_conofig_keys = [
"pluginName",
"pluginId",
"pluginDescription",
"pluginVersion",
"pluginAuthor",
]
for key in plugin_conofig_keys:
if key not in plugin_config:
os.unlink(tmp_file_path)
return json.dumps(
{"success": False, "error": "插件配置文件缺少必要字段"}
)
plugin_info = {
"id": int(time.time()), # 使用时间戳作为简单ID
"id": plugin_config["pluginId"],
"name": name,
"pluginName": plugin_config["pluginName"],
"description": description,
"version": "1.0.0",
"uploadDate": time.strftime('%Y-%m-%d', time.localtime()),
"path": str(self.plugins_dir / name)
"version": plugin_config["pluginVersion"],
"author": plugin_config["pluginAuthor"],
"uploadDate": time.strftime("%Y-%m-%d", time.localtime()),
"path": str(self.plugins_dir / name / plugin_config["pluginVersion"]),
}
# 创建版本路径并移动插件文件
version_path = self.plugins_dir / name / plugin_info["version"]
version_path.mkdir(exist_ok=True)
for file in os.listdir(os.path.dirname(msg)):
shutil.move(os.path.join(os.path.dirname(msg), file), version_path)
# 读取现有插件信息
if self.plugins_info_file.exists():
with open(self.plugins_info_file, 'r') as f:
if os.path.exists(self.plugins_info_file):
with open(self.plugins_info_file, "r") as f:
plugins = json.load(f)
else:
plugins = []
# 判断pluginId是否存在
if plugin_info["id"] in [p["id"] for p in plugins]:
os.unlink(tmp_file_path)
return json.dumps({"success": False, "error": "插件ID已存在"})
# 添加新插件
plugins.append(plugin_info)
# 保存插件信息
with open(self.plugins_info_file, 'w') as f:
with open(self.plugins_info_file, "w") as f:
json.dump(plugins, f)
# 清理临时文件
os.unlink(tmp_file_path)
return json.dumps({"success": True})
except Exception as e:
# 确保任何异常都返回有效的JSON
return json.dumps({"success": False, "error": str(e)})
return json.dumps({"success": False, "error": str(e)})
@Slot(str, result=str)
def update_plugin(self, plugin_id, name, description, file_content):
"""处理插件更新请求"""
try:
file_data = base64.b64decode(file_content)
with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as tmp_file:
tmp_file.write(file_data)
tmp_filepath = tmp_file.name
success, msg = self.extract_plugin(tmp_filepath, name)
if not success:
os.unlink(tmp_filepath)
return json.dumps({"success": False, "error": msg})
plugin_config_file = msg.replace("index.html", "plugin.config.json")
if not os.path.exists(plugin_config_file):
os.unlink(tmp_filepath)
return json.dumps({"success": False, "error": "插件配置文件不存在"})
with open(plugin_config_file, "r") as f:
plugin_config = json.load(f)
plugin_config_keys = [
"pluginName",
"pluginId",
"pluginDescription",
"pluginVersion",
"pluginAuthor",
]
for key in plugin_config_keys:
if key not in plugin_config:
os.unlink(tmp_filepath)
return json.dumps(
{"success": False, "error": "插件配置文件缺少必要字段"}
)
# 判断插件是否存在
if not os.path.exists(self.plugins_dir / name):
os.unlink(tmp_filepath)
return json.dumps({"success": False, "error": "插件不存在"})
# 判断版本是否已存在
version_path = self.plugins_dir / name / plugin_config["pluginVersion"]
if version_path.exists():
os.unlink(tmp_filepath)
return json.dumps({"success": False, "error": "版本已存在"})
# 创建版本路径并移动插件文件
version_path.mkdir(exist_ok=True)
for file in os.listdir(os.path.dirname(msg)):
shutil.move(os.path.join(os.path.dirname(msg), file), version_path)
# 更新插件信息
with open(self.plugins_info_file, "r") as f:
plugins = json.load(f)
plugin_info = next((p for p in plugins if p["id"] == plugin_id), None)
if plugin_info:
plugin_info["name"] = name
plugin_info["description"] = description
plugin_info["version"] = [
p["version"] for p in plugins if p["id"] == plugin_id
][0]
plugin_info["uploadDate"] = time.strftime("%Y-%m-%d", time.localtime())
plugin_info["path"] = str(
self.plugins_dir / name / plugin_config["pluginVersion"]
)
# 保存插件信息
with open(self.plugins_info_file, "w") as f:
json.dump(plugins, f)
# 清理临时文件
os.unlink(tmp_filepath)
return json.dumps({"success": True})
except Exception as e:
# 确保任何异常都返回有效的JSON
return json.dumps({"success": False, "error": str(e)})
# 解压插件
def extract_plugin(self, plugin_file, plugin_name):
"""解压插件"""
try:
with zipfile.ZipFile(plugin_file, 'r') as zf:
with zipfile.ZipFile(plugin_file, "r") as zf:
# 获取插件名称
plugin_dir = self.plugins_dir / plugin_name
@@ -112,8 +297,8 @@ class WebviewBridge(QObject):
if not html_files:
return False, "插件中未找到HTML文件"
index_file = html_files[0]
return True, ""
return True, str(index_file.absolute())
except Exception as e:
return False, str(e)
@@ -121,8 +306,8 @@ class WebviewBridge(QObject):
def list_plugins(self):
"""获取插件列表"""
try:
if self.plugins_info_file.exists():
with open(self.plugins_info_file, 'r') as f:
if os.path.exists(self.plugins_info_file):
with open(self.plugins_info_file, "r") as f:
plugins = json.load(f)
return json.dumps({"success": True, "plugins": plugins})
else:
@@ -130,17 +315,18 @@ class WebviewBridge(QObject):
except Exception as e:
# 确保任何异常都返回有效的JSON
return json.dumps({"success": False, "error": str(e)})
@Slot(int, result=str)
@Slot(str, result=str)
def launch_plugin(self, plugin_id):
"""启动插件"""
try:
if self.plugins_info_file.exists():
with open(self.plugins_info_file, 'r') as f:
if os.path.exists(self.plugins_info_file):
with open(self.plugins_info_file, "r") as f:
plugins = json.load(f)
# 查找插件
plugin = next((p for p in plugins if p["id"] == plugin_id), None)
print(f"Launching plugin: {plugin}")
if plugin:
# 构建插件URL
plugin_path = Path(plugin["path"])
@@ -150,10 +336,13 @@ class WebviewBridge(QObject):
index_file = html_files[0]
url = QUrl.fromLocalFile(str(index_file.absolute()))
print(f"Launching plugin: {url.toString()}")
print(f"Plugin path: {plugin_path}")
self.webview.load(url)
return json.dumps({"success": True, "url": url.toString()})
else:
return json.dumps({"success": False, "error": "插件入口文件未找到"})
return json.dumps(
{"success": False, "error": "插件入口文件未找到"}
)
else:
return json.dumps({"success": False, "error": "插件未找到"})
else:
@@ -161,7 +350,7 @@ class WebviewBridge(QObject):
except Exception as e:
# 确保任何异常都返回有效的JSON
return json.dumps({"success": False, "error": str(e)})
@Slot(str, result=str)
def save_settings(self, settings_json):
"""保存设置"""
@@ -173,14 +362,17 @@ 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('py', self.bridge)
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
+53 -121
View File
@@ -2,12 +2,21 @@
import sys
import os
from pathlib import Path
from PySide6.QtWidgets import QFrame, QVBoxLayout, QMenu, QApplication, QWidget, QPushButton, QHBoxLayout
from qframelesswindow.webengine import FramelessWebEngineView
from PySide6.QtCore import QPoint, Qt, QUrl
from ...config import config
from PySide6.QtWidgets import (
QFrame,
QVBoxLayout,
QMenu,
QApplication,
QWidget,
QPushButton,
QHBoxLayout,
)
from PySide6.QtCore import QPoint, Qt, QUrl
from qframelesswindow.webengine import FramelessWebEngineView
from .entrance import WebviewController
from ...config import config
class DevToolsWindow(FramelessWebEngineView):
@@ -16,6 +25,7 @@ class DevToolsWindow(FramelessWebEngineView):
self.setWindowTitle("开发者工具")
self.resize(600, 600)
class DevToolsView(QWidget):
def __init__(self, parent=None):
super().__init__(parent=parent)
@@ -25,6 +35,7 @@ class DevToolsView(QWidget):
# 设置窗口大小不可变
self.setFixedSize(self.size())
class WebViewInterface(QFrame):
def __init__(self, parent=None):
@@ -33,37 +44,42 @@ class WebViewInterface(QFrame):
self.dev_tools_view = None
# 添加连接状态标志
self.is_local_connection = config.get("project", {}).get("is-production", True)
# 插件目录
self.plugins_dir = Path(os.getcwd()) / "plugins"
self.plugins_dir.mkdir(exist_ok=True)
self.webView = FramelessWebEngineView(self)
self.html_path = os.getcwd().replace("\\", '/') + '/resources/webview/entrance/dist/index.html'
self.remote_url = "http://localhost:8080/"
self.html_path = (
os.getcwd().replace("\\", "/")
+ "/resources/webview/entrance/dist/index.html"
)
self.remote_url = config.get("project", {}).get(
"remote-url", "http://localhost:8080/"
)
# 创建WebviewController实例
self.page = self.webView.page()
self.controller = WebviewController(self.page, self.webView)
# 加载初始页面
self.load_current_page()
# 创建导航按钮
self.back_button = QPushButton("⬅️", self)
self.back_button.clicked.connect(self.webView.back)
self.back_button.setFixedSize(40, 30)
self.back_button.setEnabled(False)
self.forward_button = QPushButton("➡️", self)
self.forward_button.clicked.connect(self.webView.forward)
self.forward_button.setFixedSize(40, 30)
self.forward_button.setEnabled(False)
self.reload_button = QPushButton("🔃", self)
self.reload_button.clicked.connect(self.webView.reload)
self.reload_button.setFixedSize(40, 30)
self.home_button = QPushButton("🏠", self)
self.home_button.clicked.connect(self.go_home)
self.home_button.setFixedSize(40, 30)
@@ -87,7 +103,7 @@ class WebViewInterface(QFrame):
self.webView.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu)
self.webView.loadFinished.connect(self.on_load_finished)
self.webView.customContextMenuRequested.connect(self.show_context_menu)
# 应用样式
self.apply_styles()
@@ -114,8 +130,10 @@ class WebViewInterface(QFrame):
color: #adb5bd;
}
"""
nav_button_style = button_style + """
nav_button_style = (
button_style
+ """
QPushButton {
background-color: #6c757d;
}
@@ -126,8 +144,11 @@ class WebViewInterface(QFrame):
background-color: #495057;
}
"""
toggle_button_style = button_style + """
)
toggle_button_style = (
button_style
+ """
QPushButton {
background-color: #0d6efd;
}
@@ -138,7 +159,8 @@ class WebViewInterface(QFrame):
background-color: #0a58ca;
}
"""
)
self.back_button.setStyleSheet(nav_button_style)
self.forward_button.setStyleSheet(nav_button_style)
self.reload_button.setStyleSheet(nav_button_style)
@@ -178,35 +200,35 @@ class WebViewInterface(QFrame):
"""显示自定义右键菜单"""
# 创建上下文菜单
menu = QMenu(self)
# 添加标准浏览器动作
back_action = menu.addAction("后退")
back_action.triggered.connect(self.webView.back)
forward_action = menu.addAction("前进")
forward_action.triggered.connect(self.webView.forward)
reload_action = menu.addAction("重新加载")
reload_action.triggered.connect(self.webView.reload)
menu.addSeparator()
# 添加开发者工具相关动作
dev_tools_action = menu.addAction("打开开发者工具")
dev_tools_action.triggered.connect(self.toggle_dev_tools)
menu.addSeparator()
menu.addSeparator()
# 在鼠标位置显示菜单
menu.exec(self.webView.mapToGlobal(pos))
def toggle_dev_tools(self):
"""切换开发者工具窗口显示状态"""
if not self.dev_tools_view:
self.init_dev_tools()
if self.dev_tools_view is None:
return
@@ -217,97 +239,7 @@ class WebViewInterface(QFrame):
self.dev_tools_view.show()
self.dev_tools_view.dev_tools_window.show()
self.dev_tools_view.dev_tools_window.raise_() # 将窗口置于最前
def generate_plugins_page(self):
"""生成插件列表页面"""
plugins = []
for item in self.plugins_dir.iterdir():
if item.is_dir():
index_files = list(item.glob("*.html"))
if index_files:
plugins.append({
'name': item.name,
'path': f"file:///{item.absolute()}/{index_files[0].name}",
'description': f"插件 {item.name}"
})
# 生成HTML页面
html = """
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>插件管理</title>
<style>
body {
font-family: Arial, sans-serif;
margin: 20px;
background-color: #f5f5f5;
}
.header {
text-align: center;
margin-bottom: 30px;
}
.plugins-grid {
display: grid;
grid-template-columns: repeat(auto-fill, minmax(300px, 1fr));
gap: 20px;
}
.plugin-card {
background: white;
border-radius: 8px;
padding: 20px;
box-shadow: 0 2px 10px rgba(0,0,0,0.1);
}
.plugin-card h3 {
margin-top: 0;
color: #333;
}
.plugin-card p {
color: #666;
}
.plugin-card a {
display: inline-block;
margin-top: 10px;
padding: 8px 16px;
background-color: #007bff;
color: white;
text-decoration: none;
border-radius: 4px;
}
.plugin-card a:hover {
background-color: #0056b3;
}
</style>
</head>
<body>
<div class="header">
<h1>插件管理</h1>
<p>当前可用插件列表</p>
</div>
<div class="plugins-grid">
"""
if not plugins:
html += '<p>暂无插件，请上传插件。</p>'
else:
for plugin in plugins:
html += f"""
<div class="plugin-card">
<h3>{plugin['name']}</h3>
<p>{plugin['description']}</p>
<a href="{plugin['path']}" target="_blank">访问插件</a>
</div>
"""
html += """
</div>
</body>
</html>
"""
return html
def closeEvent(self, event):
"""关闭时清理资源"""
# 停止controller中的定时器
@@ -322,4 +254,4 @@ if __name__ == "__main__":
window = WebViewInterface()
window.resize(1000, 700)
window.show()
sys.exit(app.exec())
sys.exit(app.exec())