renyupeng před 3 roky
rodič
revize
a96f7a53b0

+ 29 - 4
Apietl/utlis/Utils.py

@@ -6,10 +6,9 @@
 # @Software: PyCharm
 # @contact: renyupeng@c-top.com.cn
 import argparse
-import ast
-import getopt
-import sys
-from urllib.parse import unquote
+import datetime
+from datetime import timedelta, date
+import calendar
 
 
 class Utils:
@@ -36,3 +35,29 @@ class Utils:
         date_time = args.date_time
         day_count = args.day_count
         return date_time, day_count
+
+    @staticmethod
+    def get_month_range(start_date=None):
+        if start_date is None:
+            start_date = date.today().replace(day=1)
+            _, days_in_month = calendar.monthrange(start_date.year, start_date.month)
+        end_date = start_date + timedelta(days=days_in_month)
+        return start_date, end_date
+
+    @staticmethod
+    def get_days():
+        yesterday = (datetime.date.today() + datetime.timedelta(-1)).strftime("%Y-%m-%d")
+        ll = []
+        day = timedelta(days=1)
+        startday, endday = Utils.get_month_range()
+        print(str(startday))
+        while startday < endday:
+            ll.append(str(startday))
+            startday += day
+        print('----',ll)
+        lt = ll.index(yesterday)
+        return lt
+
+
+if __name__ == '__main__':
+    print(Utils.get_days())

+ 127 - 0
OuterDataCompareInnerData/MonitorApp/MonitorAppTask.py

@@ -0,0 +1,127 @@
+# 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:
+                            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()

+ 73 - 0
OuterDataCompareInnerData/MonitorMasterData/MonitorTask.py

@@ -0,0 +1,73 @@
+# Author renyupeng
+# coding=utf-8
+# @Time    : 2021/11/17 6:48 下午
+# @Site    : 
+# @File    : MonitorTask.py
+# @Software: PyCharm
+# @contact: renyupeng@c-top.com.cn
+# @Tel 1501435553
+import logging
+from Apietl.conn.TidbConn import TidbConn
+from Apietl.conn.TidbProConn import TidbProConn
+from OuterDataCompareInnerData.Utils.sendMessage import feishuMsg
+
+
+class MonitorTask:
+    def __init__(self):
+        self.InnerConn = TidbConn.get_connection()
+        self.OuterConn = TidbProConn.get_connection()
+        self.table_list = ['ctop_account_attribution_info',
+                           'ctop_advertiser',
+                           'ctop_cwjs_agent_info',
+                           'ctop_material_ascription',
+                           'ctop_material_image_info',
+                           'ctop_material_info',
+                           'ctop_material_tag_info',
+                           'ctop_oauth_token',
+                           'ctop_product',
+                           'ctop_project',
+                           'ctop_project_member',
+                           'ctop_tag_info',
+                           'ctop_user_allocation',
+                           'sys_company',
+                           'sys_depart',
+                           'sys_depart_permission',
+                           'sys_depart_role',
+                           'sys_depart_role_permission',
+                           'sys_depart_role_user',
+                           'sys_permission',
+                           'sys_role',
+                           'sys_role_permission',
+                           'sys_user',
+                           'sys_user_agent',
+                           'sys_user_depart',
+                           'sys_user_role',
+                           'user_company'
+                           ]
+
+    def CheckTask(self):
+        try:
+            for table in self.table_list:
+                try:
+                    sql = """select count(1) from `jeecg-boot`.{table}""".format(table=table)
+                    InnertotalNum = TidbConn.quary(self.InnerConn, sql)[0][0]
+                    OutertotalNum = TidbConn.quary(self.OuterConn, sql)[0][0]
+                    if InnertotalNum != OutertotalNum:
+                        print("""{table} 内网数量为{InnertotalNum},外网数量为{OutertotalNum} 数量不想等""".format(table=table,
+                                                                                                   InnertotalNum=InnertotalNum,
+                                                                                                   OutertotalNum=OutertotalNum))
+                        feishuMsg.send_robot_msg(
+                            """{table} 内网数量为{InnertotalNum},外网数量为{OutertotalNum}数量不想等""".format(table=table,
+                                                                                                InnertotalNum=InnertotalNum,
+                                                                                                OutertotalNum=OutertotalNum))
+                except Exception as e:
+                    logging.error(e)
+            TidbConn.conn_close(conn=self.InnerConn)
+            TidbProConn.conn_close(conn=self.OuterConn)
+
+        except Exception as e:
+            logging.error(e)
+
+
+if __name__ == '__main__':
+    MonitorTask().CheckTask()

+ 175 - 0
OuterDataCompareInnerData/MonitorMediaApi/MonitorMediaApi.py

@@ -0,0 +1,175 @@
+# Author renyupeng
+# coding=utf-8
+# @Time    : 2021/11/25 2:55 下午
+# @Site    : 
+# @File    : MonitorMediaApi.py
+# @Software: PyCharm
+# @contact: renyupeng@c-top.com.cn
+# @Tel 1501435553
+import datetime
+from urllib import parse
+import pandas as pd
+from sqlalchemy import create_engine
+from OuterDataCompareInnerData.Utils.sendMessage import feishuMsg
+
+
+class MonitorMediaApi:
+    def __init__(self):
+        self.InnerStr = "mysql+pymysql://%s:%s@%s:%d/%s" % (
+            "hcst", parse.quote_plus("hcst@2021"), "192.168.0.184", 3390, "media_api")
+        self.InnerConn = create_engine(self.InnerStr,
+                                       pool_size=10,
+                                       max_overflow=5,
+                                       pool_timeout=10,
+                                       pool_recycle=3600)
+        self.OuterStr = "mysql+pymysql://%s:%s@%s:%d/%s" % (
+            "data", parse.quote_plus("hcst@2021"), "139.186.27.96", 3390, "jeecg-boot")
+        self.OuterConn = create_engine(self.OuterStr,
+                                       pool_size=10,
+                                       max_overflow=5,
+                                       pool_timeout=10,
+                                       pool_recycle=3600)
+        self.Media_List = {
+            "bytedance_ad_report_daily": "ctop_bytedance_report_plan_daily",
+            "bytedance_advertiser_report_daily": "ctop_bytedance_report_advertiser_daily",
+            "bytedance_creative_report_daily": "ctop_bytedance_report_creative_daily",
+            "bytedance_material_report_daily": "ctop_bytedance_report_material_daily",
+            "bytedance_campaign_report_daily": "ctop_bytedance_report_campaign_daily",
+            "kuaishou_account_report_daily": "ctop_kuaishou_report_daily_account",
+            "kuaishou_campaign_report_daily": "ctop_kuaishou_report_daily_campaign",
+            "kuaishou_material_video_report_daily": "ctop_kuaishou_report_daily_material",
+            "kuaishou_unit_report_daily": "ctop_kuaishou_report_daily_group"
+        }
+        self.stat_date = (datetime.date.today() + datetime.timedelta(-1)).strftime("%Y%m%d")
+        self.date = (datetime.date.today() + datetime.timedelta(-1)).strftime("%Y-%m-%d")
+
+    def compare_task(self):
+        for innertable, outertable in self.Media_List.items():
+            if 'bytedance' in innertable:
+                if innertable == 'bytedance_material_report_daily':
+                    sql = """select advertiser_id,sum(cost) as cost from media_api.{innertable} where stat_datetime={date}
+                            and  image_mode in('CREATIVE_IMAGE_MODE_VIDEO', 'CREATIVE_IMAGE_MODE_VIDEO_VERTICAL', 
+                            'CREATIVE_IMAGE_MODE_UNION_SPLASH_VIDEO')
+                            group by advertiser_id""". \
+                        format(date=self.stat_date, innertable=innertable)
+                    sql2 = """select {stat_date} as stat_datetime,account_id as advertiser_id,sum(cost) as cost from 
+                    `jeecg-boot`.{outertable}
+                    where  stat_datetime ='{date}'
+                    and image_mode in('CREATIVE_IMAGE_MODE_VIDEO', 'CREATIVE_IMAGE_MODE_VIDEO_VERTICAL', 
+                    'CREATIVE_IMAGE_MODE_UNION_SPLASH_VIDEO')
+                    group by account_id,stat_datetime""".format(date=self.date, stat_date=self.stat_date,
+                                                                outertable=outertable)
+                    InnerAdvertiserListFrame = pd.read_sql(sql, self.InnerConn)
+                    OuterAdvertiserListFrame = pd.read_sql(sql2, self.OuterConn)
+                    data = pd.merge(OuterAdvertiserListFrame, InnerAdvertiserListFrame, how='left', on='advertiser_id')
+                    result = data[data['cost_x'] != data['cost_y']]
+                    # print('线上{outertable}对内网上{innertable} 结果为'.format(innertable=innertable, outertable=outertable),
+                    #       result)
+                    if len(result.values.tolist()) == 0:
+                        message = """线上{outertable}对内网{innertable} 数据相等""".format(innertable=innertable,
+                                                                                  outertable=outertable)
+                        feishuMsg.send_robot_msg(
+                            """线上{outertable}对内网{innertable} 数据相等""".format(innertable=innertable,
+                                                                            outertable=outertable))
+                    else:
+                        messageList = ''
+                        for result in result.values.tolist():
+                            date = result[0]
+                            account_id = result[1]
+                            outerCost = result[2]
+                            innerCost = result[3]
+                            message = """广告主id={account_id} 线上消耗={outerCost},内网消耗={innerCost}""".format(date=date,
+                                                                                                        account_id=account_id,
+                                                                                                        outerCost=outerCost,
+                                                                                                        innerCost=innerCost)
+                            # print(message)
+                            message = message + '\n'
+                            messageList += message
+                        feishuMsg.send_robot_msg(
+                            """{date}线上{outertable}和内网{innertable}差异为:""".format(date=self.stat_date,
+                                                                                 outertable=outertable,
+                                                                                 innertable=innertable + '\n' + messageList)
+                        )
+
+
+
+
+
+
+                else:
+                    sql = """select advertiser_id,sum(cost) as cost from media_api.{innertable} where stat_datetime={date}
+                            group by advertiser_id""". \
+                        format(date=self.stat_date, innertable=innertable)
+                    sql2 = """select {stat_date} as stat_datetime,advertiser_id as advertiser_id,sum(cost) as cost from 
+                    `jeecg-boot`.{outertable}
+                    where  stat_datetime ='{date}'
+                    group by advertiser_id,stat_datetime""".format(date=self.date, stat_date=self.stat_date,
+                                                                   outertable=outertable)
+                    InnerAdvertiserListFrame = pd.read_sql(sql, self.InnerConn)
+                    OuterAdvertiserListFrame = pd.read_sql(sql2, self.OuterConn)
+                    data = pd.merge(OuterAdvertiserListFrame, InnerAdvertiserListFrame, how='left', on='advertiser_id')
+                    result = data[data['cost_x'] != data['cost_y']]
+
+                    if len(result.values.tolist()) == 0:
+                        feishuMsg.send_robot_msg(
+                            '线上{outertable}对内网{innertable} 数据相等'.format(innertable=innertable, outertable=outertable))
+                    else:
+                        messageList = ''
+                        for result in result.values.tolist():
+                            date = result[0]
+                            account_id = result[1]
+                            outerCost = result[2]
+                            innerCost = result[3]
+                            message = """广告主id={account_id} 线上消耗={outerCost},内网消耗={innerCost}""".format(date=date,
+                                                                                                        account_id=account_id,
+                                                                                                        outerCost=outerCost,
+                                                                                                        innerCost=innerCost)
+                            # print(message)
+                            message = message + '\n'
+                            messageList += message
+                        feishuMsg.send_robot_msg(
+                            """{date}线上{outertable}和内网{innertable}差异为:""".format(date=self.stat_date,
+                                                                                 outertable=outertable,
+                                                                                 innertable=innertable + '\n' + messageList)
+                        )
+
+            else:
+                sql = """select advertiser_id,sum(charge) as cost from media_api.{innertable} where stat_date={date}
+                        group by advertiser_id""". \
+                    format(date=self.stat_date, innertable=innertable)
+                # print(sql)
+                sql2 = """select {stat_date} as stat_date,account_id as advertiser_id,sum(charge) as cost from 
+                `jeecg-boot`.{outertable}
+                where  stat_date ='{date}'
+                group by account_id,stat_date""".format(date=self.date, stat_date=self.stat_date,
+                                                        outertable=outertable)
+                InnerAdvertiserListFrame = pd.read_sql(sql, self.InnerConn)
+                OuterAdvertiserListFrame = pd.read_sql(sql2, self.OuterConn)
+                # print(sql2)
+                data = pd.merge(OuterAdvertiserListFrame, InnerAdvertiserListFrame, how='left', on='advertiser_id')
+                result = data[data['cost_x'] != data['cost_y']]
+                if len(result.values.tolist()) == 0:
+                    feishuMsg.send_robot_msg(
+                        '线上{outertable}对内网{innertable} 数据相等'.format(innertable=innertable, outertable=outertable))
+                else:
+                    messageList = ''
+                    for result in result.values.tolist():
+                        date = result[0]
+                        account_id = result[1]
+                        outerCost = result[2]
+                        innerCost = result[3]
+                        message = """广告主id={account_id} 线上消耗={outerCost},内网消耗={innerCost}""".format(date=date,
+                                                                                                    account_id=account_id,
+                                                                                                    outerCost=outerCost,
+                                                                                                    innerCost=innerCost)
+                        # print(message)
+                        message = message + '\n'
+                        messageList += message
+                    feishuMsg.send_robot_msg(
+                        """{date}线上{outertable}和内网{innertable}差异为:""".format(date=self.stat_date, outertable=outertable,
+                                                                             innertable=innertable + '\n' + messageList)
+                    )
+
+
+if __name__ == '__main__':
+    MonitorMediaApi().compare_task()

+ 30 - 0
OuterDataCompareInnerData/Utils/SchedulerList.py

@@ -0,0 +1,30 @@
+# Author renyupeng
+# coding=utf-8
+# @Time    : 2021/11/24 1:37 下午
+# @Site    :
+# @File    : SchedulerList.py
+# @Software: PyCharm
+# @contact: renyupeng@c-top.com.cn
+# @Tel 1501435553
+
+
+class SchedulerList:
+
+    def __init__(self):
+        self.Scheduler_List = {"bytedance_ad_report_daily_dw": "application_bytedance_ad.json",
+                               "bytedance_advertiser_report_daily_dw": "application_bytedance_account.json",
+                               "bytedance_creative_report_daily_dw": "application_bytedance_creative.json",
+                               "bytedance_material_image_report_daily_dw": "application_bytedance_image.json",
+                               "bytedance_material_video_report_daily_dw": "application_bytedance_video.json",
+                               "bytedance_campaign_report_daily_dw":"application_bytedance_campaign.json",
+                               "kuaishou_account_report_daily_dw": "application_account.json",
+                               "kuaishou_campaign_report_daily_dw": "application_campaign.json",
+                               "kuaishou_material_video_report_daily_dw": "application_video.json",
+                               "kuaishou_unit_report_daily_dw": "application_unit.json"}
+    @classmethod
+    def SchedulerList(cls, table):
+        try:
+            return SchedulerList().Scheduler_List[table]
+        except KeyError:
+            return None
+

+ 46 - 0
OuterDataCompareInnerData/Utils/sendMessage.py

@@ -0,0 +1,46 @@
+# Author renyupeng
+# coding=utf-8
+# @Time    : 2021/11/3 4:00 下午
+# @Site    : 
+# @File    : sendMessage.py
+# @Software: PyCharm
+# @contact: renyupeng@c-top.com.cn
+# @Tel 1501435553
+import json
+import logging
+import traceback
+import requests
+
+
+class feishuMsg:
+    @classmethod
+    def send_robot_msg(cls, msg_content):
+        try:
+            url = 'https://open.feishu.cn/open-apis/bot/v2/hook/c567fbee-be1f-4143-987e-19decdade0d2'
+            headers = {
+                'Content-Type': 'application/json'
+            }
+            req_data = {
+                "msg_type": "text",
+                "content": {
+                    "text": str(msg_content)
+                }
+            }
+
+            data = json.dumps(req_data)
+
+            logging.info("send_robot_msg req_url:{}, req_data:{}".format(url, data))
+
+            response = requests.post(url=url, data=data, headers=headers)
+            result = response.json()
+            print(result)
+            logging.info("send_robot_msg response:{}".format(response.json()))
+
+            if result.get("StatusCode") == 0:
+                logging.info("钉钉通知发送成功")
+                return True
+            else:
+                logging.error("钉钉通知发送失败,请检查!")
+                return False
+        except Exception as e:
+            logging.error("send_robot_msg Error:{}".format(traceback.format_exc()))

+ 0 - 1
monitor/ApplicationCDCMonitor.py

@@ -69,7 +69,6 @@ class ApplicationCDCMonitor:
                         '--sink-uri="mysql://data:hcst@2021@139.186.27.96:3390" --config ../conf/{task_id}.toml '
                         '--changefeed-id={task_id} '.format(task_id=task_id))
                 else:
-                    print(back, '--------2-------')
                     if back['error'] is not None:
                         requests.delete(url='http://{}:{}{}/{}'.format(self.cdc_url, self.cdc_port,
                                                                        self.cdc_changefeeds_url, task_id))