feat: 新增通知渠道拓展 TencentBlueKing#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通知渠道
+BK_CHAT_CHANNEL = env.BK_CHAT_CHANNEL
+# 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通知渠道
+BK_CHAT_CHANNEL = os.environ.get("BK_CHAT_CHANNEL", None)
+# bk_chat通知路由
+BK_CHAT_ROUTE = os.environ.get("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.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, SendMessage
 from gcloud.periodictask.models import PeriodicTask
 
 logger = logging.getLogger("root")
@@ -39,12 +39,12 @@ def send_task_flow_message(taskflow, msg_type, node_name=""):
         title, content, email_content = title_and_content_for_atom_failed(
             taskflow, taskflow.pipeline_instance, node_name, executor
         )
-        notify_type = notify_types.get("fail", [])
+        notify_type = notify_types.get("fail", []) + notify_types.get("data", [])
     elif msg_type == "task_finished":
         title, content, email_content = title_and_content_for_flow_finished(
             taskflow, taskflow.pipeline_instance, node_name, executor
         )
-        notify_type = notify_types.get("success", [])
+        notify_type = notify_types.get("success", []) + notify_types.get("data", [])
     else:
         return False
 
@@ -53,7 +53,7 @@ def send_task_flow_message(taskflow, msg_type, node_name=""):
             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)
+    SendMessage().send_system(executor, notify_type, receivers, title, content, email_content=email_content)
 
     return True
 

gcloud/shortcuts/message/send_msg.py

@@ -10,16 +10,18 @@
 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.
 """
-
+import env
 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 or env.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 SendMessage:
+    def _get_bkchat_api(self):
+        return "{}/{}".format(BK_CHAT_API_ENTRY, "prod/im/api/v1/send_msg")
+
+    def send_system(self, executor, notify_type, receivers, title, content, email_content=None):
+        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.BK_CHAT_CHANNEL:
+            self.send_bkchat(notify_bkchat, content, email_content)
+        send_message(executor, notify_cmsi, receivers, title, content, email_content)
+
+        return True
+
+    def send_bkchat(self, notify, content, email_content=None):
+        # 兼容旧数据
+        if not email_content:
+            email_content = content
+
+        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