用户分层营销使用yinghe版画像

eda/smart_rank/dist_update_user_portrait_service.py

@@ -261,6 +261,11 @@ def get_user_service_portrait(cl_id, all_word_tags, all_tag_tag_type, all_3tag_2
                         compute_validate(x.days_diff_now)/get_action_tag_count(user_df_service, x.time)))), axis=1)
             tag_score_sum = user_df_service.groupby(by=["tag2", "tag2_type"]).agg({'tag_score': sum}).reset_index().sort_values(by=["tag_score"],ascending=False)
             tag_score_sum['weight'] = 100 * tag_score_sum['tag_score'] / tag_score_sum['tag_score'].sum()
+            tag_score_sum["pay_type"] = tag_score_sum.apply(
+                lambda x: 3 if x.action == "api/order/validate" else (
+                    2 if x.action == "api/settlement/alipay_callback" else 1
+                ), axis=1
+            )
             gmkv_tag_score_sum = tag_score_sum[["tag2", "tag_score", "weight"]][:size].to_dict('record')
             # 写gmkv
             gm_kv_cli = redis.Redis(host="172.16.40.135", port=5379, db=2, socket_timeout=2000)
@@ -268,14 +273,26 @@ def get_user_service_portrait(cl_id, all_word_tags, all_tag_tag_type, all_3tag_2
             tag_id_list_json = json.dumps(gmkv_tag_score_sum)
             gm_kv_cli.set(cl_id_portrait_key, tag_id_list_json)
             gm_kv_cli.expire(cl_id_portrait_key, time=30 * 24 * 60 * 60)
-            # 写tidb
+            # 写tidb，redis同步
             stat_date = datetime.datetime.today().strftime('%Y-%m-%d')
             replace_sql = """replace into user_service_portrait_tags (stat_date, cl_id, tag_list) values("{stat_date}","{cl_id}","{tag_list}")"""\
                 .format(stat_date=stat_date, cl_id=cl_id, tag_list=gmkv_tag_score_sum)
             cur_jerry_test.execute(replace_sql)
             db_jerry_test.commit()
+            # 写tidb 用户分层营销
+            score_result = tag_score_sum[["tag2", "cl_id", "tag_score", "weight", "pay_type"]]
+            score_result.rename(columns={"cl_id": "device_id"}, inplace=True)
+            delete_sql = "delete from api_market_personas where device_id='{}'".format(cl_id)
+            cur_jerry_test.execute(delete_sql)
+            db_jerry_test.commit()
+
+            for index, row in score_result.iterrows():
+                insert_sql = "insert into api_market_personas values (null, {}, '{}', {}, {}, {})".format(
+                    row['tag_id'], row['device_id'], row['score'], row['weight'], row['pay_type'])
+                cur_jerry_test.execute(insert_sql)
+                db_jerry_test.commit()
             db_jerry_test.close()
-            return gmkv_tag_score_sum
+            return "sucess"
     except Exception as e:
         print(e)