open-webui/backend/open_webui/apps/webui/utils.py

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"{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_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.")