#!/usr/bin/env python # -*- coding: utf-8 -*- # Copyright (c) 2024 Huawei Device Co., Ltd. # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import subprocess import sys import stat import os import argparse import shutil import json import time import re import urllib.request def _get_args(): parser = argparse.ArgumentParser(add_help=True) parser.add_argument("-op", "--out_path", default=r"./", type=str, help="path of out.", ) parser.add_argument("-rp", "--root_path", default=r"./", type=str, help="path of root. default: ./", ) parser.add_argument("-cl", "--components_list", default="", type=str, help="components_list , " "pass in the components' name, separated by commas , " "example: A,B,C . " "default: none", ) parser.add_argument("-bt", "--build_type", default=0, type=int, help="build_type ,default: 0", ) parser.add_argument("-on", "--organization_name", default='ohos', type=str, help="organization_name ,default: '' ", ) parser.add_argument("-os", "--os_arg", default=r"linux", type=str, help="path of output file. default: linux", ) parser.add_argument("-ba", "--build_arch", default=r"x86", type=str, help="build_arch_arg. default: x86", ) parser.add_argument("-lt", "--local_test", default=0, type=int, help="local test ,default: not local , 0", ) args = parser.parse_args() return args def create_directories(paths): for path in paths: os.makedirs(path, exist_ok=True) def copy_files(src_dst_pairs): for src, dst in src_dst_pairs: shutil.copy2(src, dst) def generate_common_configs(): return """import("//build/ohos.gni") config("musl_common_configs") { visibility = [ ":*" ] include_dirs = [ "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1/__algorithm", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1/__bit", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1/__charconv", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1/__chrono", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1/__compare", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1/__concepts", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1/__debug_utils", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1/__concepts", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1/__filesystem", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1/__format", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1/__functional", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1/__fwd", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1/__ios", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1/__iterator", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1/__memory", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1/__numeric", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1/__random", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1/__ranges", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1/__string", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1/__support", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1/__thread", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1/__type_traits", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1/__utility", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1/__variant", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1/experimental", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/c++/v1/ext", "//prebuilts/clang/ohos/linux-x86_64/15.0.4/llvm/include/arm-linux-ohos/c++/v1", ] cflags_c = [ "-Wno-error=bitwise-op-parentheses", "-Wno-error=shift-op-parentheses", ] } """ def generate_soft_libc_musl_shared_configs(): return """ config("soft_libc_musl_shared_configs") { visibility = [ ":*" ] include_dirs = [ "innerapis/includes", ] } ohos_prebuilt_shared_library("soft_libc_musl_shared") { public_configs = [":soft_libc_musl_shared_configs",":musl_common_configs"] public_external_deps = [ ] source = "innerapis/libs/libc.so" part_name = "musl" subsystem_name = "thirdparty" } """ def generate_soft_libcrypt_configs(): return """ config("soft_libcrypt_configs") { visibility = [ ":*" ] include_dirs = [ "innerapis/includes", ] } ohos_prebuilt_static_library("soft_libcrypt") { public_configs = [":soft_libcrypt_configs",":musl_common_configs"] public_external_deps = [ ] source = "innerapis/libs/libcrypt.a" part_name = "musl" subsystem_name = "thirdparty" } """ def generate_soft_libdl_configs(): return """ config("soft_libdl_configs") { visibility = [ ":*" ] include_dirs = [ "innerapis/includes", ] } ohos_prebuilt_static_library("soft_libdl") { public_configs = [":soft_libdl_configs",":musl_common_configs"] public_external_deps = [ ] source = "innerapis/libs/libdl.a" part_name = "musl" subsystem_name = "thirdparty" } """ def generate_soft_libm_configs(): return """ config("soft_libm_configs") { visibility = [ ":*" ] include_dirs = [ "innerapis/includes", ] } ohos_prebuilt_static_library("soft_libm") { public_configs = [":soft_libm_configs",":musl_common_configs"] public_external_deps = [ ] source = "innerapis/libs/libm.a" part_name = "musl" subsystem_name = "thirdparty" } """ def generate_soft_libpthread_configs(): return """ config("soft_libpthread_configs") { visibility = [ ":*" ] include_dirs = [ "innerapis/includes", ] } ohos_prebuilt_static_library("soft_libpthread") { public_configs = [":soft_libpthread_configs",":musl_common_configs"] public_external_deps = [ ] source = "innerapis/libs/libpthread.a" part_name = "musl" subsystem_name = "thirdparty" } """ def generate_soft_libresolv_configs(): return """ config("soft_libresolv_configs") { visibility = [ ":*" ] include_dirs = [ "innerapis/includes", ] } ohos_prebuilt_static_library("soft_libresolv") { public_configs = [":soft_libresolv_configs",":musl_common_configs"] public_external_deps = [ ] source = "innerapis/libs/libresolv.a" part_name = "musl" subsystem_name = "thirdparty" } """ def generate_soft_librt_configs(): return """ config("soft_librt_configs") { visibility = [ ":*" ] include_dirs = [ "innerapis/includes", ] } ohos_prebuilt_static_library("soft_librt") { public_configs = [":soft_librt_configs",":musl_common_configs"] public_external_deps = [ ] source = "innerapis/libs/librt.a" part_name = "musl" subsystem_name = "thirdparty" } """ def generate_soft_libutil_configs(): return """ config("soft_libutil_configs") { visibility = [ ":*" ] include_dirs = [ "innerapis/includes", ] } ohos_prebuilt_static_library("soft_libutil") { public_configs = [":soft_libutil_configs",":musl_common_configs"] public_external_deps = [ ] source = "innerapis/libs/libutil.a" part_name = "musl" subsystem_name = "thirdparty" } """ def generate_soft_libxnet_configs(): return """ config("soft_libxnet_configs") { visibility = [ ":*" ] include_dirs = [ "innerapis/includes", ] } ohos_prebuilt_static_library("soft_libxnet") { public_configs = [":soft_libxnet_configs",":musl_common_configs"] public_external_deps = [ ] source = "innerapis/libs/libxnet.a" part_name = "musl" subsystem_name = "thirdparty" } """ def generate_group_copy_libs_block(): return """ group("copy_libs") { lib_files = [ "libc.a", "libc.so", "libcrypt.a", "libdl.a", "libm.a", "libpthread.a", "libresolv.a", "librt.a", "libutil.a", "libxnet.a", "crtn.o", "crti.o", "crt1.o", "rcrt1.o", "Scrt1.o", ] sources = [] outputs = [] deps = [] foreach(file, lib_files) { copy("copy_${file}") { sources += [ "innerapis/libs/${file}" ] outputs += [ "${target_out_dir}/usr/lib/arm-linux-ohos/${file}" ] } deps += [ ":copy_${file}" ] } } group("soft_shared_libs") { public_configs = [":musl_common_configs", ":soft_libxnet_configs"] deps = [ ":soft_libc_musl_shared", ":soft_libcrypt", ":soft_libdl", ":soft_libm", ":soft_libpthread", ":soft_libresolv", ":soft_librt", ":soft_libutil", ":soft_libxnet", ] } """ def generate_group_musl_headers_block(): return """ group("musl_headers") { public_configs = [ ":musl_common_configs", ":soft_libxnet_configs", ] } """ def generate_gn_file_content(part_data): gn_content = [] gn_content.append(generate_common_configs()) gn_content.append(generate_soft_libc_musl_shared_configs()) gn_content.append(generate_soft_libcrypt_configs()) gn_content.append(generate_soft_libdl_configs()) gn_content.append(generate_soft_libm_configs()) gn_content.append(generate_soft_libpthread_configs()) gn_content.append(generate_soft_libresolv_configs()) gn_content.append(generate_soft_librt_configs()) gn_content.append(generate_soft_libutil_configs()) gn_content.append(generate_soft_libxnet_configs()) gn_content.append(generate_group_copy_libs_block()) gn_content.append(generate_group_musl_headers_block()) return '\n'.join(gn_content) def write_gn_file(gn_path, content): with open(gn_path, 'w') as gn_file: gn_file.write(content) def copy_musl_libs_includes(musl_obj_path, innerapi_target_path): for folder_name in ['include', 'lib']: src_folder_path = os.path.join(musl_obj_path, folder_name, 'arm-linux-ohos') dst_folder_path = os.path.join(innerapi_target_path, folder_name + 's') dst_folder_path_1 = os.path.join(innerapi_target_path, 'musl_headers', folder_name + 's') dst_folder_path_2 = os.path.join(innerapi_target_path, 'soft_libc_musl_static', folder_name + 's') dst_folder_path_3 = os.path.join(innerapi_target_path, 'soft_shared_libs', folder_name + 's') if os.path.exists(src_folder_path): shutil.copytree(src_folder_path, dst_folder_path, dirs_exist_ok=True) shutil.copytree(src_folder_path, dst_folder_path_1, dirs_exist_ok=True) shutil.copytree(src_folder_path, dst_folder_path_2, dirs_exist_ok=True) shutil.copytree(src_folder_path, dst_folder_path_3, dirs_exist_ok=True) def write_musl_bundle(musl_bundle_path): # 向musl的bundle.json文件里写入下面的接口 additional_innerkits = [{"name": "//third_party/musl:soft_shared_libs"}, {"name": "//third_party/musl:musl_headers"}] with open(musl_bundle_path, "r") as file: musl_bundle_content = json.load(file) musl_innerkits = musl_bundle_content["component"]["build"]["inner_kits"] musl_innerkits.extend(additional_innerkits) with open(musl_bundle_path, "w", encoding='utf-8') as file: json.dump(musl_bundle_content, file, ensure_ascii=False, indent=2) def process_musl(part_data, parts_path_info, part_name, subsystem_name, components_json): musl_obj_path = os.path.join(part_data.get('out_path'), 'obj', 'third_party', 'musl', 'usr') musl_dst_path = os.path.join(part_data.get('out_path'), 'component_package', 'third_party', 'musl') musl_src_path = os.path.join(part_data.get('root_path'), 'third_party', 'musl') create_directories([musl_dst_path]) # Copy necessary files to the musl destination path files_to_copy = [ 'configure', 'dynamic.list', 'libc.map.txt', 'musl_config.gni' ] copy_files([(os.path.join(musl_src_path, file_name), os.path.join(musl_dst_path, file_name)) for file_name in files_to_copy]) # Generate and write the GN file gn_path = os.path.join(musl_dst_path, 'BUILD.gn') gn_content = generate_gn_file_content(part_data) write_gn_file(gn_path, gn_content) innerapi_target_path = os.path.join(musl_dst_path, 'innerapis') copy_musl_libs_includes(musl_obj_path, innerapi_target_path) modules = _parse_module_list(part_data) print('modules', modules) if len(modules) == 0: return _public_deps_list = [] # ... Additional logic for processing modules, copying docs, and finishing the component build ... _copy_required_docs(part_data, _public_deps_list) musl_bundle_path = os.path.join(musl_dst_path, "bundle.json") write_musl_bundle(musl_bundle_path) _finish_component_build(part_data) def _create_bundle_json(bundle_path, bundle_content): bundle = {} with open(bundle_path, "w", encoding="utf-8") as f1: json.dump(bundle_content, f1, indent=2) def _generate_rust_bundle_content(): bundle_content = { "name": "@ohos/rust", "description": "third party rust tools, provide multiply functions about compiler", "version": "3.1.0-snapshot", "license": "Apache License 2.0", "publishAs": "binary", "segment": {"destPath": "third_party/rust/crates"}, "dirs": {"./": ["*"]}, "scripts": {}, "component": { "name": "rust", "subsystem": "thirdparty", "syscap": [], "features": [], "adapted_system_type": ["standard"], "rom": "", "ram": "", "hisysevent_config": [], "deps": { "components": [], "third_party": [] }, "build": { "sub_component": [], "inner_api": [], "test": [] } }, "os": "linux", "buildArch": "x86" } return bundle_content def process_rust(part_data, parts_path_info, part_name, subsystem_name, components_json): rust_src_path = os.path.join(part_data.get('root_path'), 'third_party', 'rust', 'crates') dst_path = os.path.join(part_data.get("out_path"), "component_package", part_data.get("part_path")) copy_directory_contents(rust_src_path, dst_path) gn_path = os.path.join(dst_path, "bundle.json") bundle_content = _generate_rust_bundle_content() _create_bundle_json(gn_path, bundle_content) _copy_license(part_data) _copy_readme(part_data) def copy_directory_contents(src_path, dst_path): if not os.path.exists(dst_path): os.makedirs(dst_path) for item in os.listdir(src_path): src = os.path.join(src_path, item) dst = os.path.join(dst_path, item) if os.path.isdir(src): if os.path.exists(dst): shutil.rmtree(dst) shutil.copytree(src, dst) elif os.path.isfile(src): shutil.copy2(src, dst) def generate_developer_test_bundle_base_info(): return { "name": "@ohos/developer_test", "description": "developer_test", "version": "3.1.0-snapshot", "license": "Apache License 2.0", "publishAs": "binary", "segment": {"destPath": "test/testfwk/developer_test"}, "repository": "", "dirs": {"./": ["*"]}, "scripts": {}, "os": "linux", "buildArch": "x86" } def generate_developer_test_component_info(): return { "name": "developer_test", "subsystem": "testfwk", "syscap": [], "features": [], "adapted_system_type": ["mini", "small", "standard"], "rom": "0KB", "ram": "0KB", "deps": {} } def generate_developer_test_build_info(): return { "sub_component": [ "//test/testfwk/developer_test/examples/app_info:app_info", "//test/testfwk/developer_test/examples/detector:detector", "//test/testfwk/developer_test/examples/calculator:calculator", "//test/testfwk/developer_test/examples/calculator:calculator_static" ], "inner_kits": [ { "name": "//test/testfwk/developer_test/aw/cxx/distributed:distributedtest_lib", "header": { "header_base": [ "//test/testfwk/developer_test/aw/cxx/distributed/utils", "//test/testfwk/developer_test/aw/cxx/distributed" ], "header_files": [ "csv_transform_xml.h", "distributed.h", "distributed_agent.h", "distributed_cfg.h", "distributed_major.h" ] } }, { "name": "//test/testfwk/developer_test/aw/cxx/hwext:performance_test_static", "header": { "header_base": "//test/testfwk/developer_test/aw/cxx/hwext", "header_files": "perf.h" } } ], "test": [ "//test/testfwk/developer_test/examples/app_info/test:unittest", "//test/testfwk/developer_test/examples/calculator/test:unittest", "//test/testfwk/developer_test/examples/calculator/test:fuzztest", "//test/testfwk/developer_test/examples/calculator/test:benchmarktest", "//test/testfwk/developer_test/examples/detector/test:unittest", "//test/testfwk/developer_test/examples/sleep/test:performance", "//test/testfwk/developer_test/examples/distributedb/test:distributedtest", "//test/testfwk/developer_test/examples/stagetest/actsbundlemanagerstagetest:unittest" ] } def _generate_developer_test_bundle_content(): bundle_content = generate_developer_test_bundle_base_info() component_info = generate_developer_test_component_info() build_info = generate_developer_test_build_info() bundle_content["component"] = component_info component_info["build"] = build_info return bundle_content def process_developer_test(part_data, parts_path_info, part_name, subsystem_name, components_json): developer_test_src_path = os.path.join(part_data.get('root_path'), 'test', 'testfwk', 'developer_test') dst_path = os.path.join(part_data.get("out_path"), "component_package", part_data.get("part_path")) copy_directory_contents(developer_test_src_path, dst_path) gn_path = os.path.join(dst_path, "bundle.json") bundle_content = _generate_developer_test_bundle_content() _create_bundle_json(gn_path, bundle_content) _copy_license(part_data) _copy_readme(part_data) _finish_component_build(part_data) def process_skia(part_data, parts_path_info, part_name, subsystem_name, components_json): skia_piex_path = os.path.join(part_data.get('root_path'), 'third_party', 'skia', 'third_party', 'externals', 'piex') skia_libjpeg_path = os.path.join(part_data.get('root_path'), 'third_party', 'skia', 'third_party', 'externals', 'libjpeg-turbo') skia_component_piex_path = os.path.join(part_data.get("out_path"), "component_package", part_data.get("part_path"), 'innerapis', 'piex', 'includes') skia_component_libjpeg_path = os.path.join(part_data.get("out_path"), "component_package", part_data.get("part_path"), 'innerapis', 'libjpeg', 'includes') copy_directory_contents(skia_piex_path, skia_component_piex_path) copy_directory_contents(skia_libjpeg_path, skia_component_libjpeg_path) part_path = _get_parts_path(parts_path_info, part_name) if part_path is None: return part_data.update({"subsystem_name": subsystem_name, "part_name": part_name, "part_path": part_path}) modules = _parse_module_list(part_data) if len(modules) == 0: return is_component_build = False _public_deps_list = [] for module in modules: module_deps_list = _handle_module(part_data, components_json, module) if module_deps_list: _public_deps_list.extend(module_deps_list) is_component_build = True _copy_required_docs(part_data, _public_deps_list) _finish_component_build(part_data) def process_variants_default(part_data, parts_path_info, part_name, subsystem_name, components_json): # 减少代码重复调用 preloader_path = os.path.join(part_data.get('root_path'), 'out', 'preloader', 'rk3568') variants_default_source_files = [ os.path.join(preloader_path, 'build_config.json'), os.path.join(part_data.get('root_path'), 'build', 'indep_configs', 'variants', 'common', 'default_deps.json'), os.path.join(preloader_path, 'features.json'), os.path.join(preloader_path, 'parts_config.json'), os.path.join(preloader_path, 'system', 'etc', 'syscap.json'), os.path.join(preloader_path, 'system', 'etc', 'param', 'syscap.para'), os.path.join(preloader_path, 'system', 'etc', 'SystemCapability.json') ] variants_root = os.path.join(part_data.get('out_path'), 'component_package', 'variants', 'variants_default') variants_component_path = os.path.join(variants_root, 'config') try: os.makedirs(variants_component_path, exist_ok=True) for source_file in variants_default_source_files: if not os.path.exists(source_file): raise FileNotFoundError(f"Source file not found: {source_file}") shutil.copy2(source_file, variants_component_path) print("All confiauration files copied successfully") # 处理bundle.json文件和license文件，readme文件 bundle_content = generate_variants_default_bundle_info() bundle_path = os.path.join(variants_root, 'bundle.json') _create_bundle_json(bundle_path, bundle_content) # 创建LICENSE文件、readme.md variants_default_license_path = os.path.join(variants_root, 'LICENSE') variants_default_readme_path = os.path.join(variants_root, 'README.md') with open(variants_default_license_path, 'w') as file: file.write("license") with open(variants_default_readme_path, 'w') as file: file.write("readme") _finish_component_build(part_data) except Exception as e: print(f"Error processing variants_default: {str(e)}") raise def generate_variants_default_bundle_info(): return { "name": "@ohos/variants_default", "description": "", "version": "3.1.0-snapshot", "license": "Apache License 2.0", "publishAs": "binary", "segment": { "destPath": "variants/variants_default" }, "dirs": { "config": [ "config/*" ] }, "scripts": {}, "component": { "name": "variants_default", "subsystem": "build", "syscap": [], "features": [], "adapted_system_type": [], "rom": "", "ram": "", "deps": { "components": [ "musl", "linux", "googletest" ], "third_party": [] }, "build": { "sub_component": [], "inner_kits": [], "test": [] } }, "os": "linux", "buildArch": "x86", "dependencies": {} } def write_hilog_gn(part_data, module): gn_path = os.path.join(part_data.get("out_path"), "component_package", part_data.get("part_path"), "innerapis", module, "BUILD.gn") if os.path.exists(gn_path): os.remove(gn_path) fd = os.open(gn_path, os.O_WRONLY | os.O_CREAT, mode=0o640) fp = os.fdopen(fd, 'w') fp.write("""import("//build/ohos.gni") config("hilog_rust_configs") { visibility = [ ":*" ] include_dirs = [ "includes", ] } ohos_rust_shared_library("hilog_rust") { sources = [ "src/lib.rs" ] deps = [ "../libhilog:libhilog" ] crate_name = "hilog_rust" crate_type = "dylib" rustflags = [ "-Zstack-protector=all" ] subsystem_name = "hiviewdfx" part_name = "hilog" }""") print("_generate_build_gn has done ") fp.close() def _hilog_rust_handle(part_data, module, components_json): public_deps_list = [] if not _is_innerkit(components_json, part_data.get("part_name"), module): return public_deps_list json_data = _get_json_data(part_data, module) _lib_special_handler(part_data.get("part_name"), module, part_data) lib_exists, _ = _copy_lib(part_data, json_data, module) if lib_exists is False: return public_deps_list includes = _handle_includes_data(json_data) deps = _handle_deps_data(json_data) _copy_includes(part_data, module, includes) _list = _generate_build_gn(part_data, module, json_data, deps, components_json, public_deps_list) write_hilog_gn(part_data, module) _toolchain_gn_copy(part_data, module, json_data['out_name']) hilog_rust_out = os.path.join(part_data.get("out_path"), "component_package", part_data.get("part_path"), "innerapis", module) hilog_rust_dir = os.path.join(part_data.get("root_path"), part_data.get("part_path"), "interfaces", "rust") folder_to_copy = os.path.join(hilog_rust_dir, "src") # 替换为实际的文件夹名称 file_to_copy = os.path.join(hilog_rust_dir, "Cargo.toml") # 替换为实际的文件名称 # 检查文件夹和文件是否存在 if os.path.exists(folder_to_copy) and os.path.exists(file_to_copy): # 复制文件夹 shutil.copytree(folder_to_copy, os.path.join(hilog_rust_out, os.path.basename(folder_to_copy))) # 复制文件 shutil.copy(file_to_copy, os.path.join(hilog_rust_out, os.path.basename(file_to_copy))) else: print("文件夹或文件不存在，无法复制。") return _list def process_hilog(part_data, parts_path_info, part_name, subsystem_name, components_json): # 只处理一个模块 # 处理分类B的逻辑 part_path = _get_parts_path(parts_path_info, part_name) if part_path is None: return part_data.update({"subsystem_name": subsystem_name, "part_name": part_name, "part_path": part_path}) modules = _parse_module_list(part_data) print('modules', modules) if len(modules) == 0: return is_component_build = False _public_deps_list = [] for module in modules: module_deps_list = _handle_module(part_data, components_json, module) if module == 'hilog_rust': _hilog_rust_handle(part_data, module, components_json) if module_deps_list: _public_deps_list.extend(module_deps_list) is_component_build = True if is_component_build: _copy_required_docs(part_data, _public_deps_list) _finish_component_build(part_data) def generate_hisysevent_gn(part_data, module): gn_path = os.path.join(part_data.get("out_path"), "component_package", part_data.get("part_path"), "innerapis", module, "BUILD.gn") if os.path.exists(gn_path): os.remove(gn_path) fd = os.open(gn_path, os.O_WRONLY | os.O_CREAT, mode=0o640) fp = os.fdopen(fd, 'w') fp.write("""import("//build/ohos.gni") ohos_rust_shared_library("hisysevent_rust") { sources = [ "src/lib.rs", "src/macros.rs", "src/sys_event.rs", "src/sys_event_manager.rs", "src/utils.rs", ] external_deps = [ "hisysevent:hisysevent_c_wrapper", "hisysevent:libhisysevent", "hisysevent:libhisyseventmanager", ] crate_name = "hisysevent" crate_type = "dylib" rustflags = [ "-Zstack-protector=all" ] part_name = "hisysevent" subsystem_name = "hiviewdfx" } """) print("_generate_build_gn has done ") fp.close() def _hisysevent_rust_handle(part_data, module, components_json): public_deps_list = [] if not _is_innerkit(components_json, part_data.get("part_name"), module): return public_deps_list json_data = _get_json_data(part_data, module) _lib_special_handler(part_data.get("part_name"), module, part_data) lib_exists, _ = _copy_lib(part_data, json_data, module) if lib_exists is False: return public_deps_list includes = _handle_includes_data(json_data) deps = _handle_deps_data(json_data) _copy_includes(part_data, module, includes) _list = _generate_build_gn(part_data, module, json_data, deps, components_json, public_deps_list) generate_hisysevent_gn(part_data, module) _toolchain_gn_copy(part_data, module, json_data['out_name']) hisysevent_rust_out = os.path.join(part_data.get("out_path"), "component_package", part_data.get("part_path"), "innerapis", module) hisysevent_rust_dir = os.path.join(part_data.get("root_path"), part_data.get("part_path"), "interfaces", "innerkits", "rust") folder_to_copy = os.path.join(hisysevent_rust_dir, "src") # 替换为实际的文件夹名称 file_to_copy = os.path.join(hisysevent_rust_dir, "Cargo.toml") # 替换为实际的文件名称 # 检查文件夹和文件是否存在 if os.path.exists(folder_to_copy) and os.path.exists(file_to_copy): # 复制文件夹 shutil.copytree(folder_to_copy, os.path.join(hisysevent_rust_out, os.path.basename(folder_to_copy))) # 复制文件 shutil.copy(file_to_copy, os.path.join(hisysevent_rust_out, os.path.basename(file_to_copy))) else: print("文件夹或文件不存在，无法复制。") return _list def process_hisysevent(part_data, parts_path_info, part_name, subsystem_name, components_json): # 只处理一个模块 part_path = _get_parts_path(parts_path_info, part_name) if part_path is None: return part_data.update({"subsystem_name": subsystem_name, "part_name": part_name, "part_path": part_path}) modules = _parse_module_list(part_data) print('modules', modules) if len(modules) == 0: return is_component_build = False _public_deps_list = [] for module in modules: module_deps_list = _handle_module(part_data, components_json, module) if module == 'hisysevent_rust': _hisysevent_rust_handle(part_data, module, components_json) if module_deps_list: _public_deps_list.extend(module_deps_list) is_component_build = True if is_component_build: _copy_required_docs(part_data, _public_deps_list) _finish_component_build(part_data) def _generate_runtime_core_build_gn(): gn_path = os.path.join(args.get("out_path"), "component_package", args.get("part_path"), "innerapis", module, "BUILD.gn") fd = os.open(gn_path, os.O_WRONLY | os.O_CREAT, mode=0o640) fp = os.fdopen(fd, 'w') _generate_import(fp) _generate_configs(fp, module) _generate_prebuilt_shared_library(fp, json_data.get('type'), module) _generate_public_configs(fp, module) _list = _generate_public_external_deps(fp, module, deps, components_json, public_deps_list) _generate_other(fp, args, json_data, module) _generate_end(fp) print("_generate_build_gn has done ") fp.close() return _list def _handle_module_runtime_core(args, components_json, module): public_deps_list = [] if _is_innerkit(components_json, args.get("part_name"), module) == False: return public_deps_list json_data = _get_json_data(args, module) _lib_special_handler(args.get("part_name"), module, args) lib_exists, is_ohos_ets_copy = _copy_lib(args, json_data, module) if lib_exists is False: return public_deps_list includes = _handle_includes_data(json_data) deps = _handle_deps_data(json_data) _copy_includes(args, module, includes) _list = _generate_build_gn(args, module, json_data, deps, components_json, public_deps_list, is_ohos_ets_copy) _toolchain_gn_copy(args, module, json_data['out_name']) return _list def process_runtime_core(part_data, parts_path_info, part_name, subsystem_name, components_json): # 处理分类runtime_core的逻辑 part_path = _get_parts_path(parts_path_info, part_name) if part_path is None: return part_data.update({"subsystem_name": subsystem_name, "part_name": part_name, "part_path": part_path}) modules = _parse_module_list(part_data) print('modules', modules) if len(modules) == 0: return is_component_build = False _public_deps_list = [] for module in modules: module_deps_list = _handle_module_runtime_core(part_data, components_json, module) if module_deps_list: _public_deps_list.extend(module_deps_list) is_component_build = True if is_component_build: _copy_required_docs(part_data, _public_deps_list) _finish_component_build(part_data) def process_drivers_interface_display(part_data, parts_path_info, part_name, subsystem_name, components_json): part_path = _get_parts_path(parts_path_info, part_name) if part_path is None: return part_data.update({"subsystem_name": subsystem_name, "part_name": part_name, "part_path": part_path}) modules = _parse_module_list(part_data) print('modules', modules) if len(modules) == 0: return is_component_build = False _public_deps_list = [] for module in modules: module_deps_list = _handle_module(part_data, components_json, module) if module_deps_list: _public_deps_list.extend(module_deps_list) is_component_build = True lib_out_dir = os.path.join(part_data.get("out_path"), "component_package", part_data.get("part_path"), "innerapis", "display_commontype_idl_headers", "libs") if not os.path.exists(lib_out_dir): os.makedirs(lib_out_dir) file_path = os.path.join(lib_out_dir, 'libdisplay_commontype_idl_headers') with open(file_path, 'wb') as file: pass if is_component_build: _copy_required_docs(part_data, _public_deps_list) _finish_component_build(part_data) def process_drivers_interface_usb(part_data, parts_path_info, part_name, subsystem_name, components_json): part_path = _get_parts_path(parts_path_info, part_name) if part_path is None: return part_data.update({"subsystem_name": subsystem_name, "part_name": part_name, "part_path": part_path}) modules = _parse_module_list(part_data) print('modules', modules) if len(modules) == 0: return is_component_build = False _public_deps_list = [] for module in modules: module_deps_list = _handle_module(part_data, components_json, module) if module_deps_list: _public_deps_list.extend(module_deps_list) is_component_build = True lib_out_dir = os.path.join(part_data.get("out_path"), "component_package", part_data.get("part_path"), "innerapis", "usb_idl_headers_1.1", "libs") if not os.path.exists(lib_out_dir): os.makedirs(lib_out_dir) file_path = os.path.join(lib_out_dir, 'libusb_idl_headers_1.1') with open(file_path, 'wb') as file: pass if is_component_build: _copy_required_docs(part_data, _public_deps_list) _finish_component_build(part_data) def process_drivers_interface_ril(part_data, parts_path_info, part_name, subsystem_name, components_json): part_path = _get_parts_path(parts_path_info, part_name) if part_path is None: return part_data.update({"subsystem_name": subsystem_name, "part_name": part_name, "part_path": part_path}) modules = _parse_module_list(part_data) print('modules', modules) if len(modules) == 0: return is_component_build = False _public_deps_list = [] for module in modules: module_deps_list = _handle_module(part_data, components_json, module) if module_deps_list: _public_deps_list.extend(module_deps_list) is_component_build = True lib_out_dir = os.path.join(part_data.get("out_path"), "component_package", part_data.get("part_path"), "innerapis", "ril_idl_headers", "libs") if not os.path.exists(lib_out_dir): os.makedirs(lib_out_dir) file_path = os.path.join(lib_out_dir, 'libril_idl_headers') with open(file_path, 'wb') as file: pass if is_component_build: _copy_required_docs(part_data, _public_deps_list) _finish_component_build(part_data) # 函数映射字典 function_map = { 'musl': process_musl, "developer_test": process_developer_test, # 同rust "drivers_interface_display": process_drivers_interface_display, # 驱动的, 新建一个libs目录/ innerapi同名文件 "runtime_core": process_runtime_core, # 编译参数, 所有下面的innerapi的cflags都不 "drivers_interface_usb": process_drivers_interface_usb, # 同驱动 "drivers_interface_ril": process_drivers_interface_ril, # 同驱动 "skia": process_skia, "variants_default": process_variants_default, } def _process_part(args, parts_path_info, part_name, subsystem_name, components_json): # 使用映射字典来调用对应的函数 if part_name in function_map.keys(): function_map[part_name](args, parts_path_info, part_name, subsystem_name, components_json) else: print(f"没有找到处理分类{part_name}的函数。") def _check_label(public_deps, value): innerapis = value["innerapis"] for _innerapi in innerapis: if _innerapi: label = _innerapi.get("label") if public_deps == label: return label.split(':')[-1] continue return "" def _get_public_external_deps(data, public_deps): if not isinstance(data, dict): return "" for key, value in data.items(): if not isinstance(value, dict): continue _data = _check_label(public_deps, value) if _data: return f"{key}:{_data}" continue return "" def _is_innerkit(data, part, module): if not isinstance(data, dict): return False part_data = data.get(part) if not isinstance(part_data, dict): return False module_list = [] for i in part_data["innerapis"]: if i: module_list.append(i["name"]) if module in module_list: return True return False def _get_components_json(out_path): jsondata = "" json_path = os.path.join(out_path + "/build_configs/parts_info/components.json") with os.fdopen(os.open(json_path, os.O_RDWR | os.O_CREAT, stat.S_IWUSR | stat.S_IRUSR), 'r', encoding='utf-8') as f: try: jsondata = json.load(f) except Exception as e: print('--_get_components_json parse json error--') return jsondata def _handle_one_layer_json(json_key, json_data, desc_list): data_list = json_data.get(json_key) if isinstance(data_list, list) and len(json_data.get(json_key)) >= 1: desc_list.extend(data_list) else: desc_list.append(json_data.get(json_key)) def _get_external_public_config(_path, _config_name): py_args = _get_args() out_path = py_args.out_path _json_path = os.path.join(out_path, 'external_public_configs', _path, f'{_config_name}.json') try: with os.fdopen(os.open(_json_path, os.O_RDWR | os.O_CREAT, stat.S_IWUSR | stat.S_IRUSR), 'r', encoding='utf-8') as f: jsondata = json.load(f) except Exception as e: print('_json_path: ', _json_path) print('--_get_external_public_config parse json error--') return [] include_dirs = jsondata.get('include_dirs') return include_dirs def _handle_two_layer_json(json_key, json_data, desc_list): value_depth = len(json_data.get(json_key)) for i in range(value_depth): _label = json_data.get(json_key)[i].get('label') _include_dirs = json_data.get(json_key)[i].get('include_dirs') if _include_dirs: desc_list.extend(_include_dirs) else: full_path = _label.split('//')[-1] _path = full_path.split(':')[0] _config_name = full_path.split(':')[-1] _include_dirs = _get_external_public_config(_path, _config_name) if _include_dirs: desc_list.extend(_include_dirs) def _get_json_data(args, module): json_path = os.path.join(args.get("out_path"), args.get("subsystem_name"), args.get("part_name"), "publicinfo", module + ".json") with os.fdopen(os.open(json_path, os.O_RDWR | os.O_CREAT, stat.S_IWUSR | stat.S_IRUSR), 'r', encoding='utf-8') as f: try: file_content = f.read() jsondata = json.loads(file_content) except Exception as e: print(json_path) print('--_get_json_data parse json error--', e) return jsondata def _handle_deps_data(json_data): dep_list = [] if json_data.get('public_deps'): _handle_one_layer_json('public_deps', json_data, dep_list) return dep_list def _handle_includes_data(json_data): include_list = [] if json_data.get('public_configs'): _handle_two_layer_json('public_configs', json_data, include_list) if json_data.get('all_dependent_configs'): _handle_two_layer_json('all_dependent_configs', json_data, include_list) return include_list def _get_static_lib_path(args, json_data): label = json_data.get('label') split_label = label.split("//")[1].split(":")[0] real_static_lib_path = os.path.join(args.get("out_path"), "obj", split_label, json_data.get('out_name')) return real_static_lib_path def _copy_dir(src_path, target_path): if not os.path.isdir(src_path): return False for file in os.listdir(src_path): path = os.path.join(src_path, file) if os.path.isdir(path): if file.startswith("."): continue path1 = os.path.join(target_path, file) _copy_dir(path, path1) else: _, file_extension = os.path.splitext(file) if file_extension not in [".h", ".hpp", ".in", ".inc", ".inl"]: continue if not os.path.exists(target_path): os.makedirs(target_path) # 打包时存在同名头文件时，使用先遍历到的 if not os.path.exists(os.path.join(target_path, file)): shutil.copy2(path, os.path.join(target_path, file)) return True def _get_target_include(part_name, include): # 需要多层提取include头文件的白名单路径 multilayer_include_config_path = os.path.join('build', 'indep_configs', 'config', 'multilayer_include_config.json') with open(multilayer_include_config_path, "r") as f: part_whitch_use_multilayer_include = json.load(f) if part_name in part_whitch_use_multilayer_include: target_include = include.replace("//", "") if target_include.endswith("/"): target_include = target_include[:-1] return target_include else: return "" def _copy_includes(args, module, includes: list): if module == 'ipc_single': includes = [ "//foundation/communication/ipc/interfaces/innerkits/ipc_core/include", "//foundation/communication/ipc/ipc/native/src/core/include", "//foundation/communication/ipc/ipc/native/src/mock/include", ] includes_out_dir = os.path.join(args.get("out_path"), "component_package", args.get("part_path"), "innerapis", module, "includes") for i in args.get("toolchain_info").keys(): toolchain_includes_out_dir = os.path.join(args.get("out_path"), "component_package", args.get("part_path"), "innerapis", module, i, "includes") toolchain_lib_out_dir = os.path.join(args.get("out_path"), "component_package", args.get("part_path"), "innerapis", module, i, "libs") if not os.path.exists(toolchain_includes_out_dir) and os.path.exists(toolchain_lib_out_dir): os.makedirs(toolchain_includes_out_dir) else: continue for include in includes: relative_target_include = _get_target_include(args.get("part_name"), include) toolchain_real_include_out_dir = os.path.join(toolchain_includes_out_dir, relative_target_include) if not os.path.exists(toolchain_real_include_out_dir): os.makedirs(toolchain_real_include_out_dir) part_path = args.get("part_path") _sub_include = include.split(f"{part_path}/")[-1] split_include = include.split("//")[1] real_include_path = os.path.join(args.get("root_path"), split_include) _copy_dir(real_include_path, toolchain_real_include_out_dir) if not os.path.exists(includes_out_dir): os.makedirs(includes_out_dir) for include in includes: relative_target_include = _get_target_include(args.get("part_name"), include) includes_real_out_dir = os.path.join(includes_out_dir, relative_target_include) if not os.path.exists(includes_real_out_dir): os.makedirs(includes_real_out_dir) part_path = args.get("part_path") _sub_include = include.split(f"{part_path}/")[-1] split_include = include.split("//")[1] real_include_path = os.path.join(args.get("root_path"), split_include) _copy_dir(real_include_path, includes_real_out_dir) print("_copy_includes has done ") def _copy_toolchain_lib(file_name, root, _name, lib_out_dir): if not file_name.startswith('.') and file_name.startswith(_name): if not os.path.exists(lib_out_dir): os.makedirs(lib_out_dir) file = os.path.join(root, file_name) shutil.copy(file, lib_out_dir) def _toolchain_lib_handler(args, toolchain_path, _name, module, toolchain_name): lib_out_dir = os.path.join(args.get("out_path"), "component_package", args.get("part_path"), "innerapis", module, toolchain_name, "libs") if os.path.isfile(toolchain_path): if not os.path.exists(lib_out_dir): os.makedirs(lib_out_dir) shutil.copy(toolchain_path, lib_out_dir) else: for root, dirs, files in os.walk(toolchain_path): for file_name in files: _copy_toolchain_lib(file_name, root, _name, lib_out_dir) def _toolchain_static_file_path_mapping(subsystem_name, args, i): if subsystem_name == "thirdparty": subsystem_name = "third_party" toolchain_path = os.path.join(args.get("out_path"), i, 'obj', subsystem_name, args.get("part_name")) return toolchain_path def replace_default_toolchains_in_output(path, default_toolchain="ohos_clang_arm64/"): return path.replace(default_toolchain, "") def _copy_lib(args, json_data, module): so_path = "" lib_status = True _target_type = json_data.get('type') subsystem_name = args.get("subsystem_name") is_ohos_ets_copy = False ets_outputs = [] # 根据 type 字段和 module 选择正确的 so_path if _target_type == 'copy' and module == 'ipc_core': so_path = os.path.join(subsystem_name, args.get("part_name"), 'libipc_single.z.so') elif _target_type == "rust_library" or _target_type == "rust_proc_macro": # 选择包含 'lib.unstripped' 的路径 outputs = json_data.get('outputs', []) for output in outputs: if 'lib.unstripped' in output: output = replace_default_toolchains_in_output(output) so_path = output break # 如果没有找到包含 'lib.unstripped' 的路径，则选择最后一个路径 if not so_path and outputs: so_path = outputs[-1] else: # 对于非 rust_library 类型，选择不包含 'lib.unstripped' 的路径 outputs = json_data.get('outputs', []) for output in outputs: if 'ohos_ets' in output: is_ohos_ets_copy = True ets_outputs.append(output) elif '.unstripped' not in output: output = replace_default_toolchains_in_output(output) so_path = output break # 如果所有路径都包含 'lib.unstripped' 或者没有 outputs，则使用 out_name if not so_path: so_path = json_data.get('out_name') if is_ohos_ets_copy: for output in ets_outputs: if not copy_so_file(args, module, output, _target_type): lib_status = False elif so_path: lib_status = copy_so_file(args, module, so_path, _target_type) if lib_status and _target_type == 'static_library': copy_static_deps_file(args, json_data.get('label'), module, so_path) return lib_status, is_ohos_ets_copy def _do_copy_static_deps_file(args, out_path, lib_path, toolchain): lib_status = False static_lib_path = os.path.join(out_path, lib_path) lib_out_dir = os.path.join(out_path, "component_package", args.get("part_path"), "common", toolchain, "deps") lib_status = _copy_file(static_lib_path, lib_out_dir) or lib_status return lib_status def is_not_basic_lib(lib_path): if "obj/third_party/musl" in lib_path or "prebuilts/clang" in lib_path: return False else: return True def read_deps_from_ninja_file(ninja_file, prefix): print("ninja file: ", ninja_file) with open(ninja_file, 'r') as f: for line in f: if line.strip().startswith(prefix): deps_libs = line.strip().split(' ') return deps_libs return [] def copy_static_deps_file(args, label, module, so_path): toolchains = set(args.get("toolchain_info").keys()) toolchains.add("") lib_status = False out_path = args.get("out_path") for toolchain in toolchains: ninja_file = os.path.join(out_path, toolchain, "obj", (label.split(':')[0]).split('//')[1], label.split(':')[1] + ".ninja") if not os.path.exists(ninja_file): continue prefix = "build " + os.path.join(toolchain, so_path) deps_libs = read_deps_from_ninja_file(ninja_file, prefix) static_deps = [] toolchain_module = toolchain + "_" + module for lib in deps_libs: lib_name = os.path.basename(lib) if lib_name in static_deps: print("lib_name: {} already in static_deps".format(lib_name)) continue if lib.endswith(".a") and lib != so_path and is_not_basic_lib(lib): static_deps.append(lib_name) lib_status = _do_copy_static_deps_file(args, out_path, lib, toolchain) or lib_status args.get("static_deps")[toolchain_module] = static_deps print("copy static deps: ", static_deps) return lib_status def copy_so_file(args, module, so_path, target_type): lib_status = False out_path = args.get("out_path") so_path_with_out_path = os.path.join(out_path, so_path) lib_out_dir = os.path.join(out_path, "component_package", args.get("part_path"), "innerapis", module, "libs") if args.get("toolchain_info").keys(): for toolchain_name in args.get("toolchain_info").keys(): lib_out_dir_with_toolchain = os.path.join(args.get("out_path"), "component_package", args.get("part_path"), "innerapis", module, toolchain_name, "libs") so_path_with_toolchain = os.path.join(args.get("out_path"), toolchain_name, so_path) unzipped_so_path_with_toolchain = so_path_with_toolchain.replace(".z.", ".") if toolchain_name in so_path: lib_status = _copy_file(so_path_with_out_path, lib_out_dir_with_toolchain, target_type) or lib_status elif os.path.isfile(so_path_with_toolchain): lib_status = _copy_file(so_path_with_toolchain, lib_out_dir_with_toolchain, target_type) or lib_status elif os.path.isfile(unzipped_so_path_with_toolchain): lib_status = _copy_file(unzipped_so_path_with_toolchain, lib_out_dir_with_toolchain, target_type) or lib_status lib_status = _copy_file(so_path_with_out_path, lib_out_dir, target_type) or lib_status return lib_status def _copy_file(so_path, lib_out_dir, target_type=""): # 处理静态库依赖 if lib_out_dir.endswith("deps") or lib_out_dir.endswith("deps/"): if not os.path.isfile(so_path): print("WARNING: {} is not a file!".format(so_path)) return False if not os.path.exists(lib_out_dir): os.makedirs(lib_out_dir) shutil.copy(so_path, lib_out_dir) return True if target_type != 'copy' and not os.path.isfile(so_path): print("WARNING: {} is not a file!".format(so_path)) return False if os.path.exists(lib_out_dir): shutil.rmtree(lib_out_dir) if os.path.isfile(so_path): if not os.path.exists(lib_out_dir): os.makedirs(lib_out_dir) shutil.copy(so_path, lib_out_dir) elif os.path.exists(so_path): dir_name = os.path.basename(so_path) new_lib_out_dir = os.path.join(lib_out_dir, dir_name) shutil.copytree(so_path, new_lib_out_dir) return True def _dirs_handler(bundlejson_out): dirs = dict() dirs['./'] = [] directory = bundlejson_out for filename in os.listdir(directory): filepath = os.path.join(directory, filename) if os.path.isfile(filepath): dirs['./'].append(filename) else: dirs[filename] = [f"{filename}/*"] delete_list = ['LICENSE', 'README.md', 'README_zh.md', 'README_en.md', 'bundle.json'] for delete_txt in delete_list: if delete_txt in dirs['./']: dirs['./'].remove(delete_txt) if dirs['./'] == []: del dirs['./'] return dirs def _copy_bundlejson(args, public_deps_list): bundlejson_out = os.path.join(args.get("out_path"), "component_package", args.get("part_path")) print("bundlejson_out : ", bundlejson_out) if not os.path.exists(bundlejson_out): os.makedirs(bundlejson_out) bundlejson = os.path.join(args.get("root_path"), args.get("part_path"), "bundle.json") dependencies_dict = {} sorted_dict = {} for public_deps in public_deps_list: _public_dep_part_name = public_deps.split(':')[0] if _public_dep_part_name != args.get("part_name"): _public_dep = f"@{args.get('organization_name')}/{_public_dep_part_name}" dependencies_dict.update({_public_dep: "*"}) sorted_dict = dict(sorted(dependencies_dict.items())) if os.path.isfile(bundlejson): with open(bundlejson, 'r') as f: bundle_data = json.load(f) bundle_data['publishAs'] = 'binary' bundle_data.update({'os': args.get('os')}) bundle_data.update({'buildArch': args.get('buildArch')}) dirs = _dirs_handler(bundlejson_out) bundle_data['dirs'] = dirs bundle_data['version'] = str(bundle_data['version']) if bundle_data['version'] == '': bundle_data['version'] = '1.0.0' pattern = r'^(\d+)\.(\d+)(-[a-zA-Z]+)?$' # 正则表达式匹配a.b[-后缀]格式的字符串 match = re.match(pattern, bundle_data['version']) if match: a = match.group(1) b = match.group(2) suffix = match.group(3) if match.group(3) else "" bundle_data['version'] = f"{a}.{b}.0{suffix}" if args.get('build_type') in [0, 1]: bundle_data['version'] += '-snapshot' if args.get('organization_name'): _name_pattern = r'@(.*.)/' bundle_data['name'] = re.sub(_name_pattern, '@' + args.get('organization_name') + '/', bundle_data['name']) if bundle_data.get('scripts'): bundle_data.update({'scripts': {}}) if bundle_data.get('licensePath'): del bundle_data['licensePath'] if bundle_data.get('readmePath'): del bundle_data['readmePath'] bundle_data['dependencies'] = sorted_dict if os.path.isfile(os.path.join(bundlejson_out, "bundle.json")): os.remove(os.path.join(bundlejson_out, "bundle.json")) with os.fdopen(os.open(os.path.join(bundlejson_out, "bundle.json"), os.O_WRONLY | os.O_CREAT, mode=0o640), "w", encoding='utf-8') as fd: json.dump(bundle_data, fd, indent=4, ensure_ascii=False) def _copy_license(args): license_out = os.path.join(args.get("out_path"), "component_package", args.get("part_path")) print("license_out : ", license_out) if not os.path.exists(license_out): os.makedirs(license_out) license_file = os.path.join(args.get("root_path"), args.get("part_path"), "LICENSE") if os.path.isfile(license_file): shutil.copy(license_file, license_out) else: license_default = os.path.join(args.get("root_path"), "build", "LICENSE") shutil.copy(license_default, license_out) bundlejson_out = os.path.join(args.get("out_path"), "component_package", args.get("part_path"), 'bundle.json') with open(bundlejson_out, 'r') as f: bundle_data = json.load(f) bundle_data.update({"license": "Apache License 2.0"}) if os.path.isfile(bundlejson_out): os.remove(bundlejson_out) with os.fdopen(os.open(bundlejson_out, os.O_WRONLY | os.O_CREAT, mode=0o640), "w", encoding='utf-8') as fd: json.dump(bundle_data, fd, indent=4, ensure_ascii=False) def _copy_readme(args): readme_out = os.path.join(args.get("out_path"), "component_package", args.get("part_path")) print("readme_out : ", readme_out) if not os.path.exists(readme_out): os.makedirs(readme_out) readme = os.path.join(args.get("root_path"), args.get("part_path"), "README.md") readme_zh = os.path.join(args.get("root_path"), args.get("part_path"), "README_zh.md") readme_en = os.path.join(args.get("root_path"), args.get("part_path"), "README_en.md") readme_out_file = os.path.join(readme_out, "README.md") if os.path.isfile(readme): shutil.copy(readme, readme_out) elif os.path.isfile(readme_zh): shutil.copy(readme_zh, readme_out_file) elif os.path.isfile(readme_en): shutil.copy(readme_en, readme_out_file) else: try: with os.fdopen(os.open(readme_out_file, os.O_WRONLY | os.O_CREAT, mode=0o640), 'w') as fp: fp.write('READ.ME') except FileExistsError: pass def _generate_import(fp, is_ohos_ets_copy=False): fp.write('import("//build/ohos.gni")\n') if is_ohos_ets_copy: fp.write('import("//build/templates/common/copy.gni")\n') def _gcc_flags_info_handle(json_data): def should_process_key(k): return k not in ["label", "include_dirs"] def process_config(config): result = {} for k, v in config.items(): if should_process_key(k): result.setdefault(k, []).extend(v) return result _flags_info = {} _public_configs = json_data.get('public_configs') if _public_configs: for config in _public_configs: config_info = process_config(config) for k, v in config_info.items(): _flags_info.setdefault(k, []).extend(v) return _flags_info def _generate_configs(fp, module, json_data, _part_name): includes = _handle_includes_data(json_data) target_includes = [] fp.write('\nconfig("' + module + '_configs") {\n') fp.write(' visibility = [ ":*" ]\n') fp.write(' include_dirs = [\n') for include in includes: target_include = _get_target_include(_part_name, include) if target_include not in target_includes: target_includes.append(target_include) for include_dir in target_includes: include_dir = os.path.join('includes', include_dir) fp.write(' "{}",\n'.format(include_dir)) if module == 'ability_runtime': fp.write(' "includes/context",\n') fp.write(' "includes/app",\n') fp.write(' ]\n') if _part_name == 'runtime_core': fp.write(' }\n') return _flags_info = _gcc_flags_info_handle(json_data) if _flags_info: for k, _list in _flags_info.items(): fp.write(f' {k} = [\n') for j in _list: # 保留所有 \ 转义符号 j_escaped = j.replace('"', '\\"') fp.write(f' "{j_escaped}",\n') fp.write(' ]\n') fp.write(' }\n') def _generate_prebuilt_target(fp, target_type, module, is_ohos_ets_copy=False): if target_type == 'static_library': fp.write('ohos_prebuilt_static_library("' + module + '") {\n') elif target_type == 'executable': fp.write('ohos_prebuilt_executable("' + module + '") {\n') elif module != 'ipc_core' and (target_type == 'etc' or target_type == 'copy'): if is_ohos_ets_copy: fp.write('ohos_copy("' + module + '") {\n') else: fp.write('ohos_prebuilt_etc("' + module + '") {\n') elif target_type == 'rust_library' or target_type == 'rust_proc_macro': fp.write('ohos_prebuilt_rust_library("' + module + '") {\n') else: fp.write('ohos_prebuilt_shared_library("' + module + '") {\n') def _generate_public_configs(fp, module): fp.write(f' public_configs = [":{module}_configs"]\n') # 目前特殊处理的依赖关系映射 _DEPENDENCIES_MAP = { ('samgr', 'samgr_proxy'): ["ipc:ipc_core"], ('ets_runtime', 'libark_jsruntime'): ["runtime_core:libarkfile_static"], } def _public_deps_special_handler(module, args): _part_name = args.get('part_name') # 使用映射字典来获取依赖列表 return _DEPENDENCIES_MAP.get((_part_name, module), []) def _generate_public_external_deps(fp, module, deps: list, components_json, public_deps_list: list, args): fp.write(' public_external_deps = [\n') for dep in deps: public_external_deps = _get_public_external_deps(components_json, dep) if len(public_external_deps) > 0: fp.write(f""" "{public_external_deps}",\n""") public_deps_list.append(public_external_deps) for _public_external_deps in _public_deps_special_handler(module, args): fp.write(f""" "{_public_external_deps}",\n""") public_deps_list.append(_public_external_deps) fp.write(' ]\n') return public_deps_list def _find_ohos_ets_dir(outputs): for path in outputs: parts = path.split('/') for part in parts: if "ohos_ets" in part: return '/'.join(parts[:-1]) return None def _generate_other(fp, args, json_data, module, is_ohos_ets_copy=False): outputs = json_data.get('outputs', []) ohos_ets_dir = _find_ohos_ets_dir(outputs) if is_ohos_ets_copy: sources_arr = [f"libs/{path.split('/')[-1]}" for path in outputs] sources_str = str(sources_arr).replace("'", '"') fp.write(f' sources = {sources_str}\n') fp.write(' outputs = [ "$root_out_dir/' + ohos_ets_dir + '/{{source_file_part}}" ]\n') fp.write(' part_name = "' + args.get("part_name") + '"\n') fp.write(' subsystem_name = "' + args.get("subsystem_name") + '"\n') else: if json_data.get('type') == 'copy' and module == 'ipc_core': so_name = 'libipc_single.z.so' else: so_name = json_data.get('out_name') for output in outputs: so_name = output.split('/')[-1] if json_data.get('type') == 'copy' and module != 'ipc_core': fp.write(' copy_linkable_file = true \n') fp.write(' source = "libs/' + so_name + '"\n') fp.write(' part_name = "' + args.get("part_name") + '"\n') fp.write(' subsystem_name = "' + args.get("subsystem_name") + '"\n') def _generate_end(fp): fp.write('}') def convert_rustdeps_to_innerapi(dep, components_json): # 分割路径和模块名 dep_parts = dep.split(':', 1) if len(dep_parts) != 2: return False, "" # 格式不正确，不是 innerapi path, module = dep_parts path = path.lstrip('//') # 遍历 components.json 中的每个部件 for component, info in components_json.items(): if path.startswith(info['path']) and _check_dep_in_innerapi(info.get('innerapis', []), dep): return True, f"{component}:{module}" return False, "" def _check_dep_in_innerapi(innerapis, dep): for innerapi in innerapis: if innerapi['label'] == (dep.split('(')[0] if ('(' in dep) else dep): return True return False def _generate_rust_deps(fp, json_data, components_json): rust_deps = json_data.get("rust_deps") external_deps = [] for _dep in rust_deps: has_innerapi, innerapi = convert_rustdeps_to_innerapi(_dep, components_json) if has_innerapi: external_deps.append(innerapi) fp.write(' external_deps = [\n') for external_dep in external_deps: fp.write(f""" "{external_dep}",\n""") fp.write(' ]\n') def _copy_rust_crate_info(fp, json_data): fp.write(f' rust_crate_name = \"{json_data.get("rust_crate_name")}\"\n') fp.write(f' rust_crate_type = \"{json_data.get("rust_crate_type")}\"\n') def _get_static_deps(args, module, toolchain): default_toolchain_module = toolchain + "_" + module return args.get("static_deps").get(default_toolchain_module, []) def _generate_static_public_deps_string(args, deps: list, toolchain: str): public_deps_str = "" if not deps: return "" public_deps_str += ' public_deps = [\n' for dep in deps: public_deps_str += f""" ":{dep}", \n""" public_deps_str += ' ]\n' return public_deps_str def _generate_static_deps_target_string(args, deps: list, toolchain: str): # target_path: part_name/innerapis/${innerapi_name}/${toolchain}/BUILD.gn # static_lib_path: part_name/common/${toolchain}/deps if toolchain: source_prefix = os.path.join("../../../common", toolchain, "deps/") else: source_prefix = os.path.join("../../common", toolchain, "deps/") output_prefix = os.path.join("common", toolchain, "deps/") target_string = "" for dep in deps: target_string += '\n' target_string += 'ohos_prebuilt_static_library("' + dep + '") {\n' target_string += ' source = "' + source_prefix + dep + '"\n' target_string += ' output = "' + output_prefix + dep + '"\n' target_string += ' part_name = "' + args.get("part_name") + '"\n' target_string += ' subsystem_name = "' + args.get("subsystem_name") + '"\n' target_string += '}' return target_string def _generate_static_deps_target(fp, args, deps: list, toolchain): target_string = _generate_static_deps_target_string(args, deps, toolchain) fp.write(target_string) def _generate_static_public_deps(fp, args, deps: list, toolchain): public_deps_string = _generate_static_public_deps_string(args, deps, toolchain) fp.write(public_deps_string) def _generate_build_gn(args, module, json_data, deps: list, components_json, public_deps_list, is_ohos_ets_copy=False): gn_path = os.path.join(args.get("out_path"), "component_package", args.get("part_path"), "innerapis", module, "BUILD.gn") static_deps_files = _get_static_deps(args, module, "") # 处理静态库依赖 fd = os.open(gn_path, os.O_WRONLY | os.O_CREAT, mode=0o640) fp = os.fdopen(fd, 'w') _generate_import(fp, is_ohos_ets_copy) _generate_configs(fp, module, json_data, args.get('part_name')) _target_type = json_data.get('type') _generate_prebuilt_target(fp, _target_type, module, is_ohos_ets_copy) _generate_public_configs(fp, module) _list = _generate_public_external_deps(fp, module, deps, components_json, public_deps_list, args) _generate_static_public_deps(fp, args, static_deps_files, "") # 处理静态库依赖 if _target_type == "rust_library" or _target_type == "rust_proc_macro": _copy_rust_crate_info(fp, json_data) _generate_rust_deps(fp, json_data, components_json) _generate_other(fp, args, json_data, module, is_ohos_ets_copy) _generate_end(fp) _generate_static_deps_target(fp, args, static_deps_files, "") # 处理静态库依赖 print(f"{module}_generate_build_gn has done ") fp.close() return _list def _toolchain_gn_modify(args, module, toolchain_name, gn_path, so_name, toolchain_gn_file): if os.path.isfile(gn_path) and so_name: with open(gn_path, 'r') as f: _gn = f.read() pattern = r"libs/(.*.)" toolchain_gn = re.sub(pattern, 'libs/' + so_name + '\"', _gn) # 处理静态库依赖传递 static_deps = _get_static_deps(args, module, toolchain_name) public_deps_str = _generate_static_public_deps_string(args, static_deps, toolchain_name) static_deps_target_str = _generate_static_deps_target_string(args, static_deps, toolchain_name) public_deps_pattern = r" public_deps\s*=\s*\[\s*([^]]*)\s*\]" toolchain_gn = re.sub(public_deps_pattern, public_deps_str, toolchain_gn, re.DOTALL) static_deps_target_pattern = re.compile(r'ohos_prebuilt_static_library\("([^"]+\.a)"\)\s*\{[^}]*\}', re.DOTALL) toolchain_gn = static_deps_target_pattern.sub("", toolchain_gn) # toolchain_gn = re.sub(r"[\n]{2,}", "\n", toolchain_gn, re.DOTALL) # 删除多余换行符 toolchain_gn += "\n" toolchain_gn += static_deps_target_str fd = os.open(toolchain_gn_file, os.O_WRONLY | os.O_CREAT, mode=0o640) fp = os.fdopen(fd, 'w') fp.write(toolchain_gn) fp.close() def _get_toolchain_gn_file(lib_out_dir, out_name): unzipped_out_name = out_name.replace(".z.", ".") if os.path.exists(os.path.join(lib_out_dir, out_name)): return out_name elif os.path.exists(os.path.join(lib_out_dir, unzipped_out_name)): return unzipped_out_name else: print('Output file not found in toolchain dir.') return '' def _toolchain_gn_copy(args, module, out_name): gn_path = os.path.join(args.get("out_path"), "component_package", args.get("part_path"), "innerapis", module, "BUILD.gn") for i in args.get("toolchain_info").keys(): lib_out_dir = os.path.join(args.get("out_path"), "component_package", args.get("part_path"), "innerapis", module, i, "libs") so_name = _get_toolchain_gn_file(lib_out_dir, out_name) if not so_name: continue toolchain_gn_file = os.path.join(args.get("out_path"), "component_package", args.get("part_path"), "innerapis", module, i, "BUILD.gn") if not os.path.exists(toolchain_gn_file): os.mknod(toolchain_gn_file) _toolchain_gn_modify(args, module, i, gn_path, so_name, toolchain_gn_file) def _parse_module_list(args): module_list = [] publicinfo_path = os.path.join(args.get("out_path"), args.get("subsystem_name"), args.get("part_name"), "publicinfo") print('publicinfo_path', publicinfo_path) if os.path.exists(publicinfo_path) is False: return module_list publicinfo_dir = os.listdir(publicinfo_path) for filename in publicinfo_dir: if filename.endswith(".json"): module_name = filename.split(".json")[0] module_list.append(module_name) print('filename', filename) print('module_list', module_list) return module_list def _lib_special_handler(part_name, module, args): if part_name == 'mksh': mksh_file_path = os.path.join(args.get('out_path'), 'startup', 'init', 'sh') sh_out = os.path.join(args.get("out_path"), "thirdparty", "mksh") if os.path.isfile(mksh_file_path): shutil.copy(mksh_file_path, sh_out) if module == 'blkid': blkid_file_path = os.path.join(args.get('out_path'), 'filemanagement', 'storage_service', 'blkid') blkid_out = os.path.join(args.get("out_path"), "thirdparty", "e2fsprogs") if os.path.isfile(blkid_file_path): shutil.copy(blkid_file_path, blkid_out) if module == 'grpc_cpp_plugin': blkid_file_path = os.path.join(args.get('out_path'), 'clang_x64', 'thirdparty', 'grpc', 'grpc_cpp_plugin') blkid_out = os.path.join(args.get("out_path"), "thirdparty", "grpc") if os.path.isfile(blkid_file_path): shutil.copy(blkid_file_path, blkid_out) def _handle_module(args, components_json, module): public_deps_list = [] if _is_innerkit(components_json, args.get("part_name"), module) == False: return public_deps_list json_data = _get_json_data(args, module) _lib_special_handler(args.get("part_name"), module, args) libstatus, is_ohos_ets_copy = _copy_lib(args, json_data, module) includes = _handle_includes_data(json_data) deps = _handle_deps_data(json_data) _copy_includes(args, module, includes) _list = _generate_build_gn(args, module, json_data, deps, components_json, public_deps_list, is_ohos_ets_copy) _toolchain_gn_copy(args, module, json_data['out_name']) return _list def _copy_required_docs(args, public_deps_list): _copy_bundlejson(args, public_deps_list) _copy_license(args) _copy_readme(args) def _finish_component_build(args): if args.get("build_type") in [0, 1]: _hpm_status = _hpm_pack(args) if _hpm_status: _copy_hpm_pack(args) def _generate_component_package(args, components_json): modules = _parse_module_list(args) print('modules', modules) if len(modules) == 0: return is_component_build = False _public_deps_list = [] for module in modules: module_deps_list = _handle_module(args, components_json, module) if module_deps_list: _public_deps_list.extend(module_deps_list) is_component_build = True if is_component_build: _copy_required_docs(args, _public_deps_list) _finish_component_build(args) def _get_part_subsystem(components_json: dict): jsondata = dict() try: for component, v in components_json.items(): jsondata[component] = v.get('subsystem') except Exception as e: print('--_get_part_subsystem parse json error--') return jsondata def _get_parts_path_info(components_json): jsondata = dict() try: for component, v in components_json.items(): jsondata[component] = v.get('path') except Exception as e: print('--_get_part_subsystem parse json error--') return jsondata def _get_toolchain_info(root_path): jsondata = "" json_path = os.path.join(root_path + "/build/indep_configs/variants/common/toolchain.json") with os.fdopen(os.open(json_path, os.O_RDWR | os.O_CREAT, stat.S_IWUSR | stat.S_IRUSR), 'r', encoding='utf-8') as f: try: jsondata = json.load(f) except Exception as e: print('--_get_toolchain_info parse json error--') return jsondata def _get_parts_path(json_data, part_name): parts_path = None if json_data.get(part_name) is not None: parts_path = json_data[part_name] return parts_path def _hpm_pack(args): part_path = os.path.join(args.get("out_path"), "component_package", args.get("part_path")) cmd = ['hpm', 'pack'] try: subprocess.run(cmd, shell=False, cwd=part_path) except Exception as e: print("{} pack fail".format(args.get("part_name"))) return 0 print("{} pack succ".format(args.get("part_name"))) return 1 def _copy_hpm_pack(args): hpm_packages_path = args.get('hpm_packages_path') part_path = os.path.join(args.get("out_path"), "component_package", args.get("part_path")) dirs = os.listdir(part_path) tgz_file_name = '' for file in dirs: if file.endswith(".tgz"): tgz_file_name = file tgz_file_out = os.path.join(part_path, tgz_file_name) if tgz_file_name: shutil.copy(tgz_file_out, hpm_packages_path) def _make_hpm_packages_dir(root_path): _out_path = os.path.join(root_path, 'out') hpm_packages_path = os.path.join(_out_path, 'hpm_packages') os.makedirs(hpm_packages_path, exist_ok=True) return hpm_packages_path def _del_exist_component_package(out_path): _component_package_path = os.path.join(out_path, 'component_package') if os.path.isdir(_component_package_path): try: print('del dir component_package start..') shutil.rmtree(_component_package_path) print('del dir component_package end..') except Exception as e: print('del dir component_package FAILED') def _get_component_check(local_test) -> list: check_list = [] if local_test == 0: contents = urllib.request.urlopen( "https://ci.openharmony.cn/api/daily_build/component/check/list").read().decode( encoding="utf-8") _check_json = json.loads(contents) try: check_list.extend(_check_json["data"]["dep_list"]) check_list.extend(_check_json["data"]["indep_list"]) except Exception as e: print("Call the component check API something wrong, plz check the API return..") check_list = list(set(check_list)) check_list = sorted(check_list) return check_list def _package_interface(args, parts_path_info, part_name, subsystem_name, components_json): part_path = _get_parts_path(parts_path_info, part_name) if part_path is None: return args.update({"subsystem_name": subsystem_name, "part_name": part_name, "part_path": part_path}) if part_name in [ "musl", # 从obj/third_party/musl/usr 下提取到includes和libs "developer_test", # 同rust "drivers_interface_display", # 驱动的, 新建一个libs目录/ innerapi同名文件 "runtime_core", # 编译参数, 所有下面的innerapi的cflags都不 "drivers_interface_usb", # 同驱动 "drivers_interface_ril", # 同驱动 "skia", "variants_default", ]: _process_part(args, parts_path_info, part_name, subsystem_name, components_json) else: _generate_component_package(args, components_json) def _get_exclusion_list(root_path): part_black_list_path = os.path.join(root_path, "build", "indep_configs", "config", "binary_package_exclusion_list.json") data = [] try: with open(part_black_list_path, 'r') as f: data = json.load(f) except FileNotFoundError: print(f"can not find file: {part_black_list_path}.") except Exception as e: print(f"{part_black_list_path}: \n {e}") return data def additional_comoponents_json(): return {"rust": { "innerapis": [], "path": "third_party/rust", "subsystem": "thirdparty", "variants": [] }, "developer_test": { "innerapis": [], "path": "test/testfwk/developer_test", "subsystem": "testfwk", "variants": [] }, "variants_default": { "innerapis": [], "path": "variants/variants_default", "subsystem": "build", "variants": [] }, } def generate_component_package(out_path, root_path, components_list=None, build_type=0, organization_name='ohos', os_arg='linux', build_arch_arg='x86', local_test=0): """ Args: out_path: output path of code default : out/rk3568 root_path: root path of code default : oh/ components_list: list of all components that need to be built build_type: build type 0: default pack,do not change organization_name 1: pack ,change organization_name 2: do not pack,do not change organization_name organization_name: default ohos, if diff then change os_arg: default : linux build_arch_arg: default : x86 local_test: 1 to open local test , default 0 to close Returns: """ start_time = time.time() components_json = _get_components_json(out_path) additional_comoponents_json_data = additional_comoponents_json() components_json.update(additional_comoponents_json_data) part_subsystem = _get_part_subsystem(components_json) parts_path_info = _get_parts_path_info(components_json) hpm_packages_path = _make_hpm_packages_dir(root_path) toolchain_info = _get_toolchain_info(root_path) ### add exclusion_list = _get_exclusion_list(root_path) # 黑名单列表 # 如果没有提供 components_list，则默认为所有非黑名单组件 all_components = components_json.keys() if local_test == 1: components_list = components_list.split(",") if components_list else all_components elif local_test == 0: if components_list: components_list = [component for component in components_list.split(",") if component not in exclusion_list] else: components_list = [component for component in all_components if component not in exclusion_list] # 如果 components_list 为空，则退出程序 if not components_list: sys.exit("stop for no target to pack..") print('components_list', components_list) # del component_package _del_exist_component_package(out_path) args = {"out_path": out_path, "root_path": root_path, "os": os_arg, "buildArch": build_arch_arg, "hpm_packages_path": hpm_packages_path, "build_type": build_type, "organization_name": organization_name, "toolchain_info": toolchain_info, "static_deps": {} } for key, value in part_subsystem.items(): part_name = key subsystem_name = value if not components_list or part_name in components_list: _package_interface(args, parts_path_info, part_name, subsystem_name, components_json) end_time = time.time() run_time = end_time - start_time print("generate_component_package out_path", out_path) print(f"Generating binary product package takes time：{run_time}") def main(): py_args = _get_args() generate_component_package(py_args.out_path, py_args.root_path, components_list=py_args.components_list, build_type=py_args.build_type, organization_name=py_args.organization_name, os_arg=py_args.os_arg, build_arch_arg=py_args.build_arch, local_test=py_args.local_test) if __name__ == '__main__': main()