Преглед изворни кода

查询feedback数据到feedback_history 去重之后

李莎 пре 2 година
родитељ
комит
2114ab9080
2 измењених фајлова са 23 додато и 12 уклоњено
  1. 0 3
      MyUtils/MysqlUtils.py
  2. 23 9
      start.py

+ 0 - 3
MyUtils/MysqlUtils.py

@@ -73,9 +73,7 @@ class MysqlUtils():
             con = self.pool.connection()
             cur = con.cursor()
             cur.execute(sql1)
-            con.commit()
             cur.execute(sql2)
-            con.commit()
             cur.execute(sql3)
             con.commit()
         except Exception as e:
@@ -94,7 +92,6 @@ class MysqlUtils():
             con = self.pool.connection()
             cur = con.cursor()
             cur.execute(sql1)
-            con.commit()
             cur.execute(sql2)
             con.commit()
         except Exception as e:

+ 23 - 9
start.py

@@ -1,10 +1,8 @@
 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 json
 import time
 import os
 
@@ -32,6 +30,9 @@ SELETE_SQL = "SELECT " \
              "t.create_time" \
              " HAVING COUNT(*) > 1"
 
+SELETE_FEEDBACK_COUNT_SQL = "select count(*) from %s.feedback where create_time >= '%s' and create_time < '%s'"
+
+
 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 (" \
@@ -89,17 +90,22 @@ with open("config.json", "r") as f:
 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))
-    duplicate_data = 'Y'
     if data_infos:
+        # print("%s 时间段%s至%s,开始插入重复数据到feedback_history表"%(datetime_now(),create_time_start,create_time_end))
         for data in data_infos:
             project_id = data[0]
             object_id = data[1]
@@ -111,18 +117,20 @@ for day in day_range:
             item_id = data[7]
             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))
-            #获取重复数据的id
             if data_infos_id:
                 for data_id in data_infos_id:
-                    print("%s 插入数据%s,标签为%s"%(datetime_now(),data_id[0],duplicate_data))
+                    # 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,duplicate_data,data_id[0])
+                    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))
     #获取没有重复的数据
@@ -152,19 +160,25 @@ for day in day_range:
             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))
+            # 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())
+        # 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()