feat(backend): github ci 加上检查-makemigrations 检查、import检查 #15341

teng00123 · teng00123 · commit 489234b92514 · 2026-01-08T03:01:57.000Z
diff --git a/dbm-ui/backend/core/translation/exceptions.py b/dbm-ui/backend/core/translation/exceptions.py
@@ -25,3 +25,7 @@ class UnTranslatedFileExistException(CoreBaseException):
 
 class LanguageSpecificFStringException(CoreBaseException):
     MESSAGE = _("存在包含特定翻译语言的f-string")
+
+
+class IllegalImportException(CoreBaseException):
+    MESSAGE = _("存在非法导入")
diff --git a/dbm-ui/backend/core/translation/language_finder.py b/dbm-ui/backend/core/translation/language_finder.py
@@ -26,7 +26,11 @@
     Language,
     LanguageFindMode,
 )
-from backend.core.translation.exceptions import LanguageSpecificFStringException, UnTranslatedFileExistException
+from backend.core.translation.exceptions import (
+    IllegalImportException,
+    LanguageSpecificFStringException,
+    UnTranslatedFileExistException,
+)
 
 logger = logging.getLogger("root")
 
@@ -54,7 +58,8 @@ class NodeTranslateInit(ast.NodeVisitor):
     获取当前文件导入的翻译函数，并检查文件中的format字符
     """
 
-    ImportPath = "django.utils.translation"
+    TranslateImportPath = "django.utils.translation"
+    IllegalImportPaths = ["backend.db_periodic_task.local_tasks"]
 
     def __init__(self, ignored_string, string_regex, file_path, *args, **kwargs):
         super(NodeTranslateInit, self).__init__(*args, **kwargs)
@@ -63,6 +68,8 @@ def __init__(self, ignored_string, string_regex, file_path, *args, **kwargs):
         self.string_regex = string_regex
         self.trans_func_names = []
         self.formatted_strings = []
+        self.illegal_imports = []
+        self.illegal_import_paths = self.IllegalImportPaths
 
     @classmethod
     def get_node_module(cls, node):
@@ -74,10 +81,39 @@ def get_node_module(cls, node):
     def _check_special_language(self, string):
         return check_special_language(self.ignored_string, self.file_path, self.string_regex, string)
 
+    def visit_Import(self, node):
+        """检查直接import语句是否包含非法导入"""
+        for name in node.names:
+            if any(name.name.startswith(path) for path in self.illegal_import_paths):
+                self.illegal_imports.append(
+                    {
+                        "line": node.lineno,
+                        "col": node.col_offset,
+                        "module": name.name,
+                        "file_path": self.file_path,
+                        "type": "import",
+                    }
+                )
+                logger.warning(f"Illegal import found: {name.name} at line {node.lineno} in {self.file_path}")
+
     def visit_ImportFrom(self, node):
         """游走到import，缓存翻译函数"""
 
-        if node.module != self.ImportPath:
+        # 检查是否导入了 from backend.db_periodic_task
+        if node.module and any(node.module.startswith(path) for path in self.illegal_import_paths):
+            self.illegal_imports.append(
+                {
+                    "line": node.lineno,
+                    "col": node.col_offset,
+                    "module": node.module,
+                    "names": [name.name for name in node.names],
+                    "file_path": self.file_path,
+                    "type": "from",
+                }
+            )
+            logger.warning(f"Illegal import from local_tasks at line {node.lineno} in {self.file_path}")
+
+        if node.module != self.TranslateImportPath:
             return
 
         for name in node.names:
@@ -92,7 +128,7 @@ def visit_Str(self, node):
         # 标记Module，后续可能添加import函数
         module_node = self.get_node_module(node)
         for import_from in module_node.body:
-            if isinstance(import_from, ast.ImportFrom) and import_from.module == self.ImportPath:
+            if isinstance(import_from, ast.ImportFrom) and import_from.module == self.TranslateImportPath:
                 return
 
         module_node.language_flag = True
@@ -225,7 +261,7 @@ def visit_Module(self, node):
         node.body.insert(
             0,
             ast.ImportFrom(
-                module=NodeTranslateInit.ImportPath, names=[ast.alias(name="gettext", asname="_")], level=0
+                module=NodeTranslateInit.TranslateImportPath, names=[ast.alias(name="gettext", asname="_")], level=0
             ),
         )
         return node
@@ -248,6 +284,7 @@ class LanguageFinder:
     # 写入待翻译信息的文件名
     TRANSLATE_INFO_FILE: str = "backend/locale/translate_info.json"
     FORMATTED_STRINGS_FILE: str = "backend/locale/formatted_string_info.json"
+    ILLEGAL_IMPORTS_FILE: str = "backend/locale/illegal_imports_info.json"
 
     def __init__(
         self,
@@ -277,6 +314,7 @@ def __init__(
             "mock.py",
             "mock_data.py",
             "backend/bk_dataview/grafana",
+            "local_tasks",
         }
         if exclude_dir_or_file_list:
             _exclude_dir_or_file_list.update(set(exclude_dir_or_file_list))
@@ -296,6 +334,8 @@ def __init__(
         self.sentences_to_be_translated = {}
         # 缓存format的字符串
         self.formatted_strings = {}
+        # 缓存非法导入信息
+        self.illegal_imports = {}
 
     def check_file(self, file_path):
         """
@@ -316,6 +356,11 @@ def check_file(self, file_path):
             string_regex=LANGUAGE_REGEX_MAP[self.translate_language],
         )
         tree.visit(nodes)
+
+        # 收集非法导入信息
+        if tree.illegal_imports:
+            self.illegal_imports[file_path] = tree.illegal_imports
+
         # 缓存当前文件的format语句
         if tree.formatted_strings:
             self.formatted_strings[file_path] = tree.formatted_strings
@@ -393,7 +438,7 @@ def run(self):
 
                     # 导入翻译函数包, 默认为gettext, TODO: 做成一个参数传递?
                     if getattr(translate_nodes, "language_flag", False):
-                        file_lines.insert(1, f"from {NodeTranslateInit.ImportPath} import gettext as _\n")
+                        file_lines.insert(1, f"from {NodeTranslateInit.TranslateImportPath} import gettext as _\n")
 
                     with open(each_file, "w") as f:
                         f.write("".join(file_lines))
@@ -413,6 +458,18 @@ def run(self):
                     f"There are f-strings containing the specific translation language in the project:"
                     f"{json.dumps(self.formatted_strings, ensure_ascii=False, indent=4)}"
                 )
+            if self.illegal_imports:
+                # TODO: 待import问题解决后，恢复此异常抛出
+                # raise IllegalImportException(
+                #     f"There are illegal imports in the project: "
+                #     f"{json.dumps(self.illegal_imports, ensure_ascii=False, indent=4)}"
+                # )
+                logger.error(
+                    IllegalImportException(
+                        f"There are illegal imports in the project: "
+                        f"{json.dumps(self.illegal_imports, ensure_ascii=False, indent=4)}"
+                    )
+                )
         else:
             # 写入未翻译信息
             with open(self.TRANSLATE_INFO_FILE, "w+") as f:
@@ -421,3 +478,6 @@ def run(self):
             # 写入format字符串信息
             with open(self.FORMATTED_STRINGS_FILE, "w+") as f:
                 f.write(json.dumps(self.formatted_strings, ensure_ascii=False, indent=4))
+            # 写入非法导入信息
+            with open(self.ILLEGAL_IMPORTS_FILE, "w+") as f:
+                f.write(json.dumps(self.illegal_imports, ensure_ascii=False, indent=4))
diff --git a/dbm-ui/backend/core/translation/management/commands/language_finder.py b/dbm-ui/backend/core/translation/management/commands/language_finder.py
@@ -54,6 +54,7 @@ def add_arguments(self, parser):
             "there is a language-specific fstrig",
             type=str,
         )
+        parser.add_argument("-c", "--check_import", action="store_true", help="check import")
 
     def handle(self, *args, **options):
         if not options.get("path"):
@@ -71,6 +72,9 @@ def handle(self, *args, **options):
                 translate_language=options["language"],
                 mode=options["mode"],
             )
-            finder.run()
+            if options.get("check_import"):
+                finder.run(is_check_import=True)
+            else:
+                finder.run()
         else:
             raise TranslationPathNotFindException("The translation file directory does not exist")
diff --git a/dbm-ui/scripts/ci/bk_ci.sh b/dbm-ui/scripts/ci/bk_ci.sh
@@ -1,27 +1,44 @@
 #!/bin/bash
-# 当前脚本目录
-SCRIPT_DIR=$(dirname $(readlink -f "$0"))
+# ============================================================
+# DBM 项目 CI 检查主脚本
+# 功能：执行项目的完整性检查、安装和质量控制
+# ============================================================
 
-cat << EOF
-SCRIPT_DIR -> "$SCRIPT_DIR"
-EOF
+# 获取当前脚本所在目录
+SCRIPT_DIR=$(dirname $(readlink -f "$0"))
+echo "当前脚本目录: $SCRIPT_DIR"
+echo "============================================================"
 
+# 初始化失败计数器
 FAILED_COUNT=0
 
+# ============================================================
+# 执行各个检查步骤
+# ============================================================
+
+# 1. 准备服务环境
+echo "[1/3] 准备服务环境..."
 ${SCRIPT_DIR}/prepare_services.sh
 FAILED_COUNT=$[$FAILED_COUNT+$?]
 
+# 3. 安装依赖
+echo "[2/3] 安装项目依赖并进行迁移文件检查..."
 ${SCRIPT_DIR}/install.sh
 FAILED_COUNT=$[$FAILED_COUNT+$?]
 
+# 5. 代码质量检查
+echo "[3/3] 执行代码质量检查..."
 ${SCRIPT_DIR}/code_quality.sh
 FAILED_COUNT=$[$FAILED_COUNT+$?]
 
-echo "命令执行失败数量：$FAILED_COUNT"
-if [[ $FAILED_COUNT -ne 0 ]];
-then
-  echo "Error: 单元测试未通过!"
-  exit 1
+# ============================================================
+# 汇总检查结果
+# ============================================================
+echo "============================================================"
+if [[ $FAILED_COUNT -ne 0 ]]; then
+    echo "❌ CI 检查失败：共有 $FAILED_COUNT 个检查项未通过"
+    exit 1
 else
-  echo "单元测试已通过"
-fi
+    echo "🎉 所有 CI 检查通过！"
+    exit 0
+fi
diff --git a/dbm-ui/scripts/ci/install.sh b/dbm-ui/scripts/ci/install.sh
@@ -77,10 +77,36 @@ then
   FAILED_COUNT=$[$FAILED_COUNT+1]
 fi
 
+python manage.py language_finder --check_import -p backend/ -m error
+if [[ $? -ne 0 ]];
+then
+  echo -e "\033[1;31m Error: python manage.py language_finder --check_import -p backend/ -m error 执行失败！请检查导入是否含有backend.db_periodic_task.local_tasks \033[0m"
+#  FAILED_COUNT=$[$FAILED_COUNT+1]
+fi
+
+
+echo "正在检查是否有未提交的数据库迁移文件..."
+MIGRATION_OUTPUT=$(python manage.py makemigrations --dry-run --check 2>&1)
+
+# 捕获命令的退出状态
+STATUS=$?
+
+if [ $STATUS -ne 0 ]; then
+    echo "❌ 错误：检测到模型变更但未生成对应的迁移文件！"
+    echo ""
+    echo "====== 迁移详情 ======"
+    echo "$MIGRATION_OUTPUT"
+    exit 1
+else
+    echo "✅ 检查通过：没有未提交的模型变更。"
+    exit 0
+fi
+
 if [[ $FAILED_COUNT -ne 0 ]];
 then
   echo -e "\033[1;31m Error: 前置命令未通过! 前置命令执行失败数量: $FAILED_COUNT \033[0m"
   exit 1
 else
   echo "前置命令已通过"
 fi
+
