Maas-Agent/apps/plugin/views/plugin_view.py

# 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)