feat:添加模型更新通知脚本

recommend/rec_monitor_push.py

@@ -0,0 +1,89 @@
+#!/usr/bin/env python
+# coding=utf-8
+import argparse
+import json
+from datetime import datetime
+
+import requests
+
+server_robot = {
+    'webhook': 'https://open.feishu.cn/open-apis/bot/v2/hook/c28a6fea-2323-4d09-94fb-943e5ccdb61b',
+}
+
+level_header_template_map = {
+    "info": "turquoise",
+    "error": "red",
+    "warn": "yellow"
+}
+
+level_header_title_suffix_map = {
+    "info": "自动更新通知",
+    "error": "自动更新告警",
+    "warn": "自动更新告警"
+}
+
+level_task_status_map = {
+    "info": "任务执行成功",
+    "error": "任务执行失败",
+    "warn": "任务执行失败",
+}
+
+
+def send_card_msg_to_feishu(webhook, card_json):
+    """发送消息到飞书"""
+    headers = {'Content-Type': 'application/json'}
+    payload_message = {
+        "msg_type": "interactive",
+        "card": card_json
+    }
+    print(f"推送飞书消息内容: {json.dumps(payload_message)}")
+    response = requests.request('POST', url=webhook, headers=headers, data=json.dumps(payload_message))
+    print(response.text)
+
+
+def _monitor(level, model, msg: str):
+    """消息推送"""
+    now = datetime.now()
+    msg = msg.replace("\\n", "\n").replace("\\t", "\t")
+    mgs_text = f"- 当前时间: {now.strftime('%Y-%m-%d %H:%M:%S')}" \
+               f"\n- 任务状态: {level_task_status_map[level]}" \
+               f"\n- 任务描述: {msg}"
+    card_json = {
+        "schema": "2.0",
+        "header": {
+            "title": {
+                "tag": "plain_text",
+                "content": model + level_header_title_suffix_map[level]
+            },
+            "template": level_header_template_map[level]
+        },
+        "body": {
+            "elements": [
+                {
+                    "tag": "markdown",
+                    "content": mgs_text,
+                    "text_align": "left",
+                    "text_size": "normal",
+                    "element_id": "overview"
+                }
+            ]
+        }
+    }
+    send_card_msg_to_feishu(
+        webhook=server_robot.get('webhook'),
+        card_json=card_json
+    )
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser(description='告警Utils')
+    parser.add_argument('--level', type=str, help='通知级别, info, warn, error', required=True)
+    parser.add_argument('--model', type=str, help='消息', required=True)
+    parser.add_argument('--msg', type=str, help='消息', required=True)
+    args = parser.parse_args()
+
+    _monitor(
+        level=args.level,
+        model=args.model,
+        msg=args.msg
+    )