效率工具：企业微信机器人完成脚本工作-异常监控（sql篇）

news2024/11/15 11:19:52

效率工具：定时完成脚本工作（sql篇）

背景：elon在公司需要人工查看某些数据是否异常，并发送到指定企业微信群中。

处理步骤：

工具：企业微信中有机器人

可以指定webhook地址嵌入到脚本中，定时执行脚本即可向这个地址发送信息。

用python脚本完成对数据库的读取，并发送指定文本信息到webhook地址。

一、导入库文件，连接数据库

import mysql.connector
import pandas as pd 
import numpy as np
from datetime import datetime
import matplotlib.pyplot as plt

import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy import text
import urllib.parse

import time
import requests
import json
def execute_mysql_query(query):
    password = urllib.parse.quote('密码')
    engine = create_engine('mysql+mysqlconnector://用户名:' + password + '@主机地址:端口号/数据库')
    with engine.connect() as connection:
        df = pd.read_sql(query, connection)
    return df

二、读取sql文件，获取监控指标

def read_file(file_path):
    sql = open(file_path, 'r', encoding = 'utf8')
    sqltxt = sql.readlines()
    # 此时 sqltxt 为 list 类型
    # 读取之后关闭文件
    sql.close()
    # list 转 str
    sql = "".join(sqltxt)
    return sql
##1.返点监控###########################################################################################################################
#2. 执行地址下的sql文件
返点监控_是否为空_sql = read_file("D:\Marshal1\datagrip_data\关键数据指标核对\返点监控_是否为空.sql")

三、对指标进行异常判断

df_返点监控_是否为空 = execute_mysql_query(返点监控_是否为空_sql)
if df_返点监控_是否为空['计入应收时间'].notnull().any():
    已结算计划计入应收时间是否为空='是'
else :
    已结算计划计入应收时间是否为空='否'

四、将异常信息传到企业微信机器人。

 #记录启动时间
 current_time=time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())
'https://developer.work.weixin.qq.com/document/path/91770#%E6%96%87%E6%9C%AC%E7%B1%BB%E5%9E%8B 微信文档'
proxies = {}
# 1、填入webhook的key值
webhook =' https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=d13fbffb-7680-4f4c-aab5-1ea8e99f1f7b'
# 2、选择一种样式，发消息
def qywx_message():
    header = { "Content-Type": "application/json" }
    message = {
        "msgtype": "markdown",
        "markdown": {
           "content": 
           f"#{current_time}\n\n"
           f"# 监控中心\n\n"
           f"## \n"
           f"## 返点监控：\n"
           f"> 已结算计划计入应收时间是否为空：<font color=\"info\">{已结算计划计入应收时间是否为空}</font>\n"
           f"> 计划多个结算时间：<font color=\"info\">{计划多个结算时间}</font>\n"
           f"## 组织架构监控：\n"
           f"> bd中是否存在同组同名：<font color=\"info\">{bd中是否存在同组同名}</font>\n"
           f"> 媒介组织架构：<font color=\"info\">{媒介组织架构}</font>\n"
           f"> 媒介主管：<font color=\"info\">{媒介主管是否有变动}</font>\n"
           f"> BD组织架构：<font color=\"warning\">{BD组织架构}</font>\n"
           f"> BD主管： <font color=\"warning\">{BD主管是否有变动}</font>\n"
           f"## 海外联盟标签监控：\n"
           f"> DM：<font color=\"info\">{DM系统业务线}</font>\n"
           f"> KN：<font color=\"info\">{KN系统业务线}</font>\n"
           f"> LKB：<font color=\"info\">{LKB系统业务线}</font>\n"
        }}
    message_json = json.dumps(message)
    send_message = requests.post(url=webhook,data=message_json,headers=header,proxies=proxies)
    print(send_message.text)
qywx_message()