knowledge_search/scripts/run_similarity_analysis.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""相似度分析独立运行脚本"""

import sys
import os

# 将项目根目录添加到Python路径
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, project_root)

import json
import logging
import argparse
from src.analyzers.similarity_analyzer import SimilarityAnalyzer


def main():
    parser = argparse.ArgumentParser(
        description='解构特征相似度分析（独立运行）',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
使用示例:
  # 基础用法 - 处理"墨镜"特征
  python3 scripts/run_similarity_analysis.py --feature "墨镜"

  # 处理多个特征
  python3 scripts/run_similarity_analysis.py --feature "墨镜" "耳环"

  # 自定义权重配置
  python3 scripts/run_similarity_analysis.py --feature "墨镜" --weight-embedding 0.7 --weight-semantic 0.3

  # 过滤低相似度特征
  python3 scripts/run_similarity_analysis.py --feature "墨镜" --min-similarity 0.3

  # 使用配置文件
  python3 scripts/run_similarity_analysis.py --config stage8_config.json

  # 自定义输入输出路径
  python3 scripts/run_similarity_analysis.py --input output_v2/deep_analysis_custom.json --output output_v2/similarity_custom.json
        """
    )

    # 输入输出
    parser.add_argument(
        '--input',
        default='output_v2/deep_analysis_results.json',
        help='解构分析结果文件路径（默认: output_v2/deep_analysis_results.json）'
    )
    parser.add_argument(
        '--output',
        default='output_v2/similarity_analysis_results.json',
        help='输出文件路径（默认: output_v2/similarity_analysis_results.json）'
    )

    # 特征过滤
    parser.add_argument(
        '--feature',
        nargs='+',
        default=None,
        help='指定要处理的原始特征名称（可指定多个），如: --feature "墨镜" "耳环"'
    )

    # 相似度配置
    parser.add_argument(
        '--weight-embedding',
        type=float,
        default=0.5,
        help='向量模型权重（默认: 0.5）'
    )
    parser.add_argument(
        '--weight-semantic',
        type=float,
        default=0.5,
        help='LLM 模型权重（默认: 0.5）'
    )
    parser.add_argument(
        '--min-similarity',
        type=float,
        default=0.0,
        help='最小相似度阈值，低于此值的特征会被过滤（默认: 0.0，保留所有）'
    )

    # 并发配置
    parser.add_argument(
        '--max-workers',
        type=int,
        default=5,
        help='最大并发数（默认: 5）'
    )

    # 综合得分P计算配置
    parser.add_argument(
        '--evaluation-path',
        default='output_v2/evaluated_results.json',
        help='评估结果数据文件路径，用于计算综合得分P（默认: output_v2/evaluated_results.json）'
    )
    parser.add_argument(
        '--no-update-evaluation',
        action='store_true',
        help='不计算和更新综合得分P（默认会计算）'
    )

    # 配置文件
    parser.add_argument(
        '--config',
        help='从配置文件读取参数（JSON 格式）'
    )

    # 日志级别
    parser.add_argument(
        '--log-level',
        default='INFO',
        choices=['DEBUG', 'INFO', 'WARNING', 'ERROR'],
        help='日志级别（默认: INFO）'
    )

    args = parser.parse_args()

    # 配置日志
    logging.basicConfig(
        level=getattr(logging, args.log_level),
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    logger = logging.getLogger(__name__)

    # 如果提供了配置文件，从文件读取参数
    if args.config:
        logger.info(f"从配置文件读取参数: {args.config}")
        try:
            with open(args.config, 'r', encoding='utf-8') as f:
                config = json.load(f)

            # 配置文件中的参数会覆盖命令行默认值，但不会覆盖用户显式指定的命令行参数
            args.input = config.get('input', args.input)
            args.output = config.get('output', args.output)
            args.feature = config.get('feature', args.feature)
            args.weight_embedding = config.get('weight_embedding', args.weight_embedding)
            args.weight_semantic = config.get('weight_semantic', args.weight_semantic)
            args.min_similarity = config.get('min_similarity', args.min_similarity)
            args.max_workers = config.get('max_workers', args.max_workers)
            args.evaluation_path = config.get('evaluation_path', args.evaluation_path)
            if 'no_update_evaluation' in config:
                args.no_update_evaluation = config.get('no_update_evaluation', args.no_update_evaluation)

        except Exception as e:
            logger.error(f"读取配置文件失败: {e}")
            return 1

    # 验证输入文件
    if not os.path.exists(args.input):
        logger.error(f"输入文件不存在: {args.input}")
        return 1

    # 读取解构分析结果
    logger.info(f"读取解构分析结果: {args.input}")
    try:
        with open(args.input, 'r', encoding='utf-8') as f:
            deconstruction_results = json.load(f)
    except Exception as e:
        logger.error(f"读取解构分析结果失败: {e}")
        return 1

    # 打印配置信息
    logger.info("\n" + "=" * 60)
    logger.info("相似度分析配置:")
    logger.info("=" * 60)
    logger.info(f"输入文件: {args.input}")
    logger.info(f"输出文件: {args.output}")
    if args.feature:
        logger.info(f"目标特征: {', '.join(args.feature)}")
    else:
        logger.info(f"目标特征: 全部")
    logger.info(f"向量模型权重: {args.weight_embedding}")
    logger.info(f"LLM 模型权重: {args.weight_semantic}")
    logger.info(f"最小相似度阈值: {args.min_similarity}")
    logger.info(f"最大并发数: {args.max_workers}")
    logger.info(f"评估结果文件路径: {args.evaluation_path}")
    logger.info(f"计算综合得分P: {'否' if args.no_update_evaluation else '是'}")
    logger.info("=" * 60 + "\n")

    # 创建分析器
    try:
        analyzer = SimilarityAnalyzer(
            weight_embedding=args.weight_embedding,
            weight_semantic=args.weight_semantic,
            max_workers=args.max_workers,
            min_similarity=args.min_similarity,
            target_features=args.feature,
            evaluation_results_path=args.evaluation_path,
            update_evaluation_scores=not args.no_update_evaluation
        )
    except Exception as e:
        logger.error(f"创建分析器失败: {e}")
        return 1

    # 运行分析
    try:
        similarity_results = analyzer.run(deconstruction_results, output_path=args.output)

        # 打印摘要
        logger.info("\n" + "=" * 60)
        logger.info("相似度分析完成")
        logger.info("=" * 60)

        metadata = similarity_results['metadata']
        overall_stats = metadata['overall_statistics']

        logger.info(f"处理帖子数: {overall_stats['total_notes']}")
        logger.info(f"提取特征总数: {overall_stats['total_features_extracted']}")
        logger.info(f"平均特征数/帖子: {overall_stats['avg_features_per_note']}")
        logger.info(f"平均最高相似度: {overall_stats['avg_max_similarity']}")
        logger.info(f"包含高相似度特征的帖子: {overall_stats['notes_with_high_similarity']}")
        logger.info(f"总耗时: {metadata['processing_time_seconds']}秒")
        logger.info(f"结果已保存: {args.output}")
        logger.info("=" * 60 + "\n")

        # 打印 Top 5 高相似度特征示例
        if similarity_results['results']:
            logger.info("Top 5 高相似度特征示例:")
            all_features = []
            for result in similarity_results['results']:
                for feat in result['deconstructed_features'][:5]:  # 每个帖子取前5个
                    all_features.append({
                        'note_id': result['note_id'],
                        'feature_name': feat['feature_name'],
                        'dimension': feat['dimension'],
                        'similarity': feat['similarity_score']
                    })

            # 按相似度排序，取 Top 5
            all_features.sort(key=lambda x: x['similarity'], reverse=True)
            for i, feat in enumerate(all_features[:5], 1):
                logger.info(f"  {i}. [{feat['note_id'][:12]}...] "
                           f"{feat['feature_name']} ({feat['dimension']}) "
                           f"- 相似度: {feat['similarity']:.3f}")

        return 0

    except Exception as e:
        logger.error(f"相似度分析失败: {e}", exc_info=True)
        return 1


if __name__ == '__main__':
    exit(main())