feedback_duplicate_data/start.py

from MyUtils.MysqlUtils import MysqlUtils
from MyUtils.DateUtils import get_day
from MyUtils.Dingtalk import send_message_mobiles
import datetime
import pytz
import json
import time
import os

SELETE_SQL = "SELECT " \
             "t.project_id, " \
             "t.object_id," \
             "t.source_type," \
             "t.user_id," \
             "t.user_phone," \
             "t.user_name," \
             "t.value_type," \
             "t.item_id," \
             "t.create_time," \
             "COUNT(*) " \
            "FROM %s.feedback t WHERE " \
             "t.create_time >= '%s' AND t.create_time < '%s' GROUP BY " \
             "t.project_id, " \
             "t.object_id," \
             "t.source_type," \
             "t.user_id," \
             "t.user_phone," \
             "t.user_name," \
             "t.value_type," \
             "t.item_id," \
             "t.create_time" \
             " HAVING COUNT(*) > 1"

SELETE_FEEDBACK_COUNT_SQL = "select count(*) from %s.feedback where create_time >= '%s' and create_time < '%s'"


SELECT_ID_SQL = "SELECT t.id FROM %s.feedback t WHERE " \
                "t.project_id='%s' and t.object_id %s and t.source_type %s and t.user_id %s and t.user_phone %s and t.user_name %s and t.value_type='%s' and t.item_id %s and t.create_time='%s'"
INSERT_ID_DATA_SQL = "INSERT INTO %s.feedback_history (" \
                     "id,project_id,object_id,source_type,user_id,user_phone,user_name," \
                     "value_type,item_id,`value`,create_time,next_open_time,model,duration_type," \
                     "custom_plan,curr_temp,nick_name, `result`,exe_result,fb_temp,`remark`) " \
                     "SELECT * FROM %s.feedback WHERE id = '%s'"

UPDATE_ID_SQL = "update %s.feedback_history set duplicate_data = '%s' where id = '%s'"

SELECT_DATA_SQL = "SELECT t.project_id,t.object_id,t.source_type,t.user_id,t.user_phone,t.user_name,t.value_type,t.item_id,t.create_time FROM %s.feedback t WHERE " \
                  "t.create_time >= '%s' AND t.create_time < '%s' " \
                  "GROUP BY t.project_id,t.object_id,t.source_type,t.user_id,t.user_phone,t.user_name,t.value_type,t.item_id,t.create_time " \
                  "HAVING COUNT(*) =1"

INSERT_DADA_SQL = "INSERT INTO %s.feedback_history (" \
                     "id,project_id,object_id,source_type,user_id,user_phone,user_name," \
                     "value_type,item_id,`value`,create_time,next_open_time,model,duration_type," \
                     "custom_plan,curr_temp,nick_name, `result`,exe_result,fb_temp,`remark`,duplicate_data) values "

DELETE_DUPLICATE_SQL = "delete from %s.feedback where id ='%s'"
DELETE_SQL = "delete from %s.feedback where create_time >= '%s' and create_time < '%s'"
DELETE_DUPLICATE_MULT_SQL = "delete from %s.feedback where id in (select * from (SELECT id FROM %s.feedback t WHERE t.create_time >= '%s' AND t.create_time < '%s' GROUP BY t.project_id,t.object_id,t.source_type,t.user_id,t.user_phone,t.user_name,t.`value`,t.item_id,t.create_time HAVING  COUNT(*) = 1) s)"

def datetime_now():
    # datetime_now = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
    #容器时间
    # tz = pytz.timezone('Asia/Shanghai')  # 东八区
    datetime_now = datetime.datetime.fromtimestamp(int(time.time()),
                                    pytz.timezone('Asia/Shanghai')).strftime('%Y-%m-%d %H:%M:%S')
    return datetime_now

def None_data(data):
    if data is None:
        data = "is NULL"
    else:
        data = "= '%s'" % data
    return data


def data_sql(SQL,tag):
    data_infos = MysqlUtil.query(SQL % (my_database, create_time_start, create_time_end))
    if data_infos:
        print("%s 时间段%s至%s,tag为%s,开始插入数据到feedback_history表" % (datetime_now(), create_time_start, create_time_end,tag))
        for data in data_infos:
            project_id = data[0]
            object_id = None_data(data[1])
            source_type = None_data(data[2])
            user_id = None_data(data[3])
            user_phone = None_data(data[4])
            user_name = None_data(data[5])
            value_type = data[6]
            item_id = None_data(data[7])
            create_time = data[8]
            create_time = datetime.datetime.strftime(create_time, "%Y-%m-%d %H:%M:%S")
            # 获取数据的id
            data_infos_id = MysqlUtil.query(SELECT_ID_SQL % (
            my_database, project_id, object_id, source_type, user_id, user_phone, user_name, value_type, item_id,
            create_time))
            if data_infos_id:
                result = None
                for data_id in data_infos_id:
                    # print("%s 开始插入数据%s" % (datetime_now(), data_id[0]))
                    # #将数据插入到history表中
                    sql1 = INSERT_ID_DATA_SQL % (my_database, my_database, data_id[0])
                    # #打标签”Y，N“
                    sql2 = UPDATE_ID_SQL % (my_database, tag, data_id[0])
                    # # 删除feedback数据
                    # sql3 = DELETE_DUPLICATE_SQL % (my_database, data_id[0])
                    result = MysqlUtil.update_mult(sql1, sql2)
                    if result == True:
                        duplicate_data_sum.append(data_id[0])
                if tag == "Y" and result == True:
                    MysqlUtil.update(UPDATE_ID_SQL % (my_database, "N", data_infos_id[0][0]))
    else:
        print("%s tag为%s,时间段%s至%s没有数据" % (datetime_now(),tag, create_time_start, create_time_end))




with open("config.json", "r") as f:
    data = json.load(f)
    mysql = data["mysql"]
    my_database = mysql["database"]
    dingding = data["dingding"]
    at_mobiles = data["at_mobiles"]

# mysql = {
#     "host": os.getenv("host"),
#     "port": int(os.getenv("port")),
#     "user": os.getenv("user"),
#     "passwd": os.getenv("passwd"),
#     "database": os.getenv("database")
# }
# my_database = os.getenv("database")
# dingding = os.getenv("dingding")
# at_mobiles = os.getenv("at_mobiles")

while True:
    time_now = datetime.datetime.fromtimestamp(int(time.time()),
                                               pytz.timezone('Asia/Shanghai')).strftime('%H:%M:%S')
    # time_now = time.strftime("%H:%M:%S", time.localtime())  # 刷新
    am_set_time = "06:00:00"
    if time_now == am_set_time:  # 此处设置每天定时的时间

        # #连接mysql
        MysqlUtil = MysqlUtils(**mysql)
        # starttime = '20230301000000'
        # endtime = '20230319000000'

        # yesterday = (datetime.datetime.today() - datetime.timedelta(days=1)).strftime("%Y%m%d")+"000000"
        # today = datetime.datetime.today().strftime("%Y%m%d")+"000000"
        today = datetime.datetime.fromtimestamp(int(time.time()),
                                        pytz.timezone('Asia/Shanghai')).strftime("%Y%m%d" ) +"000000"
        yesterday = (datetime.datetime.fromtimestamp(int(time.time()),
                                        pytz.timezone('Asia/Shanghai')) - datetime.timedelta(days=1)).strftime("%Y%m%d" )  +"000000"

        first_14day = (datetime.datetime.fromtimestamp(int(time.time()),
                                        pytz.timezone('Asia/Shanghai')) - datetime.timedelta(days=14)).strftime("%Y%m%d" )  +"000000"
        first_15day = (datetime.datetime.fromtimestamp(int(time.time()),
                                        pytz.timezone('Asia/Shanghai')) - datetime.timedelta(days=15)).strftime("%Y%m%d" )  +"000000"

        day_range = get_day(yesterday,today)

        for day in day_range:
            duplicate_data_sum = []
            create_time_start,create_time_end = day[0],day[1]
            feedback_sum = MysqlUtil.query(SELETE_FEEDBACK_COUNT_SQL%(my_database,create_time_start,create_time_end))[0][0]
            #获取重复数据
            data_sql(SELETE_SQL,"Y")
            #获取不重复数据
            data_sql(SELECT_DATA_SQL,"N")

            #对比查询记录和插入记录是否一致
            if feedback_sum == len(duplicate_data_sum):
                print("%s 时间段%s至%s,处理数据%s条，成功！"%(datetime_now(),create_time_start,create_time_end,feedback_sum))
                # message = ("【feedback表打唯一数据标签】 %s 时间段%s至%s，处理数据%s条，成功 \n") % (datetime_now(),create_time_start,create_time_end,feedback_sum)
                # send_message_mobiles(message,dingding,at_mobiles)

                #删除14天之前的数据，保留最新2周的数据
                del_sql = DELETE_SQL % (my_database,first_15day,first_14day)
                MysqlUtil.update(del_sql)

            else:
                print("%s 时间段%s至%s,查询数据%s条，插入数据%s条,失败！"%(datetime_now(),create_time_start,create_time_end,feedback_sum,len(duplicate_data_sum)))
                message = ("【feedback表打唯一数据标签】 %s 时间段%s至%s,查询数据%s条，插入数据%s条,失败 \n")%(datetime_now(),create_time_start,create_time_end,feedback_sum,len(duplicate_data_sum))
                send_message_mobiles(message,dingding,at_mobiles)
        MysqlUtil.close()

        time.sleep(2)