"""감성분석 실행 탭. Job 관리, Pre-flight 체크, 파이프라인 시각화. """ import streamlit as st import pandas as pd from core.api_client import ChainShiftClient from core.job_realtime import ( get_active_jobs, get_recent_jobs, format_job_duration, get_status_emoji, get_status_label, ) def render(data: dict): """실행 탭 렌더링.""" st.markdown("##### 🚀 감성분석 실행") st.caption("캠페인의 감성분석 Job을 시작하고 진행 상황을 확인합니다") if not data or (not data.get("api_key") and not data.get("access_token")): st.warning("인증 정보가 설정되지 않았습니다.") return analysis_client = ChainShiftClient(api_key=data.get("api_key"), access_token=data.get("access_token")) # --- Job Status --- if st.button("🔄 새로고침", key="sentiment:manual_refresh_jobs"): st.rerun() try: active_jobs = get_active_jobs(campaign_id=data["campaign_id"], limit=5) recent_jobs = get_recent_jobs(campaign_id=data["campaign_id"], limit=20) active_ids = {j["id"] for j in active_jobs} jobs_list = active_jobs + [j for j in recent_jobs if j["id"] not in active_ids] except Exception as e: active_jobs = [] jobs_list = [] st.warning(f"Job 목록 로드 실패: {e}") has_active = len(active_jobs) > 0 # --- Pipeline Visualization --- st.markdown("---") st.markdown("###### 📊 데이터 파이프라인") _render_pipeline(data) # --- Brand Status --- _render_brand_status(analysis_client, data.get("campaign_id")) # --- Current status + Start button --- st.markdown("---") st.markdown("###### ▶️ 분석 실행") if has_active: _render_active_job(active_jobs[0], analysis_client) else: _render_inactive_state(jobs_list, analysis_client, data.get("campaign_id")) # --- Job history --- st.markdown("---") st.markdown("###### 📋 Job 이력") if jobs_list: _render_job_history(jobs_list) else: st.caption("Job 이력이 없습니다.") def _fetch_brands(client: ChainShiftClient, campaign_id: int) -> list[dict]: """캠페인 브랜드 목록 조회 (session_state 캐시).""" cache_key = f"campaign_brands_{campaign_id}" if cache_key in st.session_state: return st.session_state[cache_key] try: brands = client.get_campaign_brands(campaign_id) st.session_state[cache_key] = brands return brands except Exception: return [] def _render_brand_status(client: ChainShiftClient, campaign_id: int): """캠페인 브랜드 현황 표시.""" brands = _fetch_brands(client, campaign_id) if not brands: return in_house = [b for b in brands if b.get("brand_type") in ("PRIMARY", "USER")] competitor = [b for b in brands if b.get("brand_type") == "SECONDARY"] with st.expander(f"🏷️ 캠페인 브랜드 현황 (자사 {len(in_house)}개 / 경쟁사 {len(competitor)}개)", expanded=False): col1, col2 = st.columns(2) with col1: st.markdown(f"**자사 브랜드** ({len(in_house)}개)") if in_house: for b in in_house: synonyms = b.get("synonyms") or [] syn_text = f" \n유사어: {', '.join(synonyms)}" if synonyms else "" st.markdown(f"- **{b['name']}**{syn_text}") else: st.caption("등록된 자사 브랜드 없음") with col2: st.markdown(f"**경쟁사 브랜드** ({len(competitor)}개)") if competitor: for b in competitor: synonyms = b.get("synonyms") or [] syn_text = f" \n유사어: {', '.join(synonyms)}" if synonyms else "" st.markdown(f"- **{b['name']}**{syn_text}") else: st.caption("등록된 경쟁사 브랜드 없음") def _render_pipeline(data: dict): """파이프라인 시각화.""" pipe_col1, pipe_col2, pipe_col3, pipe_col4 = st.columns(4) overview_total = data.get("overview_total_answers") or 0 overview_ih_neg = data.get("overview_nudge_candidates") or 0 overview_llm_done = data.get("overview_llm_verified") or 0 fp_rate = data.get("overview_false_positive_rate") or 0 overview_llm_confirmed = overview_llm_done - int(overview_llm_done * fp_rate) with pipe_col1: st.markdown(f"""
📥
1. 데이터 수집
{overview_total:,}
AI 답변
""", unsafe_allow_html=True) with pipe_col2: ih_rate = (overview_ih_neg / overview_total * 100) if overview_total > 0 else 0 st.markdown(f"""
🔍
2. DeBERTa 분석
{overview_ih_neg:,}
부정 감지 ({ih_rate:.1f}%)
""", unsafe_allow_html=True) with pipe_col3: verify_rate = (overview_llm_done / overview_ih_neg * 100) if overview_ih_neg > 0 else 0 st.markdown(f"""
🤖
3. LLM 검증
{overview_llm_done:,}
완료 ({verify_rate:.0f}%)
""", unsafe_allow_html=True) with pipe_col4: confirm_rate = (overview_llm_confirmed / overview_llm_done * 100) if overview_llm_done > 0 else 0 st.markdown(f"""
🎯
4. 정탐
{overview_llm_confirmed:,}
정탐률 {confirm_rate:.1f}%
""", unsafe_allow_html=True) def _render_active_job(active: dict, client: ChainShiftClient): """활성 Job 렌더링.""" progress = active.get("progress", 0) status = active.get("status", "") status_emoji = get_status_emoji(status) status_label = get_status_label(status) duration = format_job_duration(active) message = active.get("message", "처리 중...") total_answers = active.get("total_answers", 0) processed = active.get("processed_answers", 0) st.markdown(f"""
{status_emoji} {status_label}
{progress}%
소요시간: {duration}
{message}
처리: {processed:,} / {total_answers:,} 답변
""", unsafe_allow_html=True) st.progress(progress / 100) if st.button("⛔ 분석 취소", key="sentiment:cancel_job", type="secondary"): try: client.cancel_analysis_job(active["id"]) st.success("취소 요청 완료") st.rerun() except Exception as e: st.error(f"취소 실패: {e}") def _render_inactive_state(jobs_list: list, client: ChainShiftClient, campaign_id: int): """비활성 상태 렌더링.""" # --- 최근 분석 상태 --- if jobs_list: latest = jobs_list[0] latest_status = latest.get("status", "") latest_emoji = get_status_emoji(latest_status) latest_label = get_status_label(latest_status) latest_duration = format_job_duration(latest) completed_at = latest.get("completed_at") or latest.get("created_at") or "" if completed_at: completed_at = completed_at[:19].replace("T", " ") if latest_status == "completed": st.success(f"{latest_emoji} 최근 분석: **{latest_label}** (소요: {latest_duration}, {completed_at})") elif latest_status == "failed": st.error(f"{latest_emoji} 최근 분석: **{latest_label}** - {(latest.get('error_message') or '알 수 없는 오류')[:50]}") else: st.info(f"{latest_emoji} 최근 분석: **{latest_label}** ({completed_at})") else: st.info("아직 실행된 분석이 없습니다.") # --- 분석 설정 --- st.markdown("###### 분석 설정") col1, col2 = st.columns(2) with col1: run_brand = st.checkbox("자사/경쟁사 감성 분석", value=True, key="run:brand") with col2: run_keyword = st.checkbox("키워드 감성 분석", value=False, key="run:keyword") # 키워드 입력 (keyword scope 선택 시) keywords = [] if run_keyword: keywords_input = st.text_input( "분석 키워드 (쉼표 구분)", key="run:keywords", placeholder="사료, 소화, 알러지", ) keywords = [k.strip() for k in keywords_input.split(",") if k.strip()] # LLM 검증 옵션 include_llm = st.checkbox( "2차 LLM 검증 포함", value=False, key="run:llm", help="부정 감지 결과를 LLM으로 교차 검증합니다 (시간 추가)", ) # 분석 시작 버튼 (keyword 선택 시 키워드 입력 필수) can_start = run_brand or (run_keyword and len(keywords) > 0) if st.button( "▶️ 분석 시작", type="primary", key="sentiment:start_analysis", disabled=not can_start, ): scope = [] if run_brand: scope.append("brand") if run_keyword: scope.append("keyword") options = { "scope": scope, "keywords": keywords if run_keyword else [], "include_llm_verification": include_llm, } try: client.start_analysis_job(campaign_id, options=options) st.success("분석 Job이 생성되었습니다!") st.rerun() except Exception as e: st.error(f"분석 시작 실패: {e}") # --- 연구 분석 스캐폴딩 --- st.markdown("---") st.markdown("###### 연구 분석 (준비 중)") st.info( "연구 분석은 데이터팀 Athena 테이블 세팅 완료 후 사용 가능합니다.\n" "필요 테이블: `fanouts` (S3 스냅샷 미포함)" ) def _render_job_history(jobs_list: list): """Job 이력 테이블.""" rows = [] for j in jobs_list: status = j.get("status", "") status_emoji = get_status_emoji(status) status_label = get_status_label(status) duration = format_job_duration(j) total = j.get("total_answers", 0) nudge = j.get("nudge_candidates", 0) rows.append({ "상태": f"{status_emoji} {status_label}", "진행률": f"{j.get('progress', 0)}%", "처리량": f"{total:,}건" if total else "-", "넛지 후보": f"{nudge:,}건" if nudge else "-", "소요시간": duration, "생성일": (j.get("created_at") or "")[:19].replace("T", " "), "ID": (j.get("id") or "")[:8], }) st.dataframe(pd.DataFrame(rows), use_container_width=True, hide_index=True)