sft_project/VisualizationData.py at main · bubbleee030/sft_project · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
import os
import json
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
import logging
from datetime import datetime
from matplotlib.font_manager import fontManager
import requests


plt.rcParams['axes.unicode_minus'] = False # 解決負號顯示問題

# --- 日誌設定 ---
log_time = datetime.now().strftime('%Y%m%d_%H%M%S')
log_directory = 'logs'
os.makedirs(log_directory, exist_ok=True)
log_filename = f"visualization_{log_time}.log"
logging.basicConfig(level=logging.INFO, filename=os.path.join(log_directory, log_filename), filemode='w', format='%(asctime)s - %(levelname)s - %(message)s', encoding='utf-8')
console_handler = logging.StreamHandler(); console_handler.setLevel(logging.INFO); formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'); console_handler.setFormatter(formatter); logging.getLogger().addHandler(console_handler)

# --- 資料載入 ---
def load_full_metadata(file_path="final_classified_output/full_metadata_log.json"):
    """從單一的元數據日誌檔案載入所有資料。"""
    if not os.path.exists(file_path):
        logging.error(f"錯誤：找不到元數據日誌檔案 '{file_path}'。請先執行最新的 data_extraction 腳本。")
        return pd.DataFrame()
    logging.info(f"正在從 '{file_path}' 載入完整的元數據...")
    df = pd.read_json(file_path)
    logging.info(f"資料載入完成，總共 {len(df)} 筆記錄。")
    return df

# --- 模型名稱重命名 (與您原始功能相同) ---
def rename_model(df):
    if df.empty: return df
    model_mapping = {'gen_DeepSeek-R1': 'R1', 'gen_QwQ-32B': 'TAT'}
    logging.info(f"重新命名模型: {model_mapping}")
    df['genmodel'] = df['genmodel'].replace(model_mapping)
    return df

# --- 全新視覺化功能 ---
def visualize_status_distribution(df, output_dir):
    """視覺化整體數據狀態分佈 (有效/重複/錯誤)"""
    if df.empty: return
    logging.info("正在視覺化整體數據狀態分佈...")
    status_counts = df['status'].value_counts()
    plt.figure(figsize=(10, 8))
    plt.pie(status_counts, labels=status_counts.index, autopct='%1.1f%%', startangle=140, textprops={'fontsize': 14})
    plt.title('Overall Status Distribution', fontsize=18, pad=20)
    plt.ylabel('')
    save_path = os.path.join(output_dir, '1_overall_status_distribution.png')
    plt.savefig(save_path); plt.close()
    logging.info(f"狀態分佈圖已儲存至: {save_path}")

def visualize_error_reasons(df, output_dir):
    """視覺化格式錯誤的主要原因"""
    if df.empty: return
    logging.info("正在視覺化格式錯誤原因...")
    error_df = df[df['status'] == 'FormatError']
    if error_df.empty:
        logging.info("沒有格式錯誤的資料，跳過此圖表。")
        return
    reason_counts = error_df['error_reason'].value_counts()
    plt.figure(figsize=(12, 8))
    sns.barplot(x=reason_counts.values, y=reason_counts.index, palette='viridis')
    plt.title('Format Error Reasons Analysis', fontsize=18, pad=20)
    plt.xlabel('Count', fontsize=14)
    plt.ylabel('Error Reason', fontsize=14)
    plt.tight_layout()
    save_path = os.path.join(output_dir, '2_format_error_reasons.png')
    plt.savefig(save_path); plt.close()
    logging.info(f"錯誤原因圖已儲存至: {save_path}")

def visualize_rates_table(df, output_dir):
    """以美化的HTML表格呈現各模型組合的錯誤率與重複率"""
    if df.empty: return
    logging.info("正在產生模型組合健康度分析表...")
    df['combination'] = df['genmodel'] + " / " + df['judgemodel']
    status_pivot = pd.pivot_table(df, values='internal_id', index='combination', columns='status', aggfunc='count', fill_value=0)
    if 'Valid' not in status_pivot.columns: status_pivot['Valid'] = 0
    if 'FormatError' not in status_pivot.columns: status_pivot['FormatError'] = 0
    if 'Duplicate' not in status_pivot.columns: status_pivot['Duplicate'] = 0
    status_pivot['Total'] = status_pivot.sum(axis=1)
    status_pivot['Error_Rate'] = (status_pivot['FormatError'] / status_pivot['Total'])
    status_pivot['Dup_Rate'] = (status_pivot['Duplicate'] / status_pivot['Total'])

    styled_table = (status_pivot[['Valid', 'FormatError', 'Duplicate', 'Total', 'Error_Rate', 'Dup_Rate']]
                    .sort_values(by='Total', ascending=False)
                    .style
                    .format({'Error_Rate': '{:.2%}', 'Dup_Rate': '{:.2%}'})
                    .background_gradient(cmap='Reds', subset=['Error_Rate'])
                    .background_gradient(cmap='Blues', subset=['Dup_Rate'])
                    .bar(color='#5DADE2', subset=['Valid'])
                    .set_caption("模型組合健康度分析 (錯誤率與重複率)")
                    .set_properties(**{'text-align': 'center'}))

    save_path = os.path.join(output_dir, '3_model_health_rates_table.html')
    styled_table.to_html(save_path, encode_html="utf-8")
    logging.info(f"模型健康度分析表已儲存至: {save_path}")

# --- 保留並強化的視覺化功能 (基於有效資料) ---
def visualize_model_pairs_heatmap(df, output_dir):
    """分析並視覺化 genmodel 和 judgemodel 的組合熱度 (基於有效資料)"""
    if df.empty: return
    logging.info("正在視覺化模型組合熱力圖...")
    pivot_table = df.pivot_table(index='genmodel', columns='judgemodel', values='internal_id', aggfunc='count', fill_value=0)
    plt.figure(figsize=(14, 10))
    sns.heatmap(pivot_table, annot=True, fmt=".0f", cmap="viridis", linewidths=.5)
    plt.title('GenModel vs JudgeModel Heatmap (Valid Data Count)', fontsize=18, pad=20)
    plt.xlabel('JudgeModel', fontsize=14)
    plt.ylabel('GenModel', fontsize=14)
    plt.xticks(rotation=45, ha='right'); plt.yticks(rotation=0)
    plt.tight_layout()
    save_path = os.path.join(output_dir, '4_valid_data_heatmap.png')
    plt.savefig(save_path); plt.close()
    logging.info(f"模型組合熱力圖已儲存至: {save_path}")

def visualize_mission_combinations(df, output_dir):
    """分析並視覺化每個 mission 中，genmodel 和 judgemodel 的分佈情況 (基於有效資料)"""
    if df.empty: return
    logging.info("正在分析各 mission 的模型組合分佈...")
    missions = df['mission'].unique()
    for mission in missions:
        mission_df = df[df['mission'] == mission]
        if mission_df.empty: continue
        pivot_df = mission_df.pivot_table(index='genmodel', columns='judgemodel', values='internal_id', aggfunc='count', fill_value=0)
        if pivot_df.empty: continue
        pivot_df.plot(kind='bar', stacked=True, figsize=(14, 8), cmap='tab20c', width=0.8)
        plt.title(f'Mission {mission} Model Combination Distribution', fontsize=18, pad=20)
        plt.xlabel('GenModel', fontsize=14); plt.ylabel('Valid Data Count', fontsize=14)
        plt.xticks(rotation=45, ha='right'); plt.legend(title='JudgeModel', bbox_to_anchor=(1.02, 1), loc='upper left')
        plt.tight_layout()
        save_path = os.path.join(output_dir, f'5_mission_{mission}_distribution.png')
        plt.savefig(save_path); plt.close()
        logging.info(f"任務 '{mission}' 的模型分佈圖已儲存至: {save_path}")

# --- 主程式執行區 ---
if __name__ == "__main__":
    logging.info("--- 開始執行資料視覺化與分析腳本 ---")
    report_dir = "visualization_report"
    os.makedirs(report_dir, exist_ok=True)

    # 1. 從單一檔案載入所有元數據
    full_df = load_full_metadata()

    if not full_df.empty:
        full_df = rename_model(full_df)

        # 2. 執行基於【完整資料集】的分析
        visualize_status_distribution(full_df, report_dir)
        visualize_error_reasons(full_df, report_dir)
        visualize_rates_table(full_df, report_dir)

        # 3. 篩選出【有效資料】進行後續分析
        valid_df = full_df[full_df['status'] == 'Valid'].copy()
        if not valid_df.empty:
            logging.info(f"篩選出 {len(valid_df)} 筆有效資料進行深度分析。")
            visualize_model_pairs_heatmap(valid_df, report_dir)
            visualize_mission_combinations(valid_df, report_dir)
        else:
            logging.warning("沒有有效的資料可進行組合與任務分析。")

        logging.info("--- 所有分析與視覺化任務已完成 ---")
        print(f"\n分析報告與圖表已成功產生！請查看 '{report_dir}' 資料夾。")
    else:
        logging.warning("沒有載入任何資料，無法進行分析。")