任务失败推送到飞书

importoracledbimportrequestsimportjson# ====================== 配置区 请自行修改 ======================db_host="192.168.1.100"db_port=1521db_service=""db_user=""db_pwd=""#飞书机器人Webhook 地址FEISHU_WEBHOOK="https://"# ==============================================================# 扩展后的SQL：新增 用户名、任务状态 两个字段，一共6个字段sql=""" SELECT '1' AS "任务表名", '1' AS "用户名", '1' AS "存储过程包名", '1' as "存储过程名", '1' AS "执行日期", '1' AS "任务状态" FROM test WHERE IS_SUCCESS = '失败' """defsend_feishu_msg(text):data={"msg_type":"text","content":{"text":text}}headers={"Content-Type":"application/json"}resp=requests.post(FEISHU_WEBHOOK,data=json.dumps(data),headers=headers)print("飞书推送结果:",resp.text)defmain():dsn=oracledb.makedsn(db_host,db_port,service_name=db_service)conn=oracledb.connect(user=db_user,password=db_pwd,dsn=dsn)cur=conn.cursor()cur.execute(sql)rows=cur.fetchall()cur.close()conn.close()ifnotrows:print("✅ 今日无失败BI任务，无需告警")return# 长分割线msg="【BI任务失败告警】\n\n"task_index=1max_task=15show_list=rows[:max_task]iflen(rows)>max_task:msg+=f"⚠️ 共{len(rows)}条失败任务，仅展示前{max_task}条\n\n"forrowinshow_list:# 6个字段对应解包col1,col2,col3,col4,col5,col6=row msg+=f"{task_index}）\n"msg+=f"数据表：{col1}\n"msg+=f"执行用户：{col2}\n"msg+=f"程序包：{col3}\n"msg+=f"存储过程：{col4}\n"msg+=f"执行日期：{col5}\n"msg+=f"任务状态：{col6}\n\n"task_index+=1print("组装完成告警消息：\n",msg)send_feishu_msg(msg)if__name__=="__main__":main()