Maas-Agent/apps/plugin/services/plugin_registry.py

# coding=utf-8
"""
插件注册服务
提供插件的注册、查询、更新等核心功能
"""
from typing import List, Optional
from django.db import models
from django.db.models import QuerySet
from django.utils import timezone
from plugin.models.plugin import Plugin, PluginStatus, PluginType
from plugin.models.plugin_version import PluginVersion, PluginVersionStatus
from common.utils.logger import maxkb_logger


class PluginRegistryService:
    """插件注册服务"""

    @staticmethod
    def get_plugins(workspace_id: str = 'default', plugin_type: str = None,
                    status: str = None, is_installed: bool = None) -> QuerySet:
        """获取插件列表"""
        queryset = Plugin.objects.filter(workspace_id=workspace_id)
        if plugin_type:
            queryset = queryset.filter(plugin_type=plugin_type)
        if status:
            queryset = queryset.filter(status=status)
        if is_installed is not None:
            queryset = queryset.filter(is_installed=is_installed)
        return queryset.order_by('-create_time')

    @staticmethod
    def get_plugin(plugin_id: str) -> Optional[Plugin]:
        """获取单个插件"""
        try:
            return Plugin.objects.get(id=plugin_id)
        except Plugin.DoesNotExist:
            return None

    @staticmethod
    def get_plugin_by_code(code: str) -> Optional[Plugin]:
        """通过编码获取插件"""
        try:
            return Plugin.objects.get(code=code)
        except Plugin.DoesNotExist:
            return None

    @staticmethod
    def register_plugin(workspace_id: str, name: str, code: str,
                        plugin_type: str = PluginType.CUSTOM,
                        desc: str = '', version: str = '1.0.0',
                        author: str = '', icon: str = '',
                        schema: dict = None, config: dict = None,
                        entry_point: str = '', user_id: str = None) -> Plugin:
        """注册插件"""
        # 检查编码是否已存在
        if Plugin.objects.filter(code=code).exists():
            raise ValueError(f'插件编码 {code} 已存在')

        plugin = Plugin.objects.create(
            workspace_id=workspace_id,
            name=name,
            code=code,
            plugin_type=plugin_type,
            desc=desc,
            version=version,
            author=author,
            icon=icon,
            schema=schema or {},
            config=config or {},
            entry_point=entry_point,
            user_id=user_id
        )

        # 创建初始版本
        PluginVersion.objects.create(
            plugin=plugin,
            version=version,
            status=PluginVersionStatus.ACTIVE,
            schema=schema or {},
            config=config or {}
        )

        return plugin

    @staticmethod
    def update_plugin(plugin_id: str, **kwargs) -> Optional[Plugin]:
        """更新插件"""
        plugin = PluginRegistryService.get_plugin(plugin_id)
        if not plugin:
            return None

        for key, value in kwargs.items():
            if hasattr(plugin, key):
                setattr(plugin, key, value)

        plugin.save()
        return plugin

    @staticmethod
    def install_plugin(plugin_id: str) -> Optional[Plugin]:
        """安装插件"""
        plugin = PluginRegistryService.get_plugin(plugin_id)
        if not plugin:
            return None

        plugin.is_installed = True
        plugin.install_time = timezone.now()
        plugin.status = PluginStatus.ACTIVE
        plugin.save()
        return plugin

    @staticmethod
    def uninstall_plugin(plugin_id: str) -> Optional[Plugin]:
        """卸载插件"""
        plugin = PluginRegistryService.get_plugin(plugin_id)
        if not plugin:
            return None

        plugin.is_installed = False
        plugin.status = PluginStatus.INACTIVE
        plugin.save()
        return plugin

    @staticmethod
    def enable_plugin(plugin_id: str) -> Optional[Plugin]:
        """启用插件"""
        plugin = PluginRegistryService.get_plugin(plugin_id)
        if not plugin:
            return None

        plugin.status = PluginStatus.ACTIVE
        plugin.save()
        return plugin

    @staticmethod
    def disable_plugin(plugin_id: str) -> Optional[Plugin]:
        """禁用插件"""
        plugin = PluginRegistryService.get_plugin(plugin_id)
        if not plugin:
            return None

        plugin.status = PluginStatus.INACTIVE
        plugin.save()
        return plugin

    @staticmethod
    def delete_plugin(plugin_id: str) -> bool:
        """删除插件"""
        plugin = PluginRegistryService.get_plugin(plugin_id)
        if not plugin:
            return False

        # 删除关联的版本
        PluginVersion.objects.filter(plugin=plugin).delete()
        plugin.delete()
        return True

    @staticmethod
    def get_plugin_versions(plugin_id: str) -> QuerySet:
        """获取插件版本列表"""
        return PluginVersion.objects.filter(plugin_id=plugin_id).order_by('-create_time')

    @staticmethod
    def create_plugin_version(plugin_id: str, version: str,
                              changelog: str = '', schema: dict = None,
                              config: dict = None, file_path: str = '',
                              file_hash: str = '') -> Optional[PluginVersion]:
        """创建插件版本"""
        plugin = PluginRegistryService.get_plugin(plugin_id)
        if not plugin:
            return None

        plugin_version = PluginVersion.objects.create(
            plugin=plugin,
            version=version,
            changelog=changelog,
            schema=schema or plugin.schema,
            config=config or plugin.config,
            file_path=file_path,
            file_hash=file_hash
        )

        # 更新插件当前版本
        plugin.version = version
        plugin.save()

        return plugin_version

    @staticmethod
    def rollback_plugin_version(plugin_id: str, version_id: str) -> Optional[Plugin]:
        """回滚插件版本"""
        plugin = PluginRegistryService.get_plugin(plugin_id)
        if not plugin:
            return None

        try:
            target_version = PluginVersion.objects.get(id=version_id, plugin=plugin)
        except PluginVersion.DoesNotExist:
            return None

        # 更新插件配置为目标版本
        plugin.version = target_version.version
        plugin.schema = target_version.schema
        plugin.config = target_version.config
        plugin.save()

        # 标记回滚
        target_version.is_rollback = True
        target_version.save()

        return plugin

    @staticmethod
    def search_plugins(workspace_id: str, query: str) -> QuerySet:
        """搜索插件"""
        return Plugin.objects.filter(
            workspace_id=workspace_id
        ).filter(
            models.Q(name__icontains=query) | models.Q(code__icontains=query)
        ).order_by('-create_time')