Spaces:
Sleeping
Sleeping
File size: 2,315 Bytes
ef78361 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 | """캐시된 API 데이터 페처.
Streamlit cache를 활용한 API 호출 함수들.
"""
import streamlit as st
from .api_client import ChainShiftClient
@st.cache_data(ttl=300)
def get_campaigns(api_key: str = "", access_token: str = ""):
"""Fetch all campaigns with pagination and caching."""
client = ChainShiftClient(api_key=api_key or None, access_token=access_token or None)
all_items: list[dict] = []
page = 1
while True:
resp = client.get_campaigns(page=page, page_size=100)
items = resp.get("data", {}).get("items", [])
all_items.extend(items)
total = resp.get("data", {}).get("total", 0)
if len(all_items) >= total or not items:
break
page += 1
return {"data": {"items": all_items, "total": len(all_items)}}
@st.cache_data(ttl=60)
def get_nudge_candidates(
api_key: str = "",
campaign_id: int = 0,
page: int = 1,
page_size: int = 50,
platform: str | None = None,
confidence_tier: str | None = None,
access_token: str = "",
):
"""Fetch nudge candidates (in-house brand negative mentions)."""
client = ChainShiftClient(api_key=api_key or None, access_token=access_token or None)
return client.get_nudge_candidates(
campaign_id,
page=page,
page_size=page_size,
platform=platform if platform and platform != "전체" else None,
)
@st.cache_data(ttl=60)
def get_brand_mentions(
api_key: str = "",
campaign_id: int = 0,
brand_type: str | None = None,
polarity: str | None = None,
access_token: str = "",
):
"""Fetch brand mention analysis."""
client = ChainShiftClient(api_key=api_key or None, access_token=access_token or None)
return client.get_brand_mentions(
campaign_id,
brand_type=brand_type if brand_type and brand_type != "전체" else None,
polarity=polarity if polarity and polarity != "전체" else None,
)
@st.cache_data(ttl=60)
def get_feedback_stats(api_key: str = "", campaign_id: int = 0, access_token: str = ""):
"""Fetch feedback statistics for a campaign."""
try:
client = ChainShiftClient(api_key=api_key or None, access_token=access_token or None)
return client.get_feedback_stats(campaign_id)
except Exception:
return None
|