Spaces:

chainshiftlabs
/

chainshift-dashboard

Sleeping

chainshift-dashboard / features /reports /utils.py

GitHub Action

Sync from GitHub

ef78361 7 months ago

21.7 kB

"""리포트 탭 공통 유틸리티.
Feature별 미리보기, HTML 생성, CSV 변환 등 공통 함수.
"""
import io
import csv

import pandas as pd
import requests
import streamlit as st
import streamlit.components.v1 as components

from core.api_client import ChainShiftClient


def render_feature_section(
    client: ChainShiftClient,
    campaign_id: int,
    feature_key: str,
    title: str,
    description: str,
    start_date: str,
    end_date: str,
    api_key: str = "",
    access_token: str = "",
):
    """단일 Feature 섹션 렌더링."""
    html_state_key = f"html_content_{feature_key}_{campaign_id}"
    insights_key = f"insights_enabled_{feature_key}_{campaign_id}"

    with st.container(border=True):
        # Header
        c1, c2 = st.columns([4, 1])
        with c1:
            st.markdown(f"**{title}**")
            st.caption(description)

        # Preview Section (Lazy loaded)
        with st.expander(f"👁️ 미리보기", expanded=False):
            try:
                result = get_report_feature_data(api_key, campaign_id, feature_key, start_date, end_date, access_token=access_token)
                if result.get("success"):
                    data = result.get("data", {})
                    render_feature_preview(feature_key, data)
                else:
                    st.warning(f"데이터 로드 실패: {result.get('error', 'Unknown')}")
            except Exception as e:
                st.error(f"미리보기 오류: {e}")

        # LLM Insights checkbox
        enable_insights = st.checkbox(
            "🤖 LLM 인사이트 포함",
            value=st.session_state.get(insights_key, True),
            key=f"insights_cb_{feature_key}",
            help="컨설턴트 톤의 분석 코멘트를 추가합니다",
        )
        st.session_state[insights_key] = enable_insights

        # Generated HTML display section
        if html_state_key in st.session_state:
            html_data = st.session_state[html_state_key]
            html_content = html_data.get("content", "")
            html_url = html_data.get("url", "")

            st.success(f"✅ HTML 리포트 생성 완료" + (" (LLM 인사이트 포함)" if html_data.get("insights") else ""))

            if html_content:
                # Action buttons
                col_open, col_dl, col_csv, col_reset = st.columns(4)

                with col_open:
                    if html_url:
                        st.link_button("🔗 새 창에서 보기", html_url, use_container_width=True)
                    else:
                        st.button("🔗 새 창에서 보기", disabled=True, use_container_width=True, key=f"html_open_{feature_key}_disabled")

                with col_dl:
                    st.download_button(
                        label="📥 HTML 다운로드",
                        data=b'\xef\xbb\xbf' + html_content.lstrip('\ufeff').encode("utf-8"),
                        file_name=f"{feature_key}_{campaign_id}_{start_date}_{end_date}.html",
                        mime="text/html; charset=utf-8",
                        use_container_width=True,
                        key=f"html_dl_{feature_key}",
                    )

                with col_csv:
                    try:
                        result = get_report_feature_data(api_key, campaign_id, feature_key, start_date, end_date, access_token=access_token)
                        if result.get("success"):
                            csv_data = convert_report_data_to_csv(feature_key, result.get("data", {}))
                            st.download_button(
                                label="📊 CSV 다운로드",
                                data=csv_data.encode("utf-8-sig"),
                                file_name=f"{feature_key}_{campaign_id}_{start_date}_{end_date}.csv",
                                mime="text/csv",
                                use_container_width=True,
                                key=f"csv_{feature_key}_post",
                            )
                        else:
                            st.button("📊 CSV 다운로드", disabled=True, use_container_width=True, key=f"csv_{feature_key}_post_disabled")
                    except Exception:
                        st.button("📊 CSV 다운로드", disabled=True, use_container_width=True, key=f"csv_{feature_key}_post_error")

                with col_reset:
                    if st.button("🔄 다시 생성", key=f"html_reset_{feature_key}", use_container_width=True):
                        del st.session_state[html_state_key]
                        st.rerun()

                # Inline preview
                with st.expander("👁️ HTML 미리보기", expanded=False):
                    components.html(html_content, height=500, scrolling=True)

            elif html_url:
                # Fallback: HTML download failed, show direct link
                col_link, col_csv, col_reset = st.columns(3)
                with col_link:
                    st.link_button("🔗 리포트 열기 (외부 링크)", html_url, use_container_width=True)
                with col_csv:
                    try:
                        result = get_report_feature_data(api_key, campaign_id, feature_key, start_date, end_date, access_token=access_token)
                        if result.get("success"):
                            csv_data = convert_report_data_to_csv(feature_key, result.get("data", {}))
                            st.download_button(
                                label="📊 CSV 다운로드",
                                data=csv_data.encode("utf-8-sig"),
                                file_name=f"{feature_key}_{campaign_id}_{start_date}_{end_date}.csv",
                                mime="text/csv",
                                use_container_width=True,
                                key=f"csv_{feature_key}_fallback",
                            )
                        else:
                            st.button("📊 CSV 다운로드", disabled=True, use_container_width=True, key=f"csv_{feature_key}_fallback_disabled")
                    except Exception:
                        st.button("📊 CSV 다운로드", disabled=True, use_container_width=True, key=f"csv_{feature_key}_fallback_error")
                with col_reset:
                    if st.button("🔄 다시 생성", key=f"html_reset_{feature_key}", use_container_width=True):
                        del st.session_state[html_state_key]
                        st.rerun()

        else:
            # Generate button
            col_html, col_csv = st.columns(2)

            with col_html:
                if st.button(f"📄 HTML 생성", key=f"html_{feature_key}", use_container_width=True):
                    spinner_text = "HTML 생성 중..." + (" (LLM 인사이트 포함)" if enable_insights else "")
                    with st.spinner(spinner_text):
                        try:
                            # Use url mode to avoid Vercel 4.5MB response limit.
                            # Download HTML from Supabase Storage directly.
                            result_url = client.generate_html_report(
                                campaign_id=campaign_id,
                                start_date=start_date,
                                end_date=end_date,
                                features=[feature_key],
                                enable_insights=enable_insights,
                                output_mode="url",
                            )
                            if result_url.get("success"):
                                data = result_url.get("data") or {}
                                html_url = data.get("html_url", "") if isinstance(data, dict) else ""
                                html_content = ""
                                if html_url:
                                    try:
                                        dl_resp = requests.get(html_url, timeout=30)
                                        dl_resp.raise_for_status()
                                        dl_resp.encoding = "utf-8"
                                        html_content = dl_resp.text
                                    except Exception as dl_err:
                                        st.warning(f"HTML 다운로드 실패, URL 링크로 대체: {dl_err}")
                                if not html_url and not html_content:
                                    st.error("HTML 생성 실패: 스토리지 URL이 반환되지 않았습니다.")
                                else:
                                    st.session_state[html_state_key] = {
                                        "content": html_content,
                                        "url": html_url,
                                        "insights": enable_insights,
                                    }
                                    st.rerun()
                            else:
                                st.error("HTML 생성 실패: " + str(result_url.get("error", "Unknown")))
                        except Exception as e:
                            st.error(f"오류: {e}")

            with col_csv:
                try:
                    result = get_report_feature_data(api_key, campaign_id, feature_key, start_date, end_date, access_token=access_token)
                    if result.get("success"):
                        csv_data = convert_report_data_to_csv(feature_key, result.get("data", {}))
                        st.download_button(
                            label="📊 CSV 다운로드",
                            data=csv_data.encode("utf-8-sig"),
                            file_name=f"{feature_key}_{campaign_id}_{start_date}_{end_date}.csv",
                            mime="text/csv",
                            use_container_width=True,
                            key=f"csv_{feature_key}",
                        )
                    else:
                        st.button("📊 CSV 다운로드", disabled=True, use_container_width=True, key=f"csv_{feature_key}_disabled")
                except Exception:
                    st.button("📊 CSV 다운로드", disabled=True, use_container_width=True, key=f"csv_{feature_key}_error")


def render_feature_preview(feature_key: str, data: dict):
    """Feature별 미리보기 시각화."""
    if feature_key == "overview":
        cols = st.columns(4)
        with cols[0]:
            st.metric("총 질문 수", data.get("total_tasks", 0))
        with cols[1]:
            st.metric("총 답변 수", data.get("total_answers", 0))
        with cols[2]:
            st.metric("가시성", f"{data.get('overall_visibility_pct', 0):.1f}%")
        with cols[3]:
            dr = data.get("date_range", {})
            period = f"{dr.get('start', '?')} ~ {dr.get('end', '?')}"
            st.metric("분석 기간", period[:20])

    elif feature_key == "visibility":
        platforms = data.get("platforms", [])
        if platforms:
            rows = []
            for p in platforms:
                for b in p.get("brands", []):
                    rows.append({
                        "플랫폼": p.get("platform", ""),
                        "브랜드": b.get("brand_name", ""),
                        "가시성 (%)": b.get("visibility_pct", 0),
                    })
            if rows:
                df = pd.DataFrame(rows)
                st.dataframe(df, use_container_width=True, hide_index=True)
        else:
            st.info("플랫폼 데이터 없음")

    elif feature_key == "citations":
        sources = data.get("sources", [])[:10]
        if sources:
            df = pd.DataFrame(sources)
            cols = [c for c in ["source_host_url", "total_citations", "pct_of_total"] if c in df.columns]
            if cols:
                st.dataframe(df[cols], use_container_width=True, hide_index=True)
        else:
            st.info("인용 데이터 없음")

    elif feature_key == "citation-trends":
        sources = data.get("sources", [])
        if sources:
            rows = []
            for s in sources:
                for pt in s.get("trend", []):
                    rows.append({
                        "date": pt.get("task_date", ""),
                        "source": s.get("source_host_url", ""),
                        "citations": pt.get("citation_count", 0),
                    })
            if rows:
                df = pd.DataFrame(rows)
                pivot = df.pivot_table(index="date", columns="source", values="citations", aggfunc="sum").fillna(0)
                st.line_chart(pivot)
        else:
            st.info("시계열 데이터 없음")

    elif feature_key == "content-types":
        types = data.get("content_types", [])
        if types:
            df = pd.DataFrame(types)
            if "content_type" in df.columns and "total_citations" in df.columns:
                st.bar_chart(df.set_index("content_type")["total_citations"])
        else:
            st.info("콘텐츠 유형 데이터 없음")

    elif feature_key == "sentiment":
        in_house = data.get("in_house_brands", [])
        competitor = data.get("competitor_brands", [])

        if in_house:
            st.markdown("**🏢 자사 브랜드**")
            df_ih = pd.DataFrame(in_house)
            cols_ih = ["brand_name", "total_mentions", "positive_rate", "negative_rate"]
            cols_ih = [c for c in cols_ih if c in df_ih.columns]
            if cols_ih:
                st.dataframe(df_ih[cols_ih], use_container_width=True, hide_index=True)

        if competitor:
            st.markdown("**🎯 경쟁사 브랜드**")
            df_comp = pd.DataFrame(competitor)
            cols_comp = ["brand_name", "total_mentions", "positive_rate", "negative_rate"]
            cols_comp = [c for c in cols_comp if c in df_comp.columns]
            if cols_comp:
                st.dataframe(df_comp[cols_comp], use_container_width=True, hide_index=True)

        if not in_house and not competitor:
            brands = data.get("brands", [])
            if brands:
                df = pd.DataFrame(brands)
                cols = [c for c in ["brand_name", "brand_type", "positive_rate", "negative_rate"] if c in df.columns]
                if cols:
                    st.dataframe(df[cols], use_container_width=True, hide_index=True)
            else:
                st.info("감정 분석 데이터 없음")

    elif feature_key == "homepage-citations":
        daily_data = data.get("daily_data", [])[:10]
        if daily_data:
            rows = []
            for day in daily_data:
                for entry in day.get("entries", []):
                    rows.append({
                        "날짜": day.get("task_date", ""),
                        "플랫폼": entry.get("platform", ""),
                        "인용 횟수": entry.get("citation_count", 0),
                    })
            if rows:
                df = pd.DataFrame(rows)
                st.dataframe(df, use_container_width=True, hide_index=True)
        else:
            st.info("홈페이지 인용 데이터 없음")


@st.cache_data(ttl=300)
def get_report_feature_data(
    api_key: str,
    campaign_id: int,
    feature: str,
    start_date: str | None = None,
    end_date: str | None = None,
    access_token: str = "",
):
    """Fetch report feature data with caching."""
    client = ChainShiftClient(api_key=api_key or None, access_token=access_token or None)

    if feature == "overview":
        return client.get_report_overview(campaign_id, start_date, end_date)
    elif feature == "visibility":
        return client.get_report_visibility(campaign_id, start_date, end_date)
    elif feature == "citations":
        return client.get_report_citations(campaign_id, start_date, end_date, limit=50)
    elif feature == "citation-trends":
        return client.get_report_citation_trends(campaign_id, start_date, end_date)
    elif feature == "content-types":
        return client.get_report_content_types(campaign_id, start_date, end_date)
    elif feature == "sentiment":
        return client.get_report_sentiment(campaign_id)
    elif feature == "homepage-citations":
        return client.get_report_homepage_citations(campaign_id, start_date, end_date)
    else:
        return {"success": False, "error": f"Unknown feature: {feature}"}


def convert_report_data_to_csv(feature: str, data: dict) -> str:
    """Convert report feature data to CSV format."""
    output = io.StringIO()
    writer = csv.writer(output)

    if feature == "overview":
        dr = data.get("date_range", {})
        writer.writerow(["항목", "값"])
        writer.writerow(["캠페인 ID", data.get("campaign_id", "")])
        writer.writerow(["분석 기간", f"{dr.get('start', '')} ~ {dr.get('end', '')}"])
        writer.writerow(["총 질문 수", data.get("total_tasks", 0)])
        writer.writerow(["총 답변 수", data.get("total_answers", 0)])
        writer.writerow(["가시성 비율 (%)", data.get("overall_visibility_pct", 0)])

    elif feature == "visibility":
        writer.writerow(["플랫폼", "브랜드", "유형", "가시성 (%)", "브랜드 언급 수", "총 답변 수"])
        for platform in data.get("platforms", []):
            for brand in platform.get("brands", []):
                writer.writerow([
                    platform.get("platform", ""),
                    brand.get("brand_name", ""),
                    brand.get("brand_type", ""),
                    brand.get("visibility_pct", 0),
                    brand.get("brand_mentions", 0),
                    platform.get("total_answers", 0),
                ])

    elif feature == "citations":
        writer.writerow(["도메인", "유형", "인용 횟수", "답변 언급 수", "비율 (%)"])
        for item in data.get("sources", []):
            writer.writerow([
                item.get("source_host_url", ""),
                item.get("source_host_type", ""),
                item.get("total_citations", 0),
                item.get("total_answer_mentions", 0),
                item.get("pct_of_total", 0),
            ])

    elif feature == "citation-trends":
        writer.writerow(["인용 출처", "유형", "날짜", "인용 횟수", "답변 언급 수", "비율 (%)"])
        for source in data.get("sources", []):
            host = source.get("source_host_url", "")
            host_type = source.get("source_host_type", "")
            for point in source.get("trend", []):
                writer.writerow([
                    host,
                    host_type,
                    point.get("task_date", ""),
                    point.get("citation_count", 0),
                    point.get("answer_mention_count", 0),
                    point.get("citation_pct", 0),
                ])

    elif feature == "content-types":
        writer.writerow(["콘텐츠 유형", "인용 횟수", "답변 언급 수", "비율 (%)"])
        for item in data.get("content_types", []):
            writer.writerow([
                item.get("content_type", ""),
                item.get("total_citations", 0),
                item.get("total_answer_mentions", 0),
                item.get("pct_of_total", 0),
            ])

    elif feature == "sentiment":
        writer.writerow(["브랜드", "유형", "총 멘션", "긍정 %", "부정 %", "중립 %"])

        for item in data.get("in_house_brands", []):
            t = item.get("total_mentions", 0)
            neutral = round(item.get("neutral_count", 0) / t * 100, 1) if t > 0 else 0.0
            writer.writerow([
                item.get("brand_name", ""),
                "자사",
                t,
                f"{item.get('positive_rate', 0):.1f}",
                f"{item.get('negative_rate', 0):.1f}",
                f"{neutral:.1f}",
            ])

        for item in data.get("competitor_brands", []):
            t = item.get("total_mentions", 0)
            neutral = round(item.get("neutral_count", 0) / t * 100, 1) if t > 0 else 0.0
            writer.writerow([
                item.get("brand_name", ""),
                "경쟁사",
                item.get("total_mentions", 0),
                f"{item.get('positive_rate', 0):.1f}",
                f"{item.get('negative_rate', 0):.1f}",
                f"{neutral:.1f}",
            ])

        if not data.get("in_house_brands") and not data.get("competitor_brands"):
            for item in data.get("brands", []):
                pos = item.get("positive_rate", item.get("positive", 0))
                neg = item.get("negative_rate", item.get("negative", 0))
                neutral = 100 - pos - neg
                writer.writerow([
                    item.get("brand_name", item.get("name", "")),
                    item.get("brand_type", item.get("type", "")),
                    item.get("total_mentions", 0),
                    f"{pos:.1f}",
                    f"{neg:.1f}",
                    f"{neutral:.1f}",
                ])

    elif feature == "homepage-citations":
        writer.writerow(["날짜", "플랫폼", "인용 출처", "인용 횟수", "답변 언급 수"])
        for day in data.get("daily_data", []):
            task_date = day.get("task_date", "")
            for entry in day.get("entries", []):
                writer.writerow([
                    task_date,
                    entry.get("platform", ""),
                    entry.get("source_host_url", ""),
                    entry.get("citation_count", 0),
                    entry.get("answer_mention_count", 0),
                ])

    return output.getvalue()

	"""리포트 탭 공통 유틸리티.

	Feature별 미리보기, HTML 생성, CSV 변환 등 공통 함수.
	"""
	import io
	import csv

	import pandas as pd
	import requests
	import streamlit as st
	import streamlit.components.v1 as components

	from core.api_client import ChainShiftClient


	def render_feature_section(
	client: ChainShiftClient,
	campaign_id: int,
	feature_key: str,
	title: str,
	description: str,
	start_date: str,
	end_date: str,
	api_key: str = "",
	access_token: str = "",
	):
	"""단일 Feature 섹션 렌더링."""
	html_state_key = f"html_content_{feature_key}_{campaign_id}"
	insights_key = f"insights_enabled_{feature_key}_{campaign_id}"

	with st.container(border=True):
	# Header
	c1, c2 = st.columns([4, 1])
	with c1:
	st.markdown(f"{title}")
	st.caption(description)

	# Preview Section (Lazy loaded)
	with st.expander(f"👁️ 미리보기", expanded=False):
	try:
	result = get_report_feature_data(api_key, campaign_id, feature_key, start_date, end_date, access_token=access_token)
	if result.get("success"):
	data = result.get("data", {})
	render_feature_preview(feature_key, data)
	else:
	st.warning(f"데이터 로드 실패: {result.get('error', 'Unknown')}")
	except Exception as e:
	st.error(f"미리보기 오류: {e}")

	# LLM Insights checkbox
	enable_insights = st.checkbox(
	"🤖 LLM 인사이트 포함",
	value=st.session_state.get(insights_key, True),
	key=f"insights_cb_{feature_key}",
	help="컨설턴트 톤의 분석 코멘트를 추가합니다",
	)
	st.session_state[insights_key] = enable_insights

	# Generated HTML display section
	if html_state_key in st.session_state:
	html_data = st.session_state[html_state_key]
	html_content = html_data.get("content", "")
	html_url = html_data.get("url", "")

	st.success(f"✅ HTML 리포트 생성 완료" + (" (LLM 인사이트 포함)" if html_data.get("insights") else ""))

	if html_content:
	# Action buttons
	col_open, col_dl, col_csv, col_reset = st.columns(4)

	with col_open:
	if html_url:
	st.link_button("🔗 새 창에서 보기", html_url, use_container_width=True)
	else:
	st.button("🔗 새 창에서 보기", disabled=True, use_container_width=True, key=f"html_open_{feature_key}_disabled")

	with col_dl:
	st.download_button(
	label="📥 HTML 다운로드",
	data=b'\xef\xbb\xbf' + html_content.lstrip('\ufeff').encode("utf-8"),
	file_name=f"{feature_key}_{campaign_id}_{start_date}_{end_date}.html",
	mime="text/html; charset=utf-8",
	use_container_width=True,
	key=f"html_dl_{feature_key}",
	)

	with col_csv:
	try:
	result = get_report_feature_data(api_key, campaign_id, feature_key, start_date, end_date, access_token=access_token)
	if result.get("success"):
	csv_data = convert_report_data_to_csv(feature_key, result.get("data", {}))
	st.download_button(
	label="📊 CSV 다운로드",
	data=csv_data.encode("utf-8-sig"),
	file_name=f"{feature_key}_{campaign_id}_{start_date}_{end_date}.csv",
	mime="text/csv",
	use_container_width=True,
	key=f"csv_{feature_key}_post",
	)
	else:
	st.button("📊 CSV 다운로드", disabled=True, use_container_width=True, key=f"csv_{feature_key}_post_disabled")
	except Exception:
	st.button("📊 CSV 다운로드", disabled=True, use_container_width=True, key=f"csv_{feature_key}_post_error")

	with col_reset:
	if st.button("🔄 다시 생성", key=f"html_reset_{feature_key}", use_container_width=True):
	del st.session_state[html_state_key]
	st.rerun()

	# Inline preview
	with st.expander("👁️ HTML 미리보기", expanded=False):
	components.html(html_content, height=500, scrolling=True)

	elif html_url:
	# Fallback: HTML download failed, show direct link
	col_link, col_csv, col_reset = st.columns(3)
	with col_link:
	st.link_button("🔗 리포트 열기 (외부 링크)", html_url, use_container_width=True)
	with col_csv:
	try:
	result = get_report_feature_data(api_key, campaign_id, feature_key, start_date, end_date, access_token=access_token)
	if result.get("success"):
	csv_data = convert_report_data_to_csv(feature_key, result.get("data", {}))
	st.download_button(
	label="📊 CSV 다운로드",
	data=csv_data.encode("utf-8-sig"),
	file_name=f"{feature_key}_{campaign_id}_{start_date}_{end_date}.csv",
	mime="text/csv",
	use_container_width=True,
	key=f"csv_{feature_key}_fallback",
	)
	else:
	st.button("📊 CSV 다운로드", disabled=True, use_container_width=True, key=f"csv_{feature_key}_fallback_disabled")
	except Exception:
	st.button("📊 CSV 다운로드", disabled=True, use_container_width=True, key=f"csv_{feature_key}_fallback_error")
	with col_reset:
	if st.button("🔄 다시 생성", key=f"html_reset_{feature_key}", use_container_width=True):
	del st.session_state[html_state_key]
	st.rerun()

	else:
	# Generate button
	col_html, col_csv = st.columns(2)

	with col_html:
	if st.button(f"📄 HTML 생성", key=f"html_{feature_key}", use_container_width=True):
	spinner_text = "HTML 생성 중..." + (" (LLM 인사이트 포함)" if enable_insights else "")
	with st.spinner(spinner_text):
	try:
	# Use url mode to avoid Vercel 4.5MB response limit.
	# Download HTML from Supabase Storage directly.
	result_url = client.generate_html_report(
	campaign_id=campaign_id,
	start_date=start_date,
	end_date=end_date,
	features=[feature_key],
	enable_insights=enable_insights,
	output_mode="url",
	)
	if result_url.get("success"):
	data = result_url.get("data") or {}
	html_url = data.get("html_url", "") if isinstance(data, dict) else ""
	html_content = ""
	if html_url:
	try:
	dl_resp = requests.get(html_url, timeout=30)
	dl_resp.raise_for_status()
	dl_resp.encoding = "utf-8"
	html_content = dl_resp.text
	except Exception as dl_err:
	st.warning(f"HTML 다운로드 실패, URL 링크로 대체: {dl_err}")
	if not html_url and not html_content:
	st.error("HTML 생성 실패: 스토리지 URL이 반환되지 않았습니다.")
	else:
	st.session_state[html_state_key] = {
	"content": html_content,
	"url": html_url,
	"insights": enable_insights,
	}
	st.rerun()
	else:
	st.error("HTML 생성 실패: " + str(result_url.get("error", "Unknown")))
	except Exception as e:
	st.error(f"오류: {e}")

	with col_csv:
	try:
	result = get_report_feature_data(api_key, campaign_id, feature_key, start_date, end_date, access_token=access_token)
	if result.get("success"):
	csv_data = convert_report_data_to_csv(feature_key, result.get("data", {}))
	st.download_button(
	label="📊 CSV 다운로드",
	data=csv_data.encode("utf-8-sig"),
	file_name=f"{feature_key}_{campaign_id}_{start_date}_{end_date}.csv",
	mime="text/csv",
	use_container_width=True,
	key=f"csv_{feature_key}",
	)
	else:
	st.button("📊 CSV 다운로드", disabled=True, use_container_width=True, key=f"csv_{feature_key}_disabled")
	except Exception:
	st.button("📊 CSV 다운로드", disabled=True, use_container_width=True, key=f"csv_{feature_key}_error")


	def render_feature_preview(feature_key: str, data: dict):
	"""Feature별 미리보기 시각화."""
	if feature_key == "overview":
	cols = st.columns(4)
	with cols[0]:
	st.metric("총 질문 수", data.get("total_tasks", 0))
	with cols[1]:
	st.metric("총 답변 수", data.get("total_answers", 0))
	with cols[2]:
	st.metric("가시성", f"{data.get('overall_visibility_pct', 0):.1f}%")
	with cols[3]:
	dr = data.get("date_range", {})
	period = f"{dr.get('start', '?')} ~ {dr.get('end', '?')}"
	st.metric("분석 기간", period[:20])

	elif feature_key == "visibility":
	platforms = data.get("platforms", [])
	if platforms:
	rows = []
	for p in platforms:
	for b in p.get("brands", []):
	rows.append({
	"플랫폼": p.get("platform", ""),
	"브랜드": b.get("brand_name", ""),
	"가시성 (%)": b.get("visibility_pct", 0),
	})
	if rows:
	df = pd.DataFrame(rows)
	st.dataframe(df, use_container_width=True, hide_index=True)
	else:
	st.info("플랫폼 데이터 없음")

	elif feature_key == "citations":
	sources = data.get("sources", [])[:10]
	if sources:
	df = pd.DataFrame(sources)
	cols = [c for c in ["source_host_url", "total_citations", "pct_of_total"] if c in df.columns]
	if cols:
	st.dataframe(df[cols], use_container_width=True, hide_index=True)
	else:
	st.info("인용 데이터 없음")

	elif feature_key == "citation-trends":
	sources = data.get("sources", [])
	if sources:
	rows = []
	for s in sources:
	for pt in s.get("trend", []):
	rows.append({
	"date": pt.get("task_date", ""),
	"source": s.get("source_host_url", ""),
	"citations": pt.get("citation_count", 0),
	})
	if rows:
	df = pd.DataFrame(rows)
	pivot = df.pivot_table(index="date", columns="source", values="citations", aggfunc="sum").fillna(0)
	st.line_chart(pivot)
	else:
	st.info("시계열 데이터 없음")

	elif feature_key == "content-types":
	types = data.get("content_types", [])
	if types:
	df = pd.DataFrame(types)
	if "content_type" in df.columns and "total_citations" in df.columns:
	st.bar_chart(df.set_index("content_type")["total_citations"])
	else:
	st.info("콘텐츠 유형 데이터 없음")

	elif feature_key == "sentiment":
	in_house = data.get("in_house_brands", [])
	competitor = data.get("competitor_brands", [])

	if in_house:
	st.markdown("🏢 자사 브랜드")
	df_ih = pd.DataFrame(in_house)
	cols_ih = ["brand_name", "total_mentions", "positive_rate", "negative_rate"]
	cols_ih = [c for c in cols_ih if c in df_ih.columns]
	if cols_ih:
	st.dataframe(df_ih[cols_ih], use_container_width=True, hide_index=True)

	if competitor:
	st.markdown("🎯 경쟁사 브랜드")
	df_comp = pd.DataFrame(competitor)
	cols_comp = ["brand_name", "total_mentions", "positive_rate", "negative_rate"]
	cols_comp = [c for c in cols_comp if c in df_comp.columns]
	if cols_comp:
	st.dataframe(df_comp[cols_comp], use_container_width=True, hide_index=True)

	if not in_house and not competitor:
	brands = data.get("brands", [])
	if brands:
	df = pd.DataFrame(brands)
	cols = [c for c in ["brand_name", "brand_type", "positive_rate", "negative_rate"] if c in df.columns]
	if cols:
	st.dataframe(df[cols], use_container_width=True, hide_index=True)
	else:
	st.info("감정 분석 데이터 없음")

	elif feature_key == "homepage-citations":
	daily_data = data.get("daily_data", [])[:10]
	if daily_data:
	rows = []
	for day in daily_data:
	for entry in day.get("entries", []):
	rows.append({
	"날짜": day.get("task_date", ""),
	"플랫폼": entry.get("platform", ""),
	"인용 횟수": entry.get("citation_count", 0),
	})
	if rows:
	df = pd.DataFrame(rows)
	st.dataframe(df, use_container_width=True, hide_index=True)
	else:
	st.info("홈페이지 인용 데이터 없음")


	@st.cache_data(ttl=300)
	def get_report_feature_data(
	api_key: str,
	campaign_id: int,
	feature: str,
	start_date: str \| None = None,
	end_date: str \| None = None,
	access_token: str = "",
	):
	"""Fetch report feature data with caching."""
	client = ChainShiftClient(api_key=api_key or None, access_token=access_token or None)

	if feature == "overview":
	return client.get_report_overview(campaign_id, start_date, end_date)
	elif feature == "visibility":
	return client.get_report_visibility(campaign_id, start_date, end_date)
	elif feature == "citations":
	return client.get_report_citations(campaign_id, start_date, end_date, limit=50)
	elif feature == "citation-trends":
	return client.get_report_citation_trends(campaign_id, start_date, end_date)
	elif feature == "content-types":
	return client.get_report_content_types(campaign_id, start_date, end_date)
	elif feature == "sentiment":
	return client.get_report_sentiment(campaign_id)
	elif feature == "homepage-citations":
	return client.get_report_homepage_citations(campaign_id, start_date, end_date)
	else:
	return {"success": False, "error": f"Unknown feature: {feature}"}


	def convert_report_data_to_csv(feature: str, data: dict) -> str:
	"""Convert report feature data to CSV format."""
	output = io.StringIO()
	writer = csv.writer(output)

	if feature == "overview":
	dr = data.get("date_range", {})
	writer.writerow(["항목", "값"])
	writer.writerow(["캠페인 ID", data.get("campaign_id", "")])
	writer.writerow(["분석 기간", f"{dr.get('start', '')} ~ {dr.get('end', '')}"])
	writer.writerow(["총 질문 수", data.get("total_tasks", 0)])
	writer.writerow(["총 답변 수", data.get("total_answers", 0)])
	writer.writerow(["가시성 비율 (%)", data.get("overall_visibility_pct", 0)])

	elif feature == "visibility":
	writer.writerow(["플랫폼", "브랜드", "유형", "가시성 (%)", "브랜드 언급 수", "총 답변 수"])
	for platform in data.get("platforms", []):
	for brand in platform.get("brands", []):
	writer.writerow([
	platform.get("platform", ""),
	brand.get("brand_name", ""),
	brand.get("brand_type", ""),
	brand.get("visibility_pct", 0),
	brand.get("brand_mentions", 0),
	platform.get("total_answers", 0),
	])

	elif feature == "citations":
	writer.writerow(["도메인", "유형", "인용 횟수", "답변 언급 수", "비율 (%)"])
	for item in data.get("sources", []):
	writer.writerow([
	item.get("source_host_url", ""),
	item.get("source_host_type", ""),
	item.get("total_citations", 0),
	item.get("total_answer_mentions", 0),
	item.get("pct_of_total", 0),
	])

	elif feature == "citation-trends":
	writer.writerow(["인용 출처", "유형", "날짜", "인용 횟수", "답변 언급 수", "비율 (%)"])
	for source in data.get("sources", []):
	host = source.get("source_host_url", "")
	host_type = source.get("source_host_type", "")
	for point in source.get("trend", []):
	writer.writerow([
	host,
	host_type,
	point.get("task_date", ""),
	point.get("citation_count", 0),
	point.get("answer_mention_count", 0),
	point.get("citation_pct", 0),
	])

	elif feature == "content-types":
	writer.writerow(["콘텐츠 유형", "인용 횟수", "답변 언급 수", "비율 (%)"])
	for item in data.get("content_types", []):
	writer.writerow([
	item.get("content_type", ""),
	item.get("total_citations", 0),
	item.get("total_answer_mentions", 0),
	item.get("pct_of_total", 0),
	])

	elif feature == "sentiment":
	writer.writerow(["브랜드", "유형", "총 멘션", "긍정 %", "부정 %", "중립 %"])

	for item in data.get("in_house_brands", []):
	t = item.get("total_mentions", 0)
	neutral = round(item.get("neutral_count", 0) / t * 100, 1) if t > 0 else 0.0
	writer.writerow([
	item.get("brand_name", ""),
	"자사",
	t,
	f"{item.get('positive_rate', 0):.1f}",
	f"{item.get('negative_rate', 0):.1f}",
	f"{neutral:.1f}",
	])

	for item in data.get("competitor_brands", []):
	t = item.get("total_mentions", 0)
	neutral = round(item.get("neutral_count", 0) / t * 100, 1) if t > 0 else 0.0
	writer.writerow([
	item.get("brand_name", ""),
	"경쟁사",
	item.get("total_mentions", 0),
	f"{item.get('positive_rate', 0):.1f}",
	f"{item.get('negative_rate', 0):.1f}",
	f"{neutral:.1f}",
	])

	if not data.get("in_house_brands") and not data.get("competitor_brands"):
	for item in data.get("brands", []):
	pos = item.get("positive_rate", item.get("positive", 0))
	neg = item.get("negative_rate", item.get("negative", 0))
	neutral = 100 - pos - neg
	writer.writerow([
	item.get("brand_name", item.get("name", "")),
	item.get("brand_type", item.get("type", "")),
	item.get("total_mentions", 0),
	f"{pos:.1f}",
	f"{neg:.1f}",
	f"{neutral:.1f}",
	])

	elif feature == "homepage-citations":
	writer.writerow(["날짜", "플랫폼", "인용 출처", "인용 횟수", "답변 언급 수"])
	for day in data.get("daily_data", []):
	task_date = day.get("task_date", "")
	for entry in day.get("entries", []):
	writer.writerow([
	task_date,
	entry.get("platform", ""),
	entry.get("source_host_url", ""),
	entry.get("citation_count", 0),
	entry.get("answer_mention_count", 0),
	])

	return output.getvalue()