import os import re import subprocess import sys from importlib import util import types from open_webui.apps.webui.models.functions import Functions from open_webui.apps.webui.models.tools import Tools from open_webui.config import FUNCTIONS_DIR, TOOLS_DIR def extract_frontmatter(file_path): """ Extract frontmatter as a dictionary from the specified file path. """ frontmatter = {} frontmatter_started = False frontmatter_ended = False frontmatter_pattern = re.compile(r"^\s*([a-z_]+):\s*(.*)\s*$", re.IGNORECASE) try: with open(file_path, "r", encoding="utf-8") as file: first_line = file.readline() if first_line.strip() != '"""': # The file doesn't start with triple quotes return {} frontmatter_started = True for line in file: if '"""' in line: if frontmatter_started: frontmatter_ended = True break if frontmatter_started and not frontmatter_ended: match = frontmatter_pattern.match(line) if match: key, value = match.groups() frontmatter[key.strip()] = value.strip() except FileNotFoundError: print(f"Error: The file {file_path} does not exist.") return {} except Exception as e: print(f"An error occurred: {e}") return {} return frontmatter def replace_imports(content): """ Replace the import paths in the content. """ replacements = { "from utils": "from open_webui.utils", "from apps": "from open_webui.apps", "from main": "from open_webui.main", "from config": "from open_webui.config", } for old, new in replacements.items(): content = content.replace(old, new) return content def load_toolkit_module_by_id(toolkit_id, content=None): if content is None: tool = Tools.get_tool_by_id(toolkit_id) if not tool: raise Exception(f"Toolkit not found: {toolkit_id}") content = tool.content content = replace_imports(content) Tools.update_tool_by_id(toolkit_id, {"content": content}) module_name = f"tool_{toolkit_id}" module = types.ModuleType(module_name) sys.modules[module_name] = module try: # Executing the modified content in the created module's namespace exec(content, module.__dict__) # Extract frontmatter, assuming content can be treated directly as a string frontmatter = extract_frontmatter( content ) # Ensure this method is adaptable to handle content strings # Install required packages found within the frontmatter install_frontmatter_requirements(frontmatter.get("requirements", "")) print(f"Loaded module: {module.__name__}") # Create and return the object if the class 'Tools' is found in the module if hasattr(module, "Tools"): return module.Tools(), frontmatter else: raise Exception("No Tools class found in the module") except Exception as e: print(f"Error loading module: {toolkit_id}") del sys.modules[module_name] # Clean up raise e def load_function_module_by_id(function_id, content=None): if content is None: function = Functions.get_function_by_id(function_id) if not function: raise Exception(f"Function not found: {function_id}") content = function.content content = replace_imports(content) Functions.update_function_by_id(function_id, {"content": content}) module_name = f"function_{function_id}" module = types.ModuleType(module_name) sys.modules[module_name] = module try: # Execute the modified content in the created module's namespace exec(content, module.__dict__) # Extract the frontmatter from the content, simulate file-like behaviour frontmatter = extract_frontmatter( content ) # This function needs to handle string inputs # Install necessary requirements specified in frontmatter install_frontmatter_requirements(frontmatter.get("requirements", "")) print(f"Loaded module: {module.__name__}") # Create appropriate object based on available class type in the module if hasattr(module, "Pipe"): return module.Pipe(), "pipe", frontmatter elif hasattr(module, "Filter"): return module.Filter(), "filter", frontmatter elif hasattr(module, "Action"): return module.Action(), "action", frontmatter else: raise Exception("No Function class found in the module") except Exception as e: print(f"Error loading module: {function_id}") del sys.modules[module_name] # Cleanup by removing the module in case of error Functions.update_function_by_id(function_id, {"is_active": False}) raise e def install_frontmatter_requirements(requirements): if requirements: req_list = [req.strip() for req in requirements.split(",")] for req in req_list: print(f"Installing requirement: {req}") subprocess.check_call([sys.executable, "-m", "pip", "install", req]) else: print("No requirements found in frontmatter.")