model_script/t.py

# -*- coding: utf-8 -*-
import argparse
import json
from datetime import datetime

import pytz
import requests

server_robot = {
    'webhook': 'https://open.feishu.cn/open-apis/bot/v2/hook/926982f5-e7af-40f5-81fd-27d8f42718e4',
}

level_header_template_map = {
    "info": "turquoise",
    "error": "red",
    "warn": "yellow"
}

level_header_title_content_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 timestamp_format(timestamp: str) -> str:
    try:
        return (datetime.utcfromtimestamp(int(timestamp))
                .replace(tzinfo=pytz.UTC)
                .astimezone(pytz.timezone('Asia/Shanghai'))
                .strftime('%Y-%m-%d %H:%M:%S')
                )
    except ValueError as e:
        return timestamp


def train_info_parse(train_info_file: str) -> dict:
    train_info = {}
    with open(train_info_file, 'r') as f:
        for line in f:
            split = line.split(":")
            if len(split) == 2:
                key = split[0].strip()
                value = split[1].strip()
                train_info[key] = value
    return train_info


def seconds_convert(seconds):
    hours = seconds // 3600
    minutes = (seconds % 3600) // 60
    seconds = seconds % 60
    return f"{hours}小时 {minutes}分钟 {seconds}秒"


def join_end_time(msg_text: str, train_info: dict) -> str:
    if "结束时间" in train_info:
        msg_text = f"{msg_text}" \
                   f"\n- 结束时间: {timestamp_format(train_info['结束时间'])}"
    return msg_text


def join_start_time(msg_text: str, train_info: dict) -> str:
    if "开始时间" in train_info:
        msg_text = f"{msg_text}" \
                   f"\n- 开始时间: {timestamp_format(train_info['开始时间'])}"
    return msg_text

def join_running_time(msg_text: str, train_info: dict) -> str:

    return msg_text

def _monitor(model_train_info: str):
    """消息推送"""
    msg_text = f"- 当前时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
    train_info = train_info_parse(model_train_info)
    start = train_info['开始时间']
    result = train_info['广告模型自动更新任务结果']
    level = "error" if result == '失败' else "info"

    card_json = {
        "schema": "2.0",
        "header": {
            "title": {
                "tag": "plain_text",
                "content": level_header_title_content_map[level]
            },
            "template": level_header_template_map[level]
        },
        "body": {
            "elements": [
                {
                    "tag": "markdown",
                    "content": msg_text,
                    "text_align": "left",
                    "text_size": "normal",
                    "element_id": "overview"
                }
            ]
        }
    }
    # if top10 is not None and len(top10) > 0:
    #     collapsible_panel = {
    #         "tag": "collapsible_panel",
    #         "header": {
    #             "title": {
    #                 "tag": "markdown",
    #                 "content": "**Top10差异详情**"
    #             },
    #             "vertical_align": "center",
    #             "padding": "4px 0px 4px 8px"
    #         },
    #         "border": {
    #             "color": "grey",
    #             "corner_radius": "5px"
    #         },
    #         "element_id": "detail",
    #         "elements": [
    #             {
    #                 "tag": "markdown",
    #                 "content": top10.replace("\\n", "\n").replace("\\t", "\t"),
    #                 "element_id": "Top10CID"
    #             }
    #         ]
    #     }
    #     card_json['body']['elements'].append(collapsible_panel)

    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("--tif", type=str, help='模型模型过程记录文件', required=True)
    args = parser.parse_args()

    _monitor(model_train_info=args.tif)