106 lines
4.9 KiB
Python
106 lines
4.9 KiB
Python
"""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]
|