start.py 7.2 KB

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