페북 수집·지표·저장 파이프라인 정리

수집:
- pages + posts 두 actor 병렬 호출 (facebook-pages-scraper, facebook-posts-scraper)
- 저장 필드 슬림화: 페이지 메타에서 likes/rating/email/phone/address 제거
  (followers/reviews와 중복이거나 클리닉 raw_data에 이미 있음)
- 게시물 저장은 캡션 160자 + likes/reactions/shares/views/isVideo/timestamp만

지표 계산 위치 이동: 리포트 시점 → 수집 시점:
- recent_post_age / post_frequency / engagement 를 transform_for_storage에서
  결정적으로 산출해 DB에 박음 (재계산 불필요)
- 저장된 게시물은 LLM용 캡션·타입 2필드만 — 추가 슬림 단계 제거

리팩토링:
- services/facebook_audit.py 신설 (instagram_audit 패턴) — _build_overrides의
  인라인 클로저(_fb_page_patch)와 analysis.py의 _fb_post_metrics 분리
- collect.py / enrichment.py 가 transform_for_storage를 호출하도록

엔게이지먼트 표기:
- 범위(min~max)로 표시, 전부 0인 지표는 제외
- 댓글은 actor 미제공이라 "댓글 거의 없음" 고정 부가

콘텐츠 유형:
- top_content_type 은 캡션 본문 주제 추론이 필요해 LLM에 위임
- report_prompt.txt 에 facebook_audit.pages[].top_content_type 작성 지침 추가

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

app/integrations/apify.py

@@ -1,16 +1,29 @@
+import asyncio
 from http import HTTPMethod
 from urllib.parse import urlparse
 from common.utils import http_request
 
 APIFY_BASE = "https://api.apify.com/v2"
 
+# Instagram: profile + highlights 두 actor 직접 호출.
+IG_PROFILE_ACTOR = "coderx~instagram-profile-scraper-bio-posts"
+IG_HIGHLIGHTS_ACTOR = "igview-owner~instagram-highlights-scraper"
+
+# Facebook: pages + posts 두 actor 직접 호출.
+FB_PAGES_ACTOR = "apify~facebook-pages-scraper"
+FB_POSTS_ACTOR = "apify~facebook-posts-scraper"
+
+
+def _ig_username(url: str) -> str:
+    return urlparse(url).path.strip("/").split("/")[0] if "://" in url else url.lstrip("@")
+
 
 class ApifyClient:
     def __init__(self, token: str, wait_for_finish: int = 120):
         self.token = token
         self.wait_for_finish = wait_for_finish
 
-    async def _run_actor(self, actor_id: str, input_data: dict) -> list[dict]:
+    async def _run_actor(self, actor_id: str, input_data: dict, limit: int = 20) -> list[dict]:
         resp = await http_request(
             HTTPMethod.POST,
             url=f"{APIFY_BASE}/acts/{actor_id}/runs",
@@ -26,34 +39,46 @@ class ApifyClient:
         items_resp = await http_request(
             HTTPMethod.GET,
             url=f"{APIFY_BASE}/datasets/{dataset_id}/items",
-            params={"token": self.token, "limit": 20},
+            params={"token": self.token, "limit": limit},
             label=f"apify-dataset-{dataset_id}",
         )
         if not items_resp or not items_resp.is_success:
             return []
         return items_resp.json()
 
-    async def fetch_instagram_profile(self, url: str) -> dict | None:
+    async def fetch_instagram_profile(self, username: str) -> dict | None:
-        username = urlparse(url).path.strip("/").split("/")[0] if "://" in url else url.lstrip("@")
+        items = await self._run_actor(IG_PROFILE_ACTOR, {"usernames": [username]})
-        items = await self._run_actor("apify~instagram-profile-scraper", {"usernames": [username], "resultsLimit": 12})
         return items[0] if items else None
 
+    async def fetch_instagram_highlights(self, username: str) -> list[dict]:
+        return await self._run_actor(IG_HIGHLIGHTS_ACTOR, {"usernames": [username]})
+
     async def get_instagram_profile(self, url: str) -> dict | None:
-        profile = await self.fetch_instagram_profile(url)
+        username = _ig_username(url)
-        if not profile or profile.get("error"):
+        # profile·highlights 두 actor를 병렬 호출 (highlights 실패해도 profile만 있으면 진행)
+        profile, highlights = await asyncio.gather(
+            self.fetch_instagram_profile(username),
+            self.fetch_instagram_highlights(username),
+            return_exceptions=True,
+        )
+        if isinstance(profile, Exception) or not profile or profile.get("error"):
             return None
+        if isinstance(highlights, Exception):
+            highlights = []
         return {
             "username": profile["username"],
-            "profileImage": profile.get("profilePicUrlHD") or profile.get("profilePicUrl"),
+            "profileImage": profile.get("hdProfilePicUrl") or profile.get("profilePicUrl"),
             "followers": profile.get("followersCount", 0),
             "following": profile.get("followsCount", 0),
             "posts": profile.get("postsCount", 0),
             "bio": profile.get("biography", ""),
+            "category": profile.get("businessCategoryName") or "",
             "isBusinessAccount": profile.get("isBusinessAccount", False),
-            #"externalUrl": profile.get("externalUrl"), LLM에 혼동을 주는 듯 하여 비활성화 
+            #"externalUrl": profile.get("externalUrl"), LLM에 혼동을 주는 듯 하여 비활성화
+            "highlights": [h["highlightTitle"] for h in (highlights or []) if isinstance(h, dict) and h.get("highlightTitle")],
             "latestPosts": [
                 {
-                    "type": p.get("type"),
+                    "type": p.get("mediaType") or p.get("type"),
                     "likes": p.get("likesCount", 0),
                     "comments": p.get("commentsCount", 0),
                     "caption": (p.get("caption") or "")[:500],
@@ -63,89 +88,79 @@ class ApifyClient:
             ],
         }
 
-    async def fetch_instagram_posts(self, url: str, limit: int = 20) -> list[dict]:
+    # 인스타 post 스크래퍼는 현재 파이프라인 미사용 — 비활성화 (필요 시 복구)
-        username = urlparse(url).path.strip("/").split("/")[0] if "://" in url else url.lstrip("@")
+    # async def fetch_instagram_posts(self, url: str, limit: int = 20) -> list[dict]:
-        return await self._run_actor("apify~instagram-post-scraper", {
+    #     username = urlparse(url).path.strip("/").split("/")[0] if "://" in url else url.lstrip("@")
-            "directUrls": [f"https://www.instagram.com/{username}/"],
+    #     return await self._run_actor("apify~instagram-post-scraper", {
-            "resultsLimit": limit,
+    #         "directUrls": [f"https://www.instagram.com/{username}/"],
-        })
+    #         "resultsLimit": limit,
-
+    #     })
-    async def get_instagram_posts(self, url: str, limit: int = 20) -> dict:
+    #
-        items = await self.fetch_instagram_posts(url, limit)
+    # async def get_instagram_posts(self, url: str, limit: int = 20) -> dict:
-        posts = [
+    #     items = await self.fetch_instagram_posts(url, limit)
-            {
+    #     posts = [
-                "id": p["id"],
+    #         {
-                "type": p.get("type"),
+    #             "id": p["id"],
-                "url": p.get("url"),
+    #             "type": p.get("type"),
-                "caption": (p.get("caption") or "")[:500],
+    #             "url": p.get("url"),
-                "hashtags": p.get("hashtags", []),
+    #             "caption": (p.get("caption") or "")[:500],
-                "likesCount": p.get("likesCount", 0),
+    #             "hashtags": p.get("hashtags", []),
-                "commentsCount": p.get("commentsCount", 0),
+    #             "likesCount": p.get("likesCount", 0),
-                "timestamp": p.get("timestamp"),
+    #             "commentsCount": p.get("commentsCount", 0),
-            }
+    #             "timestamp": p.get("timestamp"),
-            for p in items
+    #         }
-        ]
+    #         for p in items
-        n = len(posts) or 1
+    #     ]
-        return {
+    #     n = len(posts) or 1
-            "posts": posts,
+    #     return {
-            "totalPosts": len(posts),
+    #         "posts": posts,
-            "avgLikes": round(sum(p["likesCount"] for p in posts) / n),
+    #         "totalPosts": len(posts),
-            "avgComments": round(sum(p["commentsCount"] for p in posts) / n),
+    #         "avgLikes": round(sum(p["likesCount"] for p in posts) / n),
-        }
+    #         "avgComments": round(sum(p["commentsCount"] for p in posts) / n),
-
+    #     }
-    async def fetch_instagram_reels(self, url: str, limit: int = 15) -> list[dict]:
-        username = urlparse(url).path.strip("/").split("/")[0] if "://" in url else url.lstrip("@")
-        return await self._run_actor("apify~instagram-reel-scraper", {
-            "directUrls": [f"https://www.instagram.com/{username}/reels/"],
-            "resultsLimit": limit,
-        })
-
-    async def get_instagram_reels(self, url: str, limit: int = 15) -> dict:
-        items = await self.fetch_instagram_reels(url, limit)
-        reels = [
-            {
-                "id": r["id"],
-                "url": r.get("url"),
-                "caption": (r.get("caption") or "")[:500],
-                "hashtags": r.get("hashtags", []),
-                "likesCount": r.get("likesCount", 0),
-                "commentsCount": r.get("commentsCount", 0),
-                "videoViewCount": r.get("videoViewCount", 0),
-                "videoPlayCount": r.get("videoPlayCount", 0),
-                "videoDuration": r.get("videoDuration", 0),
-                "timestamp": r.get("timestamp"),
-            }
-            for r in items
-        ]
-        n = len(reels) or 1
-        return {
-            "reels": reels,
-            "totalReels": len(reels),
-            "avgViews": round(sum(r["videoViewCount"] for r in reels) / n),
-            "avgPlays": round(sum(r["videoPlayCount"] for r in reels) / n),
-        }
 
     async def fetch_facebook_page(self, page_url: str) -> dict | None:
-        items = await self._run_actor("apify~facebook-pages-scraper", {"startUrls": [{"url": page_url}]})
+        items = await self._run_actor(FB_PAGES_ACTOR, {"startUrls": [{"url": page_url}]})
         return items[0] if items else None
 
+    async def fetch_facebook_posts(self, page_url: str, limit: int = 20) -> list[dict]:
+        return await self._run_actor(
+            FB_POSTS_ACTOR, {"startUrls": [{"url": page_url}], "resultsLimit": limit}, limit=limit,
+        )
+
     async def get_facebook_page(self, page_url: str) -> dict | None:
-        page = await self.fetch_facebook_page(page_url)
+        # pages·posts 두 task 병렬 호출 (posts 실패해도 page만 있으면 진행)
-        if not page:
+        page, posts = await asyncio.gather(
+            self.fetch_facebook_page(page_url),
+            self.fetch_facebook_posts(page_url),
+            return_exceptions=True,
+        )
+        if isinstance(page, Exception) or not page:
             return None
+        if isinstance(posts, Exception):
+            posts = []
         return {
             "pageName": page.get("title") or page.get("name"),
             "profileImage": page.get("profilePictureUrl") or page.get("profilePhoto") or page.get("profilePic"),
             "pageUrl": page.get("pageUrl", page_url),
             "followers": page.get("followers", 0),
-            "likes": page.get("likes", 0),
+            "following": page.get("followings", 0),
+            "reviews": page.get("ratingCount", 0),
             "categories": page.get("categories", []),
-            "email": page.get("email"),
+            "website": page.get("website") or page.get("websites"),
-            "phone": page.get("phone"),
-            "website": page.get("website"),
-            "address": page.get("address"),
             "intro": page.get("intro"),
-            "rating": page.get("rating"),
+            "latestPosts": [
+                {
+                    "text": (p.get("text") or "")[:160],
+                    "likes": p.get("likes", 0),
+                    "reactions": p.get("topReactionsCount", 0),
+                    "shares": p.get("shares", 0),
+                    "views": p.get("viewsCount") or 0,
+                    "isVideo": p.get("isVideo", False),
+                    "timestamp": p.get("time") or p.get("timestamp"),
+                }
+                for p in (posts or []) if isinstance(p, dict)
+            ],
         }
 
     async def fetch_tiktok_profile(self, url: str) -> list[dict]:

app/integrations/llm/temp-prompt/report_prompt.txt

@@ -67,15 +67,15 @@
 {channel_logos}
 - channel_logos.channel_logos[]에 각 채널의 로고 설명(logo_description)과 공식 로고 일치 여부(is_official)가 있습니다.
 - **facebook_audit.pages[].logo** 는 짧은 판정 타이틀로: is_official=true면 `"일치 (공식 로고)"`, false면 `"불일치 (비공식 변형)"`. 그리고 **facebook_audit.pages[].logo_description** 에 해당 채널의 logo_description(설명문)을 넣으세요.
-- **instagram_audit.accounts[].profile_photo** 는 해당 채널 로고를 짧게 서술 (예: `"모델 사진 (브랜드 로고 아님)"`, `"VIEW 골드 로고"`). 긴 문장 말고 짧게.
 - 위 값들은 channel_logos 데이터 기반으로만 작성하고 추측하지 마세요.
 - 채널 간 로고 불일치(is_official=false)는 brand 일관성 진단(problem_diagnosis/weaknesses)에 반영하세요.
 
 ## clinic_snapshot / 채널 audit 작성 지침 (수집 데이터 그대로, 추측 금지)
 - clinic_snapshot.name 은 {clinic_name} 을 **그대로** 사용 (강남언니 표기명 '-본원' 등으로 바꾸지 말 것).
 - clinic_snapshot 의 overall_rating/total_reviews/staff_count/location/certifications/lead_doctor 는 강남언니({gangnam_unni}) 데이터의 값을 그대로 사용.
-- instagram_audit.accounts: KR 인스타({instagram})·영문 인스타({instagram_en}) 데이터가 있으면 **각각 별도 계정**으로 넣고, handle/followers/posts/following 은 그 데이터 수치를 그대로. KR=language "KR"·label "인스타그램 KR", EN=language "EN"·label "인스타그램 EN".
+- **instagram_audit.accounts 는 반드시 빈 배열 []로 두세요.** 계정 정보는 시스템이 수집 데이터로 직접 채우니 LLM은 만들지 말고, instagram_audit.diagnosis(진단)만 작성하세요.
 - facebook_audit.pages: KR 페북({facebook})·영문 페북({facebook_en}) 데이터가 있으면 **각각 별도 페이지**로 넣고, url/page_name/followers 등은 그 데이터 그대로. language/label 동일 규칙.
+- facebook_audit.pages[].top_content_type 은 해당 페이지 latestPosts의 **캡션·미디어를 읽고** 주로 올리는 콘텐츠를 의미 기반으로 짧게 묘사하세요 (예: "Before/After 사진 + 환자 여정 Reels", "이벤트·프로모션 카드뉴스", "다국어 시술 소개"). 단순 "동영상/이미지 위주"가 아니라 **무슨 주제**인지 쓰세요. (recent_post_age·post_frequency·engagement 수치는 시스템이 덮어쓰니 대략 적어도 됩니다.)
 - 위 수치·URL·이름은 제공된 데이터에서 그대로 쓰고 절대 지어내지 마세요.
 
 ## 기타 채널 현황 (other_channels) 작성 지침

app/services/analysis.py

@@ -4,6 +4,8 @@ from common.db import fetchone, execute, fetch_raw, get_analysis_raw_data, save_
 from integrations.llm.llm_service import LLMService
 from integrations.llm.prompt import report_prompt, plan_prompt
 from integrations.llm.schemas.report import ReportOutput
+from services.instagram_audit import build_instagram_accounts
+from services.facebook_audit import build_facebook_pages
 from integrations.llm.schemas.plan import PlanOutput
 from models.status import AnalysisStatus
 
@@ -134,24 +136,13 @@ async def _build_overrides(analysis_run_id: str) -> dict:
             "review_count": lead.get("reviews"),
         }
 
-    # ── instagram ─────────────────────────────────────────────────────────────
+    # ── instagram (KR·EN 계정을 코드에서 구성 → LLM 출력 무시하고 교체) ──────────────
-    ig_patch: dict = {}
+    ig_patch = build_instagram_accounts(
-    if instagram.get("username"):  ig_patch["handle"]       = instagram["username"]
+        instagram, hospital.get("instagramEn") or {}, hospital.get("channelLogos") or {},
-    if instagram.get("posts"):     ig_patch["posts"]        = instagram["posts"]
+    )
-    if instagram.get("followers"): ig_patch["followers"]    = instagram["followers"]
-    if instagram.get("following"): ig_patch["following"]    = instagram["following"]
-    if instagram.get("bio"):       ig_patch["bio"]          = instagram["bio"]
-    if instagram.get("username"):  ig_patch["profile_link"] = f"https://www.instagram.com/{instagram['username']}/"
 
-    # ── facebook ──────────────────────────────────────────────────────────────
+    # ── facebook (KR=facebook_data, EN=hospital.facebookEn 둘 다 코드 산출, [KR, EN] 순서) ──
-    fb_patch: dict = {}
+    fb_pages = build_facebook_pages(facebook, hospital.get("facebookEn") or {})
-    if facebook.get("pageUrl"):    fb_patch["url"]           = facebook["pageUrl"]
-    if facebook.get("pageUrl"):    fb_patch["link"]          = facebook["pageUrl"]
-    if facebook.get("pageName"):   fb_patch["page_name"]     = facebook["pageName"]
-    if facebook.get("followers"):  fb_patch["followers"]     = facebook["followers"]
-    if facebook.get("intro"):      fb_patch["bio"]           = facebook["intro"]
-    if facebook.get("categories"): fb_patch["category"]     = ", ".join(facebook["categories"])
-    if facebook.get("website"):    fb_patch["linked_domain"] = facebook["website"]
 
     # ── youtube ───────────────────────────────────────────────────────────────
     yt_patch: dict = {}
@@ -178,9 +169,9 @@ async def _build_overrides(analysis_run_id: str) -> dict:
     if snapshot:
         overrides["clinic_snapshot"] = snapshot
     if ig_patch:
-        overrides["instagram_audit"] = {"accounts": [ig_patch]}
+        overrides["instagram_audit"] = {"accounts": ig_patch}
-    if fb_patch:
+    if fb_pages:
-        overrides["facebook_audit"] = {"pages": [fb_patch]}
+        overrides["facebook_audit"] = {"pages": fb_pages}
     if yt_patch:
         overrides["youtube_audit"] = yt_patch
     return overrides
@@ -200,6 +191,8 @@ def _deep_merge(base: dict, overrides: dict) -> dict:
 
 def _patch_report(result: ReportOutput, overrides: dict) -> ReportOutput:
     merged = _deep_merge(result.model_dump(), overrides)
+    # 인스타 계정은 프롬프트에서 LLM이 []로 두게 했고, 코드가 수집 데이터로 채운다 (데이터 없으면 빈 리스트)
+    merged.setdefault("instagram_audit", {})["accounts"] = (overrides.get("instagram_audit") or {}).get("accounts") or []
     return ReportOutput(**merged)
 
 

app/services/collect.py

@@ -15,6 +15,7 @@ from integrations.naver import NaverClient
 from integrations.youtube import YouTubeClient
 from integrations.firecrawl import FirecrawlClient
 from services.enrichment import collect_brand_assets, collect_extra_channels, collect_channel_logos
+from services.facebook_audit import transform_for_storage as transform_facebook
 
 logger = logging.getLogger(__name__)
 
@@ -31,6 +32,7 @@ async def collect_facebook(analysis_run_id: str, row_id: int, url: str) -> None:
     logger.info("[facebook] start run=%s url=%s", analysis_run_id, url)
     await set_facebook_status(row_id, "processing")
     data = await ApifyClient(get_env("APIFY_API_TOKEN")).get_facebook_page(url)
+    data = transform_facebook(data)
     await save_facebook_raw_data(row_id, data)
     logger.info("[facebook] done run=%s", analysis_run_id)
 

app/services/enrichment.py

@@ -8,6 +8,7 @@ from common.utils import get_env
 from integrations.apify import ApifyClient
 from integrations.vision import VisionClient
 from integrations.color_extractor import extract_brand_assets_from_site
+from services.facebook_audit import transform_for_storage as transform_facebook
 
 logger = logging.getLogger(__name__)
 
@@ -115,6 +116,8 @@ async def collect_extra_channels(
         if isinstance(res, Exception):
             logger.warning("[extra_channels] %s 수집 실패: %s", key, res)
         elif res:
+            if key == "facebookEn":
+                res = transform_facebook(res)
             results[key] = res
     if not results:
         logger.info("[extra_channels] 수집 결과 없음 run=%s", analysis_run_id)

app/services/facebook_audit.py

@@ -0,0 +1,105 @@
+"""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]

app/services/instagram_audit.py

@@ -0,0 +1,48 @@
+"""Instagram audit 계정(KR·EN)을 수집 데이터로 구성.
+fix 값(handle/followers/highlights/content_format 등)은 전부 코드에서 박는다 — LLM 출력 무시."""
+
+_MEDIA = {"GraphImage": "이미지", "GraphSidecar": "카드뉴스", "GraphVideo": "영상/릴스"}
+
+
+def _content_format(data: dict) -> str:
+    """latestPosts 미디어 타입으로 콘텐츠 포맷 도출 (표기 순서는 _MEDIA 정의 순서로 고정)."""
+    present = {_MEDIA.get(p.get("type")) for p in (data.get("latestPosts") or [])}
+    return "/".join(m for m in _MEDIA.values() if m in present)
+
+
+def _logo_desc(channel_logos: dict, channel: str) -> str:
+    """channelLogos(비전 결과)에서 해당 채널 로고 설명만 가져온다."""
+    for c in (channel_logos or {}).get("channel_logos", []):
+        if c.get("channel") == channel:
+            return c.get("logo_description") or ""
+    return ""
+
+
+def _account(data: dict, language: str, label: str, channel: str, channel_logos: dict) -> dict:
+    """스크래퍼 수집값으로 InstagramAccount 전 필드를 구성."""
+    handle = data.get("username") or ""
+    return {
+        "handle":         handle,
+        "language":       language,
+        "label":          label,
+        "posts":          data.get("posts", 0),
+        "followers":      data.get("followers", 0),
+        "following":      data.get("following", 0),
+        "category":       data.get("category", ""),
+        "profile_link":   f"https://www.instagram.com/{handle}/" if handle else "",
+        "highlights":     data.get("highlights") or [],
+        "reels_count":    0,  # 릴스 스크래퍼 미사용
+        "content_format": _content_format(data),
+        "profile_photo":  _logo_desc(channel_logos, channel),
+        "bio":            data.get("bio", ""),
+    }
+
+
+def build_instagram_accounts(instagram: dict, instagram_en: dict, channel_logos: dict) -> list[dict]:
+    """KR·EN 인스타 계정 리스트 구성 (username 있는 것만)."""
+    accounts: list[dict] = []
+    if instagram.get("username"):
+        accounts.append(_account(instagram, "KR", "인스타그램 KR", "Instagram", channel_logos))
+    if instagram_en.get("username"):
+        accounts.append(_account(instagram_en, "EN", "인스타그램 EN", "Instagram EN", channel_logos))
+    return accounts