1#!/usr/bin/env python3 2# -*- coding: utf-8 -*- 3# Copyright (c) 2025 Huawei Device Co., Ltd. 4# Licensed under the Apache License, Version 2.0 (the "License"); 5# you may not use this file except in compliance with the License. 6# You may obtain a copy of the License at 7# 8# http://www.apache.org/licenses/LICENSE-2.0 9# 10# Unless required by applicable law or agreed to in writing, software 11# distributed under the License is distributed on an "AS IS" BASIS, 12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 13# See the License for the specific language governing permissions and 14# limitations under the License. 15 16import os 17import shutil 18import subprocess 19from common_utils import ( 20 symlink_src2dest, 21 copy_folder, 22 remove_dest_path, 23 run_cmd_directly, 24 install_hpm, 25 install_hpm_in_other_platform, 26 npm_install, 27 is_system_component, 28 get_code_dir, 29 check_hpm_version, 30 save_data, 31 load_config, 32) 33import re 34import platform 35from collections import OrderedDict 36 37 38class OperateHanlder: 39 global_args = None 40 41 @staticmethod 42 def process_step(process_item: str, step_list: list, unchanged_list: list, processed_dict: dict): 43 process_result_file = os.path.join(OperateHanlder.global_args.code_dir, "prebuilts/.local_data/processed.json") 44 for step in step_list: 45 try: 46 getattr(OperateHanlder, "_" + step.get("type"))(step) 47 except Exception as e: 48 # if the process item is already being processed, but not being recorded(that means prebuilts/processed.json is not exist), 49 # in this situation, we just check if the process item is in unchanged_list, 50 # if it is, then we don't need to process it again, we can just mark it as processed. 51 if process_item in unchanged_list: 52 processed_dict[process_item] = True 53 break 54 # If an error occurs, save the processed status 55 processed_dict[process_item] = False 56 save_data(process_result_file, processed_dict) 57 raise e 58 59 @staticmethod 60 def run(operate_list: list, global_args, unchanged_list: tuple = ()): 61 OperateHanlder.global_args = global_args 62 # read and reset processed record 63 process_result_file = os.path.join(global_args.code_dir, "prebuilts/.local_data/processed.json") 64 if os.path.exists(process_result_file): 65 processed_dict = load_config(process_result_file) 66 else: 67 processed_dict = dict() 68 for key in processed_dict.keys(): 69 if key not in unchanged_list: 70 processed_dict[key] = False 71 72 # group operate_list by process item 73 item_steps_dict = OrderedDict() 74 for current_operate in operate_list: 75 current_process_item = re.match(r"(.*)_\d$", current_operate.get("step_id")).group(1) 76 if current_process_item not in item_steps_dict: 77 item_steps_dict[current_process_item] = [current_operate] 78 else: 79 item_steps_dict[current_process_item].append(current_operate) 80 81 # process each item 82 for process_item, step_list in item_steps_dict.items(): 83 process_item_without_suffix = re.sub(r"(\.[A-Za-z]+)+$", "", process_item).strip("_") 84 # If the process item is in unchanged_list and has been processed, skip it 85 if process_item in unchanged_list: 86 if process_item in processed_dict and processed_dict[process_item]: 87 print(f"==> {process_item_without_suffix} is unchanged, skip") 88 continue 89 print(f"\n==> process {process_item_without_suffix}") 90 processed_dict[process_item] = False 91 OperateHanlder.process_step(process_item, step_list, unchanged_list, processed_dict) 92 processed_dict[process_item] = True 93 # save the processed status of each item 94 save_data(process_result_file, processed_dict) 95 96 @staticmethod 97 def _symlink(operate: dict): 98 src = operate.get("src") 99 dest = operate.get("dest") 100 symlink_src2dest(src, dest) 101 102 @staticmethod 103 def _copy(operate: dict): 104 src = operate.get("src") 105 dest = operate.get("dest") 106 try: 107 shutil.copy2(src, dest) 108 except IsADirectoryError: 109 copy_folder(src, dest) 110 print(f"copy {src} ---> dest: {dest}") 111 112 @staticmethod 113 def _remove(operate: dict): 114 path = operate.get("path") 115 if isinstance(path, list): 116 for p in path: 117 remove_dest_path(p) 118 else: 119 remove_dest_path(path) 120 print(f"remove {path}") 121 122 @staticmethod 123 def _move(operate: dict): 124 src = operate.get("src") 125 dest = operate.get("dest") 126 127 filetype = operate.get("filetype", None) 128 if filetype: 129 file_list = os.listdir(src) 130 for file in file_list: 131 if file.endswith(filetype): 132 file_path = os.path.join(src, file) 133 shutil.move(file_path, dest) 134 print(f"move {file_path} ---> dest: {dest}") 135 else: 136 shutil.move(src, dest) 137 print(f"move {src} ---> dest: {dest}") 138 139 @staticmethod 140 def _shell(operate: dict): 141 cmd = operate.get("cmd") 142 run_cmd_directly(cmd) 143 144 145 @staticmethod 146 def _hpm_download(operate: dict): 147 hpm_path = os.path.join(OperateHanlder.global_args.code_dir, "prebuilts/hpm/node_modules/.bin/hpm") 148 npm_tool_path = os.path.join(OperateHanlder.global_args.code_dir, "prebuilts/build-tools/common/nodejs/current/bin/npm") 149 if check_hpm_version(hpm_path, npm_tool_path): 150 print("hpm version is ok, skip hpm download") 151 return 152 name = operate.get("name") 153 download_dir = operate.get("download_dir") 154 symlink_dest = operate.get("symlink") 155 if "@ohos/hpm-cli" == name: 156 install_hpm(npm_tool_path, download_dir) 157 symlink_src2dest(os.path.join(download_dir, "node_modules"), symlink_dest) 158 return 159 else: 160 install_hpm_in_other_platform(name, operate) 161 162 @staticmethod 163 def _npm_install(operate: dict, max_retry_times=2): 164 if OperateHanlder.global_args.type != "indep": 165 # 若不是系统组件，直接返回 166 if not is_system_component(): 167 return 168 success_installed_npm_config = [] 169 170 for retry_times in range(max_retry_times + 1): 171 try: 172 result, error = npm_install(operate, OperateHanlder.global_args, success_installed_npm_config) 173 if result: 174 return 175 print("npm install error, error info: %s", error) 176 except Exception as e: 177 print("An unexpected error occurred during npm install: %s", str(e)) 178 error = str(e) 179 180 # 重试次数超过最大限制，处理错误日志 181 for error_info in error.split("\n"): 182 if error_info.endswith("debug.log"): 183 log_path = error_info.split()[-1] 184 try: 185 # 读取日志文件内容 186 result = subprocess.run( 187 ["cat", log_path], 188 capture_output=True, 189 text=True, 190 timeout=60 191 ) 192 print("npm debug log content:\n%s", result.stdout) 193 except subprocess.TimeoutExpired: 194 print("Reading npm debug log timed out after 60 seconds.") 195 except Exception as e: 196 print("Error reading npm debug log: %s", str(e)) 197 break 198 199 # 抛出最终异常 200 raise Exception("npm install error with three times, prebuilts download exit") 201 202 @staticmethod 203 def _node_modules_copy(operate: dict): 204 if OperateHanlder.global_args.type != "indep": 205 if not is_system_component(): 206 return 207 208 copy_list = operate.get("copy_list") 209 for copy_config in copy_list: 210 src_dir = copy_config.get("src") 211 if not os.path.exists(src_dir): 212 print(f"{src_dir} not exist, skip node_modules copy.") 213 continue 214 dest_dir = copy_config.get("dest") 215 use_symlink = copy_config.get("use_symlink") 216 if os.path.exists(os.path.dirname(dest_dir)): 217 print("remove", os.path.dirname(dest_dir)) 218 shutil.rmtree(os.path.dirname(dest_dir)) 219 if use_symlink == "True" and OperateHanlder.global_args.enable_symlink == True: 220 os.makedirs(os.path.dirname(dest_dir), exist_ok=True) 221 os.symlink(src_dir, dest_dir) 222 print(f"symlink {src_dir} ---> dest: {dest_dir}") 223 else: 224 shutil.copytree(src_dir, dest_dir, symlinks=True) 225 print(f"copy {src_dir} ---> dest: {dest_dir}") 226 227 @staticmethod 228 def _download_sdk(operate: dict): 229 # 获取操作系统信息 230 system = platform.system() 231 if system == "Linux": 232 host_platform = "linux" 233 elif system == "Darwin": 234 host_platform = "darwin" 235 else: 236 print(f"Unsupported host platform: {system}") 237 exit(1) 238 239 # 获取 CPU 架构信息 240 machine = platform.machine() 241 if machine == "arm64": 242 host_cpu_prefix = "arm64" 243 elif machine == "aarch64": 244 host_cpu_prefix = "aarch64" 245 else: 246 host_cpu_prefix = "x86" 247 248 # 假设 code_dir 是当前目录，可根据实际情况修改 249 code_dir = get_code_dir() 250 prebuilts_python_dir = os.path.join(code_dir, "prebuilts", "python", f"{host_platform}-{host_cpu_prefix}") 251 python_dirs = [os.path.join(prebuilts_python_dir, d) for d in os.listdir(prebuilts_python_dir) if os.path.isdir(os.path.join(prebuilts_python_dir, d))] 252 python_dirs.sort(reverse=True) 253 if python_dirs: 254 python_path = os.path.join(python_dirs[0], "bin") 255 else: 256 raise Exception("python path not exist") 257 ohos_sdk_linux_dir = os.path.join(code_dir, "prebuilts", "ohos-sdk", "linux") 258 if not os.path.isdir(ohos_sdk_linux_dir): 259 python_executable = os.path.join(python_path, "python3") 260 script_path = os.path.join(code_dir, "build", "scripts", "download_sdk.py") 261 try: 262 subprocess.run([python_executable, script_path, "--branch", "master", "--product-name", operate.get("sdk_name"), "--api-version", str(operate.get("version"))], check=True) 263 264 except subprocess.CalledProcessError as e: 265 print(f"Error running download_sdk.py: {e}")