extract_data2.py

import os
import json
import logging
import time
import pandas as pd  
from simhash import Simhash
from datetime import datetime
from tqdm import tqdm

# --- 日誌設定 (保持不變) ---
log_time = datetime.now().strftime('%Y%m%d_%H%M%S')
log_directory = 'logs'
os.makedirs(log_directory, exist_ok=True)
log_filename = f"data_extraction_{log_time}.log"
log_filepath = os.path.join(log_directory, log_filename)

logging.basicConfig(
    level=logging.INFO,
    filename=log_filepath,
    filemode='w',
    format='%(asctime)s - %(levelname)s - %(message)s',
    encoding='utf-8'
)

logger = logging.getLogger()
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)

# --- 輔助函式 (保持不變) ---
missions = ['petition', 'od', 'press', 'qa']

def findMission(filename):
    for mission in missions:
        if mission in filename.lower(): return mission
    return 'unknown'

def load_simpCharSet(filepath):
    try:
        with open(filepath, 'r', encoding='utf-8') as f:
            chars = set(f.read().replace(" ", "").replace("\n", ""))
            logging.info(f"成功從 {filepath} 載入 {len(chars)} 個獨立的簡體字元。")
            return chars
    except FileNotFoundError:
        logging.error(f"錯誤：找不到簡體字字典檔案 {filepath}，程式即將終止。")
        exit()

def containSimp(text, simplifiedCharSet):
    if not isinstance(text, str) or not text: return None
    for char in text:
        if char in simplifiedCharSet: return char
    return None

def parse_model_from_path(normalized_path):
    try:
        path_parts = normalized_path.split('/')
        eval_results_index = path_parts.index('eval_results')
        genmodel = path_parts[eval_results_index + 1]
        judgemodel = path_parts[eval_results_index + 2]
        return genmodel, judgemodel
    except (ValueError, IndexError):
        logging.warning(f"無法從路徑 '{normalized_path}' 解析 genmodel 或 judgemodel。")
        return 'unknown', 'unknown'

# --- 【修改點】 移植自第二個腳本的統計報告產生器 ---
def generate_and_print_statistics(all_items_metadata):
    """
    基於一個包含所有處理元數據的列表，產生並印出詳細的統計報告。
    """
    if not all_items_metadata:
        logging.warning("沒有可供分析的資料。")
        return

    df = pd.DataFrame(all_items_metadata)
    # 設定 pandas 顯示選項，確保終端機能完整顯示報告
    pd.set_option('display.max_rows', 200)
    pd.set_option('display.max_columns', 20)
    pd.set_option('display.width', 120)

    report_str = "\n" + "="*60 + "\n 高維度交叉分析統計報告\n" + "="*60 + "\n\n"
    
    # 1. 整體狀態分佈
    report_str += "--- 1. 整體狀態分佈 ---\n"
    report_str += str(df['status'].value_counts()) + "\n\n"

    # 2. 各類錯誤原因分佈
    report_str += "--- 2. 各類格式錯誤原因分佈 ---\n"
    error_df = df[df['status'] == 'FormatError']
    report_str += str(error_df['error_reason'].value_counts()) + "\n\n"

    # 準備一個僅包含有效數據的 DataFrame 以供後續分析
    valid_df = df[df['status'] == 'Valid'].copy()
    if not valid_df.empty:
        valid_df['combination'] = valid_df['genmodel'] + " / " + valid_df['judgemodel']
        
        # 3. 各模型組合的「產出語言」分佈 (僅計算有效資料)
        report_str += "--- 3. 各模型組合的「產出語言」分佈 (僅計算有效資料) ---\n"
        lang_pivot = pd.pivot_table(valid_df, values='internal_id', index='combination', columns='language', aggfunc='count', fill_value=0)
        if 'Traditional' in lang_pivot.columns and 'Simplified' in lang_pivot.columns:
            lang_pivot['Total_Valid'] = lang_pivot['Traditional'] + lang_pivot['Simplified']
            lang_pivot['Trad_Ratio'] = (lang_pivot['Traditional'] / lang_pivot['Total_Valid']).map('{:.1%}'.format)
        report_str += str(lang_pivot.sort_values(by='Total_Valid' if 'Total_Valid' in lang_pivot.columns else 'Traditional', ascending=False)) + "\n\n"

        # 4. 各模型組合的「任務 (Mission)」產出分佈 (僅計算有效資料)
        report_str += "--- 4. 各模型組合的「任務」產出分佈 (僅計算有效資料) ---\n"
        mission_pivot = pd.pivot_table(valid_df, values='internal_id', index='combination', columns='mission', aggfunc='count', fill_value=0)
        report_str += str(mission_pivot) + "\n\n"
    else:
        report_str += "--- 3 & 4. 沒有有效資料可供語言或任務分析。 ---\n\n"

    # 5. 各模型組合的「錯誤率」與「重複率」分析 (使用完整的 DataFrame)
    report_str += "--- 5. 各模型組合的「錯誤率」與「重複率」分析 ---\n"
    df['combination'] = df['genmodel'] + " / " + df['judgemodel']
    status_pivot = pd.pivot_table(df, values='internal_id', index='combination', columns='status', aggfunc='count', fill_value=0)
    status_pivot['Total_Items'] = status_pivot.sum(axis=1)
    if 'FormatError' in status_pivot.columns: status_pivot['Error_Rate'] = (status_pivot['FormatError'] / status_pivot['Total_Items']).map('{:.1%}'.format)
    if 'Duplicate' in status_pivot.columns: status_pivot['Dup_Rate'] = (status_pivot['Duplicate'] / status_pivot['Total_Items']).map('{:.1%}'.format)
    report_str += str(status_pivot.sort_values(by='Total_Items', ascending=False)) + "\n\n"
    
    # 在終端機印出報告並寫入日誌
    print(report_str)
    logging.info(report_str)


# --- Step 1: 數據提取與驗證 (返回所有數據及其狀態) ---
def extract_and_validate(root_directory):
    """
    遍歷目錄，提取所有數據，並為每一筆數據進行驗證和標記。
    返回一個包含所有原始數據和其元數據（狀態、錯誤原因等）的列表。
    """
    all_items_with_metadata = []
    format_error = []
    validated_data = []
    patterns = ['petition', 'od', 'press', 'qa', 'score']
    
    logging.info(f"開始從 '{root_directory}' 目錄進行搜尋...")
    
    file_paths = [os.path.join(subdir, filename) for subdir, _, files in os.walk(root_directory) for filename in files]
    
    for file_path in tqdm(file_paths, desc="正在掃描所有檔案"):
        normalized_path = file_path.replace('\\', '/')
        filename = os.path.basename(normalized_path)
        genmodel, judgemodel = parse_model_from_path(normalized_path)
        mission = findMission(filename)
    
        if not mission:
            logging.info(f"  跳過檔案: {filename} (不包含 'mission')")
            continue
        # 檢查檔案名稱有raft就跳過
        if 'raft' in filename:
            logging.info(f"  跳過檔案: {filename} (包含 'raft')")
            continue            
        if not genmodel or not judgemodel:
            logging.info(f"  跳過檔案: {filename} (無法解析 genmodel 或 judgemodel)")
            continue
        if not any (pattern in filename.lower() for pattern in patterns):
            logging.info(f"  跳過檔案: {filename} (不符合指定檔名)")
            continue
            
        try:
            current_items_raw = []
            if filename.endswith('.jsonl'):
                logging.info(f"  正在處理 JSONL 檔案: {filename}")
                with open(normalized_path, 'r', encoding='utf-8') as f:
                    for line in f:
                        if line.strip(): current_items_raw.append(json.loads(line))
            elif filename.endswith('.json'):
                logging.info(f"  正在處理 JSON 檔案: {filename}")
                with open(normalized_path, 'r', encoding='utf-8') as f:
                    # 確保即使沒有 'eval_result...' 鍵也不會報錯
                    data = json.load(f)
                    current_items_raw = data.get("eval_result_from_first_iteration", [])

            for i, item in enumerate(current_items_raw):
                if filename.endswith('.json'):
                    model_responses = item.get('model_responses', {})
                    processed_item = {
                        'qid': item.get('qid'),
                        'prompt': item.get('question'),
                        'resp': model_responses.get('ground_truth'),
                        'model_response': model_responses.get('Llama-4-Maverick-17B-128E-Instruct-FP8') or model_responses.get('Llama-3.1-8B-Instruct'),
                        'full_output': item.get('judge_response')
                    }
                else:
                    processed_item = {
                        'qid': item.get('qid'),
                        'prompt': item.get('prompt'),
                        'resp': item.get('resp'),
                        'model_response': item.get('model_response'), 'full_output': item.get('full_output')
                    }
                
                # --- 【修改點】為每筆數據打上狀態標籤，而不是直接過濾掉 ---
                status = "Valid"
                error_reasons = []  # 改為列表收集所有錯誤
                
                processed_item['internal_id'] = f"{normalized_path}_{i}"
                processed_item.update({'mission': mission, 'genmodel': genmodel, 'judgemodel': judgemodel})

                full_output = processed_item.get('full_output', '')
                model_response = processed_item.get('model_response', '')
                required_tags = ['【給分原因】', '【分數】', '題文匹配度(MatchPoint)', '文本格式(TextFormat)']
                
                # 檢查所有可能的錯誤，不在第一個錯誤時停止
                if not isinstance(full_output, str) or not full_output.strip():
                    error_reasons.append("'full_output' 欄位遺失、非字串或為空")
                else:
                    if 'think' in full_output.lower():
                        error_reasons.append("'full_output' 包含 'think' 字樣")
                    if not all(tag in full_output for tag in required_tags):
                        missing_tags = [tag for tag in required_tags if tag not in full_output]
                        error_reasons.append(f"'full_output' 內容格式不符 (缺少關鍵tag: {', '.join(missing_tags)})")
                
                if not isinstance(model_response, str) or not model_response.strip():
                    error_reasons.append("'model_response' 欄位遺失、非字串或為空")
                
                # 根據錯誤列表設定狀態
                if error_reasons:
                    status = "FormatError"
                    error_reason = " | ".join(error_reasons)  # 用分隔符合併所有錯誤
                else:
                    error_reason = "N/A"

                processed_item.update({
                    'status': status,
                    'error_reason': error_reason,
                    'text_for_hash': full_output or ""
                })
                all_items_with_metadata.append(processed_item)
                format_error.append(processed_item) if status == "FormatError" else None
                validated_data.append(processed_item) if status == "Valid" else None
                
        except Exception as e:
            logging.error(f"處理檔案 {normalized_path} 時發生錯誤: {e}", exc_info=True)

    logging.info(f"Step 1 finished, 共提取 {len(all_items_with_metadata)} 筆原始資料, 包含 {len(validated_data)} 筆有效資料, {len(format_error)} 筆格式錯誤資料。")
    return all_items_with_metadata, validated_data, format_error


# --- Step 2: 去除重複 (基於元數據列表進行操作) ---
def find_and_remove_duplicates(all_items_metadata, threshold=2):
    """
    接收完整的元數據列表，對其中的 'Valid' 項目進行去重，
    並更新重複項的 'status'。返回更新後的列表。
    """
    if not all_items_metadata: 
        return all_items_metadata, []

    logging.info(f"Step 2 開始使用 SimHash 進行近似去重 (閾值: {threshold})...")
    
    # 僅對狀態為 'Valid' 的項目計算 SimHash
    valid_items = [item for item in all_items_metadata if item['status'] == 'Valid']
    if not valid_items:
        logging.warning("沒有有效的資料可進行去重。")
        return all_items_metadata, []

    hashes = [(item['internal_id'], Simhash(item['text_for_hash'])) 
              for item in tqdm(valid_items, desc="計算 SimHash", unit="項目")]
    
    
    id_to_item_map = {item['internal_id']: item for item in all_items_metadata}
    
    hashes.sort(key=lambda x: x[1].value)
    
    ids_to_remove = set()
    deduplicated_log = []
    
    for i in tqdm(range(len(hashes) - 1), desc="比較相似度", unit="項目"):
        if hashes[i][0] in ids_to_remove: continue
        for j in range(i + 1, len(hashes)):
            if hashes[j][0] in ids_to_remove: continue
            if hashes[j][1].value - hashes[i][1].value > threshold: break
            
            distance = hashes[i][1].distance(hashes[j][1])
            if distance <= threshold:
                ids_to_remove.add(hashes[j][0])
                removed_item = id_to_item_map[hashes[j][0]]
                kept_item = id_to_item_map[hashes[i][0]]
                removed_item['status'] = 'Duplicate'
                removed_item['error_reason'] = f'SimHash Similar (distance: {distance})'
                
                deduplicated_log.append({
                    'removed_qid': removed_item.get('qid'),
                    'reason': removed_item['error_reason'],
                    'compared_with_qid': kept_item.get('qid'),
                    'removed_item_details': removed_item
                })

    logging.info(f"Step 2 finished, 標記了 {len(deduplicated_log)} 筆近似重複資料。")
    return all_items_metadata, deduplicated_log


# --- Step 3: 語言分類 (基於元數據列表進行操作) ---
def classify(all_items_metadata, simplifiedCharSet):
    """
    接收完整的元數據列表，對 'Valid' 項目進行語言分類，
    並在元數據中新增 'language' 欄位。返回更新後的列表。
    """
    logging.info("Step 3 開始進行語言分類...")
    for item in tqdm(all_items_metadata, desc="分類語言", unit="項目"):
        # 只對最終有效的數據進行分類
        if item['status'] == 'Valid':
            full_output_simp = containSimp(item.get('full_output', ''), simplifiedCharSet)
            model_response_simp = containSimp(item.get('model_response', ''), simplifiedCharSet)
            
            if full_output_simp or model_response_simp:
                item['language'] = 'Simplified'
                item['trigger_char'] = full_output_simp or model_response_simp
                item['trigger_source'] = "full_output" if full_output_simp else "model_response"
                logging.info(f"分類項目 {item['internal_id']} 為簡體中文，觸發字元: {item['trigger_char']} (來源: {item['trigger_source']})")
            else:
                item['language'] = 'Traditional'
        else:
            # 對於無效或重複的數據，語言欄位標記為不適用
            item['language'] = 'N/A'
            
    logging.info("Step 3 finished, 語言分類完成。")
    return all_items_metadata

# --- 主程式執行區 ---
if __name__ == "__main__":
    start_time = time.time()
    project_root = './eval_results' 
    
    # 1. 提取所有數據並進行初步驗證，得到一個包含所有元數據的列表
    all_data, validated_data, format_error = extract_and_validate(project_root)
    
    # 2. 對元數據列表進行去重操作，此函數會直接修改列表中重複項的狀態
    all_data, deduplicated_log= find_and_remove_duplicates(all_data, threshold=2)

    # 3. 對元數據列表進行語言分類，此函數會為有效項添加 'language' 欄位
    simplifiedCharSet = load_simpCharSet('simplified_chars.txt')
    all_data = classify(all_data, simplifiedCharSet)

    # --- 【新增步驟】基於完整的元數據列表產生並印出統計報告 ---
    generate_and_print_statistics(all_data)

    # --- 數據儲存：從 all_data 列表中篩選出需要的數據進行儲存 ---
    output_dir = "final_classified_output_new"
    os.makedirs(output_dir, exist_ok=True)
    logging.info(f"開始將最終結果儲存至 '{output_dir}' 目錄...")

    # 準備儲存 buckets
    final_classified_data = {}
    language_buckets = {'traditional': [], 'simplified': []}
    format_error_log = []

    for item in all_data:
        # 根據最終狀態將項目分發到不同的 bucket
        if item['status'] == 'Valid':
            lang_key = item['language'].lower()
            mission_key = item.get('mission', 'unknown')
            bucket_key = f"{lang_key}_{mission_key}"
            
            if bucket_key not in final_classified_data:
                final_classified_data[bucket_key] = []
            final_classified_data[bucket_key].append(item)
            
            if lang_key in language_buckets:
                language_buckets[lang_key].append(item)

        elif item['status'] == 'FormatError':
            format_error_log.append({'reason': item['error_reason'], 'format_error_item_details': item})

    # 儲存分類好的資料檔案
    for bucket_name, data in final_classified_data.items():
        if not data: continue
        filename = os.path.join(output_dir, f"{bucket_name}.json")
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False, indent=4)
        logging.info(f"成功儲存 {len(data)} 筆資料至: {filename}")
    
    # 儲存語言分類的資料
    for lang, data in language_buckets.items():
        if not data: continue
        filename = os.path.join(output_dir, f"classified_data_{lang}.json")
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False, indent=4)
        logging.info(f"成功儲存 {len(data)} 筆語言分類資料至: {filename}")
    
    # 儲存日誌檔案
    output_files = {
        'format_error.json': format_error_log,
        'deduplicated_log.json': deduplicated_log
    }
    for filename, data in output_files.items():
        if not data: continue
        filepath = os.path.join(output_dir, filename)
        with open(filepath, 'w', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False, indent=4)
        logging.info(f"成功儲存 {len(data)} 筆資料至: {filepath}")
    
    # 統計資訊
    total_original = (len(validated_data) + len(format_error))
    total_validated = len(validated_data)
    total_duplicates = len(deduplicated_log)
    total_final = sum(len(data) for data in final_classified_data.values())

    logging.info(f'''
--- 統計資料 ---
原始資料總數: {total_original}
有效資料總數: {total_validated}
重複資料總數: {total_duplicates}
繁中資料總數: {sum(len(data) for key, data in final_classified_data.items() if key.startswith('traditional'))}
簡中資料總數: {sum(len(data) for key, data in final_classified_data.items() if key.startswith('simplified'))}
格式錯誤資料總數: {len(format_error)}
最終分類資料總數: {total_final}
包括:
繁中公文: {len(final_classified_data.get('traditional_od', []))}
繁中新聞稿: {len(final_classified_data.get('traditional_press', []))}
繁中陳情: {len(final_classified_data.get('traditional_petition', []))}
繁中問答: {len(final_classified_data.get('traditional_qa', []))}
簡中公文: {len(final_classified_data.get('simplified_od', []))}
簡中新聞稿: {len(final_classified_data.get('simplified_press', []))}
簡中陳情: {len(final_classified_data.get('simplified_petition', []))}
簡中問答: {len(final_classified_data.get('simplified_qa', []))}
未分類任務: {len(final_classified_data.get('unknown_mission', []))}
''')
          
    # 最終摘要
    end_time = time.time()
    duration = end_time - start_time
    minutes, seconds = divmod(duration, 60)
    final_summary = (
        f"\n處理完成！\n"
        f"總共花費時間: {int(minutes)} 分 {seconds:.2f} 秒。\n"
        f"所有提取的資料已成功儲存。\n"
        f"詳細的處理過程與統計報告已記錄在 '{os.path.abspath(log_filepath)}' 檔案中。"
    )
    logging.info(final_summary)
    print(final_summary)