# 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()