"""감성분석 경쟁사 분석 탭.
경쟁사 브랜드별 감성 분석 결과 및 부정 언급 분석.
"""
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'{alias_str}
')
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'LLM 검증: {verified_count}건 ({verified_rate:.0f}%)
')
alias_html = extra_lines[0] if aliases else ""
llm_html = extra_lines[-1] if verified_count > 0 else ""
card_html = (
f''
f'
{brand_name}
'
f'{alias_html}'
f'
총 언급: {total_mentions:,}건
'
f'
'
f'긍정 {positive_count}건 ({positive_rate:.1f}%)'
f'중립 {neutral_count}건 ({neutral_rate:.1f}%)'
f'부정 {negative_count}건 ({negative_rate:.1f}%)'
f'
'
f'{llm_html}'
f'
#{answer_id or index+1} — {brands_display}
{llm_badge}
{tier}
{platform} | {polarity_label}{f" ({EMOTION_KO.get(item.get('dominant_emotion', ''), item.get('dominant_emotion', ''))})" if item.get("dominant_emotion") else ""} | 확신도 {confidence:.0%}
'''
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"""
💬 질문
{html.escape(question[:500])}
""", 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''
f'{html.escape(display_answer)}
',
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''
f'{sentiment} {brand_name} (🏢 경쟁사) - {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"""
🔬 LLM 2차 검증
{badge_text}
LLM 확신도: {llm_conf:.0%}
판단 근거: {html.escape(verify_data.get("reasoning", "N/A"))}
""", 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'{highlighted_html}
',
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