dify/api/services/plugin/dependencies_analysis.py

import re
from collections.abc import Mapping
from typing import Any

from werkzeug.exceptions import NotFound

from configs import dify_config
from core.helper import marketplace
from core.plugin.entities.plugin import PluginDependency, PluginDependencyType, PluginInstallationSource
from core.plugin.impl.plugin import PluginInstaller
from core.trigger.constants import TRIGGER_PLUGIN_NODE_TYPE
from graphon.enums import BuiltinNodeTypes
from models.provider_ids import GenericProviderID, ModelProviderID, ToolProviderID

# Compile regex pattern for version extraction at module level for better performance
_VERSION_REGEX = re.compile(r":(?P<version>[0-9]+(?:\.[0-9]+){2}(?:[+-][0-9A-Za-z.-]+)?)(?:@|$)")


class DependenciesAnalysisService:
    @classmethod
    def analyze_tool_provider_reference(cls, provider: str) -> str:
        if re.fullmatch(r"[a-z0-9_-]+/[a-z0-9_-]+", provider):
            return provider
        return cls.analyze_tool_dependency(provider)

    @classmethod
    def extract_external_node_dependencies(cls, node_data: Mapping[str, Any]) -> list[str]:
        """Extract plugin IDs from workflow nodes that contain direct plugin references."""
        node_type = node_data.get("type")
        if node_type == TRIGGER_PLUGIN_NODE_TYPE:
            plugin_id = node_data.get("plugin_id")
            return [plugin_id] if isinstance(plugin_id, str) and plugin_id else []

        if node_type == BuiltinNodeTypes.DATASOURCE:
            plugin_id = node_data.get("plugin_id")
            if node_data.get("provider_type") != "local_file" and isinstance(plugin_id, str) and plugin_id:
                return [plugin_id]
            return []

        if node_type != BuiltinNodeTypes.AGENT or node_data.get("agent_node_kind") == "dify_agent":
            return []

        dependencies = []
        strategy_provider = node_data.get("agent_strategy_provider_name")
        if isinstance(strategy_provider, str) and strategy_provider:
            try:
                if re.fullmatch(r"[a-z0-9_-]+/[a-z0-9_-]+", strategy_provider):
                    dependencies.append(strategy_provider)
                else:
                    dependencies.append(GenericProviderID(strategy_provider).plugin_id)
            except (ValueError, NotFound):
                pass

        parameters = node_data.get("agent_parameters")
        if not isinstance(parameters, Mapping):
            return dependencies
        for parameter in parameters.values():
            if not isinstance(parameter, Mapping) or not isinstance(parameter.get("value"), list):
                continue
            for tool in parameter["value"]:
                if not isinstance(tool, Mapping):
                    continue
                provider_type = tool.get("provider_type", tool.get("type"))
                if provider_type not in (None, "builtin", "plugin"):
                    continue
                plugin_id = tool.get("plugin_id")
                if isinstance(plugin_id, str) and plugin_id:
                    dependencies.append(plugin_id)
                    continue
                provider = tool.get("provider_id") or tool.get("provider_name") or tool.get("provider")
                if not isinstance(provider, str):
                    continue
                try:
                    dependencies.append(cls.analyze_tool_provider_reference(provider))
                except (ValueError, NotFound):
                    pass
        return dependencies

    @classmethod
    def analyze_tool_dependency(cls, tool_id: str) -> str:
        """
        Analyze the dependency of a tool.

        Convert the tool id to the plugin_id
        """
        try:
            return ToolProviderID(tool_id).plugin_id
        except Exception as e:
            raise e

    @classmethod
    def analyze_model_provider_dependency(cls, model_provider_id: str) -> str:
        """
        Analyze the dependency of a model provider.

        Convert the model provider id to the plugin_id
        """
        try:
            return ModelProviderID(model_provider_id).plugin_id
        except Exception as e:
            raise e

    @classmethod
    def get_leaked_dependencies(cls, tenant_id: str, dependencies: list[PluginDependency]) -> list[PluginDependency]:
        """
        Check dependencies, returns the leaked dependencies in current workspace
        """
        required_plugin_unique_identifiers = []
        for dependency in dependencies:
            required_plugin_unique_identifiers.append(dependency.value.plugin_unique_identifier)

        manager = PluginInstaller()

        # get leaked dependencies
        missing_plugins = manager.fetch_missing_dependencies(tenant_id, required_plugin_unique_identifiers)
        missing_plugin_unique_identifiers = {plugin.plugin_unique_identifier: plugin for plugin in missing_plugins}

        leaked_dependencies = []
        for dependency in dependencies:
            unique_identifier = dependency.value.plugin_unique_identifier
            if unique_identifier in missing_plugin_unique_identifiers:
                # Extract version for Marketplace dependencies
                if dependency.type == PluginDependencyType.Marketplace:
                    version_match = _VERSION_REGEX.search(unique_identifier)
                    if version_match:
                        dependency.value.version = version_match.group("version")

                # Create and append the dependency (same for all types)
                leaked_dependencies.append(
                    PluginDependency(
                        type=dependency.type,
                        value=dependency.value,
                        current_identifier=missing_plugin_unique_identifiers[unique_identifier].current_identifier,
                    )
                )

        return leaked_dependencies

    @classmethod
    def generate_dependencies(cls, tenant_id: str, dependencies: list[str]) -> list[PluginDependency]:
        """
        Generate dependencies through the list of plugin ids
        """
        dependencies = list(set(dependencies))
        manager = PluginInstaller()
        plugins = manager.fetch_plugin_installation_by_ids(tenant_id, dependencies)
        result = []
        for plugin in plugins:
            if plugin.source == PluginInstallationSource.Github:
                result.append(
                    PluginDependency(
                        type=PluginDependencyType.Github,
                        value=PluginDependency.Github(
                            repo=plugin.meta["repo"],
                            version=plugin.meta["version"],
                            package=plugin.meta["package"],
                            github_plugin_unique_identifier=plugin.plugin_unique_identifier,
                        ),
                    )
                )
            elif plugin.source == PluginInstallationSource.Marketplace:
                result.append(
                    PluginDependency(
                        type=PluginDependencyType.Marketplace,
                        value=PluginDependency.Marketplace(
                            marketplace_plugin_unique_identifier=plugin.plugin_unique_identifier
                        ),
                    )
                )
            elif plugin.source == PluginInstallationSource.Package:
                result.append(
                    PluginDependency(
                        type=PluginDependencyType.Package,
                        value=PluginDependency.Package(plugin_unique_identifier=plugin.plugin_unique_identifier),
                    )
                )
            elif plugin.source == PluginInstallationSource.Remote:
                raise ValueError(
                    f"You used a remote plugin: {plugin.plugin_unique_identifier} in the app, please remove it first"
                    " if you want to export the DSL."
                )
            else:
                raise ValueError(f"Unknown plugin source: {plugin.source}")

        return result

    @classmethod
    def generate_latest_dependencies(cls, dependencies: list[str]) -> list[PluginDependency]:
        """
        Generate the latest version of dependencies
        """
        dependencies = list(set(dependencies))
        if not dify_config.MARKETPLACE_ENABLED:
            return []
        deps = marketplace.batch_fetch_plugin_manifests(dependencies)
        return [
            PluginDependency(
                type=PluginDependencyType.Marketplace,
                value=PluginDependency.Marketplace(marketplace_plugin_unique_identifier=dep.latest_package_identifier),
            )
            for dep in deps
        ]