动态调用子流程
A
Auto_18752243871
2025-10-11 12:51
123
# 实在代码编辑器使用说明:
# 1. 默认引入了获取元素、使用全局变量、调用流程块等常用工具方法
# 2. 可在流程块、子流程中使用“调用Python模块”组件调用此Python模块
# 3. 更多使用帮助可按快捷键“ Alt + Shift + F1 ”
from common.util.shared_variables import GetSharedVariable
from common.util.elements_util import elementsFormatNew
from projects.rpaRoot import SZEnv
from ...global_data import SHIZAI_ELEMENT_DICT, globalVar
from ...global_data import run_module, print
from typing import Any, Dict, Optional
import os, json
def load_module_by_name(name: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""
根据模块名动态加载模块并调用其主函数
Args:
name: 模块名称
params: 传递给模块函数的参数字典,可为None
Returns:
模块函数的返回结果(字典格式)
"""
# 加载配置并验证模块名
module_dict = _load_config_from_parent_dir()
if name not in module_dict:
raise ValueError(f"模块名 '{name}' 不存在于配置中")
module_id = module_dict[name]["module_id"]
module_path = f"child_modules.{module_id}"
resp_data = module_dict[name]["resp_data"]
req_data = module_dict[name]["req_data"] # 获取请求参数定义
# 处理params为None的情况
if params is None:
params = {}
# 根据req_data生成正确的参数元组
param_tuple = _generate_param_tuple(params, req_data)
# 执行模块
print(f"调用模块名称:{name}")
print(f"调用模块参数:{params}")
result = run_module({"module_path": module_path}, "main", SZEnv['rpa'], *param_tuple)
# 将tuple结果转换为字典
result_dict = _convert_tuple_to_dict(result, resp_data)
print(f"调用模块响应结果:{result_dict}")
return result_dict
def _generate_param_tuple(params: Dict[str, Any], req_data: tuple) -> tuple:
"""
根据req_data中定义的参数顺序生成参数元组
Args:
params: 用户传入的参数字典
req_data: 配置中定义的请求参数信息
Returns:
按正确顺序排列的参数元组
"""
param_list = []
for param_info in req_data:
param_name = param_info.get("name", "")
default_value = param_info.get("defaultValue")
# 如果用户提供了该参数,使用用户的值;否则使用默认值
if param_name in params:
param_list.append(params[param_name])
else:
param_list.append(default_value)
return tuple(param_list)
def _convert_tuple_to_dict(result: tuple, resp_data: tuple) -> Dict[str, Any]:
"""
将tuple结果转换为字典格式
Args:
result: 模块返回的tuple结果
resp_data: 配置中定义的输出参数信息
Returns:
字典格式的结果
"""
if not isinstance(result, tuple):
return {"result": result}
# 如果resp_data为空,返回带索引的字典
if not resp_data:
return {f"result_{i}": value for i, value in enumerate(result)}
# 根据resp_data中的参数名映射结果
result_dict = {}
for i, param_info in enumerate(resp_data):
if i < len(result):
param_name = param_info.get("name", f"result_{i}")
result_dict[param_name] = result[i]
else:
# 如果结果数量少于参数定义,使用默认值或None
param_name = param_info.get("name", f"result_{i}")
result_dict[param_name] = param_info.get("defaultValue")
# 如果结果数量多于参数定义,将多余的结果添加到字典
if len(result) > len(resp_data):
for i in range(len(resp_data), len(result)):
result_dict[f"result_{i}"] = result[i]
return result_dict
def _load_config_from_parent_dir(config_filename="process.template.json"):
"""
从项目根目录的父目录加载JSON配置文件
Args:
config_filename (str): 配置文件名
Returns:
dict: 解析后的JSON数据
Raises:
FileNotFoundError: 当配置文件不存在时
json.JSONDecodeError: 当JSON格式错误时
"""
try:
# 检查缓存
if globalVar['module_dict'] is not None:
return globalVar['module_dict']
# 获取当前文件的绝对路径
current_file = os.path.abspath(__file__)
# 计算项目根目录的父目录
project_root_parent = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(current_file))))
# 使用os.path.join确保路径正确性
json_path = os.path.join(project_root_parent, config_filename)
# 检查文件是否存在
if not os.path.exists(json_path):
raise FileNotFoundError(f"配置文件不存在: {json_path}")
# 读取并解析JSON文件
with open(json_path, 'r', encoding='utf-8') as f:
config_data = json.load(f)
# 提取子模块信息
child_module_dict = {}
nodes = config_data.get('nodes', [])
for node in nodes:
# 检查是否为子模块且包含必要字段
if not node.get('isGroup', True):
module_id = node['id']
detail = config_data.get(module_id, {})
definition = detail.get('definition', {})
args = definition.get('args0', [])
req_data = []
resp_data = []
for arg in args:
if arg.get('propertityDirection') == "1":
resp_data.append({
"name": arg.get('name', ''),
"defaultValue": arg.get('defaultValue', '')
})
else:
req_data.append({
"name": arg.get('name', ''),
"defaultValue": arg.get('defaultValue', '')
})
# 使用节点名称作为键,存储模块信息
child_module_dict[node['name']] = {
"module_id": module_id,
"resp_data": tuple(resp_data),
"req_data": tuple(req_data)
}
# 更新缓存
globalVar['module_dict'] = child_module_dict
return child_module_dict
except Exception as e:
raise ValueError(f"获取模块映射字典错误: {e}")

0人点赞
后可进行评论
小红书博主主页笔记采集,一键获取笔记数据
1回答
📢更新速递 | 实在Agent V7.2.0 上线——重磅更新,震撼来袭!
0回答
XPath应该怎么写
17回答
关于高级工程师考试的问题?
5回答
高级考试
3回答
扫码关注
获取专业的解决方案
帮您实现业务爆发式的增长




