123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156 |
- # Author renyupeng
- # coding=utf-8
- # @Time : 2021/11/17 6:48 下午
- # @Site :
- # @File : MonitorAppTask.py
- # @Software: PyCharm
- # @contact: renyupeng@c-top.com.cn
- # @Tel 1501435553
- import datetime
- import logging
- import os
- from Apietl.conn.TidbConn import TidbConn
- from Apietl.conn.TidbProConn import TidbProConn
- from OuterDataCompareInnerData.Utils.SchedulerList import SchedulerList
- from OuterDataCompareInnerData.Utils.sendMessage import feishuMsg
- class MonitorAppTask:
- def __init__(self):
- self.InnerConn = TidbConn.get_connection()
- self.OuterConn = TidbProConn.get_connection()
- self.table_list = [
- 'bytedance_ad_report_daily_dw'
- , 'bytedance_advertiser_report_daily_dw'
- , 'bytedance_campaign_report_daily_dw'
- , 'bytedance_creative_report_daily_dw'
- , 'bytedance_material_image_report_daily_dw'
- , 'bytedance_material_video_report_daily_dw'
- , 'kuaishou_account_report_daily_dw'
- , 'kuaishou_campaign_report_daily_dw'
- , 'kuaishou_material_video_report_daily_dw'
- , 'kuaishou_unit_report_daily_dw'
- ]
- self.stat_date = (datetime.date.today() + datetime.timedelta(-1)).strftime("%Y%m%d")
- def CheckTask(self):
- try:
- for table in self.table_list:
- try:
- if 'app' in table:
- sql = """select count(1) from `application`.{table}""".format(table=table)
- InnertotalNum = TidbConn.quary(self.InnerConn, sql)[0][0]
- OutertotalNum = TidbConn.quary(self.OuterConn, sql)[0][0]
- print("""{stat_date}日{table} 内网数量为{InnertotalNum},外网数量为{OutertotalNum}""".format(
- stat_date=self.stat_date,
- table=table,
- InnertotalNum=InnertotalNum,
- OutertotalNum=OutertotalNum))
- elif 'kuaishou' in table:
- sql = """select count(1) from `application`.{table} where stat_date={stat_date}""".format(
- table=table, stat_date=self.stat_date)
- InnertotalNum = TidbConn.quary(self.InnerConn, sql)[0][0]
- OutertotalNum = TidbConn.quary(self.OuterConn, sql)[0][0]
- print("""{stat_date}日 {table} 内网数量为{InnertotalNum},外网数量为{OutertotalNum}""".format(
- stat_date=self.stat_date,
- table=table,
- InnertotalNum=InnertotalNum,
- OutertotalNum=OutertotalNum))
- if InnertotalNum == 0:
- feishuMsg.send_robot_msg(
- """{stat_date}日 {table} 数据为0""".format(stat_date=self.stat_date, table=table))
- elif InnertotalNum != OutertotalNum:
- if table == 'kuaishou_material_video_report_daily_dw':
- sql = """select account_id,photo_id from
- application.{table} where stat_date={stat_date} """.format(
- table=table, stat_date=self.stat_date)
- InnertotalItem = TidbConn.quary(self.InnerConn, sql)
- OutertotalItem = TidbConn.quary(self.OuterConn, sql)
- set1 = set(InnertotalItem)
- set2 = set(OutertotalItem)
- diff_set = set1 - set2
- account_list = []
- for diff in diff_set:
- account_id = diff[0]
- account_list.append(account_id)
- acc_list = tuple(account_list)
- acclist = ','.join(str(i) for i in acc_list)
- feishuMsg.send_robot_msg(
- """{stat_date}日 {table} 内网数量为{InnertotalNum},外网数量为{OutertotalNum} 数据不相等 datax
- 数据同步启动""".format(
- stat_date=self.stat_date,
- table=table,
- InnertotalNum=InnertotalNum,
- OutertotalNum=OutertotalNum))
- datax_json = SchedulerList.SchedulerList(table)
- os.system(
- 'sh /data/datax/application/push_pro1.sh {stat_date} {table} {acclist} {datax_json}'.format(
- stat_date=self.stat_date, table=table, acclist=acclist, datax_json=datax_json))
- else:
- feishuMsg.send_robot_msg(
- """{stat_date}日 {table} 内网数量为{InnertotalNum},外网数量为{OutertotalNum} 数据不相等 datax
- 数据同步启动""".format(
- stat_date=self.stat_date,
- table=table,
- InnertotalNum=InnertotalNum,
- OutertotalNum=OutertotalNum))
- datax_json = SchedulerList.SchedulerList(table)
- os.system(
- 'sh /data/datax/application/push_pro.sh {stat_date} {table} {datax_json}'.format(
- stat_date=self.stat_date, table=table, datax_json=datax_json))
- else:
- feishuMsg.send_robot_msg(
- """{stat_date}日 {table} 内网数量为{InnertotalNum},外网数量为{OutertotalNum} 数据相等 """.format(
- stat_date=self.stat_date,
- table=table,
- InnertotalNum=InnertotalNum,
- OutertotalNum=OutertotalNum))
- else:
- sql = """select count(1) from `application`.{table} where stat_datetime={stat_date}""".format(
- table=table, stat_date=self.stat_date)
- InnertotalNum = TidbConn.quary(self.InnerConn, sql)[0][0]
- OutertotalNum = TidbConn.quary(self.OuterConn, sql)[0][0]
- print("""{stat_date}日 {table} 内网数量为{InnertotalNum},外网数量为{OutertotalNum}""".format(
- stat_date=self.stat_date,
- table=table,
- InnertotalNum=InnertotalNum,
- OutertotalNum=OutertotalNum))
- if InnertotalNum == 0:
- feishuMsg.send_robot_msg(
- """{stat_date}日 {table} 数据为0""".format(stat_date=self.stat_date, table=table))
- elif InnertotalNum != OutertotalNum:
- feishuMsg.send_robot_msg(
- """{stat_date}日 {table} 内网数量为{InnertotalNum},外网数量为{OutertotalNum} 数据不相等 datax
- 数据同步启动""".format(
- stat_date=self.stat_date,
- table=table,
- InnertotalNum=InnertotalNum,
- OutertotalNum=OutertotalNum))
- datax_json = SchedulerList.SchedulerList(table)
- os.system('sh /data/datax/application/push_pro.sh {stat_date} {table} {datax_json}'.format(
- stat_date=self.stat_date, table=table, datax_json=datax_json))
- else:
- feishuMsg.send_robot_msg(
- """{stat_date}日 {table} 内网数量为{InnertotalNum},外网数量为{OutertotalNum} 数据相等 """.format(
- stat_date=self.stat_date,
- table=table,
- InnertotalNum=InnertotalNum,
- OutertotalNum=OutertotalNum))
- except Exception as e:
- logging.error(e)
- return -1
- TidbConn.conn_close(conn=self.InnerConn)
- TidbProConn.conn_close(conn=self.OuterConn)
- except Exception as e:
- logging.error(e)
- return -1
- if __name__ == '__main__':
- MonitorAppTask().CheckTask()
|