feat: 新增通知渠道拓展 #7545

config/default.py

@@ -883,3 +883,8 @@ def check_engine_admin_permission(request, *args, **kwargs):
 if "BKAPP_SOPS_BROKER_URL" in os.environ:
     BROKER_URL = os.getenv("BKAPP_SOPS_BROKER_URL")
     print(f"BROKER_URL: {BROKER_URL}")
+
+# bk_chat通知
+ENABLE_BK_CHAT_CHANNEL = True if env.ENABLE_BK_CHAT_CHANNEL else False
+# bk_chat路由
+BK_CHAT_ROUTE = env.BK_CHAT_ROUTE

env.py

@@ -153,3 +153,8 @@
 # bk_audit
 BK_AUDIT_ENDPOINT = os.getenv("BK_AUDIT_ENDPOINT", None)
 BK_AUDIT_DATA_TOKEN = os.getenv("BK_AUDIT_DATA_TOKEN", None)
+
+# bk_chat通知渠道
+ENABLE_BK_CHAT_CHANNEL = False if os.getenv("BKAPP_ENABLE_BK_CHAT_CHANNEL") is None else True
+# bk_chat通知路由
+BK_CHAT_ROUTE = os.getenv("BK_CHAT_ROUTE", None)

gcloud/core/api.py

@@ -160,6 +160,14 @@ def get_footer_info(request):
 def get_msg_types(request):
     client = get_client_by_user(request.user.username)
     result = client.cmsi.get_msg_type()
+    if settings.Enable_BK_CHAT_CHANNEL:
+        bk_chat = {
+            "type": "bk_chat",
+            "icon": None,
+            "label": "bk_chat",
+            "is_active": True,
+        }
+        result["data"].append(bk_chat)
     return JsonResponse(result)
 
 

gcloud/shortcuts/message/__init__.py

@@ -19,7 +19,7 @@
     title_and_content_for_periodic_task_start_fail,
     title_and_content_for_clocked_task_create_fail,
 )
-from gcloud.shortcuts.message.send_msg import send_message
+from gcloud.shortcuts.message.send_msg import send_message, MessageHandler
 from gcloud.periodictask.models import PeriodicTask
 
 logger = logging.getLogger("root")
@@ -47,13 +47,15 @@ def send_task_flow_message(taskflow, msg_type, node_name=""):
         notify_type = notify_types.get("success", [])
     else:
         return False
-
+    # 如果有额外通知渠道，则将其与默认通知渠道进行拼接统一处理，如{'success': ['rtx'], 'data': [{'bk_chat':'xxx'}]}
+    if notify_types.get("data", []):
+        notify_type += notify_types.get("data")
     logger.info(
         "taskflow[id={flow_id}] will send {msg_type} message({notify_type}) to: {receivers}".format(
             flow_id=taskflow.id, msg_type=msg_type, notify_type=notify_type, receivers=receivers
         )
     )
-    send_message(executor, notify_type, receivers, title, content, email_content=email_content)
+    MessageHandler().send(executor, notify_type, receivers, title, content, email_content=email_content)
 
     return True
 

gcloud/shortcuts/message/send_msg.py

@@ -12,14 +12,16 @@
 """
 
 import logging
-
+import requests
 import ujson as json
 
 from gcloud.conf import settings
 
 get_client_by_user = settings.ESB_GET_CLIENT_BY_USER
 logger = logging.getLogger("root")
 
+BK_CHAT_API_ENTRY = settings.BK_CHAT_ROUTE
+
 
 def send_message(executor, notify_type, receivers, title, content, email_content=None):
     # 兼容旧数据
@@ -44,3 +46,46 @@ def send_message(executor, notify_type, receivers, title, content, email_content
                 "taskflow send message failed, kwargs={}, result={}".format(json.dumps(kwargs), json.dumps(send_result))
             )
     return True
+
+
+class MessageHandler:
+    def _get_bkchat_api(self):
+        return "{}/{}".format(BK_CHAT_API_ENTRY, "prod/im/api/v1/send_msg")
+
+    def send(self, executor, notify_type, receivers, title, content, email_content=None):
+        # 兼容旧数据
+        if not email_content:
+            email_content = content
+
+        notify_cmsi = []
+        notify_bkchat = {}
+        for notify in notify_type:
+            if isinstance(notify, dict) and notify.get("bk_chat", None):
+                notify_bkchat.update(notify)
+            else:
+                notify_cmsi.append(notify)
+        if settings.ENABLE_BK_CHAT_CHANNEL and notify_bkchat:
+            self.send_bkchat(notify_bkchat, email_content)
+        send_message(executor, notify_cmsi, receivers, title, content, email_content)
+
+        return True
+
+    def send_bkchat(self, notify, email_content=None):
+        params = {"bk_app_code": settings.APP_CODE, "bk_app_secret": settings.SECRET_KEY}
+
+        data = {
+            "im": "WEWORK",
+            "msg_type": "text",
+            "msg_param": {"content": email_content},
+            "receiver": {"receiver_type": "group", "receiver_ids": [notify.get("bk_chat")]},
+        }
+
+        result = requests.post(url=self._get_bkchat_api(), params=params, json=data)
+        send_result = result.json()
+        if send_result.get("code") == 0:
+            return True
+        else:
+            logger.error(
+                "taskflow send message failed, kwargs={}, result={}".format(json.dumps(data), json.dumps(send_result))
+            )
+            return False