from MyUtils.MysqlUtils import MysqlUtils from MyUtils.ZillionUtil import ZillionUtil from MyUtils.Dingtalk import send_message_markdown from MyUtils.DateUtils import get_day import datetime import pytz import json,pymysql 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" 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 * 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" 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_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 == None: data = "NULL" return data with open("config.json", "r") as f: data = json.load(f) mysql = data["mysql"] my_database = mysql["database"] sleep = data["sleep"] # mysql = { # "host": os.getenv("host"), # "port": int(os.getenv("port")), # "user": os.getenv("user"), # "passwd": os.getenv("passwd"), # "database": os.getenv("database") # } # database = os.getenv("database") # sleep = os.getenv("sleep") # #连接mysql MysqlUtil = MysqlUtils(**mysql) starttime = '20230101000000' endtime = '20230201000000' day_range = get_day(starttime,endtime) for day in day_range: sqls = [] create_time_start,create_time_end = day[0],day[1] #获取重复数据 data_infos = MysqlUtil.query(SELETE_SQL%(my_database,create_time_start,create_time_end)) duplicate_data = 'Y' if data_infos: for data in data_infos: project_id = data[0] object_id = data[1] source_type = data[2] user_id = data[3] user_phone = data[4] user_name = data[5] value_type = data[6] item_id = data[7] create_time = data[8] create_time = datetime.datetime.strftime(create_time,"%Y-%m-%d %H:%M:%S") 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)) #获取重复数据的id if data_infos_id: for data_id in data_infos_id: print("%s 插入数据%s,标签为%s"%(datetime_now(),data_id[0],duplicate_data)) # #将重复数据插入到history表中 sql1 = INSERT_ID_DATA_SQL%(my_database,my_database,data_id[0]) # #打重复的标签”Y“ sql2 = UPDATE_ID_SQL%(my_database,duplicate_data,data_id[0]) # 删除feedback重复数据 sql3 = DELETE_DUPLICATE_SQL%(my_database,data_id[0]) MysqlUtil.update_mult(sql1,sql2,sql3) else: print("%s 时间段%s至%s没有重复数据"%(datetime_now(),create_time_start,create_time_end)) #获取没有重复的数据 data_all_infos = MysqlUtil.query(SELECT_DATA_SQL%(my_database,create_time_start,create_time_end)) duplicate_data_N = 'N' if data_all_infos: for d in data_all_infos: id = d[0] project_id = d[1] object_id = d[2] source_type = d[3] user_id = d[4] user_phone = d[5] user_name = d[6] value_type = d[7] item_id = None_data(d[8]) value = None_data(d[9]) create_time = d[10] next_open_time = d[11] model = d[12] duration_type = None_data(d[13]) custom_plan = d[14] curr_temp = d[15] nick_name = d[16] result = None_data(d[17]) exe_result = d[18] exe_result = None_data(exe_result) fb_temp = None_data(d[19]) remark = d[20] print("%s 查询数据%s,标签为%s"%(datetime_now(),id,duplicate_data_N)) sql = "('%s','%s','%s','%s','%s','%s','%s','%s',%s,%s,'%s','%s','%s',%s,'%s','%s','%s',%s,'%s',%s,'%s','%s')"%(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_N) sqls.append(sql) else: print("%s 时间段%s至%s没有查到数据"%(datetime_now(),create_time_start,create_time_end)) if sqls: print("%s 批量插入无重复数据到feedback_history表"%datetime_now()) for i in range(0, len(sqls), 1000): sqlranges = sqls[i:i + 1000] sqlranges = INSERT_DADA_SQL%(my_database) + ",".join(sqlranges) #删除feedback里的数据 sql_1 = DELETE_DUPLICATE_MULT_SQL%(my_database,my_database,create_time_start,create_time_end) MysqlUtil.update_two(sqlranges,sql_1) MysqlUtil.close()