Spaces:

chainshiftlabs
/

chainshift-dashboard

Sleeping

chainshift-dashboard / features /sentiment /competitor.py
GitHub Action
Sync from GitHub
ef78361 7 months ago
24.9 kB
"""감성분석 경쟁사 분석 탭.
경쟁사 브랜드별 감성 분석 결과 및 부정 언급 분석.
"""
import html

import streamlit as st

from core.api_client import ChainShiftClient
from core.charts import CONFIDENCE_TIER_COLORS, EMOTION_KO, create_brand_sentiment_chart
from core.athena_client import fetch_full_answer
from core.styles import TIER_BORDER_COLORS
from core.utils import (
    get_confidence_tier,
    get_llm_tier_badge,
    highlight_evidence_spans,
    truncate_text,
)

from .data import _get_competitor_mentions


def render(data: dict):
    """경쟁사 분석 탭 렌더링."""
    st.markdown("##### 🏢 경쟁사 브랜드 부정 언급 분석")
    st.caption("AI 플랫폼에서 경쟁사 브랜드가 부정적으로 언급되는 사례를 분석합니다")

    # Initialize page state
    if "sentiment:comp_page" not in st.session_state:
        st.session_state["sentiment:comp_page"] = 1

    # Brand summary from pre-loaded data (Supabase RPC via data.py)
    brand_data = data.get("brand_data", {})
    competitor_summary = brand_data.get("competitor_summary", [])
    brand_names = ["전체"] + [b.get("brand_name", "") for b in competitor_summary if b.get("brand_name")]

    # Filters — Row 1: 감성, 플랫폼, 2차 검증, 브랜드
    f1, f2, f3, f4 = st.columns(4)
    with f1:
        polarity_filter = st.selectbox(
            "감성",
            options=["전체", "negative", "positive", "neutral"],
            format_func=lambda x: {"전체": "전체", "negative": "부정", "positive": "긍정", "neutral": "중립"}.get(x, x),
            key="sentiment:comp_polarity",
        )
    with f2:
        platform_filter = st.selectbox(
            "플랫폼",
            options=["전체", "CHATGPT", "GEMINI", "PERPLEXITY", "CLAUDE"],
            key="sentiment:comp_platform",
        )
    with f3:
        llm_filter = st.selectbox(
            "2차 검증",
            options=["전체", "정탐", "오탐", "미검증"],
            key="sentiment:comp_llm",
        )
    with f4:
        brand_filter = st.selectbox(
            "브랜드",
            options=brand_names,
            key="sentiment:comp_brand",
        )

    # Filters — Row 2: 페이지 크기 (우측 정렬)
    _, size_col = st.columns([4, 1])
    with size_col:
        page_size = st.selectbox("페이지 크기", options=[20, 50, 100], index=1, key="sentiment:comp_page_size")

    # Reset page when filter changes
    current_filters = f"{polarity_filter}_{platform_filter}_{llm_filter}_{brand_filter}_{page_size}"
    if st.session_state.get("sentiment:comp_last_filters") != current_filters:
        st.session_state["sentiment:comp_page"] = 1
        st.session_state["sentiment:comp_last_filters"] = current_filters

    current_page = st.session_state["sentiment:comp_page"]

    # Fetch competitor data with all filters (server-side via RPC)
    try:
        polarity_param = polarity_filter if polarity_filter != "전체" else None
        platform_param = platform_filter if platform_filter != "전체" else None
        brand_param = brand_filter if brand_filter != "전체" else None

        # Map 정탐/오탐/미검증 → direct bool params (server-side filtering)
        llm_verified_param: bool | None = None
        llm_is_neg_param: bool | None = None
        if llm_filter == "정탐":
            llm_verified_param = True
            llm_is_neg_param = True
        elif llm_filter == "오탐":
            llm_verified_param = True
            llm_is_neg_param = False
        elif llm_filter == "미검증":
            llm_verified_param = False

        resp_data = _get_competitor_mentions(
            "sb",
            data["campaign_id"],
            polarity=polarity_param,
            competitor_llm_verified=llm_verified_param,
            competitor_llm_is_negative=llm_is_neg_param,
            brand_name=brand_param,
            platform=platform_param,
            page=current_page,
            page_size=page_size,
        )

        recent_mentions = resp_data.get("recent_mentions", [])
        total_answers = resp_data.get("total_answers", 0)

    except Exception as e:
        st.error(f"경쟁사 데이터 로드 실패: {e}")
        return

    # Summary stats with filter info
    filter_tags = []
    if polarity_filter != "전체":
        filter_tags.append(f"감성:{polarity_filter}")
    if platform_filter != "전체":
        filter_tags.append(f"플랫폼:{platform_filter}")
    if llm_filter != "전체":
        filter_tags.append(f"LLM:{llm_filter}")
    if brand_filter != "전체":
        filter_tags.append(f"브랜드:{brand_filter}")

    # Stats and Export button row
    stat_col, export_col = st.columns([4, 1])

    with stat_col:
        if filter_tags:
            st.markdown(f"**필터 적용**: {' | '.join(filter_tags)} → **{total_answers:,}건**")
        else:
            st.markdown(f"**분석된 AI 답변**: {total_answers:,}건")

    with export_col:
        if st.button("📥 Excel 다운로드", key="sentiment:comp_export_btn"):
            with st.spinner("Excel 파일 생성 중..."):
                try:
                    client = ChainShiftClient(api_key=data.get("api_key"), access_token=data.get("access_token"))
                    # Map bool params back to string for API export endpoint
                    export_llm = None
                    if llm_verified_param is True:
                        export_llm = "verified"
                    elif llm_verified_param is False:
                        export_llm = "unverified"
                    excel_data = client.export_brand_mentions(
                        data["campaign_id"],
                        brand_type="competitor",
                        polarity=polarity_param,
                        llm_verified=export_llm,
                        brand_name=brand_param,
                    )
                    st.session_state["sentiment:comp_excel_data"] = excel_data
                    st.session_state["sentiment:comp_excel_ready"] = True
                except Exception as e:
                    st.error(f"Excel 생성 실패: {e}")

        # Download button if data is ready
        if st.session_state.get("sentiment:comp_excel_ready"):
            st.download_button(
                label="💾 저장",
                data=st.session_state["sentiment:comp_excel_data"],
                file_name=f"competitor_mentions_{data['campaign_id']}.xlsx",
                mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
                key="sentiment:comp_dl_btn",
            )

    if not competitor_summary and not recent_mentions:
        st.info("경쟁사 브랜드 언급 데이터가 없습니다")
        return

    # Brand summary cards (always shows ALL data for context)
    if competitor_summary:
        st.markdown("---")
        st.markdown("##### 🏢 경쟁사 브랜드 요약")
        st.caption("📊 전체 데이터 기준 (필터 미적용)")

        # Create columns for brand cards (max 3 per row)
        for i in range(0, len(competitor_summary), 3):
            cols = st.columns(3)
            for j, col in enumerate(cols):
                if i + j < len(competitor_summary):
                    brand = competitor_summary[i + j]
                    with col:
                        _render_brand_summary_card(brand)

        # Brand sentiment comparison chart
        st.markdown("---")
        st.markdown("##### 📊 경쟁사 브랜드별 감성 비교")
        if len(competitor_summary) > 0:
            fig = create_brand_sentiment_chart(competitor_summary)
            st.plotly_chart(fig, use_container_width=True, config={"displayModeBar": False})

    # Recent mentions list
    st.markdown("---")
    st.markdown("##### 📋 경쟁사 언급 목록")

    # Calculate pagination info
    total_pages = max(1, (total_answers + page_size - 1) // page_size)
    start_idx = (current_page - 1) * page_size + 1
    end_idx = min(current_page * page_size, total_answers)

    # Pagination header
    col_info, col_prev, col_page, col_next = st.columns([3, 1, 1, 1])

    with col_info:
        st.markdown(f"**전체 {total_answers:,}건** | 페이지 {current_page}/{total_pages} ({start_idx}-{end_idx}건)")

    with col_prev:
        if st.button("⬅️ 이전", disabled=current_page <= 1, key="sentiment:comp_prev"):
            st.session_state["sentiment:comp_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:comp_page_input",
        )
        if new_page != current_page:
            st.session_state["sentiment:comp_page"] = new_page
            st.rerun()

    with col_next:
        if st.button("다음 ➡️", disabled=current_page >= total_pages, key="sentiment:comp_next"):
            st.session_state["sentiment:comp_page"] = current_page + 1
            st.rerun()

    if not recent_mentions:
        st.info("현재 페이지에 표시할 데이터가 없습니다.")
        return

    # Render mention cards - use unique index for each card
    for i, item in enumerate(recent_mentions):
        # Create unique index combining page and position to avoid key collisions
        unique_idx = (current_page - 1) * page_size + i
        _render_mention_card(data, item, unique_idx)


def _render_brand_summary_card(brand: dict):
    """브랜드 요약 카드 렌더링."""
    brand_name = brand.get("brand_name", "Unknown")
    total_mentions = brand.get("total_mentions", 0)
    positive_count = brand.get("positive_count", 0)
    negative_count = brand.get("negative_count", 0)
    neutral_count = brand.get("neutral_count", 0)
    positive_rate = brand.get("positive_rate", 0)
    negative_rate = brand.get("negative_rate", 0)
    neutral_rate = round(neutral_count / total_mentions * 100, 1) if total_mentions > 0 else 0.0

    # Determine primary sentiment color
    if negative_rate > positive_rate:
        bg_color = "#FEF2F2"  # Light red
        border_color = "#FCA5A5"
    elif positive_rate > negative_rate:
        bg_color = "#F0FDF4"  # Light green
        border_color = "#86EFAC"
    else:
        bg_color = "#FEF3C7"  # Light yellow
        border_color = "#FCD34D"

    # Build optional lines
    extra_lines = []

    aliases = brand.get("aliases", [])
    if aliases:
        alias_str = ", ".join(aliases[:4])
        if len(aliases) > 4:
            alias_str += f" 외 {len(aliases) - 4}개"
        extra_lines.append(f'<div style="font-size:11px;color:#9CA3AF;margin-bottom:8px;">{alias_str}</div>')

    verified_count = brand.get("llm_verified_count", 0)
    if verified_count > 0:
        verified_rate = (verified_count / total_mentions * 100) if total_mentions > 0 else 0
        extra_lines.append(f'<div style="font-size:12px;color:#6B7280;margin-top:8px;">LLM 검증: {verified_count}건 ({verified_rate:.0f}%)</div>')

    alias_html = extra_lines[0] if aliases else ""
    llm_html = extra_lines[-1] if verified_count > 0 else ""

    card_html = (
        f'<div style="background:{bg_color};border:2px solid {border_color};border-radius:12px;padding:16px;margin:8px 0;">'
        f'<div style="font-weight:bold;font-size:18px;margin-bottom:4px;">{brand_name}</div>'
        f'{alias_html}'
        f'<div style="font-size:14px;color:#374151;margin-bottom:8px;">총 언급: <strong>{total_mentions:,}건</strong></div>'
        f'<div style="display:flex;gap:8px;flex-wrap:wrap;font-size:13px;">'
        f'<span style="background:#10B981;color:white;padding:2px 8px;border-radius:4px;">긍정 {positive_count}건 ({positive_rate:.1f}%)</span>'
        f'<span style="background:#6B7280;color:white;padding:2px 8px;border-radius:4px;">중립 {neutral_count}건 ({neutral_rate:.1f}%)</span>'
        f'<span style="background:#EF4444;color:white;padding:2px 8px;border-radius:4px;">부정 {negative_count}건 ({negative_rate:.1f}%)</span>'
        f'</div>'
        f'{llm_html}'
        f'</div>'
    )
    st.markdown(card_html, unsafe_allow_html=True)


def _render_mention_card(data: dict, item: dict, index: int):
    """개별 언급 카드 렌더링."""
    # Extract data
    polarity = item.get("overall_polarity", "neutral")
    confidence = item.get("overall_confidence", 0) or 0
    tier, emoji, tier_desc = get_confidence_tier(confidence)

    platform = item.get("platform", "N/A")
    question = item.get("question_content", "")
    answer = item.get("answer_content") or item.get("answer_preview") or ""
    # BrandMention already has brand_name field for the specific brand
    brand_name = item.get("brand_name", "")
    # For display, show the main brand from this mention
    competitor_brands = [brand_name] if brand_name else []

    # Polarity styling
    polarity_colors = {
        "negative": ("#FEF2F2", "#EF4444", "😞 부정"),
        "positive": ("#F0FDF4", "#10B981", "😊 긍정"),
        "neutral": ("#F5F5F4", "#6B7280", "😐 중립"),
    }
    bg_color, accent_color, polarity_label = polarity_colors.get(polarity, polarity_colors["neutral"])

    tier_color = CONFIDENCE_TIER_COLORS.get(tier, "#6B7280")
    border_color = TIER_BORDER_COLORS.get(tier, "#6B7280")

    answer_id = item.get("answer_id")
    question_display = html.escape(truncate_text(question, 200))
    answer_short = html.escape(truncate_text(answer, 150))
    brands_display = ", ".join(competitor_brands[:3]) if competitor_brands else "N/A"

    # LLM verification status (flat DB fields from get_nudge_export_data RPC)
    llm_verified = item.get("competitor_llm_verified", False)
    if llm_verified:
        llm_is_neg = item.get("competitor_llm_is_negative", False)
        if llm_is_neg:
            llm_badge = "🔴 부정 확인"
            llm_badge_color = "#DC2626"
        else:
            llm_badge = "🟢 부정 아님"
            llm_badge_color = "#059669"
    else:
        llm_badge = "⏳ 미검증"
        llm_badge_color = "#F59E0B"

    # Card header — left border strip style (matches in_house tab)
    header_html = f'''
    <div style="border-left: 3px solid {border_color}; padding: 8px 12px; margin: 4px 0;
                background: #F8FAFC; border-radius: 0 8px 8px 0;">
        <div style="display:flex; justify-content:space-between; align-items:center;">
            <span style="font-weight:600;">#{answer_id or index+1} — {brands_display}</span>
            <span>
                <span style="background:{llm_badge_color};color:white;padding:2px 6px;border-radius:4px;font-size:11px;">{llm_badge}</span>
                <span style="background:{CONFIDENCE_TIER_COLORS.get(tier,'#94A3B8')};color:white;padding:2px 6px;border-radius:4px;font-size:11px;">{tier}</span>
            </span>
        </div>
        <div style="font-size:12px;color:#6B7280;margin-top:4px;">
            {platform} | {polarity_label}{f" ({EMOTION_KO.get(item.get('dominant_emotion', ''), item.get('dominant_emotion', ''))})" if item.get("dominant_emotion") else ""} | 확신도 {confidence:.0%}
        </div>
    </div>
    '''
    st.markdown(header_html, unsafe_allow_html=True)

    # Expander for full details
    with st.expander(f"📖 상세 보기 — #{answer_id or index+1}"):
        _render_mention_detail(data, item, answer_id, answer, index)


def _render_mention_detail(data: dict, item: dict, answer_id: int | None, answer: str, index: int):
    """언급 상세 정보 렌더링."""
    confidence = item.get("overall_confidence", 0) or 0
    tier, _, _ = get_confidence_tier(confidence)

    # 1. Question context
    question = item.get("question_content", "")
    if question:
        st.markdown(f"""
<div style="border-left: 4px solid #4338CA; background: #EEF2FF; padding: 10px 12px;
            border-radius: 0 8px 8px 0; margin-bottom: 8px;">
    <div style="font-size: 11px; color: #4338CA; margin-bottom: 2px;">💬 질문</div>
    <div style="font-size: 14px; color: #1E1B4B;">{html.escape(question[:500])}</div>
</div>
""", unsafe_allow_html=True)

    # 2. AI 답변
    st.markdown("**🤖 AI 답변**")
    display_answer = _load_full_answer(answer_id, answer, index)

    # 3. 감성 분석 (ABSA)
    brand_detail = item.get("brand_sentiments") or {}
    if brand_detail and isinstance(brand_detail, dict):
        _render_competitor_absa(brand_detail)

    # 4. 인용 출처
    _render_citations(answer_id, item, index)

    # 5. LLM 2차 검증
    st.markdown("---")
    _render_llm_verification(data, item, answer_id, display_answer, index)


def _load_full_answer(answer_id: int | None, answer: str, index: int) -> str:
    """전체 답변 로드."""
    display_answer = answer or "N/A"

    if answer_id:
        full_answer_key = f"sentiment:comp_full_{answer_id}_{index}"
        load_full_key = f"sentiment:comp_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_full_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 answer or "N/A"
        label = "✅ 전체 답변 로드됨" if is_loaded else f"📄 미리보기 ({len(answer or '')}자)"
        st.caption(label)

    st.markdown(
        f'<div style="background: #FEF2F2; padding: 12px; border-radius: 8px; font-size: 14px; '
        f'white-space: pre-wrap; word-break: break-word; max-height: 400px; overflow-y: auto;">'
        f'{html.escape(display_answer)}</div>',
        unsafe_allow_html=True
    )

    return display_answer


def _render_competitor_absa(brand_detail: dict):
    """경쟁사 ABSA 결과 렌더링."""
    st.markdown("**🔍 브랜드별 감성 분석 (ABSA)**")

    # Parse competitor ABSA results
    competitor_data = brand_detail.get("competitor", [])
    competitor_absa = []
    if isinstance(competitor_data, list):
        competitor_absa = competitor_data
    elif isinstance(competitor_data, dict):
        competitor_absa = competitor_data.get("absa_results", [])

    if competitor_absa:
        for absa in competitor_absa:
            if isinstance(absa, dict):
                brand_name = absa.get("brand", "Unknown")
                sentiment = absa.get("sentiment", "N/A")
                conf = absa.get("confidence", 0)
                absa_tier, absa_emoji, _ = get_confidence_tier(conf)
                sent_color = "#10B981" if sentiment == "positive" else "#EF4444" if sentiment == "negative" else "#6B7280"
                st.markdown(
                    f'<span style="background: {sent_color}; color: white; padding: 2px 8px; border-radius: 4px; font-size: 12px; margin-right: 8px;">'
                    f'{sentiment}</span> <strong>{brand_name}</strong> (🏢 경쟁사) - {absa_emoji} 확신도 {conf:.0%} ({absa_tier})',
                    unsafe_allow_html=True
                )
    else:
        st.caption("ABSA 분석 결과 없음")


def _render_citations(answer_id: int | None, item: dict, index: int):
    """인용 출처 렌더링 (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_llm_verification(data: dict, item: dict, answer_id: int | None, display_answer: str, index: int):
    """LLM 검증 섹션 렌더링."""
    # Read from flat DB fields (raw Supabase row, not nested API dict)
    llm_verified = item.get("competitor_llm_verified", False)
    llm_is_negative = item.get("competitor_llm_is_negative")
    llm_confidence = item.get("competitor_llm_confidence")
    llm_evidence_spans = item.get("competitor_llm_evidence_spans") or []
    llm_reasoning = item.get("competitor_llm_reasoning") or ""
    llm_adjusted_tier = item.get("competitor_llm_adjusted_tier")

    verify_key = f"sentiment:comp_verify_{answer_id}_{index}"
    if verify_key not in st.session_state:
        st.session_state[verify_key] = None

    if llm_verified or st.session_state.get(verify_key):
        verify_data = st.session_state.get(verify_key) or {
            "is_negative": llm_is_negative,
            "confidence": llm_confidence,
            "evidence_spans": llm_evidence_spans,
            "reasoning": llm_reasoning,
            "adjusted_tier": llm_adjusted_tier,
        }

        badge_text, badge_color = get_llm_tier_badge(
            verify_data.get("adjusted_tier"),
            verify_data.get("is_negative")
        )
        llm_conf = verify_data.get("confidence", 0) or 0

        st.markdown(f"""
<div style="background: #F0FDF4; border: 1px solid #86EFAC; border-radius: 8px; padding: 12px; margin: 8px 0;">
    <div style="display: flex; justify-content: space-between; align-items: center; margin-bottom: 8px;">
        <span style="font-weight: bold;">🔬 LLM 2차 검증</span>
        <span style="background: {"#10B981" if badge_color == "green" else "#EF4444" if badge_color == "red" else "#F59E0B"}; color: white; padding: 4px 12px; border-radius: 20px; font-size: 12px;">{badge_text}</span>
    </div>
    <div style="font-size: 13px; color: #374151;">
        <strong>LLM 확신도:</strong> {llm_conf:.0%}<br>
        <strong>판단 근거:</strong> {html.escape(verify_data.get("reasoning", "N/A"))}
    </div>
</div>
""", unsafe_allow_html=True)

        # Evidence spans
        evidence_spans = verify_data.get("evidence_spans", [])
        if evidence_spans and display_answer:
            st.markdown("**📍 근거 문장 (하이라이트)**")
            highlighted_html = highlight_evidence_spans(display_answer, evidence_spans)
            st.markdown(
                f'<div style="background: #FFFBEB; padding: 12px; border-radius: 8px; font-size: 13px; '
                f'white-space: pre-wrap; max-height: 300px; overflow-y: auto;">{highlighted_html}</div>',
                unsafe_allow_html=True
            )
            st.caption("🔴 부정 | 🟢 긍정 | 🔵 중립 | 🟡 비교")

        # Re-verify button
        if st.button("🔄 재검증 요청", key=f"sentiment:comp_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"):
                    verified = result["data"].get("verified")
                    st.session_state[verify_key] = verified
                    st.rerun()
    else:
        st.info("아직 LLM 2차 검증이 수행되지 않았습니다.")
        if st.button("🔬 LLM 검증 요청", key=f"sentiment:comp_verify_req_{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"):
                    verified = result["data"].get("verified")
                    st.session_state[verify_key] = verified
                    st.success("검증 완료!")
                    st.rerun()


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