mirror of
https://github.com/open-webui/open-webui
synced 2024-11-24 21:13:59 +00:00
46 lines
1.5 KiB
Python
46 lines
1.5 KiB
Python
from importlib import util
|
|
import os
|
|
|
|
from config import TOOLS_DIR, FUNCTIONS_DIR
|
|
|
|
|
|
def load_toolkit_module_by_id(toolkit_id):
|
|
toolkit_path = os.path.join(TOOLS_DIR, f"{toolkit_id}.py")
|
|
spec = util.spec_from_file_location(toolkit_id, toolkit_path)
|
|
module = util.module_from_spec(spec)
|
|
|
|
try:
|
|
spec.loader.exec_module(module)
|
|
print(f"Loaded module: {module.__name__}")
|
|
if hasattr(module, "Tools"):
|
|
return module.Tools()
|
|
else:
|
|
raise Exception("No Tools class found")
|
|
except Exception as e:
|
|
print(f"Error loading module: {toolkit_id}")
|
|
# Move the file to the error folder
|
|
os.rename(toolkit_path, f"{toolkit_path}.error")
|
|
raise e
|
|
|
|
|
|
def load_function_module_by_id(function_id):
|
|
function_path = os.path.join(FUNCTIONS_DIR, f"{function_id}.py")
|
|
|
|
spec = util.spec_from_file_location(function_id, function_path)
|
|
module = util.module_from_spec(spec)
|
|
|
|
try:
|
|
spec.loader.exec_module(module)
|
|
print(f"Loaded module: {module.__name__}")
|
|
if hasattr(module, "Pipe"):
|
|
return module.Pipe(), "pipe"
|
|
elif hasattr(module, "Filter"):
|
|
return module.Filter(), "filter"
|
|
else:
|
|
raise Exception("No Function class found")
|
|
except Exception as e:
|
|
print(f"Error loading module: {function_id}")
|
|
# Move the file to the error folder
|
|
os.rename(function_path, f"{function_path}.error")
|
|
raise e
|