"""키워드 분석 Excel 내보내기 — Supabase 직접 쿼리 + Athena Full context.""" import streamlit as st def render_export( data: dict, keyword: str | None = None, sentiment: str | None = None, brand_only: bool = False, no_brand: bool = False, platform: str | None = None, llm_verified: str | None = None, llm_is_negative: bool | None = None, ): """키워드 분석 Excel 다운로드 (직접 Supabase, Vercel 우회). Args: llm_is_negative: True=정탐, False=오탐, None=전체 """ # Build filter description for UI filter_parts = [] if keyword: filter_parts.append(f"키워드: {keyword}") else: filter_parts.append("키워드: 전체") if sentiment: label = {"positive": "긍정", "neutral": "중립", "negative": "부정"}.get(sentiment, sentiment) filter_parts.append(f"감성: {label}") if brand_only: filter_parts.append("브랜드 멘션만") elif no_brand: filter_parts.append("비브랜드만") if platform: filter_parts.append(f"플랫폼: {platform}") if llm_is_negative is True: filter_parts.append("2차검증: 정탐") elif llm_is_negative is False: filter_parts.append("2차검증: 오탐") elif llm_verified: label = {"verified": "검증완료", "unverified": "미검증"}.get(llm_verified, llm_verified) filter_parts.append(f"2차검증: {label}") filter_desc = " | ".join(filter_parts) if filter_parts else "전체" st.caption(f"📥 다운로드 필터: {filter_desc}") if st.button("📥 Excel 다운로드", key="sentiment:kw_export_btn"): try: xlsx = _export_keyword_direct( data["campaign_id"], keyword=keyword, sentiment=sentiment, brand_only=brand_only, no_brand=no_brand, platform=platform, llm_verified=llm_verified, llm_is_negative=llm_is_negative, ) st.session_state["sentiment:kw_excel_data"] = xlsx st.session_state["sentiment:kw_excel_ready"] = True except Exception as e: st.error(f"다운로드 실패: {e}") if st.session_state.get("sentiment:kw_excel_ready"): st.download_button( label="💾 파일 저장", data=st.session_state["sentiment:kw_excel_data"], file_name=f"keyword_analysis_{data['campaign_id']}.xlsx", mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", key="sentiment:kw_dl_btn", ) def _export_keyword_direct( campaign_id: int, keyword: str | None = None, sentiment: str | None = None, brand_only: bool = False, no_brand: bool = False, platform: str | None = None, llm_verified: str | None = None, llm_is_negative: bool | None = None, ) -> bytes: """직접 Supabase cursor 페이지네이션으로 필터링된 키워드 데이터 추출. Vercel 10초 타임아웃 우회. 필터 적용으로 대상 행 수 감소. Full context는 Athena에서 배치 fetch하여 병합. Args: llm_is_negative: True=정탐(부정확정), False=오탐(부정아님), None=전체 """ from io import BytesIO from core.supabase_client import get_supabase_client from core.athena_client import fetch_full_answers_batch sb = get_supabase_client() batch_size = 1000 export_warn_threshold = 100_000 warned = False last_id = 0 all_rows: list[dict] = [] # Cursor-based pagination (id > last_id) — O(1) per batch progress = st.progress(0, text="데이터 로딩 중...") batch_num = 0 while True: query = ( sb.table("keyword_sentiment_results") .select( "id, answer_id, keyword, matched_sentence, keyword_sentiment, keyword_confidence, " "brand_name, brand_sentiment, brand_confidence, brand_mentions, " "llm_verified, llm_sentiment, llm_is_negative, llm_confidence, llm_reason_tags, " "llm_reason_summary, llm_reasoning, " "competitor_brand_name, competitor_llm_verified, competitor_llm_sentiment, " "competitor_llm_reason_tags, competitor_llm_reason_summary, " "citation_urls, citation_count, platform, question_content, created_at" ) .eq("campaign_id", campaign_id) .gt("id", last_id) ) # Apply filters if keyword: query = query.eq("keyword", keyword) if sentiment: query = query.eq("keyword_sentiment", sentiment) if brand_only: query = query.not_.is_("brand_name", "null") elif no_brand: query = query.is_("brand_name", "null") if platform: query = query.eq("platform", platform) if llm_is_negative is not None: query = query.eq("llm_is_negative", llm_is_negative) elif llm_verified == "verified": query = query.eq("llm_verified", True) elif llm_verified == "unverified": query = query.or_("llm_verified.is.null,llm_verified.eq.false") result = query.order("id").limit(batch_size).execute() rows = result.data or [] if not rows: break all_rows.extend(rows) last_id = rows[-1]["id"] batch_num += 1 pct = min(len(all_rows) / max(len(all_rows) + batch_size, 1), 0.99) progress.progress(pct, text=f"로딩 중... {len(all_rows):,}건") if len(rows) < batch_size: break if not warned and len(all_rows) >= export_warn_threshold: st.warning(f"대용량 Export ({len(all_rows):,}건+). 완료까지 시간이 걸릴 수 있습니다.") warned = True progress.progress(1.0, text=f"{len(all_rows):,}건 로드 완료. Full context 가져오는 중...") # Batch fetch full answers from Athena unique_ids = list({row["answer_id"] for row in all_rows if row.get("answer_id")}) full_answers: dict[int, str] = {} athena_batch_size = 500 for i in range(0, len(unique_ids), athena_batch_size): chunk = unique_ids[i:i + athena_batch_size] full_answers.update(fetch_full_answers_batch(chunk)) pct = min((i + athena_batch_size) / max(len(unique_ids), 1), 0.99) progress.progress(pct, text=f"Full context 로딩... {min(i + athena_batch_size, len(unique_ids)):,}/{len(unique_ids):,} 답변") progress.progress(1.0, text=f"Full context {len(full_answers):,}건 로드 완료!") # Build Excel from openpyxl import Workbook from openpyxl.utils import get_column_letter wb = Workbook() ws = wb.active ws.title = "Keyword Sentiment" headers = [ "키워드", "매칭 문장", "Full Context", "키워드 감성", "키워드 확신도", "브랜드명", "브랜드 감성", "브랜드 확신도", "브랜드 언급", "LLM 검증", "LLM 감성", "LLM 확신도", "LLM 태그", "LLM 요약", "LLM 근거", "경쟁사 브랜드", "경쟁사 LLM 검증", "경쟁사 LLM 감성", "경쟁사 LLM 태그", "경쟁사 LLM 요약", "인용 URL", "인용 수", "플랫폼", "질문", "Answer ID", "생성일", ] ws.append(headers) progress.progress(0.0, text="Excel 생성 중...") for i, row in enumerate(all_rows): # Brand mentions brand_mentions_str = "" bm = row.get("brand_mentions") if bm and isinstance(bm, dict): parts = [] if bm.get("in_house"): parts.append(f"자사: {', '.join(bm['in_house'])}") if bm.get("competitor"): parts.append(f"경쟁사: {', '.join(bm['competitor'])}") brand_mentions_str = "; ".join(parts) # Tags (list -> comma-separated) def _tags_str(tags): if tags and isinstance(tags, list): return ", ".join(str(t) for t in tags) return "" # Citations cite_str = "" cites = row.get("citation_urls") if cites and isinstance(cites, list): cite_str = "\n".join(str(u) for u in cites[:10]) # Full context from Athena answer_id = row.get("answer_id") full_text = full_answers.get(answer_id, "") if answer_id else "" ws.append([ row.get("keyword", ""), row.get("matched_sentence", ""), full_text, row.get("keyword_sentiment", ""), row.get("keyword_confidence"), row.get("brand_name", ""), row.get("brand_sentiment", ""), row.get("brand_confidence"), brand_mentions_str, "Y" if row.get("llm_verified") else "N", row.get("llm_sentiment", ""), row.get("llm_confidence"), _tags_str(row.get("llm_reason_tags")), row.get("llm_reason_summary", ""), (row.get("llm_reasoning") or "")[:500], row.get("competitor_brand_name", ""), "Y" if row.get("competitor_llm_verified") else "N", row.get("competitor_llm_sentiment", ""), _tags_str(row.get("competitor_llm_reason_tags")), row.get("competitor_llm_reason_summary", ""), cite_str, row.get("citation_count"), row.get("platform", ""), row.get("question_content", ""), answer_id, (row.get("created_at") or "")[:19].replace("T", " "), ]) if i % 10000 == 0: progress.progress(min(i / max(len(all_rows), 1), 0.99), text=f"Excel 생성 중... {i:,}/{len(all_rows):,}") # Column widths widths = [12, 50, 80, 10, 8, 12, 10, 8, 25, 6, 10, 8, 25, 30, 40, 12, 6, 10, 25, 30, 40, 6, 10, 40, 10, 16] for i, w in enumerate(widths, 1): if i <= len(widths): ws.column_dimensions[get_column_letter(i)].width = w progress.progress(1.0, text="파일 생성 완료!") buf = BytesIO() wb.save(buf) return buf.getvalue()