fix: 优化sqlserver单据在数据库名称信息不区分大小写 TencentBlueKing#7633

dbm-ui/backend/db_services/sqlserver/handlers.py

@@ -16,7 +16,7 @@
 from backend.components import DRSApi
 from backend.db_services.mysql.remote_service.exceptions import RemoteServiceBaseException
 from backend.db_services.mysql.remote_service.handlers import RemoteServiceHandler
-from backend.flow.consts import SYSTEM_DBS
+from backend.flow.consts import SQLSERVER_SYSTEM_DBS
 
 
 class RemoteSqlserverServiceHandler(RemoteServiceHandler):
@@ -66,9 +66,9 @@ def show_databases(
                         "databases": [
                             data["name"]
                             for data in cmd_results[0].get("table_data", [])
-                            if data["name"] not in SYSTEM_DBS
+                            if data["name"] not in SQLSERVER_SYSTEM_DBS
                         ],
-                        "system_databases": SYSTEM_DBS,
+                        "system_databases": SQLSERVER_SYSTEM_DBS,
                     }
                 )
         return cluster_databases

dbm-ui/backend/ticket/builders/common/base.py

@@ -269,6 +269,7 @@ def _validate_single_database_table_selector(
         ignore_tables: List,
         cluster_id,
         dbs_in_cluster_map: Dict[int, List],
+        is_ignore_case: bool = False,
     ) -> Tuple[bool, str]:
         """校验库表选择器中的单个数据是否合法"""
 
@@ -280,8 +281,15 @@ def _validate_single_database_table_selector(
             if ("%" in db_name) or ("?" in db_name) or ("*" in db_name):
                 continue
 
-            if db_name not in dbs_in_cluster_map.get(cluster_id, []):
-                return False, _("数据库{}不在所属集群{}中，请重新查验").format(db_name, cluster_id)
+            if is_ignore_case:
+                # 不区分大小写对比
+                if db_name.lower() not in [i.lower() for i in dbs_in_cluster_map.get(cluster_id, [])]:
+                    return False, _("数据库{}不在所属集群{}中，请重新查验[不区分大小写]").format(db_name, cluster_id)
+
+            else:
+                # 区分大小写对比
+                if db_name not in dbs_in_cluster_map.get(cluster_id, []):
+                    return False, _("数据库{}不在所属集群{}中，请重新查验").format(db_name, cluster_id)
 
         return True, ""
 
@@ -345,14 +353,17 @@ def validate_sqlserver_table_selector(cls, bk_biz_id: int, infos: Dict, role_key
                 ignore_tables=ignore_tables,
                 cluster_id=info["cluster_id"],
                 dbs_in_cluster_map=dbs_in_cluster_map,
+                is_ignore_case=True,
             )
             if not is_valid:
                 return is_valid, f"line {index}: {message}"
 
         return True, ""
 
     @classmethod
-    def validate_mysql_db_rename(cls, infos: Dict, cluster__databases: Dict[int, List[str]]):
+    def validate_mysql_db_rename(
+        cls, infos: Dict, cluster__databases: Dict[int, List[str]], is_ignore_case: bool = False
+    ):
         """
         校验mysql db重命名逻辑
         1. 不允许包含通配符
@@ -363,6 +374,7 @@ def validate_mysql_db_rename(cls, infos: Dict, cluster__databases: Dict[int, Lis
         6. 源DB必须在集群中，新DB必须不在集群中
         @param infos: 重命名信息 [{"cluster_id": 1, "from_database": "abc", "to_database": "cde"}]
         @param cluster__databases: 集群与业务库的映射
+        @param is_ignore_case: 是否区分大小写比较，默认不区分
         """
         cluster__db_name_map: Dict[int, Dict[str, List]] = defaultdict(
             lambda: {"from_database": [], "to_database": []}
@@ -378,14 +390,24 @@ def validate_mysql_db_rename(cls, infos: Dict, cluster__databases: Dict[int, Lis
 
         # 校验在同一个集群内，源DB名必须唯一，新DB名必须唯一，且源DB名不能出现在新DB名中
         for cluster_id, name_info in cluster__db_name_map.items():
-            from_database_list = name_info["from_database"]
+            check_dbs = (
+                [i.lower() for i in cluster__databases[cluster_id]]
+                if is_ignore_case
+                else cluster__databases[cluster_id]
+            )
+            from_database_list = (
+                [i.lower() for i in name_info["from_database"]] if is_ignore_case else name_info["from_database"]
+            )
+            to_database_list = (
+                [i.lower() for i in name_info["to_database"]] if is_ignore_case else name_info["to_database"]
+            )
+
             for db_name in from_database_list:
-                if db_name not in cluster__databases[cluster_id]:
+                if db_name not in check_dbs:
                     raise serializers.ValidationError(_("数据库[{}]不存在于集群{}中").format(db_name, cluster_id))
 
-            to_database_list = name_info["to_database"]
             for db_name in to_database_list:
-                if db_name in cluster__db_name_map[cluster_id]:
+                if db_name in check_dbs:
                     raise serializers.ValidationError(_("重命名数据库[{}]已存在于集群{}中").format(db_name, cluster_id))
 
             if len(set(from_database_list)) != len(from_database_list):

dbm-ui/backend/ticket/builders/sqlserver/sqlserver_dbrename.py

@@ -39,7 +39,7 @@ def validate(self, attrs):
         # DB重命名校验，逻辑同mysql
         cluster_ids = [info["cluster_id"] for info in attrs["infos"]]
         cluster__databases = sqlserver_db_function.get_cluster_database(cluster_ids)
-        CommonValidate.validate_mysql_db_rename(attrs["infos"], cluster__databases)
+        CommonValidate.validate_mysql_db_rename(attrs["infos"], cluster__databases, True)
 
         return attrs