""" Export Utilities - 통합 내보내기 컴포넌트 공통 내보내기 기능을 제공합니다: - CSV/Excel 변환 - 필터링 옵션 - 전체 답변 포함 옵션 """ import pandas as pd import streamlit as st from io import BytesIO from typing import Callable from .supabase_client import get_sentiment_data_for_export from .athena_client import fetch_full_answers_batch # openpyxl 설치 여부 확인 (Excel export용) try: import openpyxl EXCEL_AVAILABLE = True except ImportError: EXCEL_AVAILABLE = False # LLM 검증 상태 라벨 (용어 통일) LLM_STATUS_LABELS = { "all": "전체", "verified": "검증완료", "false_positive": "오탐 (부정→비부정)", # 부정 아님 "true_negative": "정탐 (부정 확정)", # 부정 확정 "unverified": "미검증", } POLARITY_LABELS = { "all": "전체", "negative": "부정", "positive": "긍정", "neutral": "중립", } def prepare_dataframe_for_export( data: list[dict], include_full_answers: bool = False, ) -> pd.DataFrame: """데이터를 DataFrame으로 변환하고 내보내기용으로 정리합니다. Args: data: 내보낼 데이터 리스트 include_full_answers: 전체 답변 포함 여부 Returns: 정리된 DataFrame """ if not data: return pd.DataFrame() df = pd.DataFrame(data) # 리스트 컬럼을 문자열로 변환 list_columns = ['in_house_brands', 'mentioned_brands', 'llm_evidence_spans'] for col in list_columns: if col in df.columns: df[col] = df[col].apply( lambda x: ', '.join(x) if isinstance(x, list) else str(x) if x else '' ) # 컬럼 순서 정리 - answer_full을 answer_preview 다음에 배치 if 'answer_full' in df.columns and 'answer_preview' in df.columns: cols = list(df.columns) cols.remove('answer_full') idx = cols.index('answer_preview') + 1 cols.insert(idx, 'answer_full') df = df[cols] return df def export_to_csv(df: pd.DataFrame) -> bytes: """DataFrame을 CSV 바이트로 변환합니다.""" return df.to_csv(index=False).encode('utf-8-sig') def export_to_excel(df: pd.DataFrame) -> bytes | None: """DataFrame을 Excel 바이트로 변환합니다. Returns: Excel 바이트 데이터, 또는 openpyxl이 없으면 None """ if not EXCEL_AVAILABLE: return None output = BytesIO() with pd.ExcelWriter(output, engine='openpyxl') as writer: df.to_excel(writer, index=False, sheet_name='Data') return output.getvalue() def render_export_component( campaign_id: int, key_prefix: str, title: str = "📥 데이터 내보내기", show_polarity_filter: bool = True, show_llm_filter: bool = True, default_polarity: str = "negative", default_llm_status: str = "all", in_house_only: bool = True, ): """통합 내보내기 컴포넌트를 렌더링합니다. Args: campaign_id: 캠페인 ID key_prefix: Streamlit 위젯 키 접두사 (중복 방지) title: 섹션 제목 show_polarity_filter: 감정 필터 표시 여부 show_llm_filter: LLM 상태 필터 표시 여부 default_polarity: 기본 감정 필터 값 default_llm_status: 기본 LLM 상태 필터 값 in_house_only: 자사 브랜드만 필터링 """ with st.expander(title, expanded=False): # 필터 옵션 filter_col1, filter_col2 = st.columns(2) with filter_col1: if show_polarity_filter: polarity_options = list(POLARITY_LABELS.keys()) polarity_labels = list(POLARITY_LABELS.values()) default_idx = polarity_options.index(default_polarity) if default_polarity in polarity_options else 0 selected_polarity = st.selectbox( "감정 필터", options=polarity_options, format_func=lambda x: POLARITY_LABELS[x], index=default_idx, key=f"{key_prefix}_polarity" ) else: selected_polarity = default_polarity with filter_col2: if show_llm_filter: llm_options = list(LLM_STATUS_LABELS.keys()) default_idx = llm_options.index(default_llm_status) if default_llm_status in llm_options else 0 selected_llm_status = st.selectbox( "LLM 검증 상태", options=llm_options, format_func=lambda x: LLM_STATUS_LABELS[x], index=default_idx, key=f"{key_prefix}_llm_status" ) else: selected_llm_status = default_llm_status # 내보내기 옵션 opt_col1, opt_col2 = st.columns(2) with opt_col1: include_full_answers = st.checkbox( "전체 답변 포함", value=False, help="Athena에서 전체 답변을 가져옵니다 (파일 크기 증가)", key=f"{key_prefix}_full_answers" ) with opt_col2: include_evidence = st.checkbox( "LLM 근거 포함", value=False, help="LLM 판단 근거(reasoning, evidence_spans)를 포함합니다", key=f"{key_prefix}_evidence" ) st.markdown("---") # 다운로드 버튼 btn_col1, btn_col2, btn_col3 = st.columns([1, 1, 2]) # 데이터 가져오기 data = get_sentiment_data_for_export( campaign_id=campaign_id, polarity=selected_polarity if selected_polarity != "all" else None, llm_status=selected_llm_status if selected_llm_status != "all" else None, in_house_only=in_house_only, include_full_answers=include_full_answers, include_evidence=include_evidence, ) if data: df = prepare_dataframe_for_export(data, include_full_answers) count = len(df) with btn_col1: csv_data = export_to_csv(df) st.download_button( label=f"📥 CSV ({count}건)", data=csv_data, file_name=f"campaign_{campaign_id}_export_{count}건.csv", mime="text/csv", key=f"{key_prefix}_csv_download" ) with btn_col2: if EXCEL_AVAILABLE: excel_data = export_to_excel(df) st.download_button( label=f"📥 Excel ({count}건)", data=excel_data, file_name=f"campaign_{campaign_id}_export_{count}건.xlsx", mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", key=f"{key_prefix}_excel_download" ) else: st.caption("Excel: openpyxl 필요") with btn_col3: st.caption(f"총 {count}건 | 필터: {POLARITY_LABELS.get(selected_polarity, '전체')} / {LLM_STATUS_LABELS.get(selected_llm_status, '전체')}") else: st.info("내보낼 데이터가 없습니다.")