镜像自地址
https://github.com/a13zhy/a13rollcall.git
已同步 2026-10-20 11:27:47 +08:00
861 行
30 KiB
Python
861 行
30 KiB
Python
import os
|
||
import json
|
||
import sys
|
||
import traceback
|
||
from datetime import datetime
|
||
from collections import defaultdict
|
||
|
||
|
||
class PluginLogger:
|
||
def __init__(self, plugin_name, log_dir):
|
||
self.plugin_name = plugin_name
|
||
self.log_dir = log_dir
|
||
self.log_file = os.path.join(log_dir, f"{plugin_name}.log")
|
||
self._ensure_dir()
|
||
|
||
def _ensure_dir(self):
|
||
if not os.path.exists(self.log_dir):
|
||
os.makedirs(self.log_dir)
|
||
|
||
def _write(self, level, message):
|
||
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
log_line = f"[{timestamp}] [{level}] {message}\n"
|
||
try:
|
||
with open(self.log_file, "a", encoding="utf-8") as f:
|
||
f.write(log_line)
|
||
except Exception:
|
||
pass
|
||
|
||
def info(self, message):
|
||
self._write("INFO", message)
|
||
|
||
def warning(self, message):
|
||
self._write("WARNING", message)
|
||
|
||
def error(self, message):
|
||
self._write("ERROR", message)
|
||
|
||
def debug(self, message):
|
||
self._write("DEBUG", message)
|
||
|
||
def get_logs(self, lines=100):
|
||
if not os.path.exists(self.log_file):
|
||
return []
|
||
try:
|
||
with open(self.log_file, "r", encoding="utf-8") as f:
|
||
all_lines = f.readlines()
|
||
return all_lines[-lines:]
|
||
except Exception:
|
||
return []
|
||
|
||
def clear(self):
|
||
if os.path.exists(self.log_file):
|
||
os.remove(self.log_file)
|
||
|
||
|
||
class PluginDataManager:
|
||
def __init__(self, plugin_name, data_dir):
|
||
self.plugin_name = plugin_name
|
||
self.data_dir = data_dir
|
||
self.config_file = os.path.join(data_dir, "config.json")
|
||
self._ensure_dir()
|
||
|
||
def _ensure_dir(self):
|
||
if not os.path.exists(self.data_dir):
|
||
os.makedirs(self.data_dir)
|
||
|
||
def load_config(self):
|
||
if not os.path.exists(self.config_file):
|
||
return {}
|
||
try:
|
||
with open(self.config_file, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except Exception:
|
||
return {}
|
||
|
||
def save_config(self, config):
|
||
try:
|
||
with open(self.config_file, "w", encoding="utf-8") as f:
|
||
json.dump(config, f, ensure_ascii=False, indent=2)
|
||
return True
|
||
except Exception:
|
||
return False
|
||
|
||
def get(self, key, default=None):
|
||
config = self.load_config()
|
||
return config.get(key, default)
|
||
|
||
def set(self, key, value):
|
||
config = self.load_config()
|
||
config[key] = value
|
||
return self.save_config(config)
|
||
|
||
def get_data_file(self, filename):
|
||
return os.path.join(self.data_dir, filename)
|
||
|
||
def read_data(self, filename):
|
||
file_path = self.get_data_file(filename)
|
||
if not os.path.exists(file_path):
|
||
return None
|
||
try:
|
||
with open(file_path, "r", encoding="utf-8") as f:
|
||
return f.read()
|
||
except Exception:
|
||
return None
|
||
|
||
def write_data(self, filename, content):
|
||
file_path = self.get_data_file(filename)
|
||
try:
|
||
with open(file_path, "w", encoding="utf-8") as f:
|
||
f.write(content)
|
||
return True
|
||
except Exception:
|
||
return False
|
||
|
||
def read_json(self, filename):
|
||
content = self.read_data(filename)
|
||
if content:
|
||
try:
|
||
return json.loads(content)
|
||
except Exception:
|
||
return None
|
||
return None
|
||
|
||
def write_json(self, filename, data):
|
||
try:
|
||
content = json.dumps(data, ensure_ascii=False, indent=2)
|
||
return self.write_data(filename, content)
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
class ArcxParser:
|
||
MAGIC = "ARCX"
|
||
VERSION = 1
|
||
|
||
@staticmethod
|
||
def parse(file_path):
|
||
if not os.path.exists(file_path):
|
||
return None, "文件不存在"
|
||
try:
|
||
with open(file_path, "r", encoding="utf-8") as f:
|
||
content = f.read()
|
||
if content.startswith("ARCX"):
|
||
return ArcxParser._parse_binary(content)
|
||
else:
|
||
data = json.loads(content)
|
||
return ArcxParser._validate(data), None
|
||
except json.JSONDecodeError as e:
|
||
return None, f"JSON 解析错误：{str(e)}"
|
||
except Exception as e:
|
||
return None, f"解析失败：{str(e)}"
|
||
|
||
@staticmethod
|
||
def _parse_binary(content):
|
||
try:
|
||
lines = content.split("\n")
|
||
meta = {}
|
||
code_start = -1
|
||
for i, line in enumerate(lines):
|
||
if line.startswith("META:"):
|
||
meta_part = line[5:].strip()
|
||
meta = json.loads(meta_part)
|
||
elif line.startswith("CODE:"):
|
||
code_start = i + 1
|
||
break
|
||
code = "\n".join(lines[code_start:]) if code_start > 0 else ""
|
||
data = {
|
||
"meta": meta,
|
||
"code": code,
|
||
"config": meta.get("config", {})
|
||
}
|
||
return ArcxParser._validate(data), None
|
||
except Exception as e:
|
||
return None, f"二进制格式解析失败：{str(e)}"
|
||
|
||
@staticmethod
|
||
def _validate(data):
|
||
required = ["meta", "code"]
|
||
for key in required:
|
||
if key not in data:
|
||
data[key] = {} if key == "meta" else ""
|
||
meta = data.get("meta", {})
|
||
meta.setdefault("name", "未命名插件")
|
||
meta.setdefault("version", "1.0.0")
|
||
meta.setdefault("author", "未知")
|
||
meta.setdefault("description", "")
|
||
meta.setdefault("type", "general")
|
||
meta.setdefault("api_version", 1)
|
||
meta.setdefault("has_settings_page", False)
|
||
meta.setdefault("settings_page_title", "")
|
||
data["meta"] = meta
|
||
data.setdefault("config", {})
|
||
data.setdefault("enabled", True)
|
||
data.setdefault("installed_at", datetime.now().isoformat())
|
||
return data
|
||
|
||
@staticmethod
|
||
def create_arcx(name, description, code, config=None, author="自定义", version="1.0.0", has_settings_page=False):
|
||
data = {
|
||
"meta": {
|
||
"name": name,
|
||
"version": version,
|
||
"author": author,
|
||
"description": description,
|
||
"type": "general",
|
||
"api_version": 1,
|
||
"has_settings_page": has_settings_page,
|
||
"settings_page_title": name,
|
||
"config": config or {}
|
||
},
|
||
"code": code,
|
||
"config": config or {}
|
||
}
|
||
return json.dumps(data, ensure_ascii=False, indent=2)
|
||
|
||
|
||
class PluginContext:
|
||
def __init__(self, plugin, app, plugin_mgr):
|
||
self.plugin = plugin
|
||
self.app = app
|
||
self.plugin_mgr = plugin_mgr
|
||
self._variables = {}
|
||
self._ui_elements = []
|
||
self._event_handlers = defaultdict(list)
|
||
self._loaded_libs = []
|
||
plugin_name = plugin["meta"]["name"]
|
||
base_dir = os.path.dirname(os.path.abspath(__file__))
|
||
self.data_dir = os.path.join(base_dir, "plugins", "data", plugin_name)
|
||
self.libs_dir = os.path.join(self.data_dir, "libs")
|
||
self.log_dir = os.path.join(base_dir, "plugins", "logs")
|
||
self.logger = PluginLogger(plugin_name, self.log_dir)
|
||
self.data = PluginDataManager(plugin_name, self.data_dir)
|
||
self._load_plugin_libs()
|
||
self.logger.info(f"插件上下文初始化完成，数据目录：{self.data_dir}")
|
||
|
||
def _load_plugin_libs(self):
|
||
"""加载插件自带的依赖库到sys.path"""
|
||
if os.path.exists(self.libs_dir):
|
||
if self.libs_dir not in sys.path:
|
||
sys.path.insert(0, self.libs_dir)
|
||
self._loaded_libs.append(self.libs_dir)
|
||
self.logger.info(f"已加载插件依赖库目录：{self.libs_dir}")
|
||
# 检查libs目录下的内容
|
||
try:
|
||
lib_items = os.listdir(self.libs_dir)
|
||
self.logger.info(f"插件依赖库包含 {len(lib_items)} 个项目：{', '.join(lib_items[:10])}")
|
||
except Exception as e:
|
||
self.logger.warning(f"读取依赖库目录失败：{e}")
|
||
|
||
def get_libs_dir(self):
|
||
"""获取插件依赖库目录"""
|
||
return self.libs_dir
|
||
|
||
def has_lib(self, lib_name):
|
||
"""检查插件是否自带某个依赖库"""
|
||
lib_path = os.path.join(self.libs_dir, lib_name)
|
||
return os.path.exists(lib_path) or os.path.exists(lib_path + ".py")
|
||
|
||
def list_libs(self):
|
||
"""列出插件自带的所有依赖库"""
|
||
if not os.path.exists(self.libs_dir):
|
||
return []
|
||
try:
|
||
return os.listdir(self.libs_dir)
|
||
except Exception:
|
||
return []
|
||
|
||
def set(self, key, value):
|
||
self._variables[key] = value
|
||
self.logger.debug(f"设置变量 {key} = {value}")
|
||
|
||
def get(self, key, default=None):
|
||
return self._variables.get(key, default)
|
||
|
||
def show_message(self, title, message):
|
||
from PyQt5.QtWidgets import QMessageBox
|
||
self.logger.info(f"显示消息框：{title}")
|
||
QMessageBox.information(None, title, message)
|
||
|
||
def show_custom_dialog(self, title, content, buttons=None):
|
||
from PyQt5.QtWidgets import QDialog, QVBoxLayout, QLabel, QPushButton, QHBoxLayout
|
||
self.logger.info(f"显示自定义对话框：{title}")
|
||
dlg = QDialog()
|
||
dlg.setWindowTitle(title)
|
||
dlg.setMinimumSize(400, 300)
|
||
layout = QVBoxLayout(dlg)
|
||
if isinstance(content, str):
|
||
label = QLabel(content)
|
||
label.setWordWrap(True)
|
||
layout.addWidget(label)
|
||
btn_layout = QHBoxLayout()
|
||
buttons = buttons or ["确定"]
|
||
for btn_text in buttons:
|
||
btn = QPushButton(btn_text)
|
||
btn.clicked.connect(dlg.accept)
|
||
btn_layout.addWidget(btn)
|
||
layout.addLayout(btn_layout)
|
||
dlg.exec_()
|
||
return True
|
||
|
||
def get_config(self, key, default=None):
|
||
return self.data.get(key, default)
|
||
|
||
def set_config(self, key, value):
|
||
self.logger.info(f"保存配置 {key} = {value}")
|
||
return self.data.set(key, value)
|
||
|
||
def is_frozen(self):
|
||
"""检测是否在打包环境（PyInstaller）中运行"""
|
||
return getattr(sys, 'frozen', False)
|
||
|
||
def get_app_dir(self):
|
||
"""获取应用程序目录（兼容打包环境和源码运行）"""
|
||
if self.is_frozen():
|
||
return os.path.dirname(sys.executable)
|
||
return os.path.dirname(os.path.abspath(__file__))
|
||
|
||
def get_python_version(self):
|
||
"""获取Python版本信息"""
|
||
return {
|
||
"version": sys.version,
|
||
"version_info": {
|
||
"major": sys.version_info.major,
|
||
"minor": sys.version_info.minor,
|
||
"micro": sys.version_info.micro,
|
||
},
|
||
"executable": sys.executable,
|
||
"frozen": self.is_frozen()
|
||
}
|
||
|
||
def check_dependency(self, module_name):
|
||
"""检查依赖模块是否可用"""
|
||
try:
|
||
__import__(module_name)
|
||
return True
|
||
except ImportError:
|
||
return False
|
||
|
||
def get_available_modules(self):
|
||
"""获取常用模块的可用状态"""
|
||
common_modules = [
|
||
"os", "sys", "json", "re", "math", "random", "datetime",
|
||
"subprocess", "threading", "socket", "urllib", "http",
|
||
"PyQt5", "PyQt5.QtCore", "PyQt5.QtGui", "PyQt5.QtWidgets",
|
||
"numpy", "requests", "PIL", "cv2", "pandas", "matplotlib",
|
||
"websockets", "flask", "serial", "pygame"
|
||
]
|
||
available = {}
|
||
for module in common_modules:
|
||
available[module] = self.check_dependency(module)
|
||
return available
|
||
|
||
def get_resource_path(self, relative_path):
|
||
"""获取资源文件路径（兼容打包环境）"""
|
||
if self.is_frozen():
|
||
base_path = sys._MEIPASS if hasattr(sys, '_MEIPASS') else self.get_app_dir()
|
||
return os.path.join(base_path, relative_path)
|
||
return os.path.join(self.get_app_dir(), relative_path)
|
||
|
||
def get_students(self):
|
||
if hasattr(self.app, '_students'):
|
||
return self.app._students
|
||
return []
|
||
|
||
def get_current_student(self):
|
||
if hasattr(self.app, '_current_student'):
|
||
return self.app._current_student
|
||
return None
|
||
|
||
def get_last_draw_result(self):
|
||
if hasattr(self.app, '_last_draw_result') and self.app._last_draw_result:
|
||
return self.app._last_draw_result
|
||
return None
|
||
|
||
def get_window_state(self):
|
||
is_drawing = False
|
||
is_minimized = False
|
||
is_quick_mode = False
|
||
state = "main"
|
||
current_window = "主界面"
|
||
|
||
if hasattr(self.app, '_is_drawing'):
|
||
is_drawing = self.app._is_drawing
|
||
if hasattr(self.app, 'isMinimized'):
|
||
try:
|
||
is_minimized = self.app.isMinimized()
|
||
except Exception:
|
||
pass
|
||
if hasattr(self.app, '_quick_mode'):
|
||
is_quick_mode = self.app._quick_mode
|
||
|
||
if is_minimized:
|
||
state = "minimized"
|
||
current_window = "最小化"
|
||
elif is_drawing:
|
||
state = "drawing"
|
||
current_window = "正在抽取"
|
||
elif is_quick_mode:
|
||
state = "quick"
|
||
current_window = "快捷抽取"
|
||
else:
|
||
state = "main"
|
||
current_window = "主界面"
|
||
|
||
return {
|
||
"state": state,
|
||
"is_drawing": is_drawing,
|
||
"is_minimized": is_minimized,
|
||
"is_quick_mode": is_quick_mode,
|
||
"current_window": current_window
|
||
}
|
||
|
||
def trigger_draw(self):
|
||
self.logger.info("触发抽取")
|
||
if hasattr(self.app, '_start_draw'):
|
||
self.app._start_draw()
|
||
|
||
def add_status_bar_text(self, text):
|
||
self.logger.debug(f"状态栏文本：{text}")
|
||
if hasattr(self.app, 'statusBar'):
|
||
self.app.statusBar().showMessage(text, 5000)
|
||
|
||
def log(self, message):
|
||
self.logger.info(message)
|
||
|
||
def register_event_handler(self, event_name, handler):
|
||
self._event_handlers[event_name].append(handler)
|
||
self.logger.info(f"注册事件处理器：{event_name}")
|
||
|
||
def emit_event(self, event_name, *args, **kwargs):
|
||
self.logger.debug(f"触发事件：{event_name}")
|
||
for handler in self._event_handlers.get(event_name, []):
|
||
try:
|
||
handler(*args, **kwargs)
|
||
except Exception as e:
|
||
self.logger.error(f"事件处理器执行错误：{str(e)}")
|
||
|
||
def get_logs(self, lines=100):
|
||
return self.logger.get_logs(lines)
|
||
|
||
def clear_logs(self):
|
||
self.logger.clear()
|
||
self.logger.info("日志已清空")
|
||
|
||
def get_app_config(self, key, default=None):
|
||
if hasattr(self.app, 'config_mgr') and self.app.config_mgr:
|
||
try:
|
||
value = self.app.config_mgr.get(key)
|
||
if value is not None:
|
||
return value
|
||
except Exception:
|
||
pass
|
||
return default
|
||
|
||
def set_app_config(self, key, value):
|
||
if hasattr(self.app, 'config_mgr') and self.app.config_mgr:
|
||
self.app.config_mgr.set(key, value)
|
||
self.logger.info(f"修改应用配置 {key} = {value}")
|
||
return True
|
||
return False
|
||
|
||
def restart_application(self, delay=1000):
|
||
self.logger.info("请求重启应用")
|
||
if hasattr(self.app, '_restart_application'):
|
||
from PyQt5.QtCore import QTimer
|
||
QTimer.singleShot(delay, self.app._restart_application)
|
||
return True
|
||
return False
|
||
|
||
def get_theme(self):
|
||
if hasattr(self.app, 'theme_mgr') and self.app.theme_mgr:
|
||
if hasattr(self.app.theme_mgr, 'current_theme'):
|
||
return self.app.theme_mgr.current_theme
|
||
if hasattr(self.app.theme_mgr, '_current_theme'):
|
||
return self.app.theme_mgr._current_theme
|
||
return "light"
|
||
|
||
def set_theme(self, theme_name):
|
||
if hasattr(self.app, 'theme_mgr') and self.app.theme_mgr:
|
||
self.app.theme_mgr.set_theme(theme_name)
|
||
self.logger.info(f"切换主题到 {theme_name}")
|
||
return True
|
||
return False
|
||
|
||
def get_class_name(self):
|
||
if hasattr(self.app, '_get_class_name'):
|
||
return self.app._get_class_name()
|
||
return "课堂点名"
|
||
|
||
def show_toast(self, message, duration=2000):
|
||
self.logger.info(f"显示提示：{message}")
|
||
if hasattr(self.app, '_show_toast'):
|
||
self.app._show_toast(message, duration)
|
||
return True
|
||
return False
|
||
|
||
def render_markdown(self, markdown_text):
|
||
try:
|
||
import sys
|
||
import os
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
from main_qt import MarkdownRenderer
|
||
return MarkdownRenderer.render(markdown_text)
|
||
except Exception as e:
|
||
self.logger.error(f"Markdown 渲染失败：{str(e)}")
|
||
return markdown_text
|
||
|
||
def create_markdown_view(self, parent=None):
|
||
try:
|
||
import sys
|
||
import os
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
from main_qt import MarkdownView
|
||
return MarkdownView(parent)
|
||
except Exception as e:
|
||
self.logger.error(f"创建 Markdown 视图失败：{str(e)}")
|
||
from PyQt5.QtWidgets import QTextEdit
|
||
view = QTextEdit(parent)
|
||
view.setReadOnly(True)
|
||
return view
|
||
|
||
def get_plugin_pages(self):
|
||
meta = self.plugin.get("meta", {})
|
||
return meta.get("pages", [])
|
||
|
||
def build_plugin_page(self, page_key, parent=None):
|
||
try:
|
||
code = self.plugin.get("code", "")
|
||
if not code.strip():
|
||
self.logger.error(f"插件代码为空")
|
||
return None
|
||
exec_globals = self._get_exec_globals()
|
||
try:
|
||
exec(code, exec_globals)
|
||
except Exception as e:
|
||
self.logger.error(f"执行插件代码失败：{str(e)}\n{traceback.format_exc()}")
|
||
return None
|
||
func_name = f"build_page_{page_key}"
|
||
if func_name not in exec_globals:
|
||
available = [k for k in exec_globals.keys() if k.startswith('build_')]
|
||
self.logger.error(f"未找到函数 {func_name}，可用函数：{available}")
|
||
return None
|
||
func = exec_globals[func_name]
|
||
if not callable(func):
|
||
self.logger.error(f"{func_name} 不是可调用对象")
|
||
return None
|
||
try:
|
||
if parent:
|
||
return func(self, parent)
|
||
return func(self)
|
||
except Exception as e:
|
||
self.logger.error(f"调用 {func_name} 失败：{str(e)}\n{traceback.format_exc()}")
|
||
return None
|
||
except Exception as e:
|
||
self.logger.error(f"构建插件页面 {page_key} 失败：{str(e)}\n{traceback.format_exc()}")
|
||
return None
|
||
|
||
def _get_exec_globals(self):
|
||
return {
|
||
'context': self,
|
||
'plugin': self.plugin,
|
||
'app': self.app,
|
||
'print': self.logger.info,
|
||
'__name__': 'plugin_exec',
|
||
}
|
||
|
||
|
||
class PluginManager:
|
||
def __init__(self, config_mgr):
|
||
self.config_mgr = config_mgr
|
||
self.base_dir = os.path.dirname(os.path.abspath(__file__))
|
||
self.plugins_dir = os.path.join(self.base_dir, "plugins")
|
||
self.plugins_data_dir = os.path.join(self.plugins_dir, "data")
|
||
self.plugins_log_dir = os.path.join(self.plugins_dir, "logs")
|
||
self.plugins = {}
|
||
self.contexts = {}
|
||
self._ensure_dirs()
|
||
self.load_all_plugins()
|
||
|
||
def _ensure_dirs(self):
|
||
for d in [self.plugins_dir, self.plugins_data_dir, self.plugins_log_dir]:
|
||
if not os.path.exists(d):
|
||
os.makedirs(d)
|
||
|
||
def _get_state_file(self):
|
||
return os.path.join(self.plugins_dir, "plugin_states.json")
|
||
|
||
def _save_plugin_state(self, name, enabled):
|
||
try:
|
||
states = {}
|
||
state_file = self._get_state_file()
|
||
if os.path.exists(state_file):
|
||
with open(state_file, "r", encoding="utf-8") as f:
|
||
states = json.load(f)
|
||
states[name] = {"enabled": enabled, "updated": datetime.now().strftime("%Y-%m-%d %H:%M:%S")}
|
||
with open(state_file, "w", encoding="utf-8") as f:
|
||
json.dump(states, f, ensure_ascii=False, indent=2)
|
||
except Exception as e:
|
||
print(f"保存插件状态失败：{e}")
|
||
|
||
def _load_plugin_states(self):
|
||
try:
|
||
state_file = self._get_state_file()
|
||
if not os.path.exists(state_file):
|
||
return {}
|
||
with open(state_file, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except Exception as e:
|
||
print(f"加载插件状态失败：{e}")
|
||
return {}
|
||
|
||
def load_all_plugins(self):
|
||
self.plugins = {}
|
||
if not os.path.exists(self.plugins_dir):
|
||
return
|
||
states = self._load_plugin_states()
|
||
for filename in os.listdir(self.plugins_dir):
|
||
if filename.endswith(".arcx"):
|
||
file_path = os.path.join(self.plugins_dir, filename)
|
||
plugin, error = ArcxParser.parse(file_path)
|
||
if plugin:
|
||
plugin["file_path"] = file_path
|
||
plugin["filename"] = filename
|
||
plugin_name = plugin["meta"]["name"]
|
||
if plugin_name in states:
|
||
plugin["enabled"] = states[plugin_name].get("enabled", True)
|
||
else:
|
||
plugin["enabled"] = True
|
||
self.plugins[plugin_name] = plugin
|
||
else:
|
||
print(f"插件加载失败 {filename}: {error}")
|
||
|
||
def get_plugin(self, name):
|
||
return self.plugins.get(name)
|
||
|
||
def get_all_plugins(self):
|
||
return list(self.plugins.values())
|
||
|
||
def get_enabled_plugins(self):
|
||
return [p for p in self.plugins.values() if p.get("enabled", True)]
|
||
|
||
def get_plugins_with_settings(self):
|
||
return [p for p in self.get_enabled_plugins() if p["meta"].get("has_settings_page", False)]
|
||
|
||
def get_plugin_pages(self, plugin_name):
|
||
plugin = self.get_plugin(plugin_name)
|
||
if not plugin:
|
||
return []
|
||
meta = plugin.get("meta", {})
|
||
pages = meta.get("pages", [])
|
||
if not pages and meta.get("has_settings_page", False):
|
||
pages = [{"key": "settings", "title": "设置", "icon": "settings"}]
|
||
return pages
|
||
|
||
def build_plugin_page(self, plugin_name, page_key, app, parent=None):
|
||
context = self.get_context(plugin_name, app)
|
||
if not context:
|
||
return None
|
||
return context.build_plugin_page(page_key, parent)
|
||
|
||
def enable_plugin(self, name):
|
||
if name in self.plugins:
|
||
self.plugins[name]["enabled"] = True
|
||
self._call_lifecycle(name, "on_enable")
|
||
return True
|
||
return False
|
||
|
||
def disable_plugin(self, name):
|
||
if name in self.plugins:
|
||
self._call_lifecycle(name, "on_disable")
|
||
self.plugins[name]["enabled"] = False
|
||
return True
|
||
return False
|
||
|
||
def install_plugin(self, source_path):
|
||
if not os.path.exists(source_path):
|
||
return False, "源文件不存在"
|
||
plugin, error = ArcxParser.parse(source_path)
|
||
if not plugin:
|
||
return False, error
|
||
filename = os.path.basename(source_path)
|
||
dest_path = os.path.join(self.plugins_dir, filename)
|
||
import shutil
|
||
shutil.copy2(source_path, dest_path)
|
||
plugin["file_path"] = dest_path
|
||
plugin["filename"] = filename
|
||
self.plugins[plugin["meta"]["name"]] = plugin
|
||
self._call_lifecycle(plugin["meta"]["name"], "on_install")
|
||
return True, f"插件 {plugin['meta']['name']} 安装成功"
|
||
|
||
def uninstall_plugin(self, name):
|
||
if name not in self.plugins:
|
||
return False, "插件不存在"
|
||
self._call_lifecycle(name, "on_uninstall")
|
||
plugin = self.plugins[name]
|
||
file_path = plugin.get("file_path")
|
||
if file_path and os.path.exists(file_path):
|
||
os.remove(file_path)
|
||
data_dir = os.path.join(self.plugins_data_dir, name)
|
||
if os.path.exists(data_dir):
|
||
import shutil
|
||
shutil.rmtree(data_dir, ignore_errors=True)
|
||
del self.plugins[name]
|
||
if name in self.contexts:
|
||
del self.contexts[name]
|
||
return True, f"插件 {name} 已卸载"
|
||
|
||
def get_context(self, name, app):
|
||
if name not in self.plugins:
|
||
return None
|
||
if name not in self.contexts:
|
||
self.contexts[name] = PluginContext(self.plugins[name], app, self)
|
||
return self.contexts[name]
|
||
|
||
def execute_plugin(self, name, app, entry_point="main"):
|
||
plugin = self.get_plugin(name)
|
||
if not plugin:
|
||
return None, "插件不存在"
|
||
if not plugin.get("enabled", True):
|
||
return None, "插件已禁用"
|
||
context = self.get_context(name, app)
|
||
try:
|
||
code = plugin.get("code", "")
|
||
if not code.strip():
|
||
return None, "插件代码为空"
|
||
exec_globals = {
|
||
"context": context,
|
||
"plugin": plugin,
|
||
"app": app,
|
||
"config": plugin.get("config", {}),
|
||
"print": context.log,
|
||
"__name__": f"plugin_{name}"
|
||
}
|
||
context.logger.info(f"开始执行插件，入口：{entry_point}")
|
||
exec(code, exec_globals)
|
||
if entry_point in exec_globals and callable(exec_globals[entry_point]):
|
||
result = exec_globals[entry_point](context)
|
||
context.logger.info(f"插件执行完成，返回：{result}")
|
||
return result, None
|
||
context.logger.info("插件执行完成（无入口函数）")
|
||
return None, None
|
||
except Exception as e:
|
||
error_msg = f"插件执行错误：{str(e)}\n{traceback.format_exc()}"
|
||
context.logger.error(error_msg)
|
||
return None, error_msg
|
||
|
||
def _call_lifecycle(self, name, method_name):
|
||
plugin = self.get_plugin(name)
|
||
if not plugin or not plugin.get("enabled", True):
|
||
return
|
||
context = self.contexts.get(name)
|
||
if not context:
|
||
return
|
||
try:
|
||
code = plugin.get("code", "")
|
||
exec_globals = {
|
||
"context": context,
|
||
"plugin": plugin,
|
||
"print": context.log,
|
||
"__name__": f"plugin_{name}"
|
||
}
|
||
exec(code, exec_globals)
|
||
if method_name in exec_globals and callable(exec_globals[method_name]):
|
||
context.logger.info(f"调用生命周期方法：{method_name}")
|
||
exec_globals[method_name](context)
|
||
except Exception as e:
|
||
context.logger.error(f"生命周期方法 {method_name} 执行错误：{str(e)}")
|
||
|
||
def build_plugin_settings_page(self, name, app):
|
||
plugin = self.get_plugin(name)
|
||
if not plugin:
|
||
return None
|
||
context = self.get_context(name, app)
|
||
try:
|
||
code = plugin.get("code", "")
|
||
exec_globals = {
|
||
"context": context,
|
||
"plugin": plugin,
|
||
"print": context.log,
|
||
"__name__": f"plugin_{name}"
|
||
}
|
||
exec(code, exec_globals)
|
||
if "build_settings_page" in exec_globals and callable(exec_globals["build_settings_page"]):
|
||
context.logger.info("构建设置页面")
|
||
return exec_globals["build_settings_page"](context)
|
||
except Exception as e:
|
||
context.logger.error(f"构建设置页面错误：{str(e)}")
|
||
return None
|
||
|
||
def create_sample_plugin(self):
|
||
sample_code = '''
|
||
def main(context):
|
||
context.show_message("示例插件", "这是一个示例插件！\\n\\n你可以在这里编写自己的插件逻辑。")
|
||
students = context.get_students()
|
||
context.log(f"当前有 {len(students)} 名学生")
|
||
context.set_config("last_run", str(datetime.now()))
|
||
return {"status": "success", "student_count": len(students)}
|
||
|
||
|
||
def build_settings_page(context):
|
||
from PyQt5.QtWidgets import QWidget, QVBoxLayout, QLabel, QPushButton, QLineEdit
|
||
page = QWidget()
|
||
layout = QVBoxLayout(page)
|
||
layout.setContentsMargins(20, 20, 20, 20)
|
||
layout.setSpacing(12)
|
||
|
||
title = QLabel("示例插件设置")
|
||
title.setStyleSheet("font-size: 18px; font-weight: 700; color: #0067c0;")
|
||
layout.addWidget(title)
|
||
|
||
desc = QLabel("在这里可以配置示例插件的各项参数。\\n配置会自动保存到插件数据目录的 config.json 中。")
|
||
desc.setWordWrap(True)
|
||
desc.setStyleSheet("color: #666666; font-size: 12px;")
|
||
layout.addWidget(desc)
|
||
|
||
input_label = QLabel("欢迎语：")
|
||
input_label.setStyleSheet("font-size: 13px; font-weight: 600;")
|
||
layout.addWidget(input_label)
|
||
|
||
welcome_input = QLineEdit()
|
||
welcome_input.setText(context.get_config("welcome_message", "欢迎使用示例插件！"))
|
||
layout.addWidget(welcome_input)
|
||
|
||
def save_config():
|
||
context.set_config("welcome_message", welcome_input.text())
|
||
context.show_message("保存成功", "配置已保存！")
|
||
|
||
save_btn = QPushButton("保存配置")
|
||
save_btn.setStyleSheet("QPushButton { background: #0067c0; color: white; border: none; border-radius: 6px; padding: 10px 20px; font-weight: 600; }")
|
||
save_btn.clicked.connect(save_config)
|
||
layout.addWidget(save_btn)
|
||
|
||
layout.addStretch()
|
||
return page
|
||
|
||
|
||
def on_enable(context):
|
||
context.log("插件已启用")
|
||
|
||
|
||
def on_disable(context):
|
||
context.log("插件已禁用")
|
||
|
||
|
||
def on_install(context):
|
||
context.log("插件已安装")
|
||
|
||
|
||
def on_uninstall(context):
|
||
context.log("插件已卸载")
|
||
'''
|
||
return ArcxParser.create_arcx(
|
||
name="示例插件",
|
||
description="这是一个示例插件，展示插件的基本结构、设置页面、日志系统和数据存储。",
|
||
code=sample_code,
|
||
config={"welcome_message": "欢迎使用示例插件！", "show_student_count": True},
|
||
author="A13",
|
||
version="1.0.0",
|
||
has_settings_page=True
|
||
)
|