diff --git a/app/common/utils.py b/app/common/utils.py index 5d64de6..0a5eaf4 100644 --- a/app/common/utils.py +++ b/app/common/utils.py @@ -1,6 +1,7 @@ import os import asyncio import logging +from datetime import datetime, timezone from http import HTTPMethod import httpx @@ -9,6 +10,27 @@ logger = logging.getLogger(__name__) REQUEST_TIMEOUT = 60 +def parse_ts(v) -> datetime | None: + """수집기마다 다른 timestamp 포맷을 통일된 datetime으로 변환. + 파싱 실패 시 None. + """ + # 숫자면 epoch (Unix timestamp) — apify가 가끔 epoch로 줌 + if isinstance(v, (int, float)): + return datetime.fromtimestamp(v, tz=timezone.utc) + if isinstance(v, str): + # 1순위: ISO 8601 (대부분 apify/firecrawl 출력) + try: + return datetime.fromisoformat(v.replace("Z", "+00:00")) + except ValueError: + pass + # 2순위: RFC 2822 (네이버 블로그 RSS 등 — 표준 라이브러리 파서로) + try: + from email.utils import parsedate_to_datetime + return parsedate_to_datetime(v) + except (TypeError, ValueError): + return None + return None + def get_env(key: str) -> str: v = os.environ.get(key, "") diff --git a/app/integrations/llm/temp-prompt/report_prompt.txt b/app/integrations/llm/temp-prompt/report_prompt.txt index fb30246..761168f 100644 --- a/app/integrations/llm/temp-prompt/report_prompt.txt +++ b/app/integrations/llm/temp-prompt/report_prompt.txt @@ -111,5 +111,5 @@ - 데이터가 null인 계정은 항목을 만들지 마세요. icon은 instagram/facebook/video 등 플랫폼에 맞게 설정. - strengths와 weaknesses는 각 3개 이상 작성하세요. - roadmap은 우선순위 순으로 실행 가능한 액션으로 작성하세요. -- kpis는 실제 수집된 수치 기반으로 현실적인 측정 가능 지표로 작성하세요. +- kpi_dashboard는 코드가 결정적으로 산출해 후처리 강제 치환하므로 LLM 출력 무시됩니다. 빈 배열 또는 placeholder로 두세요. - conversion_strategy의 actions는 구체적인 실행 방안으로 작성하세요. diff --git a/app/services/analysis.py b/app/services/analysis.py index 96dcaa1..2b3e597 100644 --- a/app/services/analysis.py +++ b/app/services/analysis.py @@ -8,6 +8,7 @@ from integrations.llm.prompt import report_prompt, plan_prompt, youtube_diagnosi from integrations.llm.schemas.report import ReportOutput, ClinicSnapshot, YouTubeAudit from services.instagram_audit import build_instagram_accounts from services.facebook_audit import build_facebook_pages +from services.kpi_dashboard import build_kpi_dashboard from integrations.llm.schemas.plan import PlanOutput from models.status import AnalysisStatus @@ -287,6 +288,8 @@ async def _build_overrides(analysis_run_id: str) -> dict: # ── facebook (KR=facebook_data, EN=hospital.facebookEn 둘 다 코드 산출, [KR, EN] 순서) ── fb_pages = build_facebook_pages(facebook, hospital.get("facebookEn") or {}) + # ── KPI dashboard: 7개 mockup 라이프사이클 공식으로 코드가 결정. LLM 출력은 무시. ────── + kpi = build_kpi_dashboard(instagram, facebook, youtube, gangnam_unni, hospital, naver_blog) overrides: dict = {} if snapshot: @@ -297,6 +300,8 @@ async def _build_overrides(analysis_run_id: str) -> dict: overrides["facebook_audit"] = {"pages": fb_pages} if yt_patch: overrides["youtube_audit"] = yt_patch + if kpi: + overrides["kpi_dashboard"] = kpi return overrides @@ -328,6 +333,9 @@ def _patch_report(result: ReportOutput, overrides: dict) -> ReportOutput: for i, patch in enumerate(fb_pages): if i < len(base_pages): base_pages[i].update(patch) + # KPI dashboard 강제 치환 — 코드가 계산한 라이프사이클 공식 그대로. + if overrides.get("kpi_dashboard"): + merged["kpi_dashboard"] = overrides["kpi_dashboard"] return ReportOutput(**merged) diff --git a/app/services/facebook_audit.py b/app/services/facebook_audit.py index 890437e..ed872e3 100644 --- a/app/services/facebook_audit.py +++ b/app/services/facebook_audit.py @@ -4,16 +4,7 @@ from datetime import datetime, timezone - -def _parse_ts(v) -> datetime | None: - if isinstance(v, (int, float)): - return datetime.fromtimestamp(v, tz=timezone.utc) - if isinstance(v, str): - try: - return datetime.fromisoformat(v.replace("Z", "+00:00")) - except ValueError: - return None - return None +from common.utils import parse_ts def _humanize_age(days: int) -> str: @@ -66,7 +57,7 @@ def transform_for_storage(fb: dict | None) -> dict | None: posts = fb.get("latestPosts") or [] out = {k: v for k, v in fb.items() if k != "latestPosts"} if posts: - dts = sorted((d for d in (_parse_ts(p.get("timestamp")) for p in posts) if d), reverse=True) + dts = sorted((d for d in (parse_ts(p.get("timestamp")) for p in posts) if d), reverse=True) if dts: out["recent_post_age"] = _humanize_age((datetime.now(timezone.utc) - dts[0]).days) if len(dts) > 1: diff --git a/app/services/kpi_dashboard.py b/app/services/kpi_dashboard.py new file mode 100644 index 0000000..c29450c --- /dev/null +++ b/app/services/kpi_dashboard.py @@ -0,0 +1,94 @@ +"""mockup 7개 역분석 — 채널 규모별 3개월/12개월 target 성장률 공식.""" + + +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]: + ig_en = hospital.get("instagramEn") or {} + fb_en = hospital.get("facebookEn") or {} + tiktok = hospital.get("tiktok") or {} + cafe = hospital.get("naverCafe") or {} + + kpis: list[dict] = [] + for k in [ + _follower_kpi("YouTube 구독자", youtube.get("subscribers")), + _follower_kpi("Instagram KR 팔로워", instagram.get("followers")), + _follower_kpi("Instagram EN 팔로워", ig_en.get("followers")), + _follower_kpi("Facebook KR 팔로워", facebook.get("followers")), + _follower_kpi("Facebook EN 팔로워", fb_en.get("followers")), + _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)):,}개", + }) + + return kpis