update

task/spark_test.py

@@ -29,8 +29,10 @@ from meta_base_code.utils.func_from_redis_get_portrait import *
 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')
+    db = pymysql.connect(host='172.16.30.136', port=3306, user='doris', passwd='o5gbA27hXHHm',
+                         db='doris_prod')
     cursor = db.cursor()
     cursor.execute(sql)
     result = cursor.fetchall()
@@ -76,6 +78,8 @@ spark.sql("CREATE TEMPORARY FUNCTION arrayMerge AS 'com.gmei.hive.common.udf.UDF
 task_list = []
 task_days = 3
 
+
+
 for t in range(2, task_days):
     day_num = 0 - t
     now = (datetime.datetime.now() + datetime.timedelta(days=day_num))
@@ -182,15 +186,20 @@ WHERE spam_pv.device_id IS NULL
     for count_user_count, res in enumerate(sql_res):
         # print(count, res)
         portratit_res = get_user_portrait_tag3_from_redis(res.device_id)
-
+        sql = """select cl_id, projects from kafka_tag3_log
+  where cl_id = '%s' and event_cn = 'kyc' """
         # print(count_user_count, res, portratit_res)
+        sql_res_list = con_sql(sql)
+        kyc_str_list= []
+        if sql_res:
+            print(sql_res,type(sql_res))
+            kyc_str_list = sql_res_list[0][1].split(",")
+
         temp_count = 0
         for demand in portratit_res:
             if portratit_res[demand]:
                 try:
                     for tag in portratit_res[demand][0:3]:
-                        if tag == "吸脂瘦下颌缘" and demand == "projects":
-                            print(res.device_id)
                         if tag in portrait_dict[demand]:
                            portrait_dict[demand][tag] += 1
                         else:
@@ -203,33 +212,35 @@ WHERE spam_pv.device_id IS NULL
         if not temp_count:
             count_not_has_portratit += 1
             no_portrait_device_id_list.append(res.device_id)
+        for key in kyc_str_list:
+            portrait_dict["projects"][key] -= 1
 
     print(portrait_dict)
     print(count_user_count+1,count_not_has_portratit)
     print("-------------------------------")
 
 
-    # for protratit_type in portrait_dict["projects"]:
-    #     partition_date = today_str
-    #     pid = hashlib.md5((partition_date + protratit_type).encode("utf8")).hexdigest()
-    #     action_count = portrait_dict["projects"][protratit_type]
-    #
-    #     instert_sql = """replace into new_user_project_count(
-    #                 partition_day,pid,protratit_count,protratit_type) VALUES('{partition_day}','{pid}',{protratit_count},'{protratit_type}');""".format(
-    #         partition_day=today_str, pid=pid, protratit_count=action_count
-    #         , protratit_type=protratit_type
-    #     )
-    #     print(instert_sql)
-    #     # cursor.execute("set names 'UTF8'")
-    #     db = pymysql.connect(host='172.16.40.158', port=4000, user='st_user', passwd='aqpuBLYzEV7tML5RPsN1pntUzFy',
-    #                          db='jerry_prod')
-    #     cursor = db.cursor()
-    #     res = cursor.execute(instert_sql)
-    #     db.commit()
-    #     print(res)
-
-    # cursor.executemany()
-    # db.close()
+    for protratit_type in portrait_dict["projects"]:
+        partition_date = today_str
+        pid = hashlib.md5((partition_date + protratit_type).encode("utf8")).hexdigest()
+        action_count = portrait_dict["projects"][protratit_type]
+
+        instert_sql = """replace into new_user_project_count(
+                    partition_day,pid,protratit_count,protratit_type) VALUES('{partition_day}','{pid}',{protratit_count},'{protratit_type}');""".format(
+            partition_day=today_str, pid=pid, protratit_count=action_count
+            , protratit_type=protratit_type
+        )
+        print(instert_sql)
+        # cursor.execute("set names 'UTF8'")
+        db = pymysql.connect(host='172.16.40.158', port=4000, user='st_user', passwd='aqpuBLYzEV7tML5RPsN1pntUzFy',
+                             db='jerry_prod')
+        cursor = db.cursor()
+        res = cursor.execute(instert_sql)
+        db.commit()
+        print(res)
+
+    cursor.executemany()
+    db.close()