1#!/usr/bin/env python3 2# -*- coding: utf-8 -*- 3 4# 5# Copyright (c) 2025 Huawei Device Co., Ltd. 6# Licensed under the Apache License, Version 2.0 (the "License"); 7# you may not use this file except in compliance with the License. 8# You may obtain a copy of the License at 9# 10# http://www.apache.org/licenses/LICENSE-2.0 11# 12# Unless required by applicable law or agreed to in writing, software 13# distributed under the License is distributed on an "AS IS" BASIS, 14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 15# See the License for the specific language governing permissions and 16# limitations under the License. 17# 18 19import subprocess 20import sys 21from services.interface.build_file_generator_interface import ( 22 BuildFileGeneratorInterface, 23) 24from util.log_util import LogUtil 25import os 26import json 27import time 28 29 30class PreuiltsService(BuildFileGeneratorInterface): 31 32 def __init__(self): 33 ohos_dir = self.get_ohos_dir() 34 self.last_update = os.path.join(ohos_dir, "prebuilts/.local_data/last_update.json") 35 super().__init__() 36 37 def run(self): 38 if not "--enable-prebuilts" in sys.argv: 39 return 40 if not self.check_whether_need_update(): 41 LogUtil.hb_info("you have already execute prebuilts download step and no configs changed, skip this step") 42 return 43 flags_list = self._convert_flags() 44 if "--skip-prebuilts" in flags_list: 45 print("Skip preuilts download") 46 return 47 part_names = self._get_part_names() 48 try: 49 cmd = ["/bin/bash", "build/prebuilts_config.sh", "--part-names"] 50 cmd.extend(part_names) 51 cmd_str = " ".join(cmd) 52 tips = ( 53 f"Running cmd: \"{cmd_str}\"" 54 + ", you can use --skip-preuilts to skip this step" 55 ) 56 LogUtil.hb_info(tips) 57 subprocess.run( 58 cmd, check=True, stdout=None, stderr=None # 直接输出到终端 59 ) # 直接输出到终端 60 self.write_last_update({"last_update": time.time()}) 61 except subprocess.CalledProcessError as e: 62 print(f"{cmd} execute failed: {e.returncode}") 63 raise e 64 65 def check_whether_need_update(self) -> bool: 66 last_update = self.read_last_update().get("last_update", 0) 67 if not last_update: 68 LogUtil.hb_info("No last update record found, will update prebuilts") 69 return True 70 else: 71 if self.check_file_changes(): 72 LogUtil.hb_info("Prebuilts config file has changed, will update prebuilts") 73 return True 74 else: 75 return False 76 77 def read_last_update(self): 78 if not os.path.exists(self.last_update): 79 return {} 80 try: 81 with open(self.last_update, 'r') as f: 82 return json.load(f) 83 except Exception as e: 84 LogUtil.hb_error(f"Failed to read last update file: {e}") 85 return {} 86 87 def write_last_update(self, data): 88 os.makedirs(os.path.dirname(self.last_update), exist_ok=True) 89 try: 90 with open(self.last_update, 'w') as f: 91 json.dump(data, f, indent=4) 92 except Exception as e: 93 LogUtil.hb_error(f"Failed to write last update file: {e}") 94 95 def get_ohos_dir(self): 96 cur_dir = os.getcwd() 97 while cur_dir != "/": 98 global_var = os.path.join( 99 cur_dir, 'build', 'hb', 'resources', 'global_var.py') 100 if os.path.exists(global_var): 101 return cur_dir 102 cur_dir = os.path.dirname(cur_dir) 103 raise Exception("you must run this script in ohos dir") 104 105 def get_preguilt_download_related_files_mtimes(self) -> dict: 106 dir_path = os.path.join(self.get_ohos_dir(), "build/prebuilts_service") 107 mtimes = {} 108 for root, _, files in os.walk(dir_path): 109 for file in files: 110 file_path = os.path.join(root, file) 111 mtimes[file_path] = os.path.getmtime(file_path) 112 prebuilts_config_json_path = os.path.join(self.get_ohos_dir(), "build/prebuilts_config.json") 113 prebuilts_config_py_path = os.path.join(self.get_ohos_dir(), "build/prebuilts_config.py") 114 prebuilts_config_shell_path = os.path.join(self.get_ohos_dir(), "build/prebuilts_config.sh") 115 mtimes.update({prebuilts_config_json_path: os.path.getmtime(prebuilts_config_json_path)}) 116 mtimes.update({prebuilts_config_py_path: os.path.getmtime(prebuilts_config_py_path)}) 117 mtimes.update({prebuilts_config_shell_path: os.path.getmtime(prebuilts_config_shell_path)}) 118 return mtimes 119 120 def check_file_changes(self) -> bool: 121 """ 122 check if the directory has changed by comparing file modification times. 123 :param dir_path: directory 124 :param prev_mtimes: last known modification times of files in the directory 125 :return: if the directory has changed, and the current modification times of files in the directory 126 """ 127 last_update = self.read_last_update().get("last_update", 0) 128 current_mtimes = self.get_preguilt_download_related_files_mtimes() 129 for _, mtime in current_mtimes.items(): 130 if mtime > last_update: 131 return True 132 return False 133 134 def _get_part_names(self): 135 part_name_list = [] 136 if len(sys.argv) > 2 and not sys.argv[2].startswith("-"): 137 for name in sys.argv[2:]: 138 if not name.startswith("-"): 139 part_name_list.append(name) 140 else: 141 break 142 return part_name_list 143 144 def _convert_flags(self) -> list: 145 flags_list = [] 146 for key in self.flags_dict.keys(): 147 if isinstance(self.flags_dict[key], bool) and self.flags_dict[key]: 148 flags_list.append(f"--{key}") 149 if isinstance(self.flags_dict[key], str) and self.flags_dict[key]: 150 flags_list.append(f"--{key}") 151 flags_list.append(f"{self.flags_dict[key]}") 152 if isinstance(self.flags_dict[key], list) and self.flags_dict[key]: 153 flags_list.append(f"--{key}") 154 flags_list.extend(self.flags_dict[key]) 155 return flags_list