1#!/usr/bin/env python3 2# -*- coding: utf-8 -*- 3# 4# Copyright (c) 2024 Huawei Device Co., Ltd. 5# Licensed under the Apache License, Version 2.0 (the "License"); 6# you may not use this file except in compliance with the License. 7# You may obtain a copy of the License at 8# 9# http://www.apache.org/licenses/LICENSE-2.0 10# 11# Unless required by applicable law or agreed to in writing, software 12# distributed under the License is distributed on an "AS IS" BASIS, 13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14# See the License for the specific language governing permissions and 15# limitations under the License. 16# 17 18import json 19import os 20import re 21import fnmatch 22import sys 23from abc import ABC, abstractmethod 24import xml.etree.ElementTree as ET 25 26from Utils import ChangeFileEntity, XTSTargetUtils, PathUtils, MatchConfig, HOME 27 28 29class Ci_Manager(ABC): 30 31 @abstractmethod 32 def get_targets_from_change(self, change_list: list): 33 pass 34 35 def write_result(self, target_path_set: set, target_set: set): 36 print(f"{self.__class__.__name__} 增加 build_targets : {self._build_targets}") 37 print(f"{self.__class__.__name__} 增加 build_paths : {self._build_paths}") 38 39 xts_root_target = PathUtils.get_root_target(self._xts_root_dir) 40 if xts_root_target in target_set: 41 print("编译全量代码") 42 return 43 if xts_root_target in self._build_targets: 44 target_set.add(xts_root_target) 45 target_path_set.clear() 46 print("编译全量代码") 47 return 48 49 target_set.update(set(self._build_targets)) 50 target_path_set.update(set(self._build_paths)) 51 52 53class ComponentManager(Ci_Manager): 54 55 def __init__(self, xts_root_dir, code_root_dir): 56 self._xts_root_dir = xts_root_dir 57 self._code_root_dir = code_root_dir 58 self._build_paths = [] 59 self._build_targets = [] 60 61 def get_targets_from_change(self, change_list): 62 for changeFileEntity in change_list: 63 if changeFileEntity.path in MatchConfig.get_escape_list(): 64 print(f"{changeFileEntity.name} 仓逃生,使用旧精准目标") 65 continue 66 if changeFileEntity.path not in MatchConfig.get_xts_path_list(): 67 ret = self.getTargetsPaths(changeFileEntity) 68 if ret == 1: 69 return 1 70 return 0 71 72 def getTargetsPaths(self, change_file_entity: ChangeFileEntity): 73 # 获取部件名 74 try: 75 bundle_name = self.getBundleName(change_file_entity.path) 76 except Exception as e: 77 print(f"读取{change_file_entity.name}部件仓bundle_name失败") 78 return 1 79 print(f"{self.__class__.__name__} 增加 bundle_name : {bundle_name}") 80 # 部件名(partname)获取paths 81 paths = XTSTargetUtils.getPathsByBundle([bundle_name], self._xts_root_dir) 82 if paths: 83 change_file_entity.set_already_match_utils(True) 84 self._build_paths += paths 85 return 0 86 87 def getBundleName(self, path) -> str: 88 with open(os.path.join(HOME, path, "bundle.json"), 'r') as file: 89 data = json.load(file) 90 bundle_name = data['component']['name'] 91 return bundle_name 92 93 94class XTSManager(Ci_Manager): 95 96 def __init__(self, xts_root_dir, code_root_dir): 97 self._xts_root_dir = xts_root_dir 98 self._code_root_dir = code_root_dir 99 self._build_paths = [] 100 self._build_targets = [] 101 self._need_all = False 102 103 def get_targets_from_change(self, change_list): 104 for changeFileEntity in change_list: 105 # tools仓修改，编译全量 106 if changeFileEntity.path == "test/xts/tools": 107 self._need_all = True 108 changeFileEntity.set_already_match_utils(True) 109 if changeFileEntity.path in MatchConfig.get_xts_path_list() and changeFileEntity.path in self._xts_root_dir: 110 # 只有当前编译的xts仓修改参与计算 111 ret = self.getTargetsPaths(changeFileEntity) 112 if ret == 1: 113 print(f"{changeFileEntity.name}仓修改解析失败") 114 return 1 115 if self._need_all: 116 self._build_targets += PathUtils.get_all_build_target(self._xts_root_dir) 117 return 0 118 119 # 获取path接口 120 def getTargetsPaths(self, changeFileEntity: ChangeFileEntity): 121 # 修改和新增 122 for file in changeFileEntity.add + changeFileEntity.modified: 123 # file转为绝对路径 124 file = os.path.join(self._code_root_dir, file) 125 # 筛选掉例外的目录 126 if PathUtils.isMatchRules(file, MatchConfig.get_exception_path()): 127 continue 128 # 当前文件路径或父已存在,跳过 129 if PathUtils.isTargetContains(self._build_paths, file): 130 continue 131 # 当前file对应BUILD.gn路径 132 build_File = XTSTargetUtils.get_current_Build(self._xts_root_dir, file) 133 # 计算到根目录或指定目录,直接编译全量 134 if (os.path.dirname(build_File) == self._xts_root_dir or 135 PathUtils.isMatchRules(file, MatchConfig.get_all_com_path())): 136 self._need_all = True 137 else: 138 self._build_paths.append(os.path.dirname(build_File)) 139 # 删除 140 for file in changeFileEntity.delete: 141 # file转为绝对路径 142 file = os.path.join(self._code_root_dir, file) 143 # 筛选掉例外的目录 144 if PathUtils.isMatchRules(file, MatchConfig.get_exception_path()): 145 continue 146 # 当前文件路径或父已存在,跳过 147 if PathUtils.isTargetContains(self._build_paths, file): 148 continue 149 # 当前存在的最外层路径 150 exist_path = PathUtils.get_current_exist(self._xts_root_dir, os.path.dirname(file)) 151 build_File = XTSTargetUtils.get_current_Build(self._xts_root_dir, exist_path) 152 # 计算到根目录或指定目录,直接编译全量 153 if (os.path.dirname(build_File) == self._xts_root_dir or 154 PathUtils.isMatchRules(file, MatchConfig.get_all_com_path())): 155 self._need_all = True 156 else: 157 self._build_paths.append(os.path.dirname(build_File)) 158 return 0 159 160 161class WhitelistManager(Ci_Manager): 162 163 def __init__(self, xts_root_dir, code_root_dir): 164 self._xts_root_dir = xts_root_dir 165 self._code_root_dir = code_root_dir 166 self._build_paths = [] 167 self._build_targets = [] 168 self.full_impact_flag = "FULL_IMPACT" 169 170 def get_targets_from_change(self, change_list): 171 for changeFileEntity in change_list: 172 if changeFileEntity.path in MatchConfig.get_escape_list(): 173 print(f"{changeFileEntity.name} 仓逃生,使用旧精准目标") 174 continue 175 if changeFileEntity.path not in MatchConfig.get_xts_path_list(): 176 ret = self.getTargetsandPaths(changeFileEntity) 177 if ret == 1: 178 return 1 179 return 0 180 181 def getTargetsandPaths(self, change_file_entity): 182 white_list = MatchConfig.get_white_list_repo() 183 if change_file_entity.path not in white_list: 184 return 0 185 bundles = white_list[change_file_entity.path]["add_bundle"] 186 targets = white_list[change_file_entity.path]["add_target"] 187 if targets and targets[0] == self.full_impact_flag: 188 targets = PathUtils.get_all_build_target(self._xts_root_dir) 189 change_file_entity.set_already_match_utils(True) 190 if bundles: 191 paths = XTSTargetUtils.getPathsByBundle(bundles, self._xts_root_dir) 192 if paths: 193 self._build_paths += paths 194 if targets: 195 self._build_targets += targets 196 return 0 197 198 199class OldPreciseManager(Ci_Manager): 200 201 def __init__(self, xts_root_dir, code_root_dir): 202 self._xts_root_dir = xts_root_dir 203 self._code_root_dir = code_root_dir 204 self._build_paths = [] 205 self._build_targets = [] 206 self._precise_compilation_file = os.path.join(HOME, "test", "xts", "tools", "config", 207 "precise_compilation.json") 208 self._init_old_precise_map() 209 210 def _init_old_precise_map(self): 211 self._old_precise_map = {} 212 with open(self._precise_compilation_file, 'r') as file: 213 data_list = json.load(file) 214 for item in data_list: 215 name = item['name'] 216 build_target = item['buildTarget'] 217 self._old_precise_map[name] = build_target 218 219 def get_targets_from_change(self, change_list): 220 for changeFileEntity in change_list: 221 if changeFileEntity.path not in MatchConfig.get_xts_path_list() and \ 222 not changeFileEntity.get_already_match_utils(): 223 if self._xts_root_dir.endswith("acts"): 224 ret = self.getTargets(changeFileEntity) 225 if ret == 1: 226 pass 227 else: 228 self._build_targets += PathUtils.get_all_build_target(self._xts_root_dir) 229 return 0 230 231 # 获取path接口 232 def getTargets(self, changeFileEntity: ChangeFileEntity): 233 # 获取开源仓名 234 repo_name = self.search_repo_name(changeFileEntity.path) 235 # precise_compilation.json配置文件中获取对应目标 236 if repo_name in self._old_precise_map: 237 self._build_targets.append(self._old_precise_map[repo_name]) 238 return 0 239 240 def getTargetsbyRepoName(self, repo_name): 241 with open(repo_name, 'r') as file: 242 data_list = json.load(file) 243 # 遍历列表中的每个字典 244 for item in data_list: 245 # 获取 name 和 buildTarget 的值 246 name = item['name'] 247 build_target = item['buildTarget'] 248 # 打印结果 249 print(f'Name: {name}, Build Target: {build_target}') 250 251 def search_repo_name(self, repo_path, directory=os.path.join(HOME, ".repo", "manifests")): 252 for root, dirs, files in os.walk(directory): 253 for filename in fnmatch.filter(files, '*.xml'): 254 file_path = os.path.join(root, filename) 255 for child in ET.parse(file_path).getroot().findall('project'): 256 if 'path' in child.attrib and child.attrib['path'] == repo_path: 257 if 'gitee_name' in child.attrib: 258 return child.attrib['gitee_name'] 259 if 'name' in child.attrib: 260 return child.attrib['name'] 261 return None 262 263class GetInterfaceData(Ci_Manager): 264 265 def __init__(self, xts_root_dir, code_root_dir): 266 self._xts_root_dir = xts_root_dir 267 self._code_root_dir = code_root_dir 268 self._build_paths = [] 269 self._build_targets = [] 270 self.sum_change_list_path = [] 271 self.bundle_name_list = [] 272 self.match_path_list = [] 273 self.no_match_path_list = [] 274 275 # 截取路径 276 def get_first_levels_path(self, path, num): 277 normalized_path = os.path.normpath(path) 278 parts = normalized_path.split(os.sep) 279 return os.sep.join(parts[:num]) 280 281 def path_error(self, path_list): 282 if len(path_list) > 0: 283 raise Exception('Error: interface 路径无法匹配 bundle_name, 请前往 test/xts/tools/config/ci_api_part_name.json 配置对应 path 与 bundle_name') 284 285 def get_targets_from_change(self, change_list): 286 287 # 分开处理三个 interface 仓 288 self.get_c_bundle_name(change_list, MatchConfig.get_interface_json_c_data()) 289 self.get_driver_interface_bundle_name(change_list, MatchConfig.get_interface_json_driver_interface_data()) 290 self.get_js_bundle_name(change_list, MatchConfig.get_interface_json_js_data()) 291 # 筛选出未匹配路径 292 for path in self.sum_change_list_path: 293 if path not in self.match_path_list: 294 self.no_match_path_list.append(path) 295 296 self.bundle_name_list = list(set(self.bundle_name_list)) 297 self._build_targets = list(set(self._build_targets)) 298 print('INTERFACE_BUNDLE_NAME = ', self.bundle_name_list) 299 300 # 抛出未匹配到 bundle_name 的路径 301 try: 302 self.path_error(self.no_match_path_list) 303 except Exception as e: 304 print(e) 305 for path in self.no_match_path_list: 306 print('Error: 无法匹配路径： ', path) 307 sys.exit(1) 308 309 # 根据bundle_name 查找对应 build_paths 310 self._build_paths = XTSTargetUtils.getPathsByBundle(self.bundle_name_list, self._xts_root_dir) 311 312 # 处理 interface/sdk-js 仓 313 def get_js_bundle_name(self, change_list, js_json_data): 314 # 获取 change_list interface/sdk-js 仓数据 315 for store in change_list: 316 if store.path != MatchConfig.get_interface_path_list()[0]: 317 continue 318 paths = store.add + store.modified + store.delete 319 self.sum_change_list_path += paths 320 targete_data = [data for data in js_json_data if data['path'] in paths] 321 for _data in targete_data: 322 store.set_already_match_utils(True) 323 self.match_path_list.append(_data.get('path')) 324 # 找出 bundle_name 加入 self.bundle_name_list 325 self.bundle_name_list += _data.get('bundle_name') 326 # 针对 interface/sdk-js/kits、interface/sdk-js/arkts 在配置文件中查找 build_target 327 self._build_targets += _data.get('build_targets') 328 # 根据路径匹配 329 for path in paths: 330 for source_path in js_json_data: 331 if PathUtils.is_parent_path(source_path.get('path'), path): 332 self.match_path_list.append(path) 333 store.set_already_match_utils(True) 334 self.bundle_name_list += source_path.get('bundle_name') 335 336 # 处理 driver_interface 仓 337 def get_driver_interface_bundle_name(self, change_list, driver_interface_json_data): 338 add_bundle_json_path = [] 339 for store in change_list: 340 # 获取 change_list driver_interface 仓数据 341 if store.path != MatchConfig.get_interface_path_list()[2]: 342 continue 343 for path in store.add + store.modified + store.delete: 344 self.sum_change_list_path.append(path) 345 for source_path in driver_interface_json_data: 346 # 从 drivers/interface 下第一级目录截取路径在配置文件中查找 bundle_name 347 if self.get_first_levels_path(path, 3) == source_path.get('path'): 348 store.set_already_match_utils(True) 349 self.bundle_name_list += source_path.get('bundle_name') 350 self.match_path_list.append(path) 351 352 # 查看新增目录下是否有 bundle.json 353 for path in store.add: 354 if os.path.basename(path) == 'bundle.json': 355 try: 356 # 发现 bundle.json 后进去文件寻找 bundle_name 357 with open(os.path.abspath(__file__).split('/test/')[0] + '/' + path, 'r') as d: 358 for k, v in json.load(d).items(): 359 if k == 'component': 360 add_bundle_json_path.append([path, v['name']]) 361 except FileNotFoundError: 362 print('Error: drivers/interface仓新增目录且在目录下未发现 bundle.json， 无法匹配 bundle_name， 请添加 bundle.json 文件') 363 sys.exit(1) 364 365 # 处理新增目录下其他文件 366 for path in store.add: 367 for path_name in add_bundle_json_path: 368 if self.get_first_levels_path(path, 3) == self.get_first_levels_path(path_name[0], 3): 369 self.match_path_list.append(path) 370 store.set_already_match_utils(True) 371 self.bundle_name_list.append(path_name[1]) 372 373 # 处理 interface/sdk_c 仓 374 def get_c_bundle_name(self, change_list, c_json_data): 375 for store in change_list: 376 # 获取 change_list interface/sdk_c 仓数据 377 if store.path != MatchConfig.get_interface_path_list()[1]: 378 continue 379 for path in store.add + store.modified + store.delete: 380 self.sum_change_list_path.append(path) 381 for source_path in c_json_data: 382 # 根据目录匹配 383 if PathUtils.is_parent_path(source_path.get('path'), path): 384 self.match_path_list.append(path) 385 store.set_already_match_utils(True) 386 self.bundle_name_list += source_path.get('bundle_name') 387 # 根据文件匹配 388 elif path == source_path.get('path'): 389 self.match_path_list.append(path) 390 store.set_already_match_utils(True) 391 self.bundle_name_list += source_path.get('bundle_name')