100 lines
4.1 KiB
Python
100 lines
4.1 KiB
Python
"""mockup 7개 역분석 — 채널 규모별 3개월/12개월 target 성장률 공식."""
|
|
|
|
from integrations.llm.schemas.report import KPIMetric
|
|
|
|
|
|
def _round_clean(n: int) -> int:
|
|
if n < 100: return n
|
|
if n < 1000: return round(n / 100) * 100
|
|
if n < 10_000: return round(n / 500) * 500
|
|
if n < 100_000: return round(n / 1000) * 1000
|
|
if n < 1_000_000: return round(n / 5000) * 5000
|
|
return round(n / 50_000) * 50_000
|
|
|
|
|
|
def _target_multiplier(current: int) -> tuple[float, float]:
|
|
if current < 1_000: return (2.5, 9.0)
|
|
if current < 5_000: return (1.7, 4.0)
|
|
if current < 25_000: return (1.5, 2.5)
|
|
if current < 50_000: return (1.3, 2.2)
|
|
return (1.1, 1.9)
|
|
|
|
|
|
def _follower_kpi(metric: str, val: int | None, unit: str = "명") -> dict | None:
|
|
if not val: return None
|
|
m3, m12 = _target_multiplier(val)
|
|
return {
|
|
"metric": metric,
|
|
"current": f"{val:,}{unit}",
|
|
"target_3_month": f"{_round_clean(int(val * m3)):,}{unit}",
|
|
"target_12_month": f"{_round_clean(int(val * m12)):,}{unit}",
|
|
}
|
|
|
|
|
|
def _blog_frequency(posts: list) -> tuple[str, str, str] | None:
|
|
"""RSS posts timestamp로 (current, target_3m, target_12m) 라벨 반환. target은 절대 downgrade 안 함."""
|
|
from common.utils import parse_ts
|
|
dts = sorted((d for d in (parse_ts(p.get("postDate")) for p in posts) if d), reverse=True)
|
|
if len(dts) < 2: return None
|
|
avg_gap = (dts[0] - dts[-1]).days / (len(dts) - 1)
|
|
if avg_gap > 90: current = f"방치 ({dts[0].strftime('%Y-%m')})"
|
|
elif avg_gap <= 1: current = f"주 {7 // max(int(avg_gap), 1)}회"
|
|
elif avg_gap <= 3: current = "주 2~3회"
|
|
elif avg_gap <= 14: current = "주 1~2회"
|
|
elif avg_gap <= 30: current = f"월 {max(30 // int(avg_gap), 1)}회"
|
|
else: current = "월 1회 미만"
|
|
if avg_gap > 3: return current, "주 2회", "주 3회"
|
|
if avg_gap > 2: return current, "주 3회", "주 5회"
|
|
if avg_gap > 1: return current, "주 5회", "주 7회"
|
|
return current, f"{current} 유지", f"{current} 유지"
|
|
|
|
|
|
def build_kpi_dashboard(
|
|
instagram: dict, facebook: dict, youtube: dict, gangnam_unni: dict, hospital: dict,
|
|
naver_blog: dict | None = None,
|
|
) -> list[dict]:
|
|
tiktok = hospital.get("tiktok") or {}
|
|
cafe = hospital.get("naverCafe") or {}
|
|
# print("facebook", facebook)
|
|
# print("instagram", instagram)
|
|
|
|
kpis: list[dict] = []
|
|
facebook_kpis = [_follower_kpi("Facebook " + fb["language"] + " 팔로워", fb['raw_data'].get("followers") ) for fb in facebook]
|
|
instagram_kpis = [_follower_kpi("Instagram " + ib["language"] + " 팔로워", ib['raw_data'].get("followers")) for ib in instagram]
|
|
print("facebook_kpis", facebook_kpis)
|
|
print("instagram_kpis", instagram_kpis)
|
|
|
|
kpis += facebook_kpis
|
|
kpis += instagram_kpis
|
|
for k in [
|
|
_follower_kpi("YouTube 구독자", youtube.get("subscribers")),
|
|
_follower_kpi("TikTok 팔로워", tiktok.get("followers")),
|
|
_follower_kpi("Naver Cafe 회원 수", cafe.get("memberCount")),
|
|
]:
|
|
if k: kpis.append(k)
|
|
|
|
if naver_blog:
|
|
freq = _blog_frequency(naver_blog.get("posts") or [])
|
|
if freq:
|
|
cur, t3, t12 = freq
|
|
kpis.append({
|
|
"metric": "네이버 블로그 포스팅 빈도",
|
|
"current": cur,
|
|
"target_3_month": t3,
|
|
"target_12_month": t12,
|
|
})
|
|
|
|
gu_reviews = gangnam_unni.get("totalReviews")
|
|
if gu_reviews:
|
|
if gu_reviews < 1000: rm3, rm12 = 2.0, 6.0
|
|
elif gu_reviews < 5000: rm3, rm12 = 1.10, 1.50
|
|
else: rm3, rm12 = 1.07, 1.27
|
|
kpis.append({
|
|
"metric": "강남언니 리뷰",
|
|
"current": f"{gu_reviews:,}개",
|
|
"target_3_month": f"{_round_clean(int(gu_reviews * rm3)):,}개",
|
|
"target_12_month": f"{_round_clean(int(gu_reviews * rm12)):,}개",
|
|
})
|
|
print ("kpis", kpis)
|
|
return [KPIMetric.model_validate(k).model_dump() for k in kpis]
|