"""Facebook audit 페이지(KR·EN)를 수집 데이터로 구성. 수치 지표(최근 게시일·게시 빈도·참여율)는 **수집 시점에** 결정적으로 산출해 DB에 박는다 (transform_for_storage). 콘텐츠 주제(top_content_type)는 캡션 본문 이해가 필요해 LLM이 채운다 (리포트 프롬프트 지시).""" from datetime import datetime, timezone from common.utils import parse_ts from integrations.llm.schemas.report import FacebookAudit 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, language: str, label: str) -> dict: """저장된 페북 페이지 → FacebookPage 스키마 필드 패치. 수치 지표는 수집 시점에 박혀있어 그대로 복사. language/label 은 데이터 있을 때만 명시적으로 박음 — template-copy 가 KR 값을 EN 슬롯에 잘못 상속시키는 것 방지.""" 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] if p: p["language"] = language p["label"] = label return p def build_facebook_audit(facebook: dict, facebook_en: dict, llm_pages: list[dict] | None = None) -> dict: """KR·EN 페북 페이지 구성. logo/logo_description 은 LLM Vision 결과(첫 페이지) 모든 페이지에 공통 적용, 나머지 필드는 코드가 수집 데이터로 계산.""" llm_logo = {k: v for k, v in ((llm_pages or [{}])[0]).items() if k in {"logo", "logo_description"} and v} pages = [{**llm_logo, **p} for p in ( _page_patch(facebook, "KR", "페이스북 KR"), _page_patch(facebook_en, "EN", "페이스북 EN"), ) if p] return FacebookAudit.model_validate({"pages": pages}).model_dump(exclude_unset=True)