offline tasks: fix bugs of i2i swing / hot / sessionw2v

tangwang
1 parent 9832fef6
Showing 5 changed files with 43 additions and 21 deletions Show diff stats
offline_tasks/run.sh
offline_tasks/scripts/i2i_content_similar.py
offline_tasks/scripts/i2i_deepwalk.py
offline_tasks/scripts/i2i_session_w2v.py
offline_tasks/scripts/interest_aggregation.py
@@ -6,6 +6,12 @@ cd /home/tw/recommendation/offline_tasks
 # 2. 测试连接
 # python3 test_connection.py
  
+ps -ef|grep run_all.py | awk '{print $2}' | xargs kill -9
+ps -ef|grep recommendation | awk '{print $2}' | xargs kill -9
+rm output/* -rf
+rm logs/* -rf
+
+
 # 3. 调试模式运行（小数据量）
 python3 run_all.py --lookback_days 7 --top_n 10 --debug
  
@@ -107,24 +107,37 @@ def build_feature_text(row):
     return ' '.join(features)
  
  
-def calculate_content_similarity(df, top_n=50):
+def calculate_content_similarity(df, top_n=50, logger=None):
     """
-    基于内容计算相似度
+    基于内容计算相似度（内存优化版）
     """
-    print("Building feature texts...")
+    
+    if logger:
+        logger.info("构建特征文本...")
+    else:
+        print("Building feature texts...")
     df['feature_text'] = df.apply(build_feature_text, axis=1)
  
-    print("Calculating TF-IDF...")
+    if logger:
+        logger.info("计算 TF-IDF...")
+    else:
+        print("Calculating TF-IDF...")
     vectorizer = TfidfVectorizer(max_features=1000)
     tfidf_matrix = vectorizer.fit_transform(df['feature_text'])
  
-    print("Calculating cosine similarity...")
-    # 分批计算相似度以节省内存
+    if logger:
+        logger.info(f"TF-IDF 矩阵形状: {tfidf_matrix.shape}")
+        logger.info("开始计算余弦相似度（内存优化模式）...")
+    else:
+        print("Calculating cosine similarity...")
+    
     batch_size = 1000
     result = {}
  
     for i in range(0, len(df), batch_size):
         end_i = min(i + batch_size, len(df))
+        
+        # 分批计算相似度
         batch_similarity = cosine_similarity(tfidf_matrix[i:end_i], tfidf_matrix)
  
         for j, idx in enumerate(range(i, end_i)):
@@ -144,8 +157,11 @@ def calculate_content_similarity(df, top_n=50):
  
             if similar_items:
                 result[item_id] = similar_items
-        
-        print(f"Processed {end_i}/{len(df)} products...")
+                
+        if logger:
+            logger.info(f"已处理 {end_i}/{len(df)} 个商品...")
+        else:
+            print(f"Processed {end_i}/{len(df)} products...")
  
     return result
  
@@ -257,14 +273,16 @@ def main():
     log_processing_step(logger, f"计算相似度 (方法: {args.method})")
     if args.method == 'tfidf':
         logger.info("使用 TF-IDF 方法...")
-        result = calculate_content_similarity(df, args.top_n)
+        result = calculate_content_similarity(df, args.top_n, logger=logger)
     elif args.method == 'category':
         logger.info("使用基于分类的方法...")
         result = calculate_category_based_similarity(df)
     else:  # hybrid
         logger.info("使用混合方法 (TF-IDF 70% + 分类 30%)...")
-        tfidf_sim = calculate_content_similarity(df, args.top_n)
+        tfidf_sim = calculate_content_similarity(df, args.top_n, logger=logger)
+        logger.info("计算基于分类的相似度...")
         category_sim = calculate_category_based_similarity(df)
+        logger.info("合并相似度...")
         result = merge_similarities(tfidf_sim, category_sim, weight1=0.7, weight2=0.3)
  
     logger.info(f"为 {len(result)} 个物品生成了相似度")
@@ -307,8 +325,7 @@ def main():
             output_file,
             result,
             name_mappings,
-            index_type=f'i2i:content:{args.method}',
-            logger=logger
+            description=f'i2i:content:{args.method}'
         )
  
  
@@ -372,8 +372,7 @@ def main():
             output_file,
             result,
             name_mappings,
-            index_type='i2i:deepwalk',
-            logger=logger
+            description='i2i:deepwalk'
         )
  
  
@@ -185,7 +185,8 @@ def main():
         'epochs': args.epochs,
         'top_n': args.top_n,
         'lookback_days': args.lookback_days,
-        'session_gap_minutes': args.session_gap,
+        'max_session_length': args.max_session_length,
+        'min_session_length': args.min_session_length,
         'debug': args.debug
     }
     log_algorithm_params(logger, params)
@@ -303,8 +304,7 @@ def main():
             output_file,
             result,
             name_mappings,
-            index_type='i2i:session_w2v',
-            logger=logger
+            description='i2i:session_w2v'
         )
  
  
@@ -356,16 +356,16 @@ def main():
         if args.debug and aggregations:
             for dim_key, items in aggregations.items():
                 if items:
-                    # 为每个维度生成可读索引
-                    result_dict = {dim_key: items[:args.top_n]}
+                    # 为每个维度生成可读索引 - 先排序再取前N个
+                    sorted_items = sorted(items.items(), key=lambda x: -x[1])[:args.top_n]
+                    result_dict = {dim_key: sorted_items}
                     output_file = os.path.join(OUTPUT_DIR, f'{output_prefix}_{dim_key}_{datetime.now().strftime("%Y%m%d")}.txt')
                     if os.path.exists(output_file):
                         save_readable_index(
                             output_file,
                             result_dict,
                             name_mappings,
-                            index_type=f'interest:{list_type}:{dim_key}',
-                            logger=logger
+                            description=f'interest:{list_type}:{dim_key}'
                         )
  
     # 生成全局索引（所有数据）