diff --git a/framework/fit/python/fit_cli/commands/build_cmd.py b/framework/fit/python/fit_cli/commands/build_cmd.py new file mode 100644 index 00000000..c0d43643 --- /dev/null +++ b/framework/fit/python/fit_cli/commands/build_cmd.py @@ -0,0 +1,115 @@ +import json +import shutil +import uuid +import tarfile +from datetime import datetime +from pathlib import Path +from fit_cli.utils.build import calculate_checksum, parse_python_file, type_errors + +def generate_tools_json(base_dir: Path, plugin_name: str): + """生成 tools.json""" + global type_errors + type_errors.clear() + + src_dir = base_dir / "src" + if not src_dir.exists(): + print(f"❌ 未找到插件目录 {src_dir}") + return None + + build_dir = base_dir / "build" + if not build_dir.exists(): + build_dir.mkdir(exist_ok=True) + + tools_json = { + "version": "1.0.0", + "definitionGroups": [], + "toolGroups": [] + } + # 遍历src目录下的所有.py文件 + for py_file in src_dir.glob("**/*.py"): + # 跳过__init__.py文件 + if py_file.name == "__init__.py": + continue + # 解析 Python 文件 + definition_group, tool_groups = parse_python_file(py_file) + if definition_group is not None: + tools_json["definitionGroups"].append(definition_group) + if len(tool_groups) > 0: + tools_json["toolGroups"].extend(tool_groups) + + if type_errors: + print("❌ tools.json 类型校验失败:") + for err in set(type_errors): + print(f" - {err}") + print("请修改为支持的类型:int, float, str, bool, dict, list, tuple, set, bytes") + return None # 终止构建 + + path = build_dir / "tools.json" + path.write_text(json.dumps(tools_json, indent=2, ensure_ascii=False), encoding="utf-8") + print(f"✅ 已生成 {path}") + return tools_json + + +def generate_plugin_json(base_dir: Path, plugin_name: str): + """生成 plugin.json""" + build_dir = base_dir / "build" + tar_path = build_dir / f"{plugin_name}.tar" + if not tar_path.exists(): + print(f"❌ TAR 文件 {tar_path} 不存在,请先打包源代码") + return None + + checksum = calculate_checksum(tar_path) + timestamp = datetime.now().strftime("%Y%m%d%H%M%S") + short_uuid = str(uuid.uuid4())[:8] + unique_name = f"{plugin_name}-{timestamp}-{short_uuid}" + + plugin_json = { + "checksum": checksum, + "name": plugin_name, + "description": f"{plugin_name} 插件", + "type": "python", + "uniqueness": { + "name": unique_name + } + } + path = build_dir / "plugin.json" + path.write_text(json.dumps(plugin_json, indent=2, ensure_ascii=False), encoding="utf-8") + print(f"✅ 已生成 {path}") + return plugin_json + + +def make_plugin_tar(base_dir: Path, plugin_name: str): + """打包源代码为 tar 格式""" + build_dir = base_dir / "build" + if not build_dir.exists(): + build_dir.mkdir(exist_ok=True) + + tar_path = build_dir / f"{plugin_name}.tar" + plugin_dir = base_dir + + with tarfile.open(tar_path, "w") as tar: + # 遍历插件目录下的所有文件 + for item in plugin_dir.rglob("*"): + # 排除build目录及其内容 + if "build" in item.parts: + continue + + if item.is_file(): + arcname = Path(plugin_name) / item.relative_to(plugin_dir) + tar.add(item, arcname=arcname) + print(f"✅ 已打包源代码 {tar_path}") + + +def run(args): + """build 命令入口""" + base_dir = Path("plugin") / args.name + plugin_name = args.name + + if not base_dir.exists(): + print(f"❌ 插件目录 {base_dir} 不存在,请先运行 fit_cli init {args.name}") + return + # 生成 tools.json + tools_json = generate_tools_json(base_dir, plugin_name) + if tools_json is not None: + make_plugin_tar(base_dir, plugin_name) # 打包源代码 + generate_plugin_json(base_dir, plugin_name) # 生成 plugin.json \ No newline at end of file diff --git a/framework/fit/python/fit_cli/commands/init_cmd.py b/framework/fit/python/fit_cli/commands/init_cmd.py index e3552b34..c3851de6 100644 --- a/framework/fit/python/fit_cli/commands/init_cmd.py +++ b/framework/fit/python/fit_cli/commands/init_cmd.py @@ -13,7 +13,7 @@ def hello(name: str) -> str: # 定义可供调用的函数,特别注意需要 修改函数名和参数 - 函数名(hello)应根据功能调整,例如 concat, multiply - - 参数(name: str)可以增加多个,类型也可以是 int, float 等 + - 参数(name: str)可以增加多个,类型支持 int, float, str, bool, dict, list, tuple, set, bytes, Union 等 """ return f"Hello, {name}!" # 提供函数实现逻辑 diff --git a/framework/fit/python/fit_cli/commands/package_cmd.py b/framework/fit/python/fit_cli/commands/package_cmd.py new file mode 100644 index 00000000..c4a8c7df --- /dev/null +++ b/framework/fit/python/fit_cli/commands/package_cmd.py @@ -0,0 +1,45 @@ +import zipfile +from pathlib import Path + +def package_to_zip(plugin_name: str): + """将 build 生成的文件打包为 zip""" + base_dir = Path("plugin") / plugin_name + build_dir = base_dir / "build" + + # 待打包的文件列表 + files_to_zip = [ + build_dir / f"{plugin_name}.tar", + build_dir / "tools.json", + build_dir / "plugin.json" + ] + + # 检查文件是否存在 + missing_files = [str(f) for f in files_to_zip if not f.exists()] + if missing_files: + print(f"❌ 缺少以下文件,请先执行 build 命令:{', '.join(missing_files)}") + return None + + # 打包文件 + zip_path = build_dir / f"{plugin_name}_package.zip" + with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf: + for file in files_to_zip: + zipf.write(file, arcname=file.name) + + print(f"✅ 已生成打包文件:{zip_path}") + return zip_path + + +def run(args): + """package 命令入口""" + plugin_name = args.name + base_dir = Path("plugin") / plugin_name + build_dir = base_dir / "build" + + if not base_dir.exists(): + print(f"❌ 插件目录 {base_dir} 不存在,请先运行 init 和 build 命令") + return + if not build_dir.exists(): + print(f"❌ 构建目录 {build_dir} 不存在,请先运行 build 命令") + return + + package_to_zip(plugin_name) \ No newline at end of file diff --git a/framework/fit/python/fit_cli/main.py b/framework/fit/python/fit_cli/main.py index 120207f2..995340e5 100644 --- a/framework/fit/python/fit_cli/main.py +++ b/framework/fit/python/fit_cli/main.py @@ -1,5 +1,7 @@ import argparse from fit_cli.commands import init_cmd +from fit_cli.commands import build_cmd +from fit_cli.commands import package_cmd def main(): parser = argparse.ArgumentParser(description="FIT Framework CLI 插件开发工具") @@ -10,6 +12,16 @@ def main(): parser_init.add_argument("name", help="插件目录名称") parser_init.set_defaults(func=init_cmd.run) + # build + parser_build = subparsers.add_parser("build", help="构建插件,生成 tools.json / plugin.json") + parser_build.add_argument("name", help="插件目录名称") + parser_build.set_defaults(func=build_cmd.run) + + # package + parser_package = subparsers.add_parser("package", help="将插件文件打包") + parser_package.add_argument("name", help="插件目录名称") + parser_package.set_defaults(func=package_cmd.run) + args = parser.parse_args() if hasattr(args, "func"): diff --git a/framework/fit/python/fit_cli/readme.md b/framework/fit/python/fit_cli/readme.md index 7ae782cf..4af651bb 100644 --- a/framework/fit/python/fit_cli/readme.md +++ b/framework/fit/python/fit_cli/readme.md @@ -1,14 +1,61 @@ # FIT CLI 工具 -FIT CLI 工具是基于 **FIT Framework** 的命令行开发工具,提供插件初始化、打包、发布等功能,帮助用户快速开发和管理 FIT 插件。 +FIT CLI 工具是基于 **FIT Framework** 的命令行开发工具,提供插件初始化、构建、打包等功能,帮助用户快速开发和管理 FIT 插件。 --- ## 使用方式 +FIT CLI 支持 3 个核心子命令:init(初始化)、build(构建)、package(打包),以下是详细说明。 + +### init + 以 framework/fit/python 为项目根目录,运行: ```bash python -m fit_cli init %{your_plugin_name} ``` -将会在 plugin 目录中创建 %{your_plugin_name} 目录,并生成插件模板。 +· 参数:``%{your_plugin_name}``: 自定义插件名称 + +会在 ``plugin`` 目录中创建如下结构: + + └── plugin/ + └──%{your_plugin_name}/ + └── src/ + ├── __init__.py + └── plugin.py # 插件源码模板 + +### build + +在完成插件的开发后,执行 +```bash +python -m fit_cli build %{your_plugin_name} +``` +· 参数:``%{your_plugin_name}``: 插件目录名称 + +在 ``%{your_plugin_name}`` 目录生成: + + └──%{your_plugin_name}/ + └── build/ + ├── %{your_plugin_name}.tar # 插件源码打包文件(工具包)。 + ├── tools.json # 工具的元数据。 + └── plugin.json # 插件的完整性校验与唯一性校验以及插件的基本信息。 + +开发者可根据自己的需要,修改完善``tools.json`` 和 ``plugin.json`` 文件,比如修改 ``description`` 、 ``uniqueness``等条目。 + +### package + +在完成插件的构建后,执行 +```bash +python -m fit_cli package %{your_plugin_name} +``` +· 参数:``%{your_plugin_name}``: 插件目录名称 + +在 ``plugin/%{your_plugin_name}/build/`` 目录生成最终打包文件: ``%{your_plugin_name}_package.zip`` + +--- + +## 注意事项 + +1. 运行命令前,请切换至 framework/fit/python 项目根目录。 +2. 更多详细信息和使用说明,可参考 https://github.com/ModelEngine-Group/fit-framework 官方仓库。 \ No newline at end of file diff --git a/framework/fit/python/fit_cli/utils/__init__.py b/framework/fit/python/fit_cli/utils/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/framework/fit/python/fit_cli/utils/build.py b/framework/fit/python/fit_cli/utils/build.py new file mode 100644 index 00000000..27ad085d --- /dev/null +++ b/framework/fit/python/fit_cli/utils/build.py @@ -0,0 +1,271 @@ +import ast +import hashlib +from pathlib import Path + +TYPE_MAP = { + # 基础类型 + "int": "integer", + "float": "number", + "str": "string", + "bool": "boolean", + "bytes": "string", + # 容器类型 + "dict": "object", + "Dict": "object", + "Union": "object", + "list": "array", + "List": "array", + "tuple": "array", + "Tuple": "array", + "set": "array", + "Set": "array", + # 特殊类型 + "None": "null", +} + +type_errors = [] + +def parse_type(annotation): + """解析参数类型""" + global type_errors + + if annotation is None: + type_errors.append("缺少类型注解(必须显式指定参数类型)") + return "invalid", None, True + + elif isinstance(annotation, ast.Name): + if annotation.id in TYPE_MAP: + return TYPE_MAP[annotation.id], None, True + else: + type_errors.append(f"不支持的类型: {annotation.id}") + return "invalid", None, True + + elif isinstance(annotation, ast.Constant) and annotation.value is None: + return "null", None, False + + elif isinstance(annotation, ast.Subscript): + if isinstance(annotation.value, ast.Name): + container = annotation.value.id + + # List[int] + if container in ("list", "List"): + item_type, _, _ = parse_type(annotation.slice) + if item_type == "invalid": + type_errors.append(f"不支持的列表元素类型: {annotation.slice}") + return "invalid", None, True + return "array", {"type": item_type}, True + + # Dict[str, int] → object + elif container in ("dict", "Dict"): + return "object", None, True + + # Optional[int] + elif container == "Optional": + inner_type, inner_items, _ = parse_type(annotation.slice) + if inner_type == "invalid": + type_errors.append(f"不支持的Optional类型: {annotation.slice}") + return "invalid", None, False + return inner_type, inner_items, False + + # Union[str, int] + elif container == "Union": + return "object", None, True + + # Tuple[str] + elif container in ("tuple", "Tuple"): + items = [] + if isinstance(annotation.slice, ast.Tuple): + for elt in annotation.slice.elts: + item_type, _, _ = parse_type(elt) + if item_type == "invalid": + type_errors.append(f"不支持的元组元素类型: {ast.dump(elt)}") + return "invalid", None, True + items.append({"type":item_type}) + return "array", f"{items}", True + else: + item_type, _, _ = parse_type(annotation.slice) + if item_type == "invalid": + type_errors.append(f"不支持的元组元素类型: {ast.dump(annotation.slice)}") + return "invalid", None, True + return "array", {"type":item_type}, True + + # Set[int] + elif container in ("set", "Set"): + item_type, _, _ = parse_type(annotation.slice) + if item_type == "invalid": + type_errors.append(f"不支持的集合元素类型: {annotation.slice}") + return "invalid", None, True + return "array", {"type": item_type}, True + + + else: + type_errors.append(f"不支持的容器类型: {container}") + return "invalid", None, True + + type_errors.append(f"无法识别的类型: {ast.dump(annotation)}") + return "invalid", None, True + + +def parse_parameters(args): + """解析函数参数""" + properties = {} + order = [] + required = [] + + for arg in args.args: + arg_name = arg.arg + order.append(arg_name) + arg_type, items, is_required = parse_type(arg.annotation) + # 定义参数 + prop_def = { + "defaultValue": "", + "description": f"参数 {arg_name}", + "name": arg_name, + "type": arg_type, + **({"items": items} if items else {}), + "examples": "", + "required": is_required, + } + properties[arg_name] = prop_def + if is_required: + required.append(arg_name) + return properties, order, required + + +def parse_return(annotation): + """解析返回值类型""" + if not annotation: + return {"type": "string", "convertor": ""} + + return_type, items, _ = parse_type(annotation) + ret = { + "type": return_type, + **({"items": items} if items else {}), + "convertor": "" + } + return ret + + +def parse_python_file(file_path: Path): + """解析 *.py 文件, 提取 definition / tool """ + with open(file_path, "r", encoding="utf-8") as f: + source = f.read() + tree = ast.parse(source) + + py_name = file_path.stem + definitions = [] + tool_groups = [] + + for node in tree.body: + if isinstance(node, ast.FunctionDef): + func_name = node.name + # 默认描述 + description = f"执行 {func_name} 方法" + if node.body and isinstance(node.body[0], ast.Expr): + expr_value = node.body[0].value + # 同时判断两种字符串节点类型 + if isinstance(expr_value, (ast.Str, ast.Constant)): + # 提取字符串内容 + docstring = expr_value.s if isinstance(expr_value, ast.Str) else expr_value.value + if isinstance(docstring, str): # 确保是字符串类型 + # 按换行分割,过滤空行并取第一行 + lines = [line.strip() for line in docstring.split("\n") if line.strip()] + if lines: # 若有有效行,取第一行作为描述 + description = lines[0] + + # 装饰器取 genericableId, fitableId + genericable_id, fitable_id = "", "" + for deco in node.decorator_list: + if isinstance(deco, ast.Call) and getattr(deco.func, "id", "") == "fitable": + if len(deco.args) >= 2: + genericable_id = getattr(deco.args[0], "s", "") + fitable_id = getattr(deco.args[1], "s", "") + + if not (genericable_id and fitable_id): + continue + + # 解析参数和返回值 + properties, order, required = parse_parameters(node.args) + return_schema = parse_return(node.returns) + + # definition schema + definition_schema = { + "name": func_name, + "description": description, + "parameters": { + "type": "object", + "properties": properties, + "required": required, + }, + "order": order, + "return": return_schema, + } + definitions.append({"schema": definition_schema}) + + # tool schema + tool_schema = { + "name": func_name, + "description": description, + "parameters": { + "type": "object", + "properties": { + k: { + "name": v["name"], + "type": v["type"], + **({"items": v["items"]} if "items" in v else {}), + "required": False, # 工具里参数默认非必填 + } + for k, v in properties.items() + }, + "required": [], + }, + "order": order, + "return": { + "name": "", + "description": f"{func_name} 函数的返回值", + "type": return_schema["type"], + **({"items": return_schema["items"]} if "items" in return_schema else {}), + "convertor": "", + "examples": "", + }, + } + + # tool + tool = { + "namespace": func_name, + "schema": tool_schema, + "runnables": { + "FIT": {"genericableId": genericable_id, "fitableId": fitable_id} + }, + "extensions": {"tags": ["FIT"]}, + "definitionName": func_name, + } + + # toolGroup + tool_group = { + "name": f"Impl-{func_name}", + "summary": "", + "description": "", + "extensions": {}, + "definitionGroupName": py_name, + "tools": [tool], + } + tool_groups.append(tool_group) + + definition_group = { + "name": py_name, + "summary": "", + "description": "", + "extensions": {}, + "definitions": definitions, + } + return definition_group, tool_groups + + +def calculate_checksum(file_path: Path) -> str: + """计算文件的 sha256 哈希值""" + h = hashlib.sha256() + with open(file_path, "rb") as f: + for chunk in iter(lambda: f.read(8192), b""): + h.update(chunk) + return h.hexdigest()