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[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 ]