设置工具环境变量, 将工具配置文件中的环境变量添加到当前进程环境变量中 :param tool_config: 工具配置参数 { "common": { "env_path" : {"xxx": "xxx", "xxx": "xxx", ...}, "env_value" : {"xxx": "xxx", "xxx": "xxx", ...}, "p
(self, tool_config)
| 69 | os.environ.update({env_name: os.path.expandvars(env_value)}) |
| 70 | |
| 71 | def set_tool_env(self, tool_config): |
| 72 | """设置工具环境变量, 将工具配置文件中的环境变量添加到当前进程环境变量中 |
| 73 | :param tool_config: 工具配置参数 |
| 74 | { |
| 75 | "common": |
| 76 | { |
| 77 | "env_path" : {"xxx": "xxx", "xxx": "xxx", ...}, |
| 78 | "env_value" : {"xxx": "xxx", "xxx": "xxx", ...}, |
| 79 | "path" : ["xxx", "xxx"], |
| 80 | "tool_url" : ["xxx", "xxx"] |
| 81 | }, |
| 82 | "tool_1": {...}, |
| 83 | "tool_2": {...}, |
| 84 | ... |
| 85 | } |
| 86 | :return: tool_envs,配置的环境变量 |
| 87 | """ |
| 88 | tool_envs = {} |
| 89 | |
| 90 | path_env = [] |
| 91 | for tool_name, tool_params in tool_config.items(): |
| 92 | # 添加路径类型的环境变量 |
| 93 | for env_name, full_path in tool_params["env_path"].items(): |
| 94 | if "PATH" == env_name: # PATH应该单独放在path字段中,如果放在env_path中,忽略,避免影响和覆盖原有PATH变量 |
| 95 | continue |
| 96 | if env_name not in tool_envs: |
| 97 | tool_envs[env_name] = full_path |
| 98 | |
| 99 | # 添加值类型的环境变量 |
| 100 | for env_name, env_value in tool_params["env_value"].items(): |
| 101 | if env_name not in tool_envs: |
| 102 | tool_envs[env_name] = env_value |
| 103 | |
| 104 | # 将PATH的环境变量放到一个list中 |
| 105 | for rel_path in tool_params["path"]: |
| 106 | if rel_path not in path_env: |
| 107 | path_env.append(rel_path) |
| 108 | |
| 109 | # 添加PATH环境变量 |
| 110 | path_str = os.getenv("PATH") # 先读取系统PATH环境变量 |
| 111 | if path_str: |
| 112 | path_list = path_str.split(os.pathsep) |
| 113 | else: # 增加判空,防止读取不到的情况 |
| 114 | path_list = [] |
| 115 | for full_path in path_env: |
| 116 | # path不要去重,会影响加载顺序 |
| 117 | path_list = [full_path] + path_list # 后添加的放在前面,优先搜索 |
| 118 | tool_envs["PATH"] = os.pathsep.join(path_list) |
| 119 | |
| 120 | # 调整环境变量编码 |
| 121 | self.__update_os_environ(tool_envs, False) |
| 122 | |
| 123 | return tool_envs |
| 124 | |
| 125 | def set_task_env(self, task_params): |
| 126 | """ |
no test coverage detected