start.py 8.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186
  1. from MyUtils.MysqlUtils import MysqlUtils
  2. from MyUtils.DateUtils import get_day
  3. import datetime
  4. import pytz
  5. import json
  6. import time
  7. import os
  8. SELETE_SQL = "SELECT " \
  9. "t.project_id, " \
  10. "t.object_id," \
  11. "t.source_type," \
  12. "t.user_id," \
  13. "t.user_phone," \
  14. "t.user_name," \
  15. "t.value_type," \
  16. "t.item_id," \
  17. "t.create_time," \
  18. "COUNT(*) " \
  19. "FROM %s.feedback t WHERE " \
  20. "t.create_time >= '%s' AND t.create_time < '%s' GROUP BY " \
  21. "t.project_id, " \
  22. "t.object_id," \
  23. "t.source_type," \
  24. "t.user_id," \
  25. "t.user_phone," \
  26. "t.user_name," \
  27. "t.value_type," \
  28. "t.item_id," \
  29. "t.create_time" \
  30. " HAVING COUNT(*) > 1"
  31. SELETE_FEEDBACK_COUNT_SQL = "select count(*) from %s.feedback where create_time >= '%s' and create_time < '%s'"
  32. SELECT_ID_SQL = "SELECT t.id FROM %s.feedback t WHERE " \
  33. "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'"
  34. INSERT_ID_DATA_SQL = "INSERT INTO %s.feedback_history (" \
  35. "id,project_id,object_id,source_type,user_id,user_phone,user_name," \
  36. "value_type,item_id,`value`,create_time,next_open_time,model,duration_type," \
  37. "custom_plan,curr_temp,nick_name, `result`,exe_result,fb_temp,`remark`) " \
  38. "SELECT * FROM %s.feedback WHERE id = '%s'"
  39. UPDATE_ID_SQL = "update %s.feedback_history set duplicate_data = '%s' where id = '%s'"
  40. SELECT_DATA_SQL = "SELECT * FROM %s.feedback t WHERE " \
  41. "t.create_time >= '%s' AND t.create_time < '%s' " \
  42. "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 " \
  43. "HAVING COUNT(*) =1"
  44. INSERT_DADA_SQL = "INSERT INTO %s.feedback_history (" \
  45. "id,project_id,object_id,source_type,user_id,user_phone,user_name," \
  46. "value_type,item_id,`value`,create_time,next_open_time,model,duration_type," \
  47. "custom_plan,curr_temp,nick_name, `result`,exe_result,fb_temp,`remark`,duplicate_data) values "
  48. DELETE_DUPLICATE_SQL = "delete from %s.feedback where id ='%s'"
  49. 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)"
  50. def datetime_now():
  51. # datetime_now = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
  52. #容器时间
  53. # tz = pytz.timezone('Asia/Shanghai') # 东八区
  54. datetime_now = datetime.datetime.fromtimestamp(int(time.time()),
  55. pytz.timezone('Asia/Shanghai')).strftime('%Y-%m-%d %H:%M:%S')
  56. return datetime_now
  57. def None_data(data):
  58. if data == None:
  59. data = "NULL"
  60. return data
  61. with open("config.json", "r") as f:
  62. data = json.load(f)
  63. mysql = data["mysql"]
  64. my_database = mysql["database"]
  65. sleep = data["sleep"]
  66. # mysql = {
  67. # "host": os.getenv("host"),
  68. # "port": int(os.getenv("port")),
  69. # "user": os.getenv("user"),
  70. # "passwd": os.getenv("passwd"),
  71. # "database": os.getenv("database")
  72. # }
  73. # database = os.getenv("database")
  74. # sleep = os.getenv("sleep")
  75. # #连接mysql
  76. MysqlUtil = MysqlUtils(**mysql)
  77. starttime = '20230101000000'
  78. endtime = '20230201000000'
  79. # yesterday = (datetime.datetime.today() - datetime.timedelta(days=1)).strftime("%Y%m%d")+"000000"
  80. # today = datetime.datetime.today().strftime("%Y%m%d")+"000000"
  81. day_range = get_day(starttime,endtime)
  82. for day in day_range:
  83. sqls = []
  84. duplicate_data_sum = []
  85. create_time_start,create_time_end = day[0],day[1]
  86. feedback_sum = MysqlUtil.query(SELETE_FEEDBACK_COUNT_SQL%(my_database,create_time_start,create_time_end))[0][0]
  87. #获取重复数据
  88. data_infos = MysqlUtil.query(SELETE_SQL%(my_database,create_time_start,create_time_end))
  89. if data_infos:
  90. # print("%s 时间段%s至%s,开始插入重复数据到feedback_history表"%(datetime_now(),create_time_start,create_time_end))
  91. for data in data_infos:
  92. project_id = data[0]
  93. object_id = data[1]
  94. source_type = data[2]
  95. user_id = data[3]
  96. user_phone = data[4]
  97. user_name = data[5]
  98. value_type = data[6]
  99. item_id = data[7]
  100. create_time = data[8]
  101. create_time = datetime.datetime.strftime(create_time,"%Y-%m-%d %H:%M:%S")
  102. # 获取重复数据的id
  103. 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))
  104. if data_infos_id:
  105. for data_id in data_infos_id:
  106. # print("%s 开始插入数据%s"%(datetime_now(),data_id[0]))
  107. # #将重复数据插入到history表中
  108. sql1 = INSERT_ID_DATA_SQL%(my_database,my_database,data_id[0])
  109. # #打重复的标签”Y“
  110. sql2 = UPDATE_ID_SQL%(my_database,"Y",data_id[0])
  111. # 删除feedback重复数据
  112. sql3 = DELETE_DUPLICATE_SQL%(my_database,data_id[0])
  113. MysqlUtil.update_mult(sql1,sql2,sql3)
  114. duplicate_data_sum.append(data_id[0])
  115. MysqlUtil.update(UPDATE_ID_SQL%(my_database,"N",data_infos_id[0][0]))
  116. else:
  117. print("%s 时间段%s至%s没有重复数据"%(datetime_now(),create_time_start,create_time_end))
  118. #获取没有重复的数据
  119. data_all_infos = MysqlUtil.query(SELECT_DATA_SQL%(my_database,create_time_start,create_time_end))
  120. duplicate_data_N = 'N'
  121. if data_all_infos:
  122. for d in data_all_infos:
  123. id = d[0]
  124. project_id = d[1]
  125. object_id = d[2]
  126. source_type = d[3]
  127. user_id = d[4]
  128. user_phone = d[5]
  129. user_name = d[6]
  130. value_type = d[7]
  131. item_id = None_data(d[8])
  132. value = None_data(d[9])
  133. create_time = d[10]
  134. next_open_time = d[11]
  135. model = d[12]
  136. duration_type = None_data(d[13])
  137. custom_plan = d[14]
  138. curr_temp = d[15]
  139. nick_name = d[16]
  140. result = None_data(d[17])
  141. exe_result = d[18]
  142. exe_result = None_data(exe_result)
  143. fb_temp = None_data(d[19])
  144. remark = d[20]
  145. # print("%s 查询数据%s,标签为%s"%(datetime_now(),id,duplicate_data_N))
  146. 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)
  147. sqls.append(sql)
  148. else:
  149. print("%s 时间段%s至%s没有查到数据"%(datetime_now(),create_time_start,create_time_end))
  150. if sqls:
  151. # print("%s 时间段%s至%s,开始批量插入无重复数据到feedback_history表"%(datetime_now(),create_time_start,create_time_end))
  152. for i in range(0, len(sqls), 1000):
  153. sqlranges = sqls[i:i + 1000]
  154. sqlranges = INSERT_DADA_SQL%(my_database) + ",".join(sqlranges)
  155. #删除feedback里的数据
  156. sql_1 = DELETE_DUPLICATE_MULT_SQL%(my_database,my_database,create_time_start,create_time_end)
  157. MysqlUtil.update_two(sqlranges,sql_1)
  158. #对比查询记录和插入记录是否一致
  159. feedback_his_sum = len(duplicate_data_sum)+len(sqls)
  160. if feedback_sum == feedback_his_sum:
  161. print("%s 时间段%s至%s,处理数据%s条,成功!"%(datetime_now(),create_time_start,create_time_end,feedback_sum))
  162. else:
  163. print("%s 时间段%s至%s,查询数据%s条,插入数据%s条"%(datetime_now(),create_time_start,create_time_end,feedback_sum,feedback_his_sum))
  164. MysqlUtil.close()