zhaoxi
d30c7e37a6
##前端美化和bug修复 #### 💄 美化 - **前端美化**：对于整个前端效果进行了重新设计，现在的前端看起来会更立体。 #### 🐛 修复 - **前端演示**：修复了前端展示workflow列表的bug，但是workflow的具体条目显示由于序列化导致仍然有问题。 - **密钥修复**：对于secret_key现在在使用默认情况时，会强制生成一个安全的密钥。
90 lines
4.0 KiB
Python
90 lines
4.0 KiB
Python
# Copyright 2026 zhaoxi826
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
import importlib.util
|
|
import os
|
|
import sys
|
|
from typing import Callable, Dict, List
|
|
|
|
from pretor.utils.logger import get_logger
|
|
logger = get_logger('get_tool')
|
|
_tool_cache: Dict[str, Callable] = {}
|
|
|
|
|
|
def _get_tool_func(tool_name: str) -> Callable | None:
|
|
"""检索并获取特定的 tool func 数据集合或实例对象。
|
|
根据提供的查询条件或上下文凭证，从数据库、缓存或第三方服务中读取对应的资源状态。
|
|
Args: tool_name (str): 赋予该实体的人类可读名称或标题字符串，主要用于前端 UI 展示、日志记录或模糊检索。
|
|
Returns: (Callable | None): 经由当前业务模型加工处理后所输出的具体数据实例或领域模型对象。 """
|
|
func = _tool_cache.get(tool_name, None)
|
|
if func:
|
|
return func
|
|
|
|
app_root = "/app"
|
|
tool_plugin_dir = os.path.join(app_root, "pretor", "plugin", "tool_plugin", tool_name)
|
|
|
|
if not os.path.exists(tool_plugin_dir) or not os.path.isdir(tool_plugin_dir):
|
|
logger.error(f"Tool directory not found: {tool_plugin_dir}")
|
|
return None
|
|
|
|
init_file = os.path.join(tool_plugin_dir, "__init__.py")
|
|
if not os.path.exists(init_file):
|
|
logger.error(f"Tool init file not found: {init_file}")
|
|
return None
|
|
|
|
try:
|
|
module_name = f"pretor.plugin.tool_plugin.{tool_name}"
|
|
spec = importlib.util.spec_from_file_location(module_name, init_file)
|
|
if spec is None or spec.loader is None:
|
|
logger.error(f"Failed to create spec for {module_name}")
|
|
return None
|
|
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[module_name] = module
|
|
spec.loader.exec_module(module)
|
|
|
|
func = getattr(module, tool_name, None)
|
|
|
|
if not callable(func):
|
|
logger.error(f"Tool function '{tool_name}' not found or not callable in {module_name}")
|
|
return None
|
|
_tool_cache[tool_name] = func
|
|
return func
|
|
except Exception as e:
|
|
logger.error(f"Failed to load module {module_name}: {e}")
|
|
return None
|
|
|
|
def del_tool_cache(tool_name: str) -> None:
|
|
"""执行与 del tool cache 相关的核心业务流转操作。
|
|
该方法封装了具体的算法策略或状态控制逻辑，确保操作能够在事务上下文中被原子且一致地执行。
|
|
Args: tool_name (str): 赋予该实体的人类可读名称或标题字符串，主要用于前端 UI 展示、日志记录或模糊检索。
|
|
Returns: (None): 经由当前业务模型加工处理后所输出的具体数据实例或领域模型对象。 """
|
|
if tool_name in _tool_cache:
|
|
del _tool_cache[tool_name]
|
|
|
|
def load_tools_from_list(tool_names: List[str] | None) -> List[Callable]:
|
|
"""执行与 load tools from list 相关的核心业务流转操作。
|
|
该方法封装了具体的算法策略或状态控制逻辑，确保操作能够在事务上下文中被原子且一致地执行。
|
|
Args: tool_names (List[str] | None): 赋予该实体的人类可读名称或标题字符串，主要用于前端 UI 展示、日志记录或模糊检索。
|
|
Returns: (List[Callable]): 经过筛选、排序或分页处理后的实体对象列表集合。 """
|
|
if not tool_names:
|
|
return []
|
|
|
|
tool_list = []
|
|
for tool_name in tool_names:
|
|
tool_func = _get_tool_func(tool_name)
|
|
if tool_func:
|
|
tool_list.append(tool_func)
|
|
|
|
return tool_list |