"""Facebook audit 페이지(KR·EN)를 수집 데이터로 구성. 수치 지표(최근 게시일·게시 빈도·참여율)는 **수집 시점에** 결정적으로 산출해 DB에 박는다 (transform_for_storage). 콘텐츠 주제(top_content_type)는 캡션 본문 이해가 필요해 LLM이 채운다 (리포트 프롬프트 지시).""" 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 def _humanize_age(days: int) -> str: days = max(days, 0) if days == 0: return "오늘" if days < 7: return f"{days}일 전" if days < 30: return f"{days // 7}주 전" if days < 365: return f"{days // 30}개월 전" return f"{days // 365}년 전" def _frequency_label(avg_gap_days: float) -> str: """게시물 사이 평균 간격(일) → 빈도 라벨.""" if avg_gap_days <= 1.5: return "거의 매일" if avg_gap_days <= 10: return f"주 {7 / avg_gap_days:.1f}회" if avg_gap_days <= 45: return f"월 {30 / avg_gap_days:.1f}회" return "비정기 (분기 이상 간격)" def _engagement_text(posts: list[dict]) -> str: """게시물당 좋아요/반응/공유/조회를 min~max 범위로. 전부 0인 지표는 제외. 댓글은 posts actor가 안 줘서 '댓글 거의 없음' 고정 부가 (FB 페이지는 댓글 희박이 일반적).""" def _rng(vals: list[int], label: str, unit: str) -> str | None: lo, hi = min(vals), max(vals) if hi == 0: return None return f"{label} {lo}{unit}" if lo == hi else f"{label} {lo}~{hi}{unit}" parts = [ _rng([p.get("likes", 0) for p in posts], "좋아요", "개"), _rng([p.get("reactions", 0) for p in posts], "반응", "개"), _rng([p.get("shares", 0) for p in posts], "공유", "개"), ] vid_views = [p.get("views", 0) for p in posts if p.get("isVideo")] if vid_views: parts.append(_rng(vid_views, "영상 조회", "회")) parts = [x for x in parts if x] if not parts: return "게시물당 참여 거의 없음" return "게시물당 " + " · ".join(parts) + " · 댓글 거의 없음" def transform_for_storage(fb: dict | None) -> dict | None: """apify 원본 → DB에 저장할 최종 형태. - 수치 지표(recent_post_age·post_frequency·engagement)를 그 자리에서 계산해 박음. - 게시물은 캡션·타입만 남김 (raw 숫자/timestamp는 어차피 재계산 안 하므로 버림). 수집 시점에 한 번 계산 → 리포트 생성 때는 그대로 갖다 박기만 함.""" if not isinstance(fb, dict): return fb 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) if dts: out["recent_post_age"] = _humanize_age((datetime.now(timezone.utc) - dts[0]).days) if len(dts) > 1: avg_gap = ((dts[0] - dts[-1]).days or 1) / (len(dts) - 1) out["post_frequency"] = _frequency_label(avg_gap) out["engagement"] = _engagement_text(posts) out["latestPosts"] = [ {"caption": (p.get("text") or "")[:160], "type": "video" if p.get("isVideo") else "image"} for p in posts ] else: out["latestPosts"] = [] return out def _page_patch(fb: dict) -> dict: """저장된 페북 페이지 → FacebookPage 스키마 필드 패치. 수치 지표는 수집 시점에 박혀있어 그대로 복사.""" p: dict = {} if fb.get("pageUrl"): p["url"] = p["link"] = fb["pageUrl"] if fb.get("pageName"): p["page_name"] = fb["pageName"] if fb.get("followers"): p["followers"] = fb["followers"] if fb.get("intro"): p["bio"] = fb["intro"] if fb.get("categories"): p["category"] = ", ".join(fb["categories"]) if fb.get("website"): p["linked_domain"] = fb["website"] if fb.get("reviews") is not None: p["reviews"] = fb["reviews"] if fb.get("following") is not None: p["following"] = fb["following"] for key in ("recent_post_age", "post_frequency", "engagement"): if fb.get(key): p[key] = fb[key] return p def build_facebook_pages(facebook: dict, facebook_en: dict) -> list[dict]: """KR·EN 페북 페이지 패치 리스트 구성. 프롬프트가 pages를 [KR, EN] 순서로 만들므로 동일 순서 유지. 빈 패치는 제외 (해당 채널 데이터 없음 → LLM도 페이지 안 만듦 → 인덱스 정렬 유지).""" return [pp for pp in (_page_patch(facebook), _page_patch(facebook_en)) if pp]