freshqa/freshqa_acc.py

#!/usr/bin/env python3
"""
FreshQA 정확도 계산 스크립트

이 스크립트는 FreshQA 데이터셋의 정확도를 계산하고 다양한 카테고리별로 분석합니다.
"""

import pandas as pd
import sys
import os


def load_freshqa_data(csv_path='freshqa.csv'):
    """FreshQA CSV 파일을 로드합니다."""
    try:
        # 먼저 파일을 읽어서 구조를 확인
        temp_df = pd.read_csv(csv_path)
        # print(f"파일 구조 확인: {len(temp_df)}개 행, 컬럼: {temp_df.columns.tolist()}")
        
        # rating 컬럼이 있으면 그대로 사용, 없으면 skiprows 적용
        if 'rating' in temp_df.columns:
            fresh_qa = temp_df
            # print("rating 컬럼이 있는 파일로 인식하여 전체 데이터를 사용합니다.")
        else:
            fresh_qa = pd.read_csv(csv_path, skiprows=[0, 1])
            # print("기본 FreshQA 형식으로 인식하여 skiprows를 적용합니다.")
        
        # print(f"FreshQA 데이터 로드 완료: {len(fresh_qa)}개 샘플")
        return fresh_qa
    except FileNotFoundError:
        print(f"오류: {csv_path} 파일을 찾을 수 없습니다.")
        print("현재 디렉토리에 freshqa.csv 파일이 있는지 확인해주세요.")
        sys.exit(1)
    except Exception as e:
        print(f"데이터 로드 중 오류 발생: {e}")
        sys.exit(1)


def process_freshqa_dataframe(df):
    """DataFrame을 매개변수로 받아 FreshQA 데이터를 처리합니다."""
    try:
        # print(f"DataFrame 구조 확인: {len(df)}개 행, 컬럼: {df.columns.tolist()}")
        
        # rating 컬럼이 있으면 그대로 사용, 없으면 기본값 설정
        if 'rating' in df.columns:
            # print("DataFrame에 rating 컬럼이 있어서 그대로 사용합니다.")
            processed_df = df.copy()
        else:
            # print("DataFrame에 rating 컬럼이 없어서 기본값 0으로 설정합니다.")
            processed_df = df.copy()
            processed_df['rating'] = 0  # 기본값으로 0 설정
        
        # print(f"FreshQA 데이터 처리 완료: {len(processed_df)}개 샘플")
        return processed_df
    except Exception as e:
        print(f"데이터 처리 중 오류 발생: {e}")
        raise


def update_results(df, d_acc, d_count, field_name):
    """결과 딕셔너리를 업데이트합니다."""
    if len(df) == 0:
        r = 0.0
    else:
        # rating이 문자열 'TRUE'이거나 숫자 1인 경우를 모두 처리
        if df['rating'].dtype == 'object':
            # 문자열인 경우 'TRUE' 확인
            r = len(df[df.rating == 'TRUE']) * 100 / len(df)
        else:
            # 숫자인 경우 1 확인
            r = len(df[df.rating == 1]) * 100 / len(df)
    d_acc[field_name] = r
    d_count[field_name] = len(df)


def calculate_accuracy_simple(fresh_qa):
    """FreshQA 데이터의 기본 정확도를 계산합니다 (간단한 버전)."""
    print("정확도 계산 중...")
    
    # rating 컬럼이 없으면 기본값 0으로 설정
    if 'rating' not in fresh_qa.columns:
        # print("rating 컬럼이 없어서 기본값 0으로 설정합니다.")
        fresh_qa['rating'] = 0
    
    accs = {}
    counts = {}
    
    # 전체 정확도
    update_results(fresh_qa, accs, counts, 'overall_accuracy')
    
    # split 컬럼이 있으면 분할별 정확도 계산
    if 'split' in fresh_qa.columns:
        fresh_qa_test = fresh_qa[fresh_qa.split == 'TEST']
        fresh_qa_dev = fresh_qa[fresh_qa.split == 'DEV']
        update_results(fresh_qa_test, accs, counts, 'acc_test')
        update_results(fresh_qa_dev, accs, counts, 'acc_dev')
    
    # fact_type 컬럼이 있으면 사실 유형별 정확도 계산
    if 'fact_type' in fresh_qa.columns:
        for fact_type in ['fast-changing', 'slow-changing', 'never-changing']:
            if fact_type in fresh_qa['fact_type'].values:
                sub_df = fresh_qa[fresh_qa.fact_type == fact_type]
                update_results(sub_df, accs, counts, f'{fact_type}_accuracy')
    
    # false_premise 컬럼이 있으면 False premise 정확도 계산
    if 'false_premise' in fresh_qa.columns:
        fp_df = fresh_qa[fresh_qa.false_premise == True]
        if len(fp_df) > 0:
            update_results(fp_df, accs, counts, 'false_premise_accuracy')
    
    # domain 컬럼이 있으면 도메인별 정확도 계산
    if 'domain' in fresh_qa.columns:
        # 한국어 도메인 카테고리들 (실제 CSV 파일의 domain 값들)
        korean_domains = ['정치', '스포츠', '연예', '날씨', '세계', '경제', '사회', 'IT/과학', '생활/문화', 'UNK']
        
        for domain in korean_domains:
            if domain in fresh_qa['domain'].values:
                domain_df = fresh_qa[fresh_qa.domain == domain]
                domain_test = domain_df[domain_df.split == 'TEST']
                domain_dev = domain_df[domain_df.split == 'DEV']
                
                # 도메인명을 영어로 변환 (파일명/키에 사용)
                domain_key = domain.replace('/', '_').replace(' ', '_').lower()
                if domain == 'IT/과학':
                    domain_key = 'it_science'
                elif domain == '생활/문화':
                    domain_key = 'life_culture'
                elif domain == 'UNK':
                    domain_key = 'unknown'
                
                update_results(domain_df, accs, counts, f'acc_{domain_key}')
                update_results(domain_test, accs, counts, f'acc_test_{domain_key}')
                update_results(domain_dev, accs, counts, f'acc_dev_{domain_key}')
        
        # 기존 영어 도메인들도 유지 (호환성을 위해)
        english_domains = ['politics', 'sports', 'entertainment', 'weather', 'world', 'economy', 'society', 'it_science', 'life_culture']
        for domain in english_domains:
            if domain in fresh_qa['domain'].values:
                domain_df = fresh_qa[fresh_qa.domain == domain]
                update_results(domain_df, accs, counts, f'{domain}_accuracy')
    
    # 총 질문 수 추가
    accs['total_questions'] = len(fresh_qa)
    
    return accs

def calculate_accuracy(fresh_qa):
    """FreshQA 데이터의 정확도를 계산합니다."""

    # 데이터 분할
    fresh_qa_test = fresh_qa[fresh_qa.split == 'TEST']
    fresh_qa_dev = fresh_qa[fresh_qa.split == 'DEV']
    
    accs = {}
    counts = {}
    
    # 전체 정확도
    update_results(fresh_qa, accs, counts, 'acc')
    update_results(fresh_qa_test, accs, counts, 'acc_test')
    update_results(fresh_qa_dev, accs, counts, 'acc_dev')
    
    # 사실 유형별 정확도
    for fact_type in ['fast-changing', 'slow-changing', 'never-changing']:
        sub_df = fresh_qa[(fresh_qa.false_premise == False) & (fresh_qa.fact_type == fact_type)]
        sub_df_test = sub_df[sub_df.split == 'TEST']
        sub_df_dev = sub_df[sub_df.split == 'DEV']
        
        ft = fact_type.replace('-', '_')
        update_results(sub_df, accs, counts, f'acc_{ft}')
        update_results(sub_df_test, accs, counts, f'acc_test_{ft}')
        update_results(sub_df_dev, accs, counts, f'acc_dev_{ft}')
    
    # 질문 유형별 정확도 (vp: valid premise, fp: false premise)
    for qt in ['vp', 'fp']:
        fp = True if qt == 'fp' else False
        data = fresh_qa[(fresh_qa.false_premise == fp)]
        data_test = data[data.split == 'TEST']
        data_dev = data[data.split == 'DEV']
        
        # 홉 수별 분석
        data_one_hop = data[data.num_hops == 'one-hop']
        data_one_hop_test = data_one_hop[data_one_hop.split == 'TEST']
        data_one_hop_dev = data_one_hop[data_one_hop.split == 'DEV']
        
        data_two_hop = data[data.num_hops == 'multi-hop']
        data_two_hop_test = data_two_hop[data_two_hop.split == 'TEST']
        data_two_hop_dev = data_two_hop[data_two_hop.split == 'DEV']
        
        # 연도별 분석
        data_old = data[(data.effective_year != '2022') & (data.effective_year != '2023')]
        data_old_test = data_old[data_old.split == 'TEST']
        data_old_dev = data_old[data_old.split == 'DEV']
        
        data_new = data[(data.effective_year == '2022') | (data.effective_year == '2023')]
        data_new_test = data_new[data_new.split == 'TEST']
        data_new_dev = data_new[data_new.split == 'DEV']
        
        # 기본 정확도
        update_results(data, accs, counts, f'acc_{qt}')
        update_results(data_test, accs, counts, f'acc_test_{qt}')
        update_results(data_dev, accs, counts, f'acc_dev_{qt}')
        
        # 홉 수별 정확도
        update_results(data_one_hop, accs, counts, f'acc_{qt}_one_hop')
        update_results(data_one_hop_test, accs, counts, f'acc_test_{qt}_one_hop')
        update_results(data_one_hop_dev, accs, counts, f'acc_dev_{qt}_one_hop')
        
        update_results(data_two_hop, accs, counts, f'acc_{qt}_two_hop')
        update_results(data_two_hop_test, accs, counts, f'acc_test_{qt}_two_hop')
        update_results(data_two_hop_dev, accs, counts, f'acc_dev_{qt}_two_hop')
        
        # 연도별 정확도
        update_results(data_old, accs, counts, f'acc_{qt}_old')
        update_results(data_old_test, accs, counts, f'acc_test_{qt}_old')
        update_results(data_old_dev, accs, counts, f'acc_dev_{qt}_old')
        
        update_results(data_new, accs, counts, f'acc_{qt}_new')
        update_results(data_new_test, accs, counts, f'acc_test_{qt}_new')
        update_results(data_new_dev, accs, counts, f'acc_dev_{qt}_new')
    
    # 도메인별 정확도 계산
    if 'domain' in fresh_qa.columns:
        # 한국어 도메인 카테고리들 (실제 CSV 파일의 domain 값들)
        korean_domains = ['정치', '스포츠', '연예', '날씨', '세계', '경제', '사회', 'IT/과학', '생활/문화', 'UNK']
        # 도메인명을 영어로 변환 (파일명/키에 사용)
        domain_mapping = {
                    '정치': 'politics',
                    '스포츠': 'sports',
                    '연예': 'entertainment',
                    '날씨': 'weather',
                    '세계': 'world',
                    '경제': 'economy',
                    '사회': 'society',
                    'IT/과학': 'it_science',
                    '생활/문화': 'life_culture',
                    'UNK': 'unknown'
                }
        for domain in korean_domains:
            if domain in fresh_qa['domain'].values:
                
                domain_df = fresh_qa[fresh_qa.domain == domain]
                domain_test = domain_df[domain_df.split == 'TEST']
                domain_dev = domain_df[domain_df.split == 'DEV']
                
                domain_key = domain_mapping.get(domain, domain.replace('/', '_').replace(' ', '_').lower())
                
                update_results(domain_df, accs, counts, f'acc_{domain_key}')
                update_results(domain_test, accs, counts, f'acc_test_{domain_key}')
                update_results(domain_dev, accs, counts, f'acc_dev_{domain_key}')
    
    return accs, counts


def print_results(accs, counts):
    """결과를 보기 좋게 출력합니다."""
    print("\n" + "="*80)
    print("FreshQA 정확도 분석 결과")
    print("="*80)
    
    # 전체 정확도
    print(f"\n📊 전체 정확도:")
    print(f"  전체: {accs['acc']}% ({counts['acc']}개 샘플)")
    print(f"  테스트: {accs['acc_test']}% ({counts['acc_test']}개 샘플)")
    print(f"  개발: {accs['acc_dev']}% ({counts['acc_dev']}개 샘플)")
    
    # 사실 유형별 정확도
    print(f"\n📈 사실 유형별 정확도:")
    fact_types = {
        'fast_changing': '빠르게 변하는 사실',
        'slow_changing': '천천히 변하는 사실', 
        'never_changing': '변하지 않는 사실'
    }
    
    for key, name in fact_types.items():
        print(f"  {name}:")
        print(f"    전체: {accs[f'acc_{key}']}% ({counts[f'acc_{key}']}개 샘플)")
        print(f"    테스트: {accs[f'acc_test_{key}']}% ({counts[f'acc_test_{key}']}개 샘플)")
        print(f"    개발: {accs[f'acc_dev_{key}']}% ({counts[f'acc_dev_{key}']}개 샘플)")
    
    # 질문 유형별 정확도
    print(f"\n❓ 질문 유형별 정확도:")
    question_types = {
        'vp': '유효한 전제 (Valid Premise)',
        'fp': '잘못된 전제 (False Premise)'
    }
    
    for key, name in question_types.items():
        print(f"  {name}:")
        print(f"    전체: {accs[f'acc_{key}']}% ({counts[f'acc_{key}']}개 샘플)")
        print(f"    테스트: {accs[f'acc_test_{key}']}% ({counts[f'acc_test_{key}']}개 샘플)")
        print(f"    개발: {accs[f'acc_dev_{key}']}% ({counts[f'acc_dev_{key}']}개 샘플)")
        
        # 홉 수별
        print(f"    단일 홉: {accs[f'acc_{key}_one_hop']}% ({counts[f'acc_{key}_one_hop']}개 샘플)")
        print(f"    다중 홉: {accs[f'acc_{key}_two_hop']}% ({counts[f'acc_{key}_two_hop']}개 샘플)")
        
        # 연도별
        print(f"    오래된 데이터: {accs[f'acc_{key}_old']}% ({counts[f'acc_{key}_old']}개 샘플)")
        print(f"    최신 데이터: {accs[f'acc_{key}_new']}% ({counts[f'acc_{key}_new']}개 샘플)")
    
    # 도메인별 정확도
    print(f"\n🌐 도메인별 정확도:")
    domain_mapping = {
        'politics': '정치',
        'sports': '스포츠', 
        'entertainment': '연예',
        'weather': '날씨',
        'world': '세계',
        'economy': '경제',
        'society': '사회',
        'it_science': 'IT/과학',
        'life_culture': '생활/문화',
        'unknown': 'UNK'
    }
    
    for key, name in domain_mapping.items():
        if f'acc_{key}' in accs:
            print(f"  {name}:")
            print(f"    전체: {accs[f'acc_{key}']}% ({counts[f'acc_{key}']}개 샘플)")
            if f'acc_test_{key}' in accs:
                print(f"    테스트: {accs[f'acc_test_{key}']}% ({counts[f'acc_test_{key}']}개 샘플)")
                pass
            if f'acc_dev_{key}' in accs:
                print(f"    개발: {accs[f'acc_dev_{key}']}% ({counts[f'acc_dev_{key}']}개 샘플)")
                pass
            pass
    
    print("\n" + "="*80)


def main():
    """메인 함수"""
    print("FreshQA 정확도 계산 스크립트")
    print("="*50)
    
    # CSV 파일 경로 확인
    csv_path = 'freshqa.csv'
    if len(sys.argv) > 1:
        csv_path = sys.argv[1]
    
    if not os.path.exists(csv_path):
        print(f"오류: {csv_path} 파일을 찾을 수 없습니다.")
        print("사용법: python freshqa_acc.py [csv_file_path]")
        sys.exit(1)
    
    # 데이터 로드
    fresh_qa = load_freshqa_data(csv_path)
    
    # 정확도 계산
    accs, counts = calculate_accuracy(fresh_qa)
    
    # 결과 출력
    print_results(accs, counts)
    
    # 딕셔너리 형태로도 출력 (원본 노트북과 동일)
    print(f"\n📋 딕셔너리 형태 결과:")
    print(accs)


if __name__ == "__main__":
    main()