refactor: 优化 merge_dimensions_into_conditions 方法存放位置

# Reviewed, transaction id: 70955

Author: wuw-Mercury

bkmonitor/bkmonitor/utils/alert_drilling.py

@@ -0,0 +1,52 @@
+"""
+Tencent is pleased to support the open source community by making 蓝鲸智云 - 监控平台 (BlueKing - Monitor) available.
+Copyright (C) 2017-2025 Tencent. All rights reserved.
+Licensed under the MIT License (the "License"); you may not use this file except in compliance with the License.
+You may obtain a copy of the License at http://opensource.org/licenses/MIT
+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.
+"""
+
+from typing import Any
+
+
+def merge_dimensions_into_conditions(
+    agg_condition: list[dict[str, Any]] | None,
+    dimensions: dict[str, Any],
+    dimension_fields: list[str],
+) -> list[dict[str, Any]]:
+    """合并维度信息到初始过滤条件中。
+
+    将维度信息合并到初始过滤条件中，用于告警下钻时构建查询过滤条件。
+    如果维度已存在于初始条件中，则用维度值覆盖；如果维度不存在于初始条件中，则添加为新的过滤条件。
+
+    :param agg_condition: 初始过滤条件列表
+    :param dimensions: 维度信息字典
+    :param dimension_fields: 有效维度字段列表
+    :return: 合并后的过滤条件列表
+    """
+    # 使用告警策略配置的汇聚条件作为初始过滤条件
+    filter_conditions: list[dict[str, Any]] = agg_condition or []
+
+    # 构建 key → 索引位置的映射，方便后续更新替换
+    condition_index: dict[str, int] = {condition["key"]: idx for idx, condition in enumerate(filter_conditions)}
+
+    # 遍历告警的原始维度信息，增加维度过滤条件
+    for key, value in dimensions.items():
+        if key not in dimension_fields or value is None:
+            continue
+
+        condition: dict[str, Any] = {
+            "key": key,
+            "method": "eq",
+            "value": value if isinstance(value, list) else [value or ""],
+            "condition": "and",
+        }
+        # 如果该维度已存在于汇聚条件中，则更新替换；否则添加为新过滤条件
+        if key in condition_index:
+            filter_conditions[condition_index[key]] = condition
+        else:
+            filter_conditions.append(condition)
+
+    return filter_conditions

bkmonitor/packages/fta_web/alert/utils.py

@@ -13,7 +13,6 @@
 
 from datetime import datetime, timedelta
 from dateutil.relativedelta import relativedelta
-from typing import Any
 
 
 # 生成从 start 到 end 的每日时间段
@@ -236,43 +235,3 @@ def is_include_promql(query_string: str) -> bool:
     ):
         return True
     return False
-
-
-def merge_dimensions_into_conditions(
-    agg_condition: list[dict[str, Any]] | None,
-    dimensions: dict[str, Any],
-    dimension_fields: list[str],
-) -> list[dict[str, Any]]:
-    """合并维度信息到初始过滤条件中。
-
-    将维度信息合并到初始过滤条件中，如果维度已存在于初始条件中，则用维度值覆盖；如果维度不存在于初始条件中，则添加为新的过滤条件。
-
-    :param agg_condition: 初始过滤条件列表
-    :param dimensions: 维度信息字典
-    :param dimension_fields: 有效维度字段列表
-    :return: 合并后的过滤条件列表
-    """
-    # 使用告警策略配置的汇聚条件作为初始过滤条件
-    filter_conditions: list[dict[str, Any]] = agg_condition or []
-
-    # 构建 key → 索引位置的映射，方便后续更新替换
-    condition_index: dict[str, int] = {condition["key"]: idx for idx, condition in enumerate(filter_conditions)}
-
-    # 遍历告警的原始维度信息，增加维度过滤条件
-    for key, value in dimensions.items():
-        if key not in dimension_fields or value is None:
-            continue
-
-        condition: dict[str, Any] = {
-            "key": key,
-            "method": "eq",
-            "value": value if isinstance(value, list) else [value or ""],
-            "condition": "and",
-        }
-        # 如果该维度已存在于汇聚条件中，则更新替换；否则添加为新过滤条件
-        if key in condition_index:
-            filter_conditions[condition_index[key]] = condition
-        else:
-            filter_conditions.append(condition)
-
-    return filter_conditions

bkmonitor/packages/fta_web/alert_v2/resources.py

@@ -19,12 +19,12 @@
 from bkmonitor.data_source import q_to_conditions
 from bkmonitor.data_source.unify_query.builder import QueryConfigBuilder, UnifyQuerySet
 from bkmonitor.documents import AlertDocument
+from bkmonitor.utils.alert_drilling import merge_dimensions_into_conditions
 from bkmonitor.utils.thread_backend import ThreadPool
 from constants.alert import APMTargetType, K8STargetType
 from constants.data_source import DataTypeLabel, DataSourceLabel
 from core.drf_resource import Resource, resource, api
 from fta_web.alert.resources import AlertDetailResource as BaseAlertDetailResource
-from fta_web.alert.utils import merge_dimensions_into_conditions
 from fta_web.alert_v2.target import BaseTarget, get_target_instance
 from monitor_web.data_explorer.event.constants import EventSource
 from monitor_web.data_explorer.event.resources import (
@@ -188,18 +188,24 @@ def build_generic_query(
         :param alert: 告警文档对象
         :param target: 目标对象
         :param q: 查询构建器
-        :return: 构建好的查询配置，如果不是自定义事件告警则返回 None
+        :return: 构建好的查询配置，如果不是支持的告警类型则返回 None
         """
         # 获取告警策略数据查询配置
         try:
             query_config: dict[str, Any] = alert.strategy["items"][0]["query_configs"][0]
         except (KeyError, IndexError, TypeError):
             return None
 
-        if not (
-            query_config.get("data_source_label") == DataSourceLabel.CUSTOM
-            and query_config.get("data_type_label") == DataTypeLabel.EVENT
-        ):
+        # 事件类告警包括：自定义事件、日志关键字事件
+        data_source: tuple[str, str] = (
+            query_config.get("data_source_label", ""),
+            query_config.get("data_type_label", ""),
+        )
+        supported_sources: list[tuple[str, str]] = [
+            (DataSourceLabel.CUSTOM, DataTypeLabel.EVENT),
+            (DataSourceLabel.BK_MONITOR_COLLECTOR, DataTypeLabel.LOG),
+        ]
+        if data_source not in supported_sources:
             return None
 
         result_table_id: str = query_config.get("result_table_id") or ""

bkmonitor/packages/monitor_adapter/home/alert_redirect.py

@@ -18,10 +18,10 @@
 from bkmonitor.documents import ActionInstanceDocument, AlertDocument
 from bkmonitor.models import ActionInstance
 from bkmonitor.utils import time_tools
+from bkmonitor.utils.alert_drilling import merge_dimensions_into_conditions
 from constants.apm import ApmAlertHelper, FIVE_MIN_SECONDS
 from constants.data_source import DataSourceLabel, DataTypeLabel
 from core.drf_resource import resource
-from fta_web.alert.utils import merge_dimensions_into_conditions
 
 """
 提供告警详情页面跳转