경쟁사 브랜드 요약 — 총 {total_mentions:,}건 언급, {len(brand_stats)}개 브랜드
{' '.join(brand_chips)}
""", unsafe_allow_html=True)
def render_competitor_drilldown(data: dict):
"""경쟁사 키워드 드릴다운."""
st.markdown("**경쟁사 키워드 상세 드릴다운**")
keyword_data = data.get("keyword_data", {})
keywords_list = keyword_data.get("keywords", [])
keyword_names = ["전체"] + [kw.get("keyword", "") for kw in keywords_list]
if len(keyword_names) <= 1:
st.info("키워드 데이터가 없습니다")
return
selected_raw = st.selectbox(
"키워드 선택",
options=keyword_names,
key="sentiment:comp_kw_drilldown_select",
)
selected_keyword = None if selected_raw == "전체" else selected_raw
# Filters — matching keyword drilldown pattern
f1, f2, f3 = st.columns(3)
with f1:
sentiment_filter = st.selectbox(
"감성",
options=["전체", "positive", "neutral", "negative"],
format_func=lambda x: {"전체": "전체", "positive": "긍정", "neutral": "중립", "negative": "부정"}.get(x, x),
key="sentiment:comp_drill_sentiment",
)
with f2:
platform_filter = st.selectbox(
"플랫폼",
options=["전체", "CHATGPT", "GEMINI", "PERPLEXITY", "CLAUDE"],
key="sentiment:comp_drill_platform",
)
with f3:
llm_status_filter = st.selectbox(
"2차 검증",
options=["전체", "정탐", "오탐", "미검증"],
key="sentiment:comp_drill_llm",
)
# Build server-side filter params
sentiment_param = sentiment_filter if sentiment_filter != "전체" else None
platform_param = platform_filter if platform_filter != "전체" else None
llm_is_negative_param = None
llm_verified_param = None
if llm_status_filter == "정탐":
llm_is_negative_param = True
elif llm_status_filter == "오탐":
llm_is_negative_param = False
elif llm_status_filter == "미검증":
llm_verified_param = "unverified"
# Export + page size row
dl_col, _, size_col = st.columns([2, 2, 1])
with dl_col:
_render_competitor_export(
data,
keyword=selected_keyword,
sentiment=sentiment_param,
platform=platform_param,
llm_is_negative=llm_is_negative_param,
llm_verified=llm_verified_param,
)
with size_col:
page_size = st.selectbox("페이지 크기", options=[20, 50, 100], index=1, key="sentiment:comp_drill_page_size")
# Pagination state
if "sentiment:comp_drill_page" not in st.session_state:
st.session_state["sentiment:comp_drill_page"] = 1
# Reset page on filter change
comp_filter_key = f"{selected_keyword}_{sentiment_filter}_{platform_filter}_{llm_status_filter}_{page_size}"
if st.session_state.get("sentiment:comp_drill_last_filters") != comp_filter_key:
st.session_state["sentiment:comp_drill_page"] = 1
st.session_state["sentiment:comp_drill_last_filters"] = comp_filter_key
current_page = st.session_state["sentiment:comp_drill_page"]
try:
items, total = fetch_competitor_drilldown(
campaign_id=data["campaign_id"],
keyword=selected_keyword,
sentiment=sentiment_param,
llm_verified=llm_verified_param,
llm_is_negative=llm_is_negative_param,
platform=platform_param,
page=current_page,
page_size=page_size,
)
except Exception as e:
st.error(f"경쟁사 키워드 결과 로드 실패: {e}")
return
total_pages = max(1, (total + page_size - 1) // page_size)
has_more = len(items) == page_size # count="planned" may underestimate
start_idx = (current_page - 1) * page_size + 1
end_idx = min(current_page * page_size, total)
if total_pages > 1 or has_more:
col_info, col_prev, col_page, col_next = st.columns([3, 1, 1, 1])
with col_info:
st.markdown(f"**전체 ~{total:,}건** | 페이지 {current_page}/{total_pages} ({start_idx}-{end_idx}건)")
with col_prev:
if st.button("⬅️ 이전", disabled=current_page <= 1, key="sentiment:comp_drill_prev"):
st.session_state["sentiment:comp_drill_page"] = current_page - 1
st.rerun()
with col_page:
new_page = st.number_input(
"페이지", min_value=1, max_value=max(total_pages, current_page + 1),
value=current_page, label_visibility="collapsed", key="sentiment:comp_drill_page_input",
)
if new_page != current_page:
st.session_state["sentiment:comp_drill_page"] = new_page
st.rerun()
with col_next:
if st.button("다음 ➡️", disabled=not has_more, key="sentiment:comp_drill_next"):
st.session_state["sentiment:comp_drill_page"] = current_page + 1
st.rerun()
else:
st.markdown(f"**전체 ~{total:,}건**")
if not items:
st.info("조건에 맞는 경쟁사 키워드 결과가 없습니다.")
return
for i, item in enumerate(items):
unique_idx = (current_page - 1) * page_size + i
render_keyword_result_card(data, item, unique_idx, is_competitor=True)
def fetch_competitor_drilldown(
campaign_id: int,
keyword: str | None = None,
sentiment: str | None = None,
llm_verified: str | None = None,
llm_is_negative: bool | None = None,
platform: str | None = None,
page: int = 1,
page_size: int = 50,
) -> tuple[list[dict], int]:
"""경쟁사 키워드 직접 Supabase 쿼리 (Vercel 타임아웃 우회).
Filters match fetch_keyword_drilldown for consistency.
"""
from core.supabase_client import get_supabase_client
sb = get_supabase_client()
query = (
sb.table("keyword_sentiment_results")
.select("*", count="planned")
.eq("campaign_id", campaign_id)
.is_("brand_name", "null")
)
# competitor_llm_verified filter (was hardcoded True, now conditional)
if llm_verified == "unverified":
query = query.or_("competitor_llm_verified.is.null,competitor_llm_verified.eq.false")
else:
# 전체/정탐/오탐: only show LLM-analyzed results
query = query.eq("competitor_llm_verified", True)
if keyword:
query = query.eq("keyword", keyword)
if sentiment:
query = query.eq("keyword_sentiment", sentiment)
if platform:
query = query.eq("platform", platform)
if llm_is_negative is not None:
query = query.eq("competitor_llm_is_negative", llm_is_negative)
offset = (page - 1) * page_size
query = query.order("created_at", desc=True).range(offset, offset + page_size - 1)
result = query.execute()
return result.data or [], result.count or 0
def fetch_keyword_drilldown(
campaign_id: int,
keyword: str | None = None,
sentiment: str | None = None,
llm_verified: str | None = None,
llm_is_negative: bool | None = None,
platform: str | None = None,
brand_only: bool = False,
no_brand: bool = False,
page: int = 1,
page_size: int = 50,
) -> tuple[list[dict], int]:
"""직접 Supabase 페이지네이션 쿼리 (Vercel 10s 타임아웃 우회).
Args:
llm_is_negative: True=정탐(부정확정), False=오탐(부정아님), None=전체
Returns:
(items, total_count)
"""
from core.supabase_client import get_supabase_client
sb = get_supabase_client()
query = sb.table("keyword_sentiment_results").select("*", count="planned")
query = query.eq("campaign_id", campaign_id)
if keyword:
query = query.eq("keyword", keyword)
if sentiment:
query = query.eq("keyword_sentiment", sentiment)
if platform:
query = query.eq("platform", platform)
if brand_only:
query = query.not_.is_("brand_name", "null")
elif no_brand:
query = query.is_("brand_name", "null")
if llm_is_negative is not None:
query = query.eq("llm_is_negative", llm_is_negative)
elif llm_verified == "verified":
query = query.eq("llm_verified", True)
elif llm_verified == "unverified":
query = query.or_("llm_verified.is.null,llm_verified.eq.false")
offset = (page - 1) * page_size
query = query.order("created_at", desc=True).range(offset, offset + page_size - 1)
result = query.execute()
return result.data or [], result.count or 0
def _render_competitor_export(
data: dict,
keyword: str | None = None,
sentiment: str | None = None,
platform: str | None = None,
llm_is_negative: bool | None = None,
llm_verified: str | None = None,
):
"""경쟁사 드릴다운 CSV export (키워드 탭 export_kw.py 패턴 일치)."""
filter_parts = []
if keyword:
filter_parts.append(f"키워드: {keyword}")
if sentiment:
label = {"positive": "긍정", "neutral": "중립", "negative": "부정"}.get(sentiment, sentiment)
filter_parts.append(f"감성: {label}")
if platform:
filter_parts.append(f"플랫폼: {platform}")
if llm_verified == "unverified":
filter_parts.append("미검증만")
elif llm_is_negative is True:
filter_parts.append("정탐만")
elif llm_is_negative is False:
filter_parts.append("오탐만")
if filter_parts:
st.caption(f"📥 필터: {' | '.join(filter_parts)}")
if st.button("📥 Excel 다운로드", key="sentiment:comp_drill_export_btn"):
with st.spinner("Excel 생성 중..."):
try:
xlsx = _export_competitor_drilldown(
data["campaign_id"],
keyword=keyword,
sentiment=sentiment,
platform=platform,
llm_is_negative=llm_is_negative,
llm_verified=llm_verified,
)
st.session_state["sentiment:comp_drill_excel"] = xlsx
st.session_state["sentiment:comp_drill_excel_ready"] = True
except Exception as e:
st.error(f"다운로드 실패: {e}")
if st.session_state.get("sentiment:comp_drill_excel_ready"):
st.download_button(
label="💾 파일 저장",
data=st.session_state["sentiment:comp_drill_excel"],
file_name=f"competitor_keyword_{data['campaign_id']}.xlsx",
mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
key="sentiment:comp_drill_dl_btn",
)
def _export_competitor_drilldown(
campaign_id: int,
keyword: str | None = None,
sentiment: str | None = None,
platform: str | None = None,
llm_is_negative: bool | None = None,
llm_verified: str | None = None,
) -> bytes:
"""경쟁사 드릴다운 데이터를 Excel로 추출 (cursor pagination)."""
from io import BytesIO
from core.supabase_client import get_supabase_client
sb = get_supabase_client()
batch_size = 1000
last_id = 0
all_rows: list[dict] = []
while True:
query = (
sb.table("keyword_sentiment_results")
.select(
"id, answer_id, keyword, matched_sentence, keyword_sentiment, keyword_confidence, "
"competitor_brand_name, competitor_llm_verified, competitor_llm_sentiment, "
"competitor_llm_is_negative, competitor_llm_confidence, "
"competitor_llm_reason_tags, competitor_llm_reason_summary, "
"citation_urls, citation_count, platform, question_content, created_at"
)
.eq("campaign_id", campaign_id)
.is_("brand_name", "null")
.gt("id", last_id)
)
# competitor_llm_verified filter (conditional, matching drilldown)
if llm_verified == "unverified":
query = query.or_("competitor_llm_verified.is.null,competitor_llm_verified.eq.false")
else:
query = query.eq("competitor_llm_verified", True)
if keyword:
query = query.eq("keyword", keyword)
if sentiment:
query = query.eq("keyword_sentiment", sentiment)
if platform:
query = query.eq("platform", platform)
if llm_is_negative is not None:
query = query.eq("competitor_llm_is_negative", llm_is_negative)
result = query.order("id").limit(batch_size).execute()
rows = result.data or []
if not rows:
break
all_rows.extend(rows)
last_id = rows[-1]["id"]
if len(rows) < batch_size:
break
from openpyxl import Workbook
from openpyxl.utils import get_column_letter
wb = Workbook()
ws = wb.active
ws.title = "Competitor Keywords"
headers = [
"키워드", "매칭 문장", "키워드 감성", "키워드 확신도",
"경쟁사 브랜드", "경쟁사 LLM 감성", "경쟁사 정탐/오탐",
"경쟁사 LLM 확신도", "경쟁사 LLM 태그", "경쟁사 LLM 요약",
"인용 URL", "인용 수", "플랫폼", "질문", "Answer ID", "생성일",
]
ws.append(headers)
def _tags_str(tags):
if tags and isinstance(tags, list):
return ", ".join(str(t) for t in tags)
return ""
for row in all_rows:
cites = row.get("citation_urls")
cite_str = "\n".join(str(u) for u in cites[:10]) if cites and isinstance(cites, list) else ""
neg = row.get("competitor_llm_is_negative")
neg_label = "정탐" if neg is True else "오탐" if neg is False else ""
ws.append([
row.get("keyword", ""),
row.get("matched_sentence", ""),
row.get("keyword_sentiment", ""),
row.get("keyword_confidence"),
row.get("competitor_brand_name", ""),
row.get("competitor_llm_sentiment", ""),
neg_label,
row.get("competitor_llm_confidence"),
_tags_str(row.get("competitor_llm_reason_tags")),
row.get("competitor_llm_reason_summary", ""),
cite_str,
row.get("citation_count"),
row.get("platform", ""),
row.get("question_content", ""),
row.get("answer_id"),
(row.get("created_at") or "")[:19].replace("T", " "),
])
widths = [12, 50, 10, 8, 15, 10, 8, 8, 25, 30, 40, 6, 10, 40, 10, 16]
for i, w in enumerate(widths, 1):
ws.column_dimensions[get_column_letter(i)].width = w
buf = BytesIO()
wb.save(buf)
return buf.getvalue()