knowledge_search/analyze_stage6_results.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Stage 6 评估结果统计分析
分析两层评估的过滤效果和匹配质量
"""

import json
from typing import Dict, List, Any
from collections import defaultdict


def load_stage6_results(file_path: str) -> List[Dict[str, Any]]:
    """加载Stage 6评估结果"""
    with open(file_path, 'r', encoding='utf-8') as f:
        return json.load(f)


def analyze_evaluation_results(data: List[Dict[str, Any]]) -> Dict[str, Any]:
    """分析评估结果"""

    # 全局统计
    global_stats = {
        'total_search_words': 0,
        'total_notes_evaluated': 0,
        'total_filtered': 0,
        'match_distribution': {
            '完全匹配(8-10)': 0,
            '相似匹配(6-7)': 0,
            '弱相似(5-6)': 0,
            '无匹配(≤4)': 0
        }
    }

    # 按原始特征分组统计
    feature_stats = defaultdict(lambda: {
        'search_words_count': 0,
        'total_notes': 0,
        'total_filtered': 0,
        'match_distribution': {
            '完全匹配(8-10)': 0,
            '相似匹配(6-7)': 0,
            '弱相似(5-6)': 0,
            '无匹配(≤4)': 0
        },
        'search_words': []
    })

    # 所有搜索词的详细统计
    search_word_details = []

    # 遍历所有原始特征
    for feature_result in data:
        original_feature = feature_result.get('原始特征名称', 'Unknown')

        # 从组合评估结果_分组中读取
        grouped_results = feature_result.get('组合评估结果_分组', [])

        for group in grouped_results:
            base_word = group.get('base_word', '')

            for eval_item in group.get('top10_searches', []):
                # 检查是否有评估结果
                evaluation = eval_item.get('evaluation_with_filter')
                if not evaluation:
                    continue

                search_word = eval_item.get('search_word', '')

                # 提取评估数据
                total_notes = evaluation.get('total_notes', 0)
                evaluated_notes = evaluation.get('evaluated_notes', 0)
                filtered_count = evaluation.get('filtered_count', 0)
                statistics = evaluation.get('statistics', {})

                # 更新全局统计
                global_stats['total_search_words'] += 1
                global_stats['total_notes_evaluated'] += total_notes
                global_stats['total_filtered'] += filtered_count

                for key in global_stats['match_distribution']:
                    global_stats['match_distribution'][key] += statistics.get(key, 0)

                # 更新特征统计
                feature_stats[original_feature]['search_words_count'] += 1
                feature_stats[original_feature]['total_notes'] += total_notes
                feature_stats[original_feature]['total_filtered'] += filtered_count

                for key in feature_stats[original_feature]['match_distribution']:
                    feature_stats[original_feature]['match_distribution'][key] += statistics.get(key, 0)

                # 记录搜索词详情
                search_word_info = {
                    'original_feature': original_feature,
                    'base_word': base_word,
                    'search_word': search_word,
                    'total_notes': total_notes,
                    'evaluated_notes': evaluated_notes,
                    'filtered_count': filtered_count,
                    'match_distribution': statistics,
                    'high_quality_count': statistics.get('完全匹配(8-10)', 0),
                    'similar_count': statistics.get('相似匹配(6-7)', 0)
                }

                search_word_details.append(search_word_info)
                feature_stats[original_feature]['search_words'].append(search_word_info)

    # 计算全局过滤率
    if global_stats['total_notes_evaluated'] > 0:
        global_stats['filter_rate'] = global_stats['total_filtered'] / global_stats['total_notes_evaluated']
    else:
        global_stats['filter_rate'] = 0.0

    # 计算每个特征的过滤率
    for feature_name, stats in feature_stats.items():
        if stats['total_notes'] > 0:
            stats['filter_rate'] = stats['total_filtered'] / stats['total_notes']
        else:
            stats['filter_rate'] = 0.0

    # 按高质量匹配数排序搜索词
    search_word_details.sort(key=lambda x: x['high_quality_count'], reverse=True)

    return {
        'global_stats': global_stats,
        'feature_stats': dict(feature_stats),
        'search_word_details': search_word_details
    }


def print_statistics(stats: Dict[str, Any]):
    """打印统计结果"""
    global_stats = stats['global_stats']
    feature_stats = stats['feature_stats']
    search_word_details = stats['search_word_details']

    print("=" * 80)
    print("Stage 6 评估结果统计分析")
    print("=" * 80)

    # 全局统计
    print("\n【全局统计】")
    print(f"  总搜索词数: {global_stats['total_search_words']}")
    print(f"  总评估帖子数: {global_stats['total_notes_evaluated']}")
    print(f"  总过滤帖子数: {global_stats['total_filtered']} (过滤率: {global_stats['filter_rate']*100:.1f}%)")
    print(f"\n  匹配度分布:")
    for match_type, count in global_stats['match_distribution'].items():
        print(f"    {match_type}: {count} 个帖子")

    # 按原始特征统计
    print("\n" + "=" * 80)
    print("【按原始特征统计】")
    print("=" * 80)

    for feature_name, stats in sorted(feature_stats.items()):
        print(f"\n特征: {feature_name}")
        print(f"  搜索词数: {stats['search_words_count']}")
        print(f"  总评估帖子: {stats['total_notes']}")
        print(f"  总过滤帖子: {stats['total_filtered']} (过滤率: {stats['filter_rate']*100:.1f}%)")
        print(f"  高质量匹配: {stats['match_distribution']['完全匹配(8-10)']} 个帖子")
        print(f"  相似匹配: {stats['match_distribution']['相似匹配(6-7)']} 个帖子")

        # 找出该特征下高质量匹配最多的搜索词
        best_searches = sorted(stats['search_words'], key=lambda x: x['high_quality_count'], reverse=True)[:3]
        if best_searches:
            print(f"  Top 3 最佳搜索词:")
            for idx, sw in enumerate(best_searches, 1):
                print(f"    {idx}. \"{sw['search_word']}\" - {sw['high_quality_count']}个完全匹配")

    # Top 10 最佳搜索词
    print("\n" + "=" * 80)
    print("【Top 10 最佳搜索词（按完全匹配数排序）】")
    print("=" * 80)

    for idx, sw in enumerate(search_word_details[:10], 1):
        print(f"\n{idx}. \"{sw['search_word']}\"")
        print(f"   原始特征: {sw['original_feature']}")
        print(f"   Base Word: {sw['base_word']}")
        print(f"   评估帖子: {sw['total_notes']}, 过滤: {sw['filtered_count']}")
        print(f"   完全匹配(8-10): {sw['high_quality_count']} 个")
        print(f"   相似匹配(6-7): {sw['similar_count']} 个")

    # 过滤效果分析
    print("\n" + "=" * 80)
    print("【过滤效果分析】")
    print("=" * 80)

    total_evaluated = global_stats['total_notes_evaluated']
    total_filtered = global_stats['total_filtered']
    total_remaining = total_evaluated - total_filtered

    total_high_quality = global_stats['match_distribution']['完全匹配(8-10)']
    total_similar = global_stats['match_distribution']['相似匹配(6-7)']
    total_weak = global_stats['match_distribution']['弱相似(5-6)']
    total_no_match = global_stats['match_distribution']['无匹配(≤4)']

    print(f"  评估帖子总数: {total_evaluated}")
    print(f"  第一层过滤（Query不相关）: {total_filtered} ({total_filtered/total_evaluated*100:.1f}%)")
    print(f"  通过过滤的帖子: {total_remaining} ({total_remaining/total_evaluated*100:.1f}%)")
    print(f"\n  通过过滤后的质量分布:")
    if total_remaining > 0:
        print(f"    完全匹配(8-10): {total_high_quality} ({total_high_quality/total_remaining*100:.1f}%)")
        print(f"    相似匹配(6-7): {total_similar} ({total_similar/total_remaining*100:.1f}%)")
        print(f"    弱相似(5-6): {total_weak} ({total_weak/total_remaining*100:.1f}%)")
        print(f"    无匹配(≤4): {total_no_match} ({total_no_match/total_remaining*100:.1f}%)")

    print("\n" + "=" * 80)


def save_statistics(stats: Dict[str, Any], output_path: str):
    """保存统计结果到JSON文件"""
    with open(output_path, 'w', encoding='utf-8') as f:
        json.dump(stats, f, ensure_ascii=False, indent=2)
    print(f"\n统计结果已保存到: {output_path}")


def main():
    """主函数"""
    input_file = "output_v2/stage6_with_evaluations.json"
    output_file = "output_v2/stage6_statistics.json"

    print("正在加载数据...")
    data = load_stage6_results(input_file)

    print("正在分析评估结果...")
    stats = analyze_evaluation_results(data)

    # 打印统计结果
    print_statistics(stats)

    # 保存结果
    save_statistics(stats, output_file)


if __name__ == '__main__':
    main()