将用户活跃度数据保存到数据库

sql/01_2022-08-09.sql

@@ -96,4 +96,15 @@ CREATE TABLE "public"."ai_repo_topic_detail" (
   "topic_id" int4,
   "topic_name" varchar(255) COLLATE "pg_catalog"."default",
   "created_unix" int8
-);
+);
+
+
+
+CREATE TABLE "public"."ai_user_activity" (
+    "user_id" int8,
+    "activity" int4,
+    "type" varchar(20) COLLATE "pg_catalog"."default",
+    "created_unix" int8
+  );
+
+

user_portrait_analysis/activity_analysis/data_process.py

@@ -87,6 +87,19 @@ def get_rfm_data():
     return df_rfm
 
 
+def save_train_result_to_pgsql(df):
+    start = datetime.datetime.now()
+    logger.info("开始将用户活跃度数据保存到数据库...")
+
+    engine = create_engine(f'postgresql+psycopg2://{user}:{passwd}@{host}:{port}/{database}', pool_recycle=3600)
+    conn = engine.connect()
+
+    df['type'] = 'activity'
+    df['created_unix'] = int(datetime.datetime.now().timestamp())
+    df.to_sql('ai_user_activity', conn, index=False, if_exists='replace')
+    logger.info("用户活跃度数据保存到数据库完毕，共" + str(df.shape[0]) + "条数据，总耗时" + str((datetime.datetime.now() - start).seconds) + "秒")
+
+
 if __name__ == '__main__':
     # 取之前多少天登录的数据
     latest_data_date = get_before_date(data_before_days)

user_portrait_analysis/activity_analysis/train.py

@@ -4,7 +4,7 @@ import pandas as pd
 from sklearn.cluster import KMeans
 from config import logger
 from config import RANDOM_SEED
-from data_process import get_rfm_data
+from data_process import get_rfm_data, save_train_result_to_pgsql
 from config import activity_analysis_path
 warnings.filterwarnings('ignore')
 
@@ -68,14 +68,17 @@ def train():
     # 将用户 id 和活跃度分值提取出来并创新建立一个表     
     result_csv['user_id'] = result_csv['user_id'].astype(int)
     result_csv['activity'] = result_csv['activity'].astype(int)
-    resuts_dict = dict(zip(result_csv['user_id'], result_csv['activity']))
+    result_dict = dict(zip(result_csv['user_id'], result_csv['activity']))
+
+    # 将训练结果保存到数据库
+    save_train_result_to_pgsql(result_csv)
 
     # 保存模型和结果
     pickle.dump(kmeans_model, open(activity_analysis_path + 'results/user_activity_model.pkl', 'wb'))
-    pickle.dump(resuts_dict, open(activity_analysis_path + 'results/user_activity_dict.pkl', 'wb'))
+    pickle.dump(result_dict, open(activity_analysis_path + 'results/user_activity_dict.pkl', 'wb'))
 
     logger.info('用户活跃度分析模型训练完成')
-    return resuts_dict
+    return result_dict
 
 
 if __name__ == '__main__':