• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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
19import pathlib
20import time
21import json
22import importlib
23import re
24
25
26def get_code_dir():
27    current_dir = os.path.dirname(__file__)
28    while True:
29        check_path = os.path.join(current_dir, "build", "ohos.gni")
30        if os.path.exists(check_path):
31            return current_dir
32        else:
33            new_dir = os.path.dirname(current_dir)
34            if new_dir == current_dir:
35                raise Exception(f"file {__file__} not in ohos source directory")
36            else:
37                current_dir = new_dir
38
39
40def import_rich_module():
41    module = importlib.import_module("rich.progress")
42    progress = module.Progress(
43        module.TextColumn("[bold blue]{task.fields[filename]}", justify="right"),
44        module.BarColumn(bar_width=None),
45        "[progress.percentage]{task.percentage:>3.1f}%",
46        "•",
47        module.DownloadColumn(),
48        "•",
49        module.TransferSpeedColumn(),
50        "•",
51        module.TimeRemainingColumn(),
52    )
53    return progress
54
55
56def save_data(file_path: str, data):
57    os.makedirs(os.path.dirname(file_path), exist_ok=True)
58    with open(file_path, "w") as f:
59        json.dump(data, f, indent=4)
60
61
62def load_config(config_file: str):
63    with open(config_file, "r", encoding="utf-8") as r:
64        config = json.load(r)
65        return config
66
67
68def copy_file(src: str, dest: str):
69    if not os.path.exists(dest):
70        os.makedirs(dest)
71    shutil.copy(src, dest)
72
73
74def remove_dest_path(dest_path: str):
75    if os.path.exists(dest_path) or os.path.islink(dest_path):
76        if os.path.islink(dest_path):
77            os.unlink(dest_path)
78        elif os.path.isdir(dest_path):
79            shutil.rmtree(dest_path)
80        else:
81            os.remove(dest_path)
82
83
84def copy_folder(src: str, dest: str):
85    remove_dest_path(dest)
86    shutil.copytree(src, dest)
87
88
89def symlink_src2dest(src_dir: str, dest_dir: str):
90    remove_dest_path(dest_dir)
91    os.makedirs(os.path.dirname(dest_dir), exist_ok=True)
92    os.symlink(src_dir, dest_dir)
93    print("symlink {} ---> {}".format(src_dir, dest_dir))
94
95
96def run_cmd_directly(cmd: list):
97    cmd_str = " ".join(cmd)
98    print(f"run command: {cmd_str}\n")
99    try:
100        subprocess.run(
101            cmd, check=True, stdout=None, stderr=None
102        )  # 直接输出到终端
103    except subprocess.CalledProcessError as e:
104        print(f"{cmd} execute failed: {e.returncode}")
105        raise e
106
107
108def run_cmd(cmd: list) -> tuple:
109    res = subprocess.Popen(
110        cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE
111    )
112    sout, serr = res.communicate(timeout=300)
113    return sout.rstrip().decode("utf-8"), serr, res.returncode
114
115
116def is_system_component() -> bool:
117    root_dir = get_code_dir()
118    return any(
119        pathlib.Path(root_dir, *components).exists()
120        for components in [
121            ("interface", "sdk-js"),
122            ("foundation", "arkui"),
123            ("arkcompiler",)
124        ]
125    )
126
127
128def check_hpm_version(hpm_path: str, npm_path: str) -> bool:
129    if not os.path.exists(hpm_path):
130        print(f"hpm not found at {hpm_path}, now install.")
131        return False
132    local_hpm_version = subprocess.run([hpm_path, "-V"], capture_output=True, text=True).stdout.strip()
133    cmd = npm_path + " search hpm-cli --registry https://registry.npmjs.org/"
134    cmd = cmd.split()
135    proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
136    try:
137        out, _ = proc.communicate(timeout=10)
138    except subprocess.TimeoutExpired:
139        proc.kill()
140    if proc.returncode == 0:
141        latest_hpm_version = ""
142        pattern = r'^@ohos/hpm-cli\s*\|(?:[^|]*\|){3}([^|]*)'
143        for line in out.splitlines():
144            match = re.match(pattern, line)
145            if match:
146                latest_hpm_version = match.group(1).strip()
147                break
148        if latest_hpm_version and latest_hpm_version == local_hpm_version:
149            print(f"local hpm version: {local_hpm_version}, remote latest hpm version: {latest_hpm_version}")
150            return True
151    print(f"local hpm version: {local_hpm_version}, remote latest hpm version: {latest_hpm_version}")
152    return False
153
154
155def install_hpm_in_other_platform(name: str, operate: dict):
156    download_dir = operate.get("download_dir")
157    package_path = operate.get("package_path")
158    package_lock_path = operate.get("package_lock_path")
159    hash_value = (
160        subprocess.run(
161            ["sha256sum", package_lock_path], capture_output=True, text=True
162        )
163        .stdout.strip()
164        .split(" ")[0]
165    )
166    hash_dir = os.path.join(download_dir, hash_value)
167    copy_file(package_path, hash_dir)
168    copy_file(package_lock_path, hash_dir)
169
170    if not os.path.exists(os.path.join(hash_dir, "npm-install.js")):
171        npm_install_script = os.path.join(
172            os.path.dirname(package_path), "npm-install.js"
173        )
174        if os.path.exists(npm_install_script):
175            shutil.copyfile(
176                npm_install_script, os.path.join(hash_dir, "npm-install.js")
177            )
178
179    result = subprocess.run(
180        ["npm", "install", "--prefix", hash_dir], capture_output=True, text=True
181    )
182    if result.returncode == 0:
183        print("npm install completed in the {} directory.".format(hash_dir))
184    else:
185        print("npm dependency installation failed:", result.stderr)
186
187    symlink_src = os.path.join(hash_dir, "node_modules")
188    symlink_dest = operate.get("symlink")
189
190    if name == "legacy_bin":
191        for link in operate.get("symlink", []):
192            symlink_src2dest(symlink_src, link)
193        return
194
195    if name in ["parse5"]:
196        copy_folder(symlink_src, symlink_dest)
197        return
198
199    copy_folder(symlink_src, symlink_dest)
200
201    for copy_entry in operate.get("copy", []):
202        copy_folder(
203            copy_entry["src"], copy_entry["dest"]
204        )
205
206    for copy_ext_entry in operate.get("copy_ext", []):
207        copy_folder(
208            copy_ext_entry["src"], copy_ext_entry["dest"]
209        )
210
211
212def install_hpm(npm_tool_path: str, hpm_install_dir: str):
213    content = """\
214package-lock=true
215registry=http://repo.huaweicloud.com/repository/npm
216strict-ssl=false
217lockfile=false
218"""
219    with os.fdopen(
220            os.open(
221                os.path.join(os.path.expanduser("~"), ".npmrc"),
222                os.O_WRONLY | os.O_CREAT,
223                mode=0o640,
224            ),
225            "w",
226    ) as f:
227        os.truncate(f.fileno(), 0)
228        f.write(content)
229    if not os.path.exists(hpm_install_dir):
230        os.makedirs(hpm_install_dir)
231    with os.fdopen(
232            os.open(
233                os.path.join(hpm_install_dir, "package.json"),
234                os.O_WRONLY | os.O_CREAT,
235                mode=0o640,
236            ),
237            "w",
238    ) as f:
239        os.truncate(f.fileno(), 0)
240        f.write("{}\n")
241    node_bin_path = os.path.dirname(npm_tool_path)
242    os.environ["PATH"] = f"{node_bin_path}:{os.environ['PATH']}"
243    subprocess.run(
244        [
245            npm_tool_path,
246            "install",
247            "@ohos/hpm-cli",
248            "--registry",
249            "https://repo.huaweicloud.com/repository/npm/",
250            "--prefix",
251            hpm_install_dir,
252        ]
253    )
254
255
256def npm_config(npm_tool_path: str, global_args: object) -> tuple:
257    node_path = os.path.dirname(npm_tool_path)
258    os.environ["PATH"] = "{}:{}".format(node_path, os.environ.get("PATH"))
259    if global_args.skip_ssl:
260        skip_ssl_cmd = "{} config set strict-ssl false;".format(npm_tool_path).split()
261        _, err, retcode = run_cmd(skip_ssl_cmd)
262        if retcode != 0:
263            return False, err.decode()
264    npm_clean_cmd = "{} cache clean -f".format(npm_tool_path).split()
265    npm_package_lock_cmd = "{} config set package-lock true".format(npm_tool_path).split()
266    _, err, retcode = run_cmd(npm_clean_cmd)
267    if retcode != 0:
268        return False, err.decode()
269    _, err, retcode = run_cmd(npm_package_lock_cmd)
270    if retcode != 0:
271        return False, err.decode()
272    return True, None
273
274
275def npm_install(operate: dict, global_args: object, success_installed_npm_config: list) -> tuple:
276    install_list = operate.get("npm_install_path")
277    npm_tool_path = os.path.join(global_args.code_dir, "prebuilts/build-tools/common/nodejs/current/bin/npm")
278
279    preset_is_ok, err = npm_config(npm_tool_path, global_args)
280    if not preset_is_ok:
281        return preset_is_ok, err
282
283    print("start npm install, please wait.")
284    for install_path in install_list:
285        if install_path in success_installed_npm_config:
286            continue
287        full_code_path = install_path
288        basename = os.path.basename(full_code_path)
289        node_modules_path = os.path.join(full_code_path, "node_modules")
290        npm_cache_dir = os.path.join("~/.npm/_cacache", basename)
291
292        if os.path.exists(node_modules_path):
293            print("remove node_modules %s" % node_modules_path)
294            run_cmd(("rm -rf {}".format(node_modules_path)).split())
295
296        if os.path.exists(full_code_path):
297            cmd = ["timeout", "-s", "9", "90s", npm_tool_path, "install", "--registry", global_args.npm_registry,
298                   "--cache", npm_cache_dir]
299            if global_args.host_platform == "darwin":
300                cmd = [npm_tool_path, "install", "--registry", global_args.npm_registry, "--cache", npm_cache_dir]
301            if global_args.unsafe_perm:
302                cmd.append("--unsafe-perm")
303            proc = subprocess.Popen(
304                cmd, cwd=full_code_path, stdout=subprocess.PIPE, stderr=subprocess.PIPE
305            )
306            # wait proc Popen with 0.1 second
307            time.sleep(0.1)
308            _, err = proc.communicate()
309            if proc.returncode:
310                print("in dir:{}, executing:{}".format(full_code_path, " ".join(cmd)))
311                return False, err.decode()
312            else:
313                success_installed_npm_config.append(install_path)
314                print(f"{node_modules_path} install over!")
315        else:
316            print(
317                    "npm install path {} not exist, skip".format(full_code_path)
318                )
319            if global_args.type != "indep":
320                print(
321                    "npm install path {} not exist, please check your config file".format(full_code_path)
322                )
323                raise Exception(
324                    "{} not exist, it shouldn't happen, pls check...".format(full_code_path)
325                )
326
327    return True, None