"""계층 분석 진행 현황 + 분석 결과 탭. 진행 현황: Supabase 직접 쿼리 (Vercel timeout 회피) 분석 결과: API 호출 (질문 데이터) """ import streamlit as st import pandas as pd from core.api_client import ChainShiftClient from core.job_realtime import ( get_active_hierarchy_jobs, get_recent_hierarchy_jobs, get_hierarchy_step_label, format_job_duration, get_status_emoji, get_status_label, ) # ============================================================================ # 진행 현황 탭 # ============================================================================ def render_active(base_ctx: dict): """진행 현황 탭 렌더링.""" st.markdown("##### ⏳ 진행 현황") st.caption("대기 중이거나 처리 중인 계층 분석 Job을 확인합니다") if st.button("🔄 새로고침", key="hier:refresh_active"): st.rerun() try: active_jobs = get_active_hierarchy_jobs(limit=5) except Exception as e: st.warning(f"Job 목록 로드 실패: {e}") active_jobs = [] if not active_jobs: st.info("현재 진행 중인 계층 분석 Job이 없습니다.") return for job in active_jobs: _render_active_card(job) def _render_active_card(job: dict): """활성 Job 카드 렌더링.""" progress = job.get("progress", 0) status = job.get("status", "") step = job.get("current_step") step_label = get_hierarchy_step_label(step) status_emoji = get_status_emoji(status) status_label = get_status_label(status) duration = format_job_duration(job) prompt = job.get("prompt", "") title = job.get("title") or prompt[:30] st.markdown(f"""
{status_emoji} {title}
{progress}%
소요시간: {duration}
{status_label} · {step_label}
키워드: {prompt}
""", unsafe_allow_html=True) st.progress(progress / 100) # ============================================================================ # 분석 결과 탭 # ============================================================================ def render_history(base_ctx: dict): """분석 결과 탭 렌더링.""" st.markdown("##### 📋 분석 결과") st.caption("완료된 계층 분석 Job 이력과 결과를 확인합니다") if st.button("🔄 새로고침", key="hier:refresh_history"): st.rerun() try: jobs = get_recent_hierarchy_jobs(limit=20) except Exception as e: st.warning(f"Job 이력 로드 실패: {e}") jobs = [] if not jobs: st.info("아직 계층 분석 이력이 없습니다.") return # Job 이력 테이블 rows = [] for j in jobs: status = j.get("status", "") emoji = get_status_emoji(status) label = get_status_label(status) duration = format_job_duration(j) rows.append({ "상태": f"{emoji} {label}", "제목": j.get("title") or "-", "키워드": j.get("prompt", "")[:30], "진행률": f"{j.get('progress', 0)}%", "소요시간": 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) # 완료된 Job expander: 질문 통계 + 샘플 completed_jobs = [j for j in jobs if j.get("status") == "completed"] if not completed_jobs: return st.markdown("---") st.markdown("###### 완료된 분석 상세") if not base_ctx.get("api_key") and not base_ctx.get("access_token"): st.caption("질문 상세 보기는 인증이 필요합니다.") return client = ChainShiftClient( api_key=base_ctx.get("api_key"), access_token=base_ctx.get("access_token"), ) for job in completed_jobs[:5]: job_id = job["id"] title = job.get("title") or job.get("prompt", "")[:30] created = (job.get("created_at") or "")[:10] with st.expander(f"{title} ({created})", expanded=False): _render_job_detail(client, job_id) def _render_job_detail(client: ChainShiftClient, job_id: str): """완료된 Job 상세: 질문 통계 + 샘플 질문.""" # 질문 통계 try: stats_resp = client.get_hierarchy_question_stats(job_id) stats = (stats_resp or {}).get("data") or {} except Exception as e: st.warning(f"통계 로드 실패: {e}") stats = {} if stats: total = stats.get("total_questions", 0) brand_count = stats.get("brand_mention_count", 0) persona_count = stats.get("persona_included_count", 0) col1, col2, col3 = st.columns(3) with col1: st.metric("총 질문", f"{total:,}개") with col2: st.metric("브랜드 포함", f"{brand_count:,}개") with col3: st.metric("페르소나 적용", f"{persona_count:,}개") # 여정별 분포 by_depth1 = stats.get("by_journey_depth1") or {} if by_depth1: st.markdown("**여정 유형별 분포**") depth1_labels = { "awareness_comparison": "인지/비교", "purchase": "구매", "post_purchase": "구매 후", } depth1_rows = [ {"여정": depth1_labels.get(k, k), "질문 수": v} for k, v in by_depth1.items() ] st.dataframe( pd.DataFrame(depth1_rows), use_container_width=True, hide_index=True, ) # 샘플 질문 10개 try: q_resp = client.get_hierarchy_questions(job_id, page=1, page_size=10) questions = ((q_resp or {}).get("data") or {}).get("items") or [] except Exception: questions = [] if questions: st.markdown("**샘플 질문 (최대 10개)**") for i, q in enumerate(questions, 1): journey = q.get("journey_depth2", "") question_text = q.get("question", "") brand = " 🏷️" if q.get("brand_mention") else "" st.markdown(f"{i}. [{journey}] {question_text}{brand}") elif stats: st.caption("질문 데이터가 아직 없습니다.")