275 lines
8.4 KiB
Python
275 lines
8.4 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
|
||
"""
|
||
内置函数管理模块
|
||
为任务编辑器提供系统内置函数信息
|
||
支持异步函数调用
|
||
"""
|
||
|
||
import asyncio
|
||
from typing import Dict, List, Any, Callable, Awaitable, Union, Optional
|
||
from utils.string_utils import (
|
||
is_blank, is_not_blank, is_any_blank, is_all_blank,
|
||
is_numeric, is_alpha, starts_with, ends_with,
|
||
contains, split, chop, chomp, index_of,
|
||
difference, replace, remove, right, left,
|
||
length, reverse
|
||
)
|
||
|
||
# 函数类型定义
|
||
AsyncFunc = Callable[..., Awaitable[Any]]
|
||
|
||
# 注册所有内置函数及其元数据
|
||
BUILT_IN_FUNCTIONS = [
|
||
{
|
||
"id": "isBlank()",
|
||
"name": "是否为空",
|
||
"description": "检查字符串是否为null、空字符串或只包含空白字符",
|
||
"syntax": "isBlank(str)",
|
||
"example": "isBlank(' ') // 返回 true",
|
||
"category": "string",
|
||
"func": is_blank
|
||
},
|
||
{
|
||
"id": "isNotBlank()",
|
||
"name": "是否不为空",
|
||
"description": "检查字符串是否不为空，与isBlank()相反",
|
||
"syntax": "isNotBlank(str)",
|
||
"example": "isNotBlank('hello') // 返回 true",
|
||
"category": "string",
|
||
"func": is_not_blank
|
||
},
|
||
{
|
||
"id": "isAnyBlank()",
|
||
"name": "是否任意一个为空",
|
||
"description": "检查多个字符串中是否有任意一个为空",
|
||
"syntax": "isAnyBlank(str1, str2, ...)",
|
||
"example": "isAnyBlank('hello', '') // 返回 true",
|
||
"category": "string",
|
||
"func": is_any_blank
|
||
},
|
||
{
|
||
"id": "isAllBlank()",
|
||
"name": "是否全部为空",
|
||
"description": "检查一个或多个字符串是否全部为空",
|
||
"syntax": "isAllBlank(str1, str2, ...)",
|
||
"example": "isAllBlank('', ' ') // 返回 true",
|
||
"category": "string",
|
||
"func": is_all_blank
|
||
},
|
||
{
|
||
"id": "isNumeric()",
|
||
"name": "是否为数字",
|
||
"description": "检查字符串是否为数字",
|
||
"syntax": "isNumeric(str)",
|
||
"example": "isNumeric('123') // 返回 true",
|
||
"category": "string",
|
||
"func": is_numeric
|
||
},
|
||
{
|
||
"id": "isAlpha()",
|
||
"name": "是否为字母",
|
||
"description": "检查字符串是否只包含字母",
|
||
"syntax": "isAlpha(str)",
|
||
"example": "isAlpha('abc') // 返回 true",
|
||
"category": "string",
|
||
"func": is_alpha
|
||
},
|
||
{
|
||
"id": "startsWith()",
|
||
"name": "以某字符开头",
|
||
"description": "检查字符串是否以某个字符开头",
|
||
"syntax": "startsWith(str, prefix)",
|
||
"example": "startsWith('hello', 'he') // 返回 true",
|
||
"category": "string",
|
||
"func": starts_with
|
||
},
|
||
{
|
||
"id": "endsWith()",
|
||
"name": "以某字符结尾",
|
||
"description": "判断字符串是否以某个字符结尾",
|
||
"syntax": "endsWith(str, suffix)",
|
||
"example": "endsWith('hello', 'lo') // 返回 true",
|
||
"category": "string",
|
||
"func": ends_with
|
||
},
|
||
{
|
||
"id": "contains()",
|
||
"name": "包含字符",
|
||
"description": "判断字符串是否包含某个字符",
|
||
"syntax": "contains(str, substr)",
|
||
"example": "contains('hello', 'll') // 返回 true",
|
||
"category": "string",
|
||
"func": contains
|
||
},
|
||
{
|
||
"id": "split()",
|
||
"name": "拆分字符串",
|
||
"description": "将字符串按照某个字符进行拆分，返回字符串数组",
|
||
"syntax": "split(str, separator)",
|
||
"example": "split('a,b,c', ',') // 返回 ['a', 'b', 'c']",
|
||
"category": "string",
|
||
"func": split
|
||
},
|
||
{
|
||
"id": "chop()",
|
||
"name": "去掉最后一个字符",
|
||
"description": "将字符串的最后一个字符去掉，返回去掉之后的字符串",
|
||
"syntax": "chop(str)",
|
||
"example": "chop('hello') // 返回 'hell'",
|
||
"category": "string",
|
||
"func": chop
|
||
},
|
||
{
|
||
"id": "chomp()",
|
||
"name": "去除尾部子字符串",
|
||
"description": "切除字符串最后面的换行符等",
|
||
"syntax": "chomp(str)",
|
||
"example": "chomp('hello\\n') // 返回 'hello'",
|
||
"category": "string",
|
||
"func": chomp
|
||
},
|
||
{
|
||
"id": "indexOf()",
|
||
"name": "查找索引位置",
|
||
"description": "获取某个字符在字符串中的索引位置",
|
||
"syntax": "indexOf(str, substr)",
|
||
"example": "indexOf('hello', 'l') // 返回 2",
|
||
"category": "string",
|
||
"func": index_of
|
||
},
|
||
{
|
||
"id": "difference()",
|
||
"name": "获取差异部分",
|
||
"description": "获取第二个字符串在第一个字符串中不一样的部分",
|
||
"syntax": "difference(str1, str2)",
|
||
"example": "difference('abc', 'abd') // 返回 'd'",
|
||
"category": "string",
|
||
"func": difference
|
||
},
|
||
{
|
||
"id": "replace",
|
||
"name": "替换字符",
|
||
"description": "将字符串中的某些字符替换成另外的字符",
|
||
"syntax": "replace(str, old, new)",
|
||
"example": "replace('hello', 'l', 'x') // 返回 'hexxo'",
|
||
"category": "string",
|
||
"func": replace
|
||
},
|
||
{
|
||
"id": "remove()",
|
||
"name": "删除字符",
|
||
"description": "删除字符串的某一部分",
|
||
"syntax": "remove(str, substr)",
|
||
"example": "remove('hello', 'll') // 返回 'heo'",
|
||
"category": "string",
|
||
"func": remove
|
||
},
|
||
{
|
||
"id": "right()",
|
||
"name": "截取右侧字符",
|
||
"description": "从字符串右边截取指定长度的子字符串",
|
||
"syntax": "right(str, length)",
|
||
"example": "right('hello', 2) // 返回 'lo'",
|
||
"category": "string",
|
||
"func": right
|
||
},
|
||
{
|
||
"id": "left()",
|
||
"name": "截取左侧字符",
|
||
"description": "截取字符串左侧指定长度的字符串",
|
||
"syntax": "left(str, length)",
|
||
"example": "left('hello', 2) // 返回 'he'",
|
||
"category": "string",
|
||
"func": left
|
||
},
|
||
{
|
||
"id": "length()",
|
||
"name": "获取字符串长度",
|
||
"description": "获得字符串长度",
|
||
"syntax": "length(str)",
|
||
"example": "length('hello') // 返回 5",
|
||
"category": "string",
|
||
"func": length
|
||
},
|
||
{
|
||
"id": "reverse()",
|
||
"name": "反转字符串",
|
||
"description": "反转字符串",
|
||
"syntax": "reverse(str)",
|
||
"example": "reverse('hello') // 返回 'olleh'",
|
||
"category": "string",
|
||
"func": reverse
|
||
}
|
||
]
|
||
|
||
# 创建函数字典以便根据ID快速查找
|
||
FUNCTION_MAP = {func["id"]: func for func in BUILT_IN_FUNCTIONS}
|
||
|
||
def get_function_list() -> List[Dict[str, Any]]:
|
||
"""
|
||
获取所有内置函数的元数据列表
|
||
|
||
Returns:
|
||
List[Dict[str, Any]]: 函数列表，不包含func实现
|
||
"""
|
||
return [{k: v for k, v in func.items() if k != 'func'} for func in BUILT_IN_FUNCTIONS]
|
||
|
||
def get_function_by_id(function_id: str) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
根据ID获取指定函数的元数据
|
||
|
||
Args:
|
||
function_id: 函数ID
|
||
|
||
Returns:
|
||
Optional[Dict[str, Any]]: 函数元数据，如果不存在则返回None
|
||
"""
|
||
return FUNCTION_MAP.get(function_id)
|
||
|
||
async def call_function(function_id: str, *args, **kwargs) -> Any:
|
||
"""
|
||
异步调用指定ID的函数
|
||
|
||
Args:
|
||
function_id: 函数ID
|
||
*args: 位置参数
|
||
**kwargs: 关键字参数
|
||
|
||
Returns:
|
||
Any: 调用函数的结果
|
||
|
||
Raises:
|
||
ValueError: 如果函数不存在
|
||
"""
|
||
func_data = FUNCTION_MAP.get(function_id)
|
||
if not func_data:
|
||
raise ValueError(f"内置函数 {function_id} 不存在")
|
||
|
||
# 获取异步函数并调用
|
||
async_func: AsyncFunc = func_data["func"]
|
||
return await async_func(*args, **kwargs)
|
||
|
||
def call_function_sync(function_id: str, *args, **kwargs) -> Any:
|
||
"""
|
||
同步调用指定ID的函数（在内部使用事件循环执行异步函数）
|
||
适用于非异步环境中调用异步函数
|
||
|
||
Args:
|
||
function_id: 函数ID
|
||
*args: 位置参数
|
||
**kwargs: 关键字参数
|
||
|
||
Returns:
|
||
Any: 调用函数的结果
|
||
|
||
Raises:
|
||
ValueError: 如果函数不存在
|
||
"""
|
||
# 创建一个新的事件循环来执行异步函数
|
||
loop = asyncio.new_event_loop()
|
||
try:
|
||
return loop.run_until_complete(call_function(function_id, *args, **kwargs))
|
||
finally:
|
||
loop.close() |