# coding=utf-8 """ 插件视图 提供插件的 CRUD API """ from django.utils.translation import gettext as _ from rest_framework.request import Request from rest_framework.views import APIView from common.auth import TokenAuth from common.exception.app_exception import AppApiException from common.result import result from plugin.services.plugin_registry import PluginRegistryService class PluginView(APIView): """插件管理""" authentication_classes = [TokenAuth] class List(APIView): """获取插件列表""" authentication_classes = [TokenAuth] def get(self, request: Request, workspace_id: str): plugin_type = request.query_params.get('plugin_type') status = request.query_params.get('status') is_installed = request.query_params.get('is_installed') if is_installed is not None: is_installed = is_installed == 'true' plugins = PluginRegistryService.get_plugins( workspace_id=workspace_id, plugin_type=plugin_type, status=status, is_installed=is_installed ) data = [{ 'id': str(p.id), 'name': p.name, 'code': p.code, 'desc': p.desc, 'plugin_type': p.plugin_type, 'status': p.status, 'version': p.version, 'author': p.author, 'icon': p.icon, 'is_installed': p.is_installed, 'install_time': p.install_time.isoformat() if p.install_time else None, 'create_time': p.create_time.isoformat() if p.create_time else None, } for p in plugins] return result.success(data) class Create(APIView): """创建插件""" authentication_classes = [TokenAuth] def post(self, request: Request, workspace_id: str): name = request.data.get('name', '') code = request.data.get('code', '') plugin_type = request.data.get('plugin_type', 'CUSTOM') desc = request.data.get('desc', '') version = request.data.get('version', '1.0.0') author = request.data.get('author', '') icon = request.data.get('icon', '') schema = request.data.get('schema', {}) config = request.data.get('config', {}) entry_point = request.data.get('entry_point', '') if not name or not code: raise AppApiException(400, _('插件名称和编码不能为空')) try: plugin = PluginRegistryService.register_plugin( workspace_id=workspace_id, name=name, code=code, plugin_type=plugin_type, desc=desc, version=version, author=author, icon=icon, schema=schema, config=config, entry_point=entry_point, user_id=str(request.user.id) if hasattr(request.user, 'id') else None ) except ValueError as e: raise AppApiException(400, str(e)) return result.success({ 'id': str(plugin.id), 'name': plugin.name, 'code': plugin.code, 'desc': plugin.desc, 'plugin_type': plugin.plugin_type, 'status': plugin.status, 'version': plugin.version, 'author': plugin.author, 'icon': plugin.icon, 'is_installed': plugin.is_installed, 'create_time': plugin.create_time.isoformat() if plugin.create_time else None, }) class Operate(APIView): """插件操作（更新/删除）""" authentication_classes = [TokenAuth] def get(self, request: Request, workspace_id: str, plugin_id: str): plugin = PluginRegistryService.get_plugin(plugin_id) if not plugin: raise AppApiException(404, _('插件不存在')) # 获取版本列表 versions = PluginRegistryService.get_plugin_versions(plugin_id) version_list = [{ 'id': str(v.id), 'version': v.version, 'status': v.status, 'changelog': v.changelog, 'is_rollback': v.is_rollback, 'create_time': v.create_time.isoformat() if v.create_time else None, } for v in versions] return result.success({ 'id': str(plugin.id), 'name': plugin.name, 'code': plugin.code, 'desc': plugin.desc, 'plugin_type': plugin.plugin_type, 'status': plugin.status, 'version': plugin.version, 'author': plugin.author, 'icon': plugin.icon, 'schema': plugin.schema, 'config': plugin.config, 'entry_point': plugin.entry_point, 'is_installed': plugin.is_installed, 'install_time': plugin.install_time.isoformat() if plugin.install_time else None, 'create_time': plugin.create_time.isoformat() if plugin.create_time else None, 'versions': version_list, }) def put(self, request: Request, workspace_id: str, plugin_id: str): name = request.data.get('name') desc = request.data.get('desc') plugin_type = request.data.get('plugin_type') author = request.data.get('author') icon = request.data.get('icon') schema = request.data.get('schema') config = request.data.get('config') entry_point = request.data.get('entry_point') plugin = PluginRegistryService.update_plugin( plugin_id=plugin_id, name=name, desc=desc, plugin_type=plugin_type, author=author, icon=icon, schema=schema, config=config, entry_point=entry_point ) if not plugin: raise AppApiException(404, _('插件不存在')) return result.success({ 'id': str(plugin.id), 'name': plugin.name, 'code': plugin.code, 'desc': plugin.desc, 'plugin_type': plugin.plugin_type, 'status': plugin.status, 'version': plugin.version, 'author': plugin.author, 'icon': plugin.icon, 'update_time': plugin.update_time.isoformat() if plugin.update_time else None, }) def delete(self, request: Request, workspace_id: str, plugin_id: str): success = PluginRegistryService.delete_plugin(plugin_id) if not success: raise AppApiException(404, _('插件不存在')) return result.success(True) class Install(APIView): """安装/卸载插件""" authentication_classes = [TokenAuth] def post(self, request: Request, workspace_id: str, plugin_id: str): action = request.data.get('action', 'install') if action == 'install': plugin = PluginRegistryService.install_plugin(plugin_id) elif action == 'uninstall': plugin = PluginRegistryService.uninstall_plugin(plugin_id) else: raise AppApiException(400, _('无效的操作')) if not plugin: raise AppApiException(404, _('插件不存在')) return result.success({ 'id': str(plugin.id), 'status': plugin.status, 'is_installed': plugin.is_installed, }) class EnableDisable(APIView): """启用/禁用插件""" authentication_classes = [TokenAuth] def post(self, request: Request, workspace_id: str, plugin_id: str): action = request.data.get('action', 'enable') if action == 'enable': plugin = PluginRegistryService.enable_plugin(plugin_id) elif action == 'disable': plugin = PluginRegistryService.disable_plugin(plugin_id) else: raise AppApiException(400, _('无效的操作')) if not plugin: raise AppApiException(404, _('插件不存在')) return result.success({ 'id': str(plugin.id), 'status': plugin.status, }) class Version(APIView): """版本管理""" authentication_classes = [TokenAuth] def get(self, request: Request, workspace_id: str, plugin_id: str): versions = PluginRegistryService.get_plugin_versions(plugin_id) data = [{ 'id': str(v.id), 'version': v.version, 'status': v.status, 'changelog': v.changelog, 'is_rollback': v.is_rollback, 'create_time': v.create_time.isoformat() if v.create_time else None, } for v in versions] return result.success(data) def post(self, request: Request, workspace_id: str, plugin_id: str): version = request.data.get('version', '') changelog = request.data.get('changelog', '') schema = request.data.get('schema') config = request.data.get('config') file_path = request.data.get('file_path', '') file_hash = request.data.get('file_hash', '') if not version: raise AppApiException(400, _('版本号不能为空')) plugin_version = PluginRegistryService.create_plugin_version( plugin_id=plugin_id, version=version, changelog=changelog, schema=schema, config=config, file_path=file_path, file_hash=file_hash ) if not plugin_version: raise AppApiException(404, _('插件不存在')) return result.success({ 'id': str(plugin_version.id), 'version': plugin_version.version, 'status': plugin_version.status, 'changelog': plugin_version.changelog, 'create_time': plugin_version.create_time.isoformat() if plugin_version.create_time else None, }) class Rollback(APIView): """版本回滚""" authentication_classes = [TokenAuth] def post(self, request: Request, workspace_id: str, plugin_id: str, version_id: str): plugin = PluginRegistryService.rollback_plugin_version(plugin_id, version_id) if not plugin: raise AppApiException(404, _('插件或版本不存在')) return result.success({ 'id': str(plugin.id), 'version': plugin.version, 'schema': plugin.schema, 'config': plugin.config, }) class Search(APIView): """搜索插件""" authentication_classes = [TokenAuth] def get(self, request: Request, workspace_id: str): query = request.query_params.get('query', '') if not query: raise AppApiException(400, _('搜索关键词不能为空')) plugins = PluginRegistryService.search_plugins(workspace_id, query) data = [{ 'id': str(p.id), 'name': p.name, 'code': p.code, 'desc': p.desc, 'plugin_type': p.plugin_type, 'status': p.status, 'version': p.version, 'author': p.author, 'icon': p.icon, 'is_installed': p.is_installed, } for p in plugins] return result.success(data)