"""자사 브랜드 분석 탭. 자사 브랜드 부정 언급 분석 + AI 2차 검증 결과 + 전략적 인사이트. """ import html import pandas as pd import streamlit as st from core.api_client import ChainShiftClient from core.charts import CONFIDENCE_TIER_COLORS, EMOTION_KO, create_domain_bar_chart from core.athena_client import fetch_full_answer from core.styles import TIER_BORDER_COLORS from core.supabase_client import ( get_false_positives, get_true_negatives, ) from core.utils import ( format_brands_list, get_confidence_tier, get_feedback_reason_label, get_feedback_type_emoji, get_llm_tier_badge, highlight_evidence_spans, truncate_text, ) def render(data: dict): """자사 브랜드 분석 탭 렌더링.""" # --- Section 1: 부정 언급 분석 (from insights.py) --- st.markdown("##### 🏠 자사 브랜드 부정 언급 AI 답변") st.caption("AI가 자사 브랜드에 대해 부정적으로 언급한 답변을 자동으로 감지합니다") # --- Overview Card --- _render_overview(data) # Filters — Row 1: 플랫폼, 확신도, (CEJ), 2차 검증 has_cej = bool(data.get("cej_stats")) filter_cols = st.columns(4 if has_cej else 3) with filter_cols[0]: platform_options = ["전체"] + list(data["platform_stats"].keys()) platform_filter = st.selectbox("플랫폼", options=platform_options, key="sentiment:ih_platform") with filter_cols[1]: tier_filter = st.selectbox("확신도", options=["전체", "HIGH", "MEDIUM", "LOW"], key="sentiment:ih_tier") cej_filter = "전체" if has_cej: with filter_cols[2]: cej_options = ["전체"] + list(data["cej_stats"].keys()) cej_filter = st.selectbox("CEJ 단계", options=cej_options, key="sentiment:ih_cej") with filter_cols[-1]: llm_status_filter = st.selectbox("2차 검증", options=["전체", "정탐", "오탐", "미검증"], key="sentiment:ih_llm_status") # Filters — Row 2: 페이지 크기 (우측 정렬) _, size_col = st.columns([4, 1]) with size_col: page_size = st.selectbox("페이지 크기", options=[20, 50, 100], index=1, key="sentiment:ih_page_size") # Filter candidates filtered = data["candidates"] if platform_filter != "전체": filtered = [c for c in filtered if c.get("platform") == platform_filter] if tier_filter != "전체": filtered = [c for c in filtered if get_confidence_tier(c.get("overall_confidence"))[0] == tier_filter] if cej_filter != "전체": filtered = [c for c in filtered if c.get("cej_depth1") == cej_filter] if llm_status_filter == "정탐": filtered = [c for c in filtered if c.get("llm_verified") and c.get("llm_is_negative")] elif llm_status_filter == "오탐": filtered = [c for c in filtered if c.get("llm_verified") and not c.get("llm_is_negative")] elif llm_status_filter == "미검증": filtered = [c for c in filtered if not c.get("llm_verified")] total_all = data.get("total_nudge", len(data["candidates"])) st.markdown(f"**{len(filtered)}건** 표시 중 (전체 {total_all}건)") # Export _render_export_section(data) # Pagination total_pages = max(1, (len(filtered) + page_size - 1) // page_size) if "sentiment:ih_page" not in st.session_state: st.session_state["sentiment:ih_page"] = 1 # Reset page when filters change ih_filter_key = f"{platform_filter}_{tier_filter}_{cej_filter}_{llm_status_filter}_{page_size}" if st.session_state.get("sentiment:ih_last_filters") != ih_filter_key: st.session_state["sentiment:ih_page"] = 1 st.session_state["sentiment:ih_last_filters"] = ih_filter_key current_page = st.session_state["sentiment:ih_page"] # Pagination header (always show for consistency with other tabs) start_idx = (current_page - 1) * page_size + 1 end_idx = min(current_page * page_size, len(filtered)) if total_pages > 1: col_info, col_prev, col_page, col_next = st.columns([3, 1, 1, 1]) with col_info: st.markdown(f"**전체 {len(filtered):,}건** | 페이지 {current_page}/{total_pages} ({start_idx}-{end_idx}건)") with col_prev: if st.button("⬅️ 이전", disabled=current_page <= 1, key="sentiment:ih_prev"): st.session_state["sentiment:ih_page"] = current_page - 1 st.rerun() with col_page: new_page = st.number_input( "페이지", min_value=1, max_value=total_pages, value=current_page, label_visibility="collapsed", key="sentiment:ih_page_input", ) if new_page != current_page: st.session_state["sentiment:ih_page"] = new_page st.rerun() with col_next: if st.button("다음 ➡️", disabled=current_page >= total_pages, key="sentiment:ih_next"): st.session_state["sentiment:ih_page"] = current_page + 1 st.rerun() else: st.markdown(f"**전체 {len(filtered):,}건**") # Candidate cards (paginated) page_start = (current_page - 1) * page_size page_end = page_start + page_size for i, item in enumerate(filtered[page_start:page_end]): unique_idx = page_start + i _render_candidate_card(data, item, unique_idx) # --- Section 2: AI 2차 검증 결과 (from verification.py) --- st.markdown("---") st.markdown("##### 🤖 AI 2차 검증 결과") _render_verification_section(data) # --- Section 3: 전략적 인사이트 --- st.markdown("---") _render_strategic_insights(data) def _render_overview(data: dict): """자사 브랜드 오버뷰 카드.""" candidates = data.get("candidates", []) total = data.get("total_nudge", len(candidates)) # Extract brand names from candidates all_brands: set[str] = set() for c in candidates: for b in c.get("in_house_brands", []): all_brands.add(b) brands_display = ", ".join(sorted(all_brands)[:5]) if all_brands else "N/A" if len(all_brands) > 5: brands_display += f" 외 {len(all_brands) - 5}개" # Tier distribution tier_stats = data.get("tier_stats", {}) high = tier_stats.get("HIGH", 0) medium = tier_stats.get("MEDIUM", 0) low = tier_stats.get("LOW", 0) # LLM verification stats (from RPC, not limited by PostgREST page size) llm_stats = data.get("llm_verification_stats", {}) verified_count = llm_stats.get("verified_count", 0) tp_count = llm_stats.get("true_positive_count", 0) fp_count = llm_stats.get("false_positive_count", 0) llm_text = f"{verified_count}건 완료" if verified_count > 0: llm_text += f" (정탐 {tp_count} / 오탐 {fp_count})" st.markdown(f"""
추적 브랜드: {html.escape(brands_display)}
전체 부정 감지: {total}건 🔴 HIGH: {high} | 🟡 MEDIUM: {medium} | 🟢 LOW: {low} LLM 검증: {llm_text}
""", unsafe_allow_html=True) def _render_export_section(data: dict): """Export 영역 — inline 버튼 (경쟁사/키워드 탭과 통일).""" # Map current UI filters for export exp_platform = None exp_llm_neg = None if "sentiment:ih_platform" in st.session_state: _p = st.session_state["sentiment:ih_platform"] if _p != "전체": exp_platform = _p if "sentiment:ih_llm_status" in st.session_state: _s = st.session_state["sentiment:ih_llm_status"] if _s == "정탐": exp_llm_neg = True elif _s == "오탐": exp_llm_neg = False _, export_col = st.columns([4, 1]) with export_col: if st.button("📥 Excel 다운로드", key="sentiment:ih_export_btn"): with st.spinner("Excel 파일 생성 중..."): try: client = ChainShiftClient(api_key=data.get("api_key"), access_token=data.get("access_token")) xlsx = client.export_nudge_candidates( data["campaign_id"], include_full_answers=False, include_evidence=True, llm_verified_only=False, platform=exp_platform, llm_is_negative=exp_llm_neg, ) st.session_state["sentiment:ih_excel_data"] = xlsx st.session_state["sentiment:ih_excel_ready"] = True except Exception as e: st.error(f"다운로드 실패: {e}") if st.session_state.get("sentiment:ih_excel_ready"): st.download_button( label="💾 저장", data=st.session_state["sentiment:ih_excel_data"], file_name=f"in_house_analysis_{data['campaign_id']}.xlsx", mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", key="sentiment:ih_dl_btn", ) def _render_candidate_card(data: dict, item: dict, index: int): """Individual candidate card with LLM verification inline.""" answer_id = item.get("answer_id", "N/A") confidence = item.get("overall_confidence", 0) tier, tier_emoji, tier_desc = get_confidence_tier(confidence) tier_color = TIER_BORDER_COLORS.get(tier, "#94A3B8") raw_emotion = item.get("dominant_emotion") or "" emotion = EMOTION_KO.get(raw_emotion, raw_emotion) or "N/A" platform = item.get("platform", "N/A") brands = item.get("in_house_brands", []) # LLM verification badge llm_badge = "" if item.get("llm_verified"): if item.get("llm_is_negative"): llm_badge = '🔴 부정 확인' else: llm_badge = '🟢 부정 아님' else: llm_badge = '⏳ 미검증' st.markdown(f"""
#{answer_id} — {', '.join(brands) if brands else 'N/A'} {llm_badge} {tier}
{platform} | 😞 부정 ({emotion}) | 확신도 {confidence:.0%}
""", unsafe_allow_html=True) with st.expander(f"📖 상세 보기 — #{answer_id}", expanded=False): _render_candidate_detail(data, item, answer_id, index) def _render_candidate_detail(data: dict, item: dict, answer_id: int, index: int): """Candidate detail: question, full answer, ABSA, citations, LLM card, feedback.""" # 1. Question context question = item.get("question_content", "") if question: st.markdown(f"""
💬 질문
{html.escape(question[:500])}
""", unsafe_allow_html=True) # 2. Full answer (lazy-load from Athena) st.markdown("**🤖 AI 답변**") preview = item.get("answer_preview", "") display_answer = _load_full_answer_ih(answer_id, preview, index) # 3. Brand ABSA brand_detail = item.get("brand_sentiment_detail", {}) if brand_detail: _render_brand_absa(brand_detail) # 4. Citations _render_citations_ih(answer_id, item, index) # 5. LLM verification card (styled) if item.get("llm_verified"): llm_is_negative = item.get("llm_is_negative", False) llm_confidence = item.get("llm_confidence", 0) or 0 llm_reasoning = item.get("llm_reasoning", "") llm_evidence_spans = item.get("llm_evidence_spans", []) llm_adjusted_tier = item.get("llm_adjusted_tier") badge_text, badge_color = get_llm_tier_badge(llm_adjusted_tier, llm_is_negative) badge_bg = {"green": "#10B981", "red": "#EF4444", "orange": "#F59E0B", "blue": "#3B82F6"}.get(badge_color, "#6B7280") st.markdown("---") st.markdown(f"""
🔬 LLM 2차 검증 {badge_text}
LLM 확신도: {llm_confidence:.0%}
판단 근거: {html.escape(llm_reasoning[:500]) if llm_reasoning else 'N/A'}
""", unsafe_allow_html=True) # Evidence highlighting if llm_evidence_spans and display_answer: st.markdown("**📍 근거 문장 (하이라이트)**") highlighted_html = highlight_evidence_spans(display_answer, llm_evidence_spans) st.markdown( f'
{highlighted_html}
', unsafe_allow_html=True, ) st.caption("🔴 부정 | 🟢 긍정 | 🔵 중립 | 🟡 비교") # Per-brand LLM results breakdown per_brand_results = item.get("in_house_llm_results") or [] if len(per_brand_results) > 0: _render_per_brand_llm_results(per_brand_results) # Re-verify button if st.button("🔄 재검증 요청", key=f"sentiment:ih_reverify_{answer_id}_{index}"): with st.spinner("LLM 재검증 중..."): result = _request_llm_verification(data.get("api_key", ""), answer_id, force=True, access_token=data.get("access_token")) if result and result.get("success") and result.get("data"): st.success("재검증 완료! 페이지를 새로고침하면 반영됩니다.") st.rerun() else: st.info("아직 LLM 2차 검증이 수행되지 않았습니다.") if st.button("🔬 LLM 검증 요청", key=f"sentiment:ih_verify_{answer_id}_{index}"): with st.spinner("Gemini Pro로 검증 중... (최대 30초)"): result = _request_llm_verification(data.get("api_key", ""), answer_id, access_token=data.get("access_token")) if result and result.get("success") and result.get("data"): st.success("검증 완료!") st.rerun() # 6. Feedback _render_feedback_inline(data, item, answer_id) def _load_full_answer_ih(answer_id: int, preview: str, index: int) -> str: """Lazy-load full answer from Athena for in-house tab.""" display_answer = preview or "N/A" if answer_id and answer_id != "N/A": full_answer_key = f"sentiment:ih_full_{answer_id}_{index}" load_key = f"sentiment:ih_load_{answer_id}_{index}" if full_answer_key not in st.session_state: st.session_state[full_answer_key] = None cached = st.session_state.get(full_answer_key) is_loaded = isinstance(cached, str) and len(cached) > 0 load_full = st.checkbox( "📥 전체 답변 불러오기", key=load_key, value=is_loaded, ) if load_full and not is_loaded: with st.spinner("Athena에서 전체 답변을 가져오는 중..."): try: full_content = fetch_full_answer(answer_id) if full_content: st.session_state[full_answer_key] = full_content st.rerun() else: st.warning("답변을 찾을 수 없습니다") except Exception as e: st.warning(f"전체 답변 로드 실패: {e}") display_answer = st.session_state.get(full_answer_key) or preview or "N/A" label = "✅ 전체 답변 로드됨" if is_loaded else f"📄 미리보기 ({len(preview or '')}자)" st.caption(label) st.markdown( f'
' f'{html.escape(display_answer)}
', unsafe_allow_html=True, ) return display_answer def _render_citations_ih(answer_id: int, item: dict, index: int): """In-house tab citation rendering (Supabase citation_urls).""" st.markdown("**🔗 인용 출처**") citation_urls = item.get("citation_urls", []) or [] if citation_urls: for url in citation_urls[:5]: display_url = url[:50] + "..." if len(url) > 50 else url st.markdown(f"• [{display_url}]({url})") if len(citation_urls) > 5: st.caption(f"+{len(citation_urls) - 5}개 더...") else: st.caption("인용 소스 없음") def _render_brand_absa(brand_detail: dict): """Render ABSA results per brand.""" st.markdown("**🔍 브랜드별 감성 분석 (ABSA)**") in_house = brand_detail.get("in_house", []) # Handle dual format (list or dict) if isinstance(in_house, list): for bd in in_house: brand = bd.get("brand", "N/A") sentiment = bd.get("sentiment", "N/A") confidence = bd.get("confidence", 0) color = "#DC2626" if sentiment == "negative" else "#059669" if sentiment == "positive" else "#6B7280" st.markdown( f'{brand}: {sentiment} ({confidence:.0%})', unsafe_allow_html=True, ) elif isinstance(in_house, dict): for brand, info in in_house.items(): sentiment = info.get("sentiment", "N/A") if isinstance(info, dict) else str(info) st.markdown(f"**{brand}**: {sentiment}") def _render_per_brand_llm_results(per_brand_results: list[dict]): """Render per-brand in-house LLM verification breakdown.""" st.markdown("**🏷️ 브랜드별 LLM 검증 결과**") for r in per_brand_results: brand = r.get("brand", "Unknown") is_neg = r.get("is_negative", False) conf = r.get("confidence", 0) or 0 reasoning = r.get("reasoning", "") tier = r.get("adjusted_tier", "NONE") if is_neg: badge = f'🔴 부정 확인 ({tier})' border_color = "#DC2626" else: badge = '🟢 부정 아님' border_color = "#059669" st.markdown(f"""
{html.escape(brand)} {badge}
확신도 {conf:.0%} — {html.escape(reasoning[:200]) if reasoning else 'N/A'}
""", unsafe_allow_html=True) def _render_feedback_inline(data: dict, item: dict, answer_id: int): """Inline feedback buttons.""" _token = data.get("access_token") col1, col2, col3 = st.columns(3) with col1: if st.button("👍 정확해요", key=f"sentiment:ih_fb_ok_{answer_id}"): _submit_feedback(data.get("api_key", ""), answer_id, data["campaign_id"], "correct", access_token=_token) with col2: if st.button("👎 틀려요", key=f"sentiment:ih_fb_wrong_{answer_id}"): _submit_feedback(data.get("api_key", ""), answer_id, data["campaign_id"], "wrong", access_token=_token) with col3: if st.button("🤔 애매해요", key=f"sentiment:ih_fb_ambig_{answer_id}"): _submit_feedback(data.get("api_key", ""), answer_id, data["campaign_id"], "ambiguous", access_token=_token) def _request_llm_verification( api_key: str, answer_id: int, force: bool = False, access_token: str | None = None, ) -> dict | None: """Request LLM verification for an answer.""" try: client = ChainShiftClient(api_key=api_key, access_token=access_token) return client.verify_answer(answer_id, force=force) except Exception as e: st.error(f"LLM 검증 요청 실패: {e}") return None def _submit_feedback( api_key: str, answer_id: int, campaign_id: int, feedback_type: str, access_token: str | None = None, ): """Submit feedback.""" try: client = ChainShiftClient(api_key=api_key, access_token=access_token) client.submit_feedback(answer_id, campaign_id, feedback_type) st.success("피드백이 저장되었습니다!") except Exception as e: st.error(f"피드백 저장 실패: {e}") def _render_verification_section(data: dict): """LLM 2차 검증 통합 결과 (from verification.py).""" # Use RPC-provided stats (accurate counts, no extra queries) llm_stats = data.get("llm_verification_stats", {}) total_verified = llm_stats.get("verified_count", 0) true_positive = llm_stats.get("true_positive_count", 0) false_positive = llm_stats.get("false_positive_count", 0) if total_verified > 0: fp_rate = false_positive / total_verified * 100 col1, col2, col3 = st.columns(3) with col1: st.metric("검증 완료", f"{total_verified}건") with col2: st.metric("✅ 정탐 (True Positive)", f"{true_positive}건") with col3: st.metric("❌ 오탐 (False Positive)", f"{false_positive}건", delta=f"{fp_rate:.1f}%", delta_color="inverse") # False positive list fp_tab, tp_tab, citation_tab = st.tabs(["❌ 오탐 목록", "✅ 정탐 목록", "📎 인용 분석"]) with fp_tab: _render_false_positives(data, false_positive) with tp_tab: _render_true_negatives(data, true_positive) with citation_tab: _render_citation_analysis(data) else: st.info("LLM 2차 검증 데이터가 없습니다") def _render_false_positives(data: dict, count: int): """Show false positive cases.""" if count == 0: st.success("오탐 없음") return try: fp_list, _ = get_false_positives(data["campaign_id"], page_size=20) except Exception: fp_list = [] for item in fp_list: answer_id = item.get("answer_id", "N/A") reasoning = item.get("llm_reasoning", "") st.markdown(f"""
#{answer_id} — 오탐 확정
{html.escape(reasoning[:200])}
""", unsafe_allow_html=True) def _render_true_negatives(data: dict, count: int): """Show true positive (confirmed negative) cases.""" if count == 0: st.info("정탐 없음") return try: tn_list, _ = get_true_negatives(data["campaign_id"], page_size=20) except Exception: tn_list = [] for item in tn_list: answer_id = item.get("answer_id", "N/A") reasoning = item.get("llm_reasoning", "") st.markdown(f"""
#{answer_id} — 부정 확정
{html.escape(reasoning[:200])}
""", unsafe_allow_html=True) def _render_citation_analysis(data: dict): """Citation domain analysis.""" domain_counts = data.get("domain_counts", {}) if not domain_counts: st.info("인용 데이터가 없습니다") return st.markdown("**인용 도메인 분포**") fig = create_domain_bar_chart(domain_counts) st.plotly_chart(fig, use_container_width=True, config={"displayModeBar": False}) df = pd.DataFrame( [(d, c) for d, c in sorted(domain_counts.items(), key=lambda x: -x[1])], columns=["도메인", "인용 횟수"], ) st.dataframe(df, use_container_width=True, hide_index=True) def _render_strategic_insights(data: dict): """전략적 인사이트 (BIT + CEJ).""" st.markdown("##### 📊 전략적 인사이트") bit_stats = data.get("bit_stats", {}) cej_stats = data.get("cej_stats", {}) if bit_stats: st.markdown("**BIT 사분면 분포**") df = pd.DataFrame( [(k, v) for k, v in sorted(bit_stats.items(), key=lambda x: -x[1])], columns=["사분면", "건수"], ) st.dataframe(df, use_container_width=True, hide_index=True) if cej_stats: st.markdown("**CEJ 단계별 분포**") df = pd.DataFrame( [(k, v) for k, v in sorted(cej_stats.items(), key=lambda x: -x[1])], columns=["단계", "건수"], ) st.dataframe(df, use_container_width=True, hide_index=True)