model_script/recommend/rec_monitor_push.py

#!/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
    )