offline tasks

tangwang
1 parent 8cc6477b
Showing 5 changed files with 278 additions and 59 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
@@ -4,7 +4,7 @@ cd /home/tw/recommendation/offline_tasks
 # cat UPDATE_CONFIG_GUIDE.md
 # 2. 测试连接
-python3 test_connection.py
+# python3 test_connection.py
 # 3. 调试模式运行（小数据量）
 python3 run_all.py --lookback_days 7 --top_n 10 --debug
@@ -13,7 +13,9 @@ mv output output_debug
 mkdir output
 # # 4. 生产模式运行（大数据量）
-python3 run_all.py --lookback_days 730 --top_n 50
+python3 run_all.py --lookback_days 730 --top_n 50 --debug
 # 5. 加载到Redis
 python3 scripts/load_index_to_redis.py --redis-host localhost
+
+
@@ -17,6 +17,11 @@ from db_service import create_db_connection
 from offline_tasks.config.offline_config import (
     DB_CONFIG, OUTPUT_DIR, DEFAULT_I2I_TOP_N
 )
+from offline_tasks.scripts.debug_utils import (
+    setup_debug_logger, log_dataframe_info, log_dict_stats,
+    save_readable_index, fetch_name_mappings, log_algorithm_params,
+    log_processing_step
+)
 def fetch_product_features(engine):
@@ -221,8 +226,19 @@ def main():
     args = parser.parse_args()
+    # 设置logger
+    logger = setup_debug_logger('i2i_content_similar', debug=args.debug)
+    
+    # 记录算法参数
+    params = {
+        'top_n': args.top_n,
+        'method': args.method,
+        'debug': args.debug
+    }
+    log_algorithm_params(logger, params)
+    
     # 创建数据库连接
-    print("Connecting to database...")
+    logger.info("连接数据库...")
     engine = create_db_connection(
         DB_CONFIG['host'],
         DB_CONFIG['port'],
@@ -232,34 +248,47 @@ def main():
     )
     # 获取商品特征
+    log_processing_step(logger, "获取商品特征")
     df = fetch_product_features(engine)
+    logger.info(f"获取到 {len(df)} 个商品的特征数据")
+    log_dataframe_info(logger, df, "商品特征数据")
     # 计算相似度
+    log_processing_step(logger, f"计算相似度 (方法: {args.method})")
     if args.method == 'tfidf':
-        print("\nUsing TF-IDF method...")
+        logger.info("使用 TF-IDF 方法...")
         result = calculate_content_similarity(df, args.top_n)
     elif args.method == 'category':
-        print("\nUsing category-based method...")
+        logger.info("使用基于分类的方法...")
         result = calculate_category_based_similarity(df)
     else:  # hybrid
-        print("\nUsing hybrid method...")
+        logger.info("使用混合方法 (TF-IDF 70% + 分类 30%)...")
         tfidf_sim = calculate_content_similarity(df, args.top_n)
         category_sim = calculate_category_based_similarity(df)
         result = merge_similarities(tfidf_sim, category_sim, weight1=0.7, weight2=0.3)
-    # 创建item_id到name的映射
-    item_name_map = dict(zip(df['item_id'], df['item_name']))
+    logger.info(f"为 {len(result)} 个物品生成了相似度")
     # 输出结果
+    log_processing_step(logger, "保存结果")
     output_file = args.output or os.path.join(
         OUTPUT_DIR, 
         f'i2i_content_{args.method}_{datetime.now().strftime("%Y%m%d")}.txt'
     )
-    print(f"\nWriting results to {output_file}...")
+    # 获取name mappings
+    name_mappings = {}
+    if args.debug:
+        logger.info("获取物品名称映射...")
+        name_mappings = fetch_name_mappings(engine, debug=True)
+    
+    logger.info(f"写入结果到 {output_file}...")
     with open(output_file, 'w', encoding='utf-8') as f:
         for item_id, sims in result.items():
-            item_name = item_name_map.get(item_id, 'Unknown')
+            # 使用name_mappings获取名称
+            item_name = name_mappings.get(item_id, 'Unknown')
+            if item_name == 'Unknown' and 'item_name' in df.columns:
+                item_name = df[df['item_id'] == item_id]['item_name'].iloc[0] if len(df[df['item_id'] == item_id]) > 0 else 'Unknown'
             if not sims:
                 continue
@@ -268,8 +297,19 @@ def main():
             sim_str = ','.join([f'{sim_id}:{score:.4f}' for sim_id, score in sims])
             f.write(f'{item_id}\t{item_name}\t{sim_str}\n')
-    print(f"Done! Generated content-based similarities for {len(result)} items")
-    print(f"Output saved to: {output_file}")
+    logger.info(f"完成！为 {len(result)} 个物品生成了基于内容的相似度")
+    logger.info(f"输出保存到：{output_file}")
+    
+    # 如果启用debug模式，保存可读格式
+    if args.debug:
+        log_processing_step(logger, "保存Debug可读格式")
+        save_readable_index(
+            output_file,
+            result,
+            name_mappings,
+            index_type=f'i2i:content:{args.method}',
+            logger=logger
+        )
 if __name__ == '__main__':
@@ -17,6 +17,11 @@ from offline_tasks.config.offline_config import (
     DB_CONFIG, OUTPUT_DIR, I2I_CONFIG, get_time_range,
     DEFAULT_LOOKBACK_DAYS, DEFAULT_I2I_TOP_N
 )
+from offline_tasks.scripts.debug_utils import (
+    setup_debug_logger, log_dataframe_info, log_dict_stats,
+    save_readable_index, fetch_name_mappings, log_algorithm_params,
+    log_processing_step
+)
 def build_item_graph(df, behavior_weights):
@@ -223,8 +228,26 @@ def main():
     args = parser.parse_args()
+    # 设置logger
+    logger = setup_debug_logger('i2i_deepwalk', debug=args.debug)
+    
+    # 记录算法参数
+    params = {
+        'num_walks': args.num_walks,
+        'walk_length': args.walk_length,
+        'window_size': args.window_size,
+        'vector_size': args.vector_size,
+        'min_count': args.min_count,
+        'workers': args.workers,
+        'epochs': args.epochs,
+        'top_n': args.top_n,
+        'lookback_days': args.lookback_days,
+        'debug': args.debug
+    }
+    log_algorithm_params(logger, params)
+    
     # 创建数据库连接
-    print("Connecting to database...")
+    logger.info("连接数据库...")
     engine = create_db_connection(
         DB_CONFIG['host'],
         DB_CONFIG['port'],
@@ -235,7 +258,7 @@ def main():
     # 获取时间范围
     start_date, end_date = get_time_range(args.lookback_days)
-    print(f"Fetching data from {start_date} to {end_date}...")
+    logger.info(f"获取数据范围：{start_date} 到 {end_date}")
     # SQL查询 - 获取用户行为数据
     sql_query = f"""
@@ -255,9 +278,12 @@ def main():
         AND se.anonymous_id IS NOT NULL
     """
-    print("Executing SQL query...")
+    logger.info("执行SQL查询...")
     df = pd.read_sql(sql_query, engine)
-    print(f"Fetched {len(df)} records")
+    logger.info(f"获取到 {len(df)} 条记录")
+    
+    # 记录数据信息
+    log_dataframe_info(logger, df, "用户行为数据")
     # 定义行为权重
     behavior_weights = {
@@ -267,23 +293,26 @@ def main():
         'addToCart': 3.0,
         'purchase': 10.0
     }
+    logger.debug(f"行为权重: {behavior_weights}")
     # 构建物品图
-    print("Building item graph...")
+    log_processing_step(logger, "构建物品图")
     graph = build_item_graph(df, behavior_weights)
-    print(f"Graph built with {len(graph)} nodes")
+    logger.info(f"构建物品图完成，共 {len(graph)} 个节点")
     # 保存边文件（可选）
     if args.save_graph:
         edge_file = os.path.join(OUTPUT_DIR, f'item_graph_{datetime.now().strftime("%Y%m%d")}.txt')
         save_edge_file(graph, edge_file)
+        logger.info(f"图边文件已保存到 {edge_file}")
     # 生成随机游走
-    print("Generating random walks...")
+    log_processing_step(logger, "生成随机游走")
     walks = generate_walks(graph, args.num_walks, args.walk_length)
-    print(f"Generated {len(walks)} walks")
+    logger.info(f"生成 {len(walks)} 条游走路径")
     # 训练Word2Vec模型
+    log_processing_step(logger, "训练Word2Vec模型")
     w2v_config = {
         'vector_size': args.vector_size,
         'window_size': args.window_size,
@@ -292,29 +321,39 @@ def main():
         'epochs': args.epochs,
         'sg': 1
     }
+    logger.debug(f"Word2Vec配置: {w2v_config}")
     model = train_word2vec(walks, w2v_config)
+    logger.info(f"训练完成。词汇表大小：{len(model.wv)}")
     # 保存模型（可选）
     if args.save_model:
         model_path = os.path.join(OUTPUT_DIR, f'deepwalk_model_{datetime.now().strftime("%Y%m%d")}.model')
         model.save(model_path)
-        print(f"Model saved to {model_path}")
+        logger.info(f"模型已保存到 {model_path}")
     # 生成相似度
-    print("Generating similarities...")
+    log_processing_step(logger, "生成相似度")
     result = generate_similarities(model, top_n=args.top_n)
-    
-    # 创建item_id到name的映射
-    item_name_map = dict(zip(df['item_id'].astype(str), df.groupby('item_id')['item_name'].first()))
+    logger.info(f"生成了 {len(result)} 个物品的相似度")
     # 输出结果
+    log_processing_step(logger, "保存结果")
     output_file = args.output or os.path.join(OUTPUT_DIR, f'i2i_deepwalk_{datetime.now().strftime("%Y%m%d")}.txt')
-    print(f"Writing results to {output_file}...")
+    # 获取name mappings
+    name_mappings = {}
+    if args.debug:
+        logger.info("获取物品名称映射...")
+        name_mappings = fetch_name_mappings(engine, debug=True)
+    
+    logger.info(f"写入结果到 {output_file}...")
     with open(output_file, 'w', encoding='utf-8') as f:
         for item_id, sims in result.items():
-            item_name = item_name_map.get(item_id, 'Unknown')
+            # 使用name_mappings获取名称
+            item_name = name_mappings.get(int(item_id), 'Unknown') if item_id.isdigit() else 'Unknown'
+            if item_name == 'Unknown' and 'item_name' in df.columns:
+                item_name = df[df['item_id'].astype(str) == item_id]['item_name'].iloc[0] if len(df[df['item_id'].astype(str) == item_id]) > 0 else 'Unknown'
             if not sims:
                 continue
@@ -323,8 +362,19 @@ def main():
             sim_str = ','.join([f'{sim_id}:{score:.4f}' for sim_id, score in sims])
             f.write(f'{item_id}\t{item_name}\t{sim_str}\n')
-    print(f"Done! Generated i2i similarities for {len(result)} items")
-    print(f"Output saved to: {output_file}")
+    logger.info(f"完成！为 {len(result)} 个物品生成了相似度")
+    logger.info(f"输出保存到：{output_file}")
+    
+    # 如果启用debug模式，保存可读格式
+    if args.debug:
+        log_processing_step(logger, "保存Debug可读格式")
+        save_readable_index(
+            output_file,
+            result,
+            name_mappings,
+            index_type='i2i:deepwalk',
+            logger=logger
+        )
 if __name__ == '__main__':
@@ -18,21 +18,30 @@ from offline_tasks.config.offline_config import (
     DB_CONFIG, OUTPUT_DIR, I2I_CONFIG, get_time_range,
     DEFAULT_LOOKBACK_DAYS, DEFAULT_I2I_TOP_N
 )
+from offline_tasks.scripts.debug_utils import (
+    setup_debug_logger, log_dataframe_info, log_dict_stats,
+    save_readable_index, fetch_name_mappings, log_algorithm_params,
+    log_processing_step
+)
-def prepare_session_data(df, session_gap_minutes=30):
+def prepare_session_data(df, session_gap_minutes=30, logger=None):
     """
     准备会话数据
     Args:
         df: DataFrame with columns: user_id, item_id, create_time
         session_gap_minutes: 会话间隔时间（分钟）
+        logger: Logger instance for debugging
     Returns:
         List of sessions, each session is a list of item_ids
     """
     sessions = []
+    if logger:
+        logger.debug(f"开始准备会话数据，会话间隔：{session_gap_minutes}分钟")
+    
     # 按用户和时间排序
     df = df.sort_values(['user_id', 'create_time'])
@@ -65,21 +74,33 @@ def prepare_session_data(df, session_gap_minutes=30):
     # 过滤掉长度小于2的会话
     sessions = [s for s in sessions if len(s) >= 2]
+    if logger:
+        session_lengths = [len(s) for s in sessions]
+        logger.debug(f"生成 {len(sessions)} 个会话")
+        logger.debug(f"会话长度统计：最小={min(session_lengths)}, 最大={max(session_lengths)}, "
+                    f"平均={sum(session_lengths)/len(session_lengths):.2f}")
+    
     return sessions
-def train_word2vec(sessions, config):
+def train_word2vec(sessions, config, logger=None):
     """
     训练Word2Vec模型
     Args:
         sessions: List of sessions
         config: Word2Vec配置
+        logger: Logger instance for debugging
     Returns:
         Word2Vec模型
     """
-    print(f"Training Word2Vec with {len(sessions)} sessions...")
+    if logger:
+        logger.info(f"训练Word2Vec模型，共 {len(sessions)} 个会话")
+        logger.debug(f"模型参数：vector_size={config['vector_size']}, window={config['window_size']}, "
+                    f"min_count={config['min_count']}, epochs={config['epochs']}")
+    else:
+        print(f"Training Word2Vec with {len(sessions)} sessions...")
     model = Word2Vec(
         sentences=sessions,
@@ -92,23 +113,30 @@ def train_word2vec(sessions, config):
         seed=42
     )
-    print(f"Training completed. Vocabulary size: {len(model.wv)}")
+    if logger:
+        logger.info(f"训练完成。词汇表大小：{len(model.wv)}")
+    else:
+        print(f"Training completed. Vocabulary size: {len(model.wv)}")
     return model
-def generate_similarities(model, top_n=50):
+def generate_similarities(model, top_n=50, logger=None):
     """
     生成物品相似度
     Args:
         model: Word2Vec模型
         top_n: Top N similar items
+        logger: Logger instance for debugging
     Returns:
         Dict[item_id, List[Tuple(similar_item_id, score)]]
     """
     result = {}
+    if logger:
+        logger.info(f"生成Top {top_n} 相似物品")
+    
     for item_id in model.wv.index_to_key:
         try:
             similar_items = model.wv.most_similar(item_id, topn=top_n)
@@ -116,6 +144,9 @@ def generate_similarities(model, top_n=50):
         except KeyError:
             continue
+    if logger:
+        logger.info(f"生成了 {len(result)} 个物品的相似度")
+    
     return result
@@ -146,8 +177,25 @@ def main():
     args = parser.parse_args()
+    # 设置logger
+    logger = setup_debug_logger('i2i_session_w2v', debug=args.debug)
+    
+    # 记录算法参数
+    params = {
+        'window_size': args.window_size,
+        'vector_size': args.vector_size,
+        'min_count': args.min_count,
+        'workers': args.workers,
+        'epochs': args.epochs,
+        'top_n': args.top_n,
+        'lookback_days': args.lookback_days,
+        'session_gap_minutes': args.session_gap,
+        'debug': args.debug
+    }
+    log_algorithm_params(logger, params)
+    
     # 创建数据库连接
-    print("Connecting to database...")
+    logger.info("连接数据库...")
     engine = create_db_connection(
         DB_CONFIG['host'],
         DB_CONFIG['port'],
@@ -158,7 +206,7 @@ def main():
     # 获取时间范围
     start_date, end_date = get_time_range(args.lookback_days)
-    print(f"Fetching data from {start_date} to {end_date}...")
+    logger.info(f"获取数据范围：{start_date} 到 {end_date}")
     # SQL查询 - 获取用户行为序列
     sql_query = f"""
@@ -181,19 +229,23 @@ def main():
         se.create_time
     """
-    print("Executing SQL query...")
+    logger.info("执行SQL查询...")
     df = pd.read_sql(sql_query, engine)
-    print(f"Fetched {len(df)} records")
+    logger.info(f"获取到 {len(df)} 条记录")
+    
+    # 记录数据信息
+    log_dataframe_info(logger, df, "用户行为数据")
     # 转换create_time为datetime
     df['create_time'] = pd.to_datetime(df['create_time'])
     # 准备会话数据
-    print("Preparing session data...")
-    sessions = prepare_session_data(df, session_gap_minutes=args.session_gap)
-    print(f"Generated {len(sessions)} sessions")
+    log_processing_step(logger, "准备会话数据")
+    sessions = prepare_session_data(df, session_gap_minutes=args.session_gap, logger=logger)
+    logger.info(f"生成 {len(sessions)} 个会话")
     # 训练Word2Vec模型
+    log_processing_step(logger, "训练Word2Vec模型")
     w2v_config = {
         'vector_size': args.vector_size,
         'window_size': args.window_size,
@@ -203,28 +255,35 @@ def main():
         'sg': 1
     }
-    model = train_word2vec(sessions, w2v_config)
+    model = train_word2vec(sessions, w2v_config, logger=logger)
     # 保存模型（可选）
     if args.save_model:
         model_path = os.path.join(OUTPUT_DIR, f'session_w2v_model_{datetime.now().strftime("%Y%m%d")}.model')
         model.save(model_path)
-        print(f"Model saved to {model_path}")
+        logger.info(f"模型已保存到 {model_path}")
     # 生成相似度
-    print("Generating similarities...")
-    result = generate_similarities(model, top_n=args.top_n)
-    
-    # 创建item_id到name的映射
-    item_name_map = dict(zip(df['item_id'].astype(str), df.groupby('item_id')['item_name'].first()))
+    log_processing_step(logger, "生成相似度")
+    result = generate_similarities(model, top_n=args.top_n, logger=logger)
     # 输出结果
+    log_processing_step(logger, "保存结果")
     output_file = args.output or os.path.join(OUTPUT_DIR, f'i2i_session_w2v_{datetime.now().strftime("%Y%m%d")}.txt')
-    print(f"Writing results to {output_file}...")
+    # 获取name mappings用于标准输出格式
+    name_mappings = {}
+    if args.debug:
+        logger.info("获取物品名称映射...")
+        name_mappings = fetch_name_mappings(engine, debug=True)
+    
+    logger.info(f"写入结果到 {output_file}...")
     with open(output_file, 'w', encoding='utf-8') as f:
         for item_id, sims in result.items():
-            item_name = item_name_map.get(item_id, 'Unknown')
+            # 使用name_mappings获取名称，如果没有则从df中获取
+            item_name = name_mappings.get(int(item_id), 'Unknown') if item_id.isdigit() else 'Unknown'
+            if item_name == 'Unknown' and 'item_name' in df.columns:
+                item_name = df[df['item_id'].astype(str) == item_id]['item_name'].iloc[0] if len(df[df['item_id'].astype(str) == item_id]) > 0 else 'Unknown'
             if not sims:
                 continue
@@ -233,8 +292,19 @@ def main():
             sim_str = ','.join([f'{sim_id}:{score:.4f}' for sim_id, score in sims])
             f.write(f'{item_id}\t{item_name}\t{sim_str}\n')
-    print(f"Done! Generated i2i similarities for {len(result)} items")
-    print(f"Output saved to: {output_file}")
+    logger.info(f"完成！为 {len(result)} 个物品生成了相似度")
+    logger.info(f"输出保存到：{output_file}")
+    
+    # 如果启用debug模式，保存可读格式
+    if args.debug:
+        log_processing_step(logger, "保存Debug可读格式")
+        save_readable_index(
+            output_file,
+            result,
+            name_mappings,
+            index_type='i2i:session_w2v',
+            logger=logger
+        )
 if __name__ == '__main__':
@@ -17,6 +17,11 @@ from offline_tasks.config.offline_config import (
     DB_CONFIG, OUTPUT_DIR, INTEREST_AGGREGATION_CONFIG, get_time_range,
     DEFAULT_LOOKBACK_DAYS, DEFAULT_RECENT_DAYS, DEFAULT_INTEREST_TOP_N
 )
+from offline_tasks.scripts.debug_utils import (
+    setup_debug_logger, log_dataframe_info, log_dict_stats,
+    save_readable_index, fetch_name_mappings, log_algorithm_params,
+    log_processing_step
+)
 def calculate_time_weight(event_time, reference_time, decay_factor=0.95, days_unit=30):
@@ -227,8 +232,22 @@ def main():
     args = parser.parse_args()
+    # 设置logger
+    logger = setup_debug_logger('interest_aggregation', debug=args.debug)
+    
+    # 记录算法参数
+    params = {
+        'top_n': args.top_n,
+        'lookback_days': args.lookback_days,
+        'recent_days': args.recent_days,
+        'new_days': args.new_days,
+        'decay_factor': args.decay_factor,
+        'debug': args.debug
+    }
+    log_algorithm_params(logger, params)
+    
     # 创建数据库连接
-    print("Connecting to database...")
+    logger.info("连接数据库...")
     engine = create_db_connection(
         DB_CONFIG['host'],
         DB_CONFIG['port'],
@@ -242,7 +261,9 @@ def main():
     recent_start_date, _ = get_time_range(args.recent_days)
     new_start_date, _ = get_time_range(args.new_days)
-    print(f"Fetching data from {start_date} to {end_date}...")
+    logger.info(f"获取数据范围：{start_date} 到 {end_date}")
+    logger.debug(f"热门商品起始日期：{recent_start_date}")
+    logger.debug(f"新品起始日期：{new_start_date}")
     # SQL查询 - 获取用户行为数据（包含用户特征和商品分类）
     sql_query = f"""
@@ -279,9 +300,12 @@ def main():
         se.create_time
     """
-    print("Executing SQL query...")
+    logger.info("执行SQL查询...")
     df = pd.read_sql(sql_query, engine)
-    print(f"Fetched {len(df)} records")
+    logger.info(f"获取到 {len(df)} 条记录")
+    
+    # 记录数据信息
+    log_dataframe_info(logger, df, "用户行为数据")
     # 转换时间列
     df['create_time'] = pd.to_datetime(df['create_time'])
@@ -289,37 +313,70 @@ def main():
     # 定义行为权重
     behavior_weights = INTEREST_AGGREGATION_CONFIG['behavior_weights']
+    logger.debug(f"行为权重: {behavior_weights}")
     # 准备不同类型的数据集
+    log_processing_step(logger, "准备不同类型的数据集")
     # 1. 热门商品：最近N天的高交互商品
     df_hot = df[df['create_time'] >= recent_start_date].copy()
+    logger.info(f"热门商品数据集：{len(df_hot)} 条记录")
     # 2. 加购商品：加购行为
     df_cart = df[df['event_type'].isin(['addToCart', 'addToPool'])].copy()
+    logger.info(f"加购商品数据集：{len(df_cart)} 条记录")
     # 3. 新品：商品创建时间在最近N天内
     df_new = df[df['item_create_time'] >= new_start_date].copy()
+    logger.info(f"新品数据集：{len(df_new)} 条记录")
     # 生成不同列表类型的索引
-    print("\n=== Generating indices ===")
+    log_processing_step(logger, "生成不同列表类型的索引")
     list_type_indices = generate_list_type_indices(
         df_hot, df_cart, df_new, behavior_weights
     )
+    logger.info(f"生成了 {len(list_type_indices)} 种列表类型的索引")
+    
+    # 获取name mappings用于debug输出
+    name_mappings = {}
+    if args.debug:
+        logger.info("获取物品名称映射...")
+        name_mappings = fetch_name_mappings(engine, debug=True)
     # 输出索引
+    log_processing_step(logger, "保存索引文件")
     for list_type, aggregations in list_type_indices.items():
         output_prefix = f'{args.output_prefix}_{list_type}'
+        logger.info(f"保存 {list_type} 类型的索引...")
         output_indices(aggregations, output_prefix, top_n=args.top_n)
+        
+        # 如果启用debug模式，保存可读格式
+        if args.debug and aggregations:
+            for dim_key, items in aggregations.items():
+                if items:
+                    # 为每个维度生成可读索引
+                    result_dict = {dim_key: items[:args.top_n]}
+                    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
+                        )
     # 生成全局索引（所有数据）
-    print("\nGenerating global indices...")
+    log_processing_step(logger, "生成全局索引")
     global_aggregations = aggregate_by_dimensions(
         df, behavior_weights, time_decay=True, decay_factor=args.decay_factor
     )
+    logger.info("保存全局索引...")
     output_indices(global_aggregations, f'{args.output_prefix}_global', top_n=args.top_n)
-    print("\n=== All indices generated successfully! ===")
+    logger.info("="*80)
+    logger.info("所有索引生成完成！")
+    logger.info("="*80)
 if __name__ == '__main__':