|
@@ -1,5 +1,6 @@
|
|
from MyUtils.MysqlUtils import MysqlUtils
|
|
from MyUtils.MysqlUtils import MysqlUtils
|
|
from MyUtils.DateUtils import get_day
|
|
from MyUtils.DateUtils import get_day
|
|
|
|
+from MyUtils.Dingtalk import send_message_mobiles
|
|
import datetime
|
|
import datetime
|
|
import pytz
|
|
import pytz
|
|
import json
|
|
import json
|
|
@@ -34,7 +35,7 @@ SELETE_FEEDBACK_COUNT_SQL = "select count(*) from %s.feedback where create_time
|
|
|
|
|
|
|
|
|
|
SELECT_ID_SQL = "SELECT t.id FROM %s.feedback t WHERE " \
|
|
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'"
|
|
|
|
|
|
+ "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 (" \
|
|
INSERT_ID_DATA_SQL = "INSERT INTO %s.feedback_history (" \
|
|
"id,project_id,object_id,source_type,user_id,user_phone,user_name," \
|
|
"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," \
|
|
"value_type,item_id,`value`,create_time,next_open_time,model,duration_type," \
|
|
@@ -43,9 +44,9 @@ INSERT_ID_DATA_SQL = "INSERT INTO %s.feedback_history (" \
|
|
|
|
|
|
UPDATE_ID_SQL = "update %s.feedback_history set duplicate_data = '%s' 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 " \
|
|
|
|
|
|
+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' " \
|
|
"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 " \
|
|
|
|
|
|
+ "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"
|
|
"HAVING COUNT(*) =1"
|
|
|
|
|
|
INSERT_DADA_SQL = "INSERT INTO %s.feedback_history (" \
|
|
INSERT_DADA_SQL = "INSERT INTO %s.feedback_history (" \
|
|
@@ -65,122 +66,107 @@ def datetime_now():
|
|
return datetime_now
|
|
return datetime_now
|
|
|
|
|
|
def None_data(data):
|
|
def None_data(data):
|
|
- if data == None:
|
|
|
|
- data = "NULL"
|
|
|
|
|
|
+ if data is None:
|
|
|
|
+ data = "is NULL"
|
|
|
|
+ else:
|
|
|
|
+ data = "= '%s'" % data
|
|
return data
|
|
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'
|
|
|
|
-
|
|
|
|
-# yesterday = (datetime.datetime.today() - datetime.timedelta(days=1)).strftime("%Y%m%d")+"000000"
|
|
|
|
-# today = datetime.datetime.today().strftime("%Y%m%d")+"000000"
|
|
|
|
-day_range = get_day(starttime,endtime)
|
|
|
|
-
|
|
|
|
-
|
|
|
|
-
|
|
|
|
-for day in day_range:
|
|
|
|
- sqls = []
|
|
|
|
- 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_infos = MysqlUtil.query(SELETE_SQL%(my_database,create_time_start,create_time_end))
|
|
|
|
|
|
+
|
|
|
|
+def data_sql(SQL,tag):
|
|
|
|
+ data_infos = MysqlUtil.query(SQL % (my_database, create_time_start, create_time_end))
|
|
if data_infos:
|
|
if data_infos:
|
|
- # print("%s 时间段%s至%s,开始插入重复数据到feedback_history表"%(datetime_now(),create_time_start,create_time_end))
|
|
|
|
|
|
+ print("%s 时间段%s至%s,tag为%s,开始插入数据到feedback_history表" % (datetime_now(), create_time_start, create_time_end,tag))
|
|
for data in data_infos:
|
|
for data in data_infos:
|
|
project_id = data[0]
|
|
project_id = data[0]
|
|
- object_id = data[1]
|
|
|
|
- source_type = data[2]
|
|
|
|
- user_id = data[3]
|
|
|
|
- user_phone = data[4]
|
|
|
|
- user_name = data[5]
|
|
|
|
|
|
+ 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]
|
|
value_type = data[6]
|
|
- item_id = data[7]
|
|
|
|
|
|
+ item_id = None_data(data[7])
|
|
create_time = data[8]
|
|
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))
|
|
|
|
|
|
+ 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:
|
|
if data_infos_id:
|
|
|
|
+ result = None
|
|
for data_id in data_infos_id:
|
|
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“
|
|
|
|
- sql2 = UPDATE_ID_SQL%(my_database,"Y",data_id[0])
|
|
|
|
- # 删除feedback重复数据
|
|
|
|
- sql3 = DELETE_DUPLICATE_SQL%(my_database,data_id[0])
|
|
|
|
- MysqlUtil.update_mult(sql1,sql2,sql3)
|
|
|
|
- duplicate_data_sum.append(data_id[0])
|
|
|
|
- MysqlUtil.update(UPDATE_ID_SQL%(my_database,"N",data_infos_id[0][0]))
|
|
|
|
- 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)
|
|
|
|
|
|
+ # 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, sql3)
|
|
|
|
+ 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:
|
|
else:
|
|
- print("%s 时间段%s至%s没有查到数据"%(datetime_now(),create_time_start,create_time_end))
|
|
|
|
- if sqls:
|
|
|
|
- # print("%s 时间段%s至%s,开始批量插入无重复数据到feedback_history表"%(datetime_now(),create_time_start,create_time_end))
|
|
|
|
- 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)
|
|
|
|
- #对比查询记录和插入记录是否一致
|
|
|
|
- feedback_his_sum = len(duplicate_data_sum)+len(sqls)
|
|
|
|
- if feedback_sum == feedback_his_sum:
|
|
|
|
- print("%s 时间段%s至%s,处理数据%s条,成功!"%(datetime_now(),create_time_start,create_time_end,feedback_sum))
|
|
|
|
- else:
|
|
|
|
- print("%s 时间段%s至%s,查询数据%s条,插入数据%s条"%(datetime_now(),create_time_start,create_time_end,feedback_sum,feedback_his_sum))
|
|
|
|
-
|
|
|
|
-MysqlUtil.close()
|
|
|
|
-
|
|
|
|
|
|
+ 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"]
|
|
|
|
+
|
|
|
|
+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")
|
|
|
|
+
|
|
|
|
+
|
|
|
|
+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"
|
|
|
|
+
|
|
|
|
+ day_range = get_day(yesterday,today)
|
|
|
|
+
|
|
|
|
+ for day in day_range:
|
|
|
|
+ sqls = []
|
|
|
|
+ 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,15727360513)
|
|
|
|
+
|
|
|
|
+ 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,15727360513)
|
|
|
|
+ MysqlUtil.close()
|
|
|
|
|
|
|
|
|