test/send_message.py

from Schedule.send_alarm import ram_dict, post_text, Get_Alarm_Info, Get_Renew_Info


def send_message():
    Info = Get_Alarm_Info()
    alarm_list = []
    for point in ram_dict():
        alarm_list.append(f"""## <font color=\"green\">********阿里云{point}子RAM日常监控: ********</font>\n""")
        if not Info[point]:
            alarm_list.append(f'**当前账户暂无告警**')
        else:
            for alarm in Info[point]:
                alarm_list.append(f"""***告警节点:*** <font color=\"red\">{alarm['InstanceName']}</font\
>\n**告警推送时间:** <font color=\"comment\">{alarm['AlertTime']}</font\
>\n**告警持续时间（小时）:** <font color=\"comment\">{alarm['LastTime']}</font\
>\n**触发规则:** <font color=\"red\">{alarm['RuleName']}</font\
>\n**告警值:** <font color=\"comment\">{alarm['Value']}</font>\n""")

                if alarm != Info[point][-1]:
                    alarm_list.append('-' * 40 + '\n')
        send_str = ''.join(alarm_list)
        if len(send_str.encode('utf-8')) > 4000:
            new_list = list(send_str)[:3000]
            new_str = ''.join(new_list)
            new_str = new_str + '\n**markdown长度限制，多余部分省略。。。**'
        else:
            new_str = send_str
        try:
            post_text(new_str)
            alarm_list = []
            logger.info('日常监控定时任务执行成功。')
        except Exception as e:
            logger.error('日常监控定时任务异常：%s' % str(e))