按需拉取|更新内置工具库（自定义工具不通过该方法拉取） :param print_enable: 是否打印拉取过程 :param tool_names: :return:
(self, print_enable)
| 189 | include_common=include_common) |
| 190 | |
| 191 | def git_load_tools(self, print_enable): |
| 192 | """ |
| 193 | 按需拉取|更新内置工具库（自定义工具不通过该方法拉取） |
| 194 | :param print_enable: 是否打印拉取过程 |
| 195 | :param tool_names: |
| 196 | :return: |
| 197 | """ |
| 198 | # 拉取工具 |
| 199 | tool_url_info = {} |
| 200 | for tool_name, tool_info in self.__tool_config.items(): |
| 201 | if tool_info["tool_url"]: |
| 202 | tool_url = tool_info["tool_url"] |
| 203 | if isinstance(tool_url, list): |
| 204 | for url in tool_url: |
| 205 | tool_url_info[url] = tool_info |
| 206 | else: |
| 207 | tool_url_info[tool_url] = tool_info |
| 208 | # LogPrinter.info(f">>>>>type:{type(tool_url_info)}\n tool_url_info: {tool_url_info}") |
| 209 | # import json |
| 210 | # LogPrinter.info(">>> 工具配置: %s" % json.dumps(self.__tool_config, indent=1)) |
| 211 | |
| 212 | # LogPrinter.info("开始更新工具...") |
| 213 | load_tool_list = [] |
| 214 | tool_dirname_list = [] |
| 215 | for git_url, tool_info in tool_url_info.items(): |
| 216 | tool_dirname = BaseScmUrlMgr.get_last_dir_name_from_url(git_url) |
| 217 | # 工具目录名重复，说明是同一工具，去重，避免重复拉取造成线程冲突 |
| 218 | if tool_dirname in tool_dirname_list: |
| 219 | continue |
| 220 | else: |
| 221 | tool_dirname_list.append(tool_dirname) |
| 222 | tool_dirpath = os.path.join(settings.TOOL_BASE_DIR, tool_dirname) |
| 223 | |
| 224 | load_type, dirpath_copy_from = ToolCommonLoader.load_tool_type(tool_dirpath, tool_dirname, git_url) |
| 225 | |
| 226 | load_tool_list.append({ |
| 227 | "load_type": load_type, |
| 228 | "git_url": git_url, |
| 229 | "tool_dirname": tool_dirname, |
| 230 | "tool_dirpath": tool_dirpath, |
| 231 | "copy_from": dirpath_copy_from, |
| 232 | "print_enable": print_enable, |
| 233 | "auth_info": tool_info.get("auth_info") |
| 234 | }) |
| 235 | |
| 236 | if load_tool_list: # 多线程拉取工具 |
| 237 | start_time = time.time() |
| 238 | error_list = ThreadRunner().run(load_tool_list) |
| 239 | |
| 240 | if error_list: |
| 241 | # 将线程的第一个异常抛出 |
| 242 | raise error_list[0] |
| 243 | LogPrinter.info(f"Initialize tools done.(use time: {format(time.time() - start_time, '.2f')}s)") |
| 244 | |
| 245 | def set_tool_env(self): |
| 246 | """ |
no test coverage detected