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