python脚本源码检查数据库内容并上报飞书

#-- coding: UTF-8 --
import time
import unittest
from StdSuites import data
import operator
import pymysql
import requests
import json
import datetime
def mysql_list(database,table_name):
conn = pymysql.connect(
host=‘数据库名称’,
port=3306,
user=‘用户名’,
passwd='密码‘,
db=database,
charset=‘utf8’
)
cursor = conn.cursor()
sql=“数据库语句;” %table_name
cursor.execute(sql)

获取所有记录列表

results= cursor.fetchall()
table_list=[]
for r in results:
table_list.append(list®)
# print(table_list)
cursor.close()
conn.close()
return list(table_list)

def cg():
x1 = mysql_list(‘数据库名’, 表名’)
time.sleep(5*60)
x2 = mysql_list(‘数据库名’, ‘表名’)
# return cmp(x1, x2)
# cgtest=cmp(x1, x2)
cgtest=operator.eq(x1, x2)
# print(x1)
# print(x2)
if len(x2):
if cgtest == True: #如果五分钟前后结果完全一致，输出任意一次查询结果
x3 = json.dumps(x2, cls=DateEncoder)
# print(“五分钟前后：数据一致:”)
# print("数据为： "+x3)
return(x3)
else: #如果五分钟结果不一致，输出两次查询结果中的共同值
x4 = []
for i in x1:
if i in x2:
x4.append(i)
# print(x4)
x5 = json.dumps(x4, cls=DateEncoder)
print(“五分钟前后：数据一致”)
return(x5)
else:
print(“数据无异常”)
return(1)

class DateEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, datetime.datetime):
return obj.strftime(‘%Y-%m-%d %H:%M:%S’)
elif isinstance(obj, datetime.date):
return obj.strftime(“%Y-%m-%d”)
else:
return json.JSONEncoder.default(self, obj)

def sender(msg):
payload_message = {
“msg_type”: “text”,
“content”: {
“text”: msg,
}
}
print(payload_message)
# data1=(payload_message)
data2=json.dumps(payload_message)
# print(data2)
# print(data1[2])
headers = {‘Content-Type’: ‘application/json’}
# requests.request(“POST”, url=“飞书机器人地址”,
# headers=headers, data=json.dumps(payload_message))
response=requests.request(“POST”, url=“飞书机器人地址”,
headers=headers, data=data2)
print(response.text)

if name == ‘main’:

cgtest=cg()
# print(cgtest)
if cgtest==1:
print(“无错误”)
else:
sender(‘数据异常：具体信息为： ‘+’\n’+cgtest)