update

task/recommend_strategy_fix.py

@@ -35,9 +35,7 @@ cursor = db.cursor()
 
 def con_sql(sql):
     # 从数据库的表里获取数据
-
-    db = pymysql.connect(host='172.16.40.158', port=4000, user='st_user', passwd='aqpuBLYzEV7tML5RPsN1pntUzFy',
-                         db='jerry_prod')
+    db = pymysql.connect(host='172.16.40.158', port=4000, user='st_user', passwd='aqpuBLYzEV7tML5RPsN1pntUzFy', db='jerry_prod')
     cursor = db.cursor()
     cursor.execute(sql)
     result = cursor.fetchall()
@@ -190,7 +188,7 @@ for t in range(0, task_days):
                       app_session_id
             )a
             LATERAL VIEW explode (a.recommend_type) v as recommend_type
-            group by partition_date,cl_id) t0
+            group by partition_date,cl_id) t0 having  session_pv0 >= 4
             on t1.device_id = t0.cl_id 
  LEFT JOIN
         (--精准曝光，卡片id和session_id去重
@@ -781,7 +779,7 @@ for t in range(0, task_days):
         on t2.partition_date=dev.partition_date and t2.cl_id=dev.device_id
     WHERE spam_pv.device_id IS NULL
       and dev.device_id is null
-    GROUP BY t1.partition_date,t1.device_os_type,t1.active_type,t2.card_content_type,t2.recommend_type  having t0.session_pv0 >= 4 or t3.session_pv >0
+    GROUP BY t1.partition_date,t1.device_os_type,t1.active_type,t2.card_content_type,t2.recommend_type
     order by day_id,device_os_type,active_type,card_content_type,recommend_type
     """.format(partition_day=today_str)
     ctr_df = spark.sql(ctr_sql)
@@ -797,6 +795,16 @@ for t in range(0, task_days):
 
     device_df.show(1, False)
     sql_res = device_df.collect()
+
+    data_dict = {}
+    has_click_pv_sql = 'select pid,card_click from recommend_strategy_d where partition_day = "%s"' % today_str
+    all_data_res = con_sql(has_click_pv_sql)
+    for data in all_data_res:
+        data_dict[data[0]] = data[1]
+    if not data_dict:
+        import sys
+        sys.exit(0)
+    sql_res = {}
     print("-----------------------------------------------------------------------------")
     for res in sql_res:
         # print(res)
@@ -829,6 +837,12 @@ for t in range(0, task_days):
         second_jump_ctr = res.second_jump_ctr
         click_ctr = res.click_ctr
         pid = hashlib.md5((day_id + device_os_type + active_type + card_content_type + recommend_type).encode("utf8")).hexdigest()
+        try:
+            card_click_diff = card_click - data_dict.get(pid)
+            card_click += card_click_diff
+            card_exposure += card_exposure
+        except:
+            print("add error")
         instert_sql = """replace into recommend_strategy_d_fix(
         day_id,device_os_type,active_type,card_content_type,recommend_type,card_click,card_exposure,avg_page_stay,navbar_search,
         highlight_word,self_welfare_card,recommend_welfare_card,recommend_content_card,recommend_special_card,transfer_card,video_consultation,