start.py 7.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172
  1. from MyUtils.MysqlUtils import MysqlUtils
  2. from MyUtils.DateUtils import get_day
  3. from MyUtils.Dingtalk import send_message_mobiles
  4. import datetime
  5. import pytz
  6. import json
  7. import time
  8. import os
  9. SELETE_SQL = "SELECT " \
  10. "t.project_id, " \
  11. "t.object_id," \
  12. "t.source_type," \
  13. "t.user_id," \
  14. "t.user_phone," \
  15. "t.user_name," \
  16. "t.value_type," \
  17. "t.item_id," \
  18. "t.create_time," \
  19. "COUNT(*) " \
  20. "FROM %s.feedback t WHERE " \
  21. "t.create_time >= '%s' AND t.create_time < '%s' GROUP BY " \
  22. "t.project_id, " \
  23. "t.object_id," \
  24. "t.source_type," \
  25. "t.user_id," \
  26. "t.user_phone," \
  27. "t.user_name," \
  28. "t.value_type," \
  29. "t.item_id," \
  30. "t.create_time" \
  31. " HAVING COUNT(*) > 1"
  32. SELETE_FEEDBACK_COUNT_SQL = "select count(*) from %s.feedback where create_time >= '%s' and create_time < '%s'"
  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 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 " \
  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_type,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 is None:
  60. data = "is NULL"
  61. else:
  62. data = "= '%s'" % data
  63. return data
  64. def data_sql(SQL,tag):
  65. data_infos = MysqlUtil.query(SQL % (my_database, create_time_start, create_time_end))
  66. if data_infos:
  67. print("%s 时间段%s至%s,tag为%s,开始插入数据到feedback_history表" % (datetime_now(), create_time_start, create_time_end,tag))
  68. for data in data_infos:
  69. project_id = data[0]
  70. object_id = None_data(data[1])
  71. source_type = None_data(data[2])
  72. user_id = None_data(data[3])
  73. user_phone = None_data(data[4])
  74. user_name = None_data(data[5])
  75. value_type = data[6]
  76. item_id = None_data(data[7])
  77. create_time = data[8]
  78. create_time = datetime.datetime.strftime(create_time, "%Y-%m-%d %H:%M:%S")
  79. # 获取数据的id
  80. data_infos_id = MysqlUtil.query(SELECT_ID_SQL % (
  81. my_database, project_id, object_id, source_type, user_id, user_phone, user_name, value_type, item_id,
  82. create_time))
  83. if data_infos_id:
  84. result = None
  85. for data_id in data_infos_id:
  86. # print("%s 开始插入数据%s" % (datetime_now(), data_id[0]))
  87. # #将数据插入到history表中
  88. sql1 = INSERT_ID_DATA_SQL % (my_database, my_database, data_id[0])
  89. # #打标签”Y,N“
  90. sql2 = UPDATE_ID_SQL % (my_database, tag, data_id[0])
  91. # 删除feedback数据
  92. sql3 = DELETE_DUPLICATE_SQL % (my_database, data_id[0])
  93. result = MysqlUtil.update_mult(sql1, sql2, sql3)
  94. if result == True:
  95. duplicate_data_sum.append(data_id[0])
  96. if tag == "Y" and result == True:
  97. MysqlUtil.update(UPDATE_ID_SQL % (my_database, "N", data_infos_id[0][0]))
  98. else:
  99. print("%s tag为%s,时间段%s至%s没有数据" % (datetime_now(),tag, create_time_start, create_time_end))
  100. # with open("config.json", "r") as f:
  101. # data = json.load(f)
  102. # mysql = data["mysql"]
  103. # my_database = mysql["database"]
  104. # dingding = data["dingding"]
  105. mysql = {
  106. "host": os.getenv("host"),
  107. "port": int(os.getenv("port")),
  108. "user": os.getenv("user"),
  109. "passwd": os.getenv("passwd"),
  110. "database": os.getenv("database")
  111. }
  112. my_database = os.getenv("database")
  113. dingding = os.getenv("dingding")
  114. at_mobiles = os.getenv("at_mobiles")
  115. while True:
  116. time_now = datetime.datetime.fromtimestamp(int(time.time()),
  117. pytz.timezone('Asia/Shanghai')).strftime('%H:%M:%S')
  118. # time_now = time.strftime("%H:%M:%S", time.localtime()) # 刷新
  119. am_set_time = "06:00:00"
  120. if time_now == am_set_time: # 此处设置每天定时的时间
  121. # #连接mysql
  122. MysqlUtil = MysqlUtils(**mysql)
  123. # starttime = '20230301000000'
  124. # endtime = '20230319000000'
  125. yesterday = (datetime.datetime.today() - datetime.timedelta(days=1)).strftime("%Y%m%d")+"000000"
  126. today = datetime.datetime.today().strftime("%Y%m%d")+"000000"
  127. day_range = get_day(yesterday,today)
  128. for day in day_range:
  129. sqls = []
  130. duplicate_data_sum = []
  131. create_time_start,create_time_end = day[0],day[1]
  132. feedback_sum = MysqlUtil.query(SELETE_FEEDBACK_COUNT_SQL%(my_database,create_time_start,create_time_end))[0][0]
  133. #获取重复数据
  134. data_sql(SELETE_SQL,"Y")
  135. #获取不重复数据
  136. data_sql(SELECT_DATA_SQL,"N")
  137. #对比查询记录和插入记录是否一致
  138. if feedback_sum == len(duplicate_data_sum):
  139. print("%s 时间段%s至%s,处理数据%s条,成功!"%(datetime_now(),create_time_start,create_time_end,feedback_sum))
  140. # message = ("【feedback表打唯一数据标签】 %s 时间段%s至%s,处理数据%s条,成功 \n") % (datetime_now(),create_time_start,create_time_end,feedback_sum)
  141. # send_message_mobiles(message,dingding,at_mobiles)
  142. else:
  143. print("%s 时间段%s至%s,查询数据%s条,插入数据%s条,失败!"%(datetime_now(),create_time_start,create_time_end,feedback_sum,len(duplicate_data_sum)))
  144. message = ("【feedback表打唯一数据标签】 %s 时间段%s至%s,查询数据%s条,插入数据%s条,失败 \n")%(datetime_now(),create_time_start,create_time_end,feedback_sum,len(duplicate_data_sum))
  145. send_message_mobiles(message,dingding,at_mobiles)
  146. MysqlUtil.close()