tools: fix CI git diff path parsing

Author: yan hu

tools/ci/compile_bsp_with_drivers.py

@@ -8,11 +8,15 @@
 # 2023-06-27     dejavudwh    the first version
 #
 
-import subprocess
 import logging
 import os
 import sys
 
+try:
+    from git_utils import get_changed_files
+except ImportError:
+    from tools.ci.git_utils import get_changed_files
+
 CONFIG_BSP_USING_X = ["CONFIG_BSP_USING_UART", "CONFIG_BSP_USING_I2C", "CONFIG_BSP_USING_SPI", "CONFIG_BSP_USING_ADC", "CONFIG_BSP_USING_DAC"]
 
 def init_logger():
@@ -24,25 +28,35 @@ def init_logger():
                         )
 
 def diff():
-    result = subprocess.run(['git', 'diff', '--name-only', 'HEAD', 'origin/master', '--diff-filter=ACMR', '--no-renames', '--full-index'], stdout = subprocess.PIPE)
-    file_list = result.stdout.decode().strip().split('\n')
+    try:
+        file_list = get_changed_files(
+            "HEAD",
+            "origin/master",
+            diff_filter="ACMR",
+            no_renames=True,
+            full_index=True,
+        )
+    except RuntimeError as e:
+        logging.error(e)
+        return set()
+
     logging.info(file_list)
     bsp_paths = set()
     for file in file_list:
         if "bsp/" in file:
             logging.info("Modifed file: {}".format(file))
             bsp_paths.add(file)
-    
+
     dirs = set()
     for dir in bsp_paths:
-        dir = os.path.dirname(dir)    
+        dir = os.path.dirname(dir)
         while "bsp/" in dir:
             files = os.listdir(dir)
             if ".config" in files and "rt-thread.elf" not in files and not dir.endswith("bsp"):
                 logging.info("Found bsp path: {}".format(dir))
                 dirs.add(dir)
                 break
-            new_dir = os.path.dirname(dir)    
+            new_dir = os.path.dirname(dir)
             dir = new_dir
 
     return dirs

tools/ci/format_ignore.py

@@ -11,7 +11,11 @@
 import yaml
 import logging
 import os
-import subprocess
+
+try:
+    from git_utils import get_changed_files
+except ImportError:
+    from tools.ci.git_utils import get_changed_files
 
 def init_logger():
     log_format = "[%(filename)s %(lineno)d %(levelname)s] %(message)s "
@@ -72,13 +76,23 @@ def __exclude_file(self, file_path):
         return 1
 
     def get_new_file(self):
-        result = subprocess.run(['git', 'diff', '--name-only', 'HEAD', 'origin/master', '--diff-filter=ACMR', '--no-renames', '--full-index'], stdout = subprocess.PIPE)
-        file_list = result.stdout.decode().strip().split('\n')
+        try:
+            file_list = get_changed_files(
+                "HEAD",
+                "origin/master",
+                diff_filter="ACMR",
+                no_renames=True,
+                full_index=True,
+            )
+        except RuntimeError as e:
+            logging.error(e)
+            return []
+
         new_files = []
         for line in file_list:
             logging.info("modified file -> {}".format(line))
             result = self.__exclude_file(line)
             if result != 0:
                 new_files.append(line)
-        
-        return new_files
+
+        return new_files

tools/ci/git_diff_show.py

@@ -17,6 +17,16 @@
 
 import json
 from typing import List
+
+try:
+    from git_utils import get_blob_size
+    from git_utils import get_merge_base as git_get_merge_base
+    from git_utils import get_name_status_lines
+except ImportError:
+    from tools.ci.git_utils import get_blob_size
+    from tools.ci.git_utils import get_merge_base as git_get_merge_base
+    from tools.ci.git_utils import get_name_status_lines
+
 class FileDiff:
     def __init__(self, path: str, status: str, size_change: int = 0, old_size: int = 0, new_size: int = 0):
         self.path = path
@@ -49,14 +59,13 @@ def get_diff_files(self) -> List[FileDiff]:
             sys.exit(1)
 
         # 获取差异文件列表
-        diff_cmd = f"git diff --name-status {merge_base} HEAD"
-        print(diff_cmd)
+        print("git diff --name-status {} HEAD".format(merge_base))
         try:
-            output = subprocess.check_output(diff_cmd.split(), stderr=subprocess.STDOUT)
-            output = output.decode(self.encoding).strip()
+            lines = get_name_status_lines(merge_base, "HEAD", self.encoding)
+            output = "\n".join(lines)
             print(output)
-        except subprocess.CalledProcessError as e:
-            print(f"Error executing git diff: {e.output.decode(self.encoding)}")
+        except RuntimeError as e:
+            print("Error executing git diff: {}".format(e))
             sys.exit(1)
 
         if not output:
@@ -110,24 +119,15 @@ def get_diff_files(self) -> List[FileDiff]:
     def get_merge_base(self) -> str:
         """获取当前分支和目标分支的最近共同祖先"""
         try:
-            cmd = f"git merge-base {self.target_branch} HEAD"
-            print(cmd)
-            output = subprocess.check_output(cmd.split(), stderr=subprocess.STDOUT)
-            return output.decode(self.encoding).strip()
-        except subprocess.CalledProcessError as e:
-            print(f"Error executing git merge-base: {e.output.decode(self.encoding)}")
+            print("git merge-base {} HEAD".format(self.target_branch))
+            return git_get_merge_base(self.target_branch, "HEAD", self.encoding)
+        except RuntimeError as e:
+            print("Error executing git merge-base: {}".format(e))
             return None
 
     def get_file_size(self, path: str, ref: str) -> int:
-        """获取指定分支上文件的大小"""
-        try:
-            # 使用 git cat-file 来获取文件内容，然后计算其大小
-            cmd = f"git cat-file blob {ref}:{path}"
-            output = subprocess.check_output(cmd.split(), stderr=subprocess.STDOUT)
-            return len(output)
-        except subprocess.CalledProcessError:
-            # 如果文件不存在或无法获取，返回0
-            return 0
+        """Get the size of a file at the specified git ref."""
+        return get_blob_size(ref, path, self.encoding)
 
 def format_size(size: int) -> str:
     """将字节大小转换为人类可读的格式"""

tools/ci/git_utils.py

@@ -0,0 +1,112 @@
+#
+# Copyright (c) 2006-2026, RT-Thread Development Team
+#
+# SPDX-License-Identifier: Apache-2.0
+#
+
+import locale
+import subprocess
+
+
+def _default_encoding():
+    return locale.getpreferredencoding()
+
+
+def _decode_output(data, encoding=None):
+    if data is None:
+        return ""
+
+    if encoding is None:
+        encoding = _default_encoding()
+
+    return data.decode(encoding, errors="replace")
+
+
+def _run_git(args, encoding=None):
+    result = subprocess.run(
+        ["git"] + args,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        check=False,
+    )
+    stdout = _decode_output(getattr(result, "stdout", b""), encoding)
+    stderr = _decode_output(getattr(result, "stderr", b""), encoding)
+    returncode = getattr(result, "returncode", 0)
+
+    if returncode != 0:
+        raise RuntimeError(
+            "git {} failed with code {}: {}".format(
+                " ".join(args),
+                returncode,
+                stderr.strip(),
+            )
+        )
+
+    return stdout
+
+
+def split_nonempty_lines(output):
+    return [line.strip() for line in output.splitlines() if line.strip()]
+
+
+def get_changed_files(base, head, diff_filter="ACMR", no_renames=True, full_index=True, encoding=None):
+    args = [
+        "diff",
+        "--name-only",
+        base,
+        head,
+        "--diff-filter={}".format(diff_filter),
+    ]
+
+    if no_renames:
+        args.append("--no-renames")
+    if full_index:
+        args.append("--full-index")
+
+    output = _run_git(args, encoding=encoding)
+    return split_nonempty_lines(output)
+
+
+def get_name_status_lines(base, head="HEAD", encoding=None):
+    output = _run_git(
+        ["diff", "--name-status", base, head],
+        encoding=encoding,
+    )
+    return split_nonempty_lines(output)
+
+
+def get_merge_base(target_branch, head="HEAD", encoding=None):
+    output = _run_git(
+        ["merge-base", target_branch, head],
+        encoding=encoding,
+    )
+    return output.strip()
+
+
+def _parse_ls_tree_size(output, encoding=None):
+    text = _decode_output(output, encoding).strip()
+    if not text:
+        return 0
+
+    fields = text.split(None, 4)
+    if len(fields) < 4:
+        return 0
+
+    try:
+        return int(fields[3])
+    except ValueError:
+        return 0
+
+
+def get_blob_size(ref, path, encoding=None):
+    result = subprocess.run(
+        ["git", "ls-tree", "-l", ref, "--", path],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        check=False,
+    )
+
+    if result.returncode != 0:
+        return 0
+
+    return _parse_ls_tree_size(result.stdout, encoding)