refactor(report): build_overrides + patch_report 통합 / list wholesale merge

- _build_overrides 가 result 받아 deep_merge 까지 처리, _patch_report 제거
- _deep_merge: list by-index → wholesale 치환 (EN 슬롯 누락/라벨 섞임 차단)
- build_facebook_audit: template-copy 대신 LLM logo/logo_description 만 두 페이지에 공통 적용
- _page_patch: language/label 명시 박음 (KR/EN 교차 오염 방지)
- FacebookPage/InstagramAccount/YouTubeAudit: 불필요한 Optional 제거, has_whatsapp/top_content_type 만 Optional 유지
- build_instagram_audit/build_facebook_audit: dict 반환 (overrides[k] = patch 단순 박기)

app/api/analysis.py

@@ -5,23 +5,50 @@ from common.deps import verify_api_key
 from common.db.hospital import select_hospital
 from common.db.source import select_source_mainpage, insert_source, insert_raw_info
 from common.db.run import insert_run, select_run_status
+from common.utils import _normalize_homepage, _with_scheme
 from models.analysis import AnalysisCreate, AnalysisStartResponse, AnalysisStatusResponse
 from models.file import FileListItem, FileType, FileUploadResponse
 from models.status import AnalysisStatus, SourceType
 from services.pipeline import run_pipeline
 from services.file_data import get_analysis_files_response, handle_analysis_file_upload, soft_delete_analysis_file
+from mock_urls import MOCK_CLINICS
 
 router = APIRouter(prefix="/api/analysis", tags=["analysis"], dependencies=[Depends(verify_api_key)])
 logger = logging.getLogger(__name__)
 
 
+# 클라가 일부만 보내거나 빈 값이면 mock_urls 의 동일 homepage 매칭으로 채워줌 (메인 + 부가 채널 동일 규칙).
+def _channels_from_mockurls(homepage_url: str) -> dict:
+    target = _normalize_homepage(homepage_url)
+    if not target:
+        return {}
+    for c in MOCK_CLINICS:
+        urls = c["urls"]
+        if _normalize_homepage(urls.get("homepage", "")) == target:
+            return {
+                # main
+                "instagram":    _with_scheme(urls.get("instagram")),
+                "facebook":     _with_scheme(urls.get("facebook")),
+                "naver_blog":   _with_scheme(urls.get("naverBlog")),
+                "youtube":      _with_scheme(urls.get("youtube")),
+                "gangnam_unni": _with_scheme(urls.get("gangnamUnni")),
+                # extra
+                "tiktok":       _with_scheme(urls.get("tiktok")),
+                "instagram_en": _with_scheme(urls.get("instagramEn")),
+                "facebook_en":  _with_scheme(urls.get("facebookEn")),
+                "kakao_talk":   _with_scheme(urls.get("kakaoTalk")),
+                "naver_cafe":   _with_scheme(urls.get("naverCafe")),
+            }
+    return {}
+
+
 @router.post("", status_code=status.HTTP_202_ACCEPTED, response_model=AnalysisStartResponse)
 async def start_analysis(body: AnalysisCreate, background_tasks: BackgroundTasks):
     logger.info("POST /api/analysis clinic_id=%s", body.clinic_id)
     analysis_run_id = str(uuid6.uuid7())
     hospital_id = body.clinic_id
 
-    # 사실 hospital과 owner_user_id 비교 후 검증이 필요한 거지만 일단 PoC 니까. 나중에 바꿉니다.
+    # 사실 hospital 과 owner_user_id 비교 후 검증이 필요한 거지만 일단 PoC 니까. 나중에 바꿉니다.
     hospital = await select_hospital(hospital_id)
     if not hospital:
         raise HTTPException(status_code=409, detail="Clinic not found")
@@ -31,19 +58,40 @@ async def start_analysis(body: AnalysisCreate, background_tasks: BackgroundTasks
     mainpage = await select_source_mainpage(hospital_id)
     if mainpage:
         await insert_raw_info(mainpage["source_id"], analysis_run_id, data_tag=SourceType.MAINPAGE)
+        # branding (HTML/CSS + Vision 로고 매칭) — mainpage 와 같은 homepage URL 을 source 로 사용.
+        branding_id = await insert_source(hospital_id, SourceType.BRANDING, mainpage["url"], language="KR")
+        await insert_raw_info(branding_id, analysis_run_id, data_tag=SourceType.BRANDING)
 
-    channels = [
-        (SourceType.INSTAGRAM,    body.channels.instagram),
-        (SourceType.FACEBOOK,     body.channels.facebook),
-        (SourceType.NAVER_BLOG,   body.channels.naver_blog),
-        (SourceType.YOUTUBE,      body.channels.youtube),
-        (SourceType.GANGNAM_UNNI, body.channels.gangnam_unni),
+    # 클라가 안 보낸 채널은 mock_urls 에서 homepage 매칭으로 보충 (main + extra 동일 규칙).
+    mock = _channels_from_mockurls((mainpage or {}).get("url") or "")
+
+    # 메인 5채널 (KR). _with_scheme 으로 'gangnamunni.com/...' 같이 scheme/www 없이 와도 보강.
+    main_channels = [
+        (SourceType.INSTAGRAM,    _with_scheme(body.channels.instagram)    or mock.get("instagram")),
+        (SourceType.FACEBOOK,     _with_scheme(body.channels.facebook)     or mock.get("facebook")),
+        (SourceType.NAVER_BLOG,   _with_scheme(body.channels.naver_blog)   or mock.get("naver_blog")),
+        (SourceType.YOUTUBE,      _with_scheme(body.channels.youtube)      or mock.get("youtube")),
+        (SourceType.GANGNAM_UNNI, _with_scheme(body.channels.gangnam_unni) or mock.get("gangnam_unni")),
     ]
-    for source_type, url in channels:
+    for source_type, url in main_channels:
         if url:
-            source_id = await insert_source(hospital_id, source_type, url)
+            source_id = await insert_source(hospital_id, source_type, url, language="KR")
             await insert_raw_info(source_id, analysis_run_id, data_tag=source_type)
 
+    # 부가 채널 — instagram_en/facebook_en 은 동일 source_type 에 language='EN' 으로 구분, 나머지는 자체 source_type.
+    extra_channels = [
+        (SourceType.INSTAGRAM, "EN", _with_scheme(body.channels.instagram_en) or mock.get("instagram_en")),
+        (SourceType.FACEBOOK,  "EN", _with_scheme(body.channels.facebook_en)  or mock.get("facebook_en")),
+        (SourceType.TIKTOK,    "KR", _with_scheme(body.channels.tiktok)       or mock.get("tiktok")),
+        (SourceType.KAKAOTALK, "KR", _with_scheme(body.channels.kakao_talk)   or mock.get("kakao_talk")),
+        (SourceType.NAVER_CAFE, "KR", _with_scheme(body.channels.naver_cafe)  or mock.get("naver_cafe")),
+    ]
+    for source_type, language, url in extra_channels:
+        if url:
+            source_id = await insert_source(hospital_id, source_type, url, language=language)
+            await insert_raw_info(source_id, analysis_run_id, data_tag=source_type)
+
+    logger.info("[analysis] main+extra channels resolved (mock_matched=%s)", bool(mock))
     background_tasks.add_task(run_pipeline, analysis_run_id)
 
     return AnalysisStartResponse(

app/api/plan.py

@@ -2,9 +2,12 @@ import json
 import logging
 from fastapi import APIRouter, Depends, HTTPException, Response
 from common.db.run import select_run_with_clinic
+from common.db.source import select_run_source_raw
 from common.deps import verify_api_key
+from common.utils import _with_scheme
 from integrations.llm.schemas.plan import PlanOutput
 from models.plan import PlanApiResponse
+from models.status import SourceType
 
 router = APIRouter(prefix="/api/plan", tags=["plan"], dependencies=[Depends(verify_api_key)])
 logger = logging.getLogger(__name__)
@@ -20,11 +23,14 @@ async def get_plan(run_id: str):
         return Response(status_code=204)
     data = json.loads(row["plan_data"]) if isinstance(row["plan_data"], str) else row["plan_data"]
     plan = PlanOutput(**data)
+    # 강남언니에서 긁어온 이름이 있으면 우선 (hospital_baseinfo 의 정식 이름보다 강남언니가 더 광고용 표기).
+    gu = await select_run_source_raw(run_id, SourceType.GANGNAM_UNNI) or {}
+    clinic_name = gu.get("name") or row["hospital_name"]
     return PlanApiResponse(
         id=run_id,
-        clinic_name=row["hospital_name"],
+        clinic_name=clinic_name,
         clinic_name_en=row["hospital_name_en"],
         created_at=str(row["created_at"]),
-        target_url=row["target_url"],
+        target_url=_with_scheme(row["target_url"]),
         **plan.model_dump(),
     )

app/api/report.py

@@ -3,6 +3,7 @@ import logging
 from fastapi import APIRouter, Depends, HTTPException, Response
 from common.db.run import select_run_with_clinic
 from common.deps import verify_api_key
+from common.utils import _with_scheme
 from integrations.llm.schemas.report import ReportOutput
 from models.report import MarketingReportResponse
 
@@ -25,6 +26,6 @@ async def get_report(run_id: str):
         clinic_name=row["hospital_name"],
         clinic_name_en=row["hospital_name_en"],
         created_at=str(row["created_at"]),
-        target_url=row["target_url"],
+        target_url=_with_scheme(row["target_url"]),
         **llm_output.model_dump(exclude={"id", "created_at", "target_url"}),
     )

app/common/db/__init__.py

@@ -1,14 +1,16 @@
 from common.db.base import execute, fetchone, fetchall
 from common.db.hospital import select_hospital, update_hospital_status, insert_hospital, update_hospital
 from common.db.source import (
-    insert_source, select_source_mainpage,
-    insert_raw_info, update_raw_info_status, update_raw_info,
+    insert_source, select_source_mainpage, select_source_by_type,
+    insert_raw_info, update_raw_info_status, update_raw_info, update_raw_info_merge,
+    update_raw_info_logo_url, select_mainpage_logo_url, select_branding_info_id,
     select_raw_info_data,
-    select_run_sources, select_run_raw_data, select_run_mainpage_url,
+    select_run_sources, select_run_raw_data, select_run_source_raw,
+    select_run_mainpage_url,
 )
 from common.db.run import (
     insert_run, select_run, select_run_status, update_run_status,
-    update_run_report, update_run_plan, select_run_with_clinic,
+    update_run_report, update_run_plan, select_run_with_clinic, select_run_report_data,
 )
 from common.db.market import upsert_market_status, upsert_market_result, select_market
 from common.db.file_data import insert_file, select_run_files, select_file, delete_file

app/common/db/run.py

@@ -22,6 +22,18 @@ async def select_run(analysis_run_id: str) -> dict | None:
     )
 
 
+async def select_run_report_data(analysis_run_id: str) -> dict | None:
+    """report 결과가 필요할 때만 호출. raw JSON 파싱해서 dict 반환."""
+    import json
+    row = await fetchone(
+        "SELECT report_data FROM analysis_runs WHERE analysis_run_id = %s",
+        (analysis_run_id,),
+    )
+    if not row or not row["report_data"]:
+        return None
+    return json.loads(row["report_data"]) if isinstance(row["report_data"], str) else row["report_data"]
+
+
 async def select_run_status(analysis_run_id: str) -> str | None:
     row = await fetchone(
         "SELECT status FROM analysis_runs WHERE analysis_run_id = %s",

app/common/db/source.py

@@ -17,7 +17,7 @@ async def insert_source(
 
 async def select_source_mainpage(hospital_id: str) -> dict | None:
     return await fetchone(
-        "SELECT source_id FROM remote_source WHERE hospital_id = %s AND source_type = 'mainpage'",
+        "SELECT source_id, url FROM remote_source WHERE hospital_id = %s AND source_type = 'mainpage'",
         (hospital_id,),
     )
 
@@ -64,7 +64,7 @@ async def select_run_sources(analysis_run_id: str) -> list[dict]:
 
 async def select_run_raw_data(analysis_run_id: str) -> dict:
     rows = await fetchall(
-        "SELECT rs.source_type, ri.raw_data"
+        "SELECT rs.source_type, rs.language, ri.raw_data, ri.logo_url"
         " FROM raw_info ri JOIN remote_source rs USING (source_id)"
         " WHERE ri.analysis_run_id = %s",
         (analysis_run_id,),
@@ -72,10 +72,87 @@ async def select_run_raw_data(analysis_run_id: str) -> dict:
     result: dict = {}
     for row in rows:
         raw = row["raw_data"]
-        result[row["source_type"]] = json.loads(raw) if isinstance(raw, str) else raw
+        key = row["source_type"]
+        if (row.get("language") or "").upper() == "EN":
+            key = f"{key}_en"
+        data = json.loads(raw) if isinstance(raw, str) else (raw or {})
+        if isinstance(data, dict) and row.get("logo_url"):
+            data["_logo_url"] = row["logo_url"]
+        result[key] = data
     return result
 
 
+async def select_run_source_raw(
+    analysis_run_id: str, source_type: str, language: str | None = None,
+) -> dict | None:
+    sql = (
+        "SELECT ri.raw_data FROM raw_info ri JOIN remote_source rs USING (source_id)"
+        " WHERE ri.analysis_run_id = %s AND rs.source_type = %s"
+    )
+    args: tuple = (analysis_run_id, source_type)
+    if language:
+        sql += " AND rs.language = %s"
+        args = (*args, language)
+    sql += " LIMIT 1"
+    row = await fetchone(sql, args)
+    if not row or not row["raw_data"]:
+        return None
+    return json.loads(row["raw_data"]) if isinstance(row["raw_data"], str) else row["raw_data"]
+
+
+async def update_raw_info_logo_url(info_id: int, logo_url: str) -> None:
+    """raw_info.logo_url 컬럼에 로고 URL 저장 (JSON raw_data 와 분리해 컬럼 인덱스/조회 용이)."""
+    await execute(
+        "UPDATE raw_info SET logo_url = %s WHERE info_id = %s",
+        (logo_url, info_id),
+    )
+
+
+async def select_branding_info_id(analysis_run_id: str) -> int | None:
+    row = await fetchone(
+        "SELECT ri.info_id FROM raw_info ri JOIN remote_source rs USING (source_id)"
+        " WHERE ri.analysis_run_id = %s AND rs.source_type = 'branding' LIMIT 1",
+        (analysis_run_id,),
+    )
+    return (row or {}).get("info_id")
+
+
+async def select_mainpage_logo_url(analysis_run_id: str) -> str | None:
+    row = await fetchone(
+        "SELECT ri.logo_url FROM raw_info ri JOIN remote_source rs USING (source_id)"
+        " WHERE ri.analysis_run_id = %s AND rs.source_type = 'mainpage' LIMIT 1",
+        (analysis_run_id,),
+    )
+    return (row or {}).get("logo_url")
+
+
+async def update_raw_info_merge(info_id: int, patch: dict) -> None:
+    """raw_info.raw_data 를 read-modify-write 로 top-level 머지.
+    한 source 가 단계별로 (예: branding 의 brandAssets → channelLogos) 키를 덧붙일 때 사용."""
+    row = await fetchone("SELECT raw_data FROM raw_info WHERE info_id = %s", (info_id,))
+    if not row:
+        return
+    raw = row["raw_data"]
+    data = json.loads(raw) if isinstance(raw, str) else (raw or {})
+    data.update(patch)
+    await execute(
+        "UPDATE raw_info SET raw_data = %s, status = 'done' WHERE info_id = %s",
+        (json.dumps(data, ensure_ascii=False), info_id),
+    )
+
+
+async def select_source_by_type(
+    hospital_id: str, source_type: str, language: str | None = None,
+) -> dict | None:
+    sql = "SELECT source_id, url FROM remote_source WHERE hospital_id = %s AND source_type = %s"
+    args: tuple = (hospital_id, source_type)
+    if language:
+        sql += " AND language = %s"
+        args = (*args, language)
+    sql += " LIMIT 1"
+    return await fetchone(sql, args)
+
+
 async def select_run_mainpage_url(analysis_run_id: str) -> str:
     row = await fetchone(
         "SELECT rs.url FROM raw_info ri JOIN remote_source rs USING (source_id)"

app/common/utils.py

@@ -1,11 +1,36 @@
 import os
 import asyncio
+import logging
+from datetime import datetime, timezone
 from http import HTTPMethod
 import httpx
 
+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, "")
@@ -37,3 +62,48 @@ async def http_request(
                     print(f"  [error] {label} → {e}")
                     return None
     return None
+
+
+async def _run_optional_step(coro, label: str) -> None:
+    """부가 단계 실행 헬퍼: 예외를 삼키고 경고 로그만 남겨 호출측 흐름이 멈추지 않게 격리."""
+    try:
+        await coro
+    except Exception as e:
+        logger.warning("%s 실패 (무시하고 진행): %s", label, e)
+
+
+def _normalize_homepage(url: str) -> str:
+    """URL을 scheme/www/끝슬래시 제거 + 소문자로 정규화 (homepage 매칭용)."""
+    u = (url or "").strip().lower()
+    for p in ("https://", "http://"):
+        if u.startswith(p):
+            u = u[len(p):]
+    if u.startswith("www."):
+        u = u[4:]
+    return u.rstrip("/")
+
+
+# SSL 인증서가 www.* 에만 유효한 도메인 — bare 도메인이면 사용자 클릭 시 브라우저 SSL warning 뜸.
+_WWW_REQUIRED = ("gangnamunni.com", "facebook.com", "instagram.com", "toxnfill.com")
+
+
+def _with_scheme(u: str | None) -> str | None:
+    """scheme 없는 URL에 https:// 보정 (수집기/링크 표시용). 빈 값은 None.
+    + 중첩된 https://가 끼어있으면 마지막 URL만 추출 (LLM이 가끔 'https://www.X/https://Y' 같이 만듦).
+    + SSL 엄격 도메인(gangnamunni/facebook/instagram)은 www. 자동 보강."""
+    if not u:
+        return None
+    u = u.strip()
+    # 'https://www.facebook.com/https://facebook.com/X' 같은 중첩 → 마지막 'http(s)://' 부터 잘라 사용
+    last = max(u.rfind("https://"), u.rfind("http://"))
+    if last > 0:
+        u = u[last:]
+    if "://" not in u:
+        u = "https://" + u
+    # scheme 뒤가 www. 없이 SSL 엄격 도메인이면 www. 추가
+    for dom in _WWW_REQUIRED:
+        for scheme in ("https://", "http://"):
+            if u.startswith(scheme + dom):
+                u = scheme + "www." + u[len(scheme):]
+                break
+    return u

app/integrations/apify.py

@@ -1,16 +1,32 @@
+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"
+
+# TikTok
+TIKTOK_ACTOR = "clockworks~tiktok-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 +42,53 @@ 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:
-        username = urlparse(url).path.strip("/").split("/")[0] if "://" in url else url.lstrip("@")
-        print(username)
-        items = await self._run_actor("apify~instagram-profile-scraper", {"usernames": [username], "resultsLimit": 12})
+    async def fetch_instagram_profile(self, username: str) -> dict | None:
+        items = await self._run_actor(IG_PROFILE_ACTOR, {"usernames": [username]})
         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)
-        if not profile or profile.get("error"):
+        username = _ig_username(url)
+        # 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 = []
+        # 프로필상 하이라이트가 있다고 하면(highlight_reel_count>0) 빈 결과일 때 최대 2회 재시도.
+        if not highlights and (profile.get("highlight_reel_count", 0) or profile.get("highlightReelCount", 0)) > 0:
+            for _ in range(2):
+                retry = await self.fetch_instagram_highlights(username)
+                if retry:
+                    highlights = retry
+                    break
         return {
             "username": profile["username"],
+            "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에 혼동을 주는 듯 하여 비활성화
+            "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,87 +98,119 @@ class ApifyClient:
             ],
         }
 
-    async def fetch_instagram_posts(self, url: str, limit: int = 20) -> list[dict]:
-        username = urlparse(url).path.strip("/").split("/")[0] if "://" in url else url.lstrip("@")
-        return await self._run_actor("apify~instagram-post-scraper", {
-            "username": [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)
-        posts = [
-            {
-                "id": p["id"],
-                "type": p.get("type"),
-                "url": p.get("url"),
-                "caption": (p.get("caption") or "")[:500],
-                "hashtags": p.get("hashtags", []),
-                "likesCount": p.get("likesCount", 0),
-                "commentsCount": p.get("commentsCount", 0),
-                "timestamp": p.get("timestamp"),
-            }
-            for p in items
-        ]
-        n = len(posts) or 1
-        return {
-            "posts": posts,
-            "totalPosts": len(posts),
-            "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("@")
-        print(username)
-        return await self._run_actor("apify~instagram-reel-scraper", {
-            "username": [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),
-        }
+    # 인스타 post 스크래퍼는 현재 파이프라인 미사용 — 비활성화 (필요 시 복구)
+    # async def fetch_instagram_posts(self, url: str, limit: int = 20) -> list[dict]:
+    #     username = urlparse(url).path.strip("/").split("/")[0] if "://" in url else url.lstrip("@")
+    #     return await self._run_actor("apify~instagram-post-scraper", {
+    #         "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)
+    #     posts = [
+    #         {
+    #             "id": p["id"],
+    #             "type": p.get("type"),
+    #             "url": p.get("url"),
+    #             "caption": (p.get("caption") or "")[:500],
+    #             "hashtags": p.get("hashtags", []),
+    #             "likesCount": p.get("likesCount", 0),
+    #             "commentsCount": p.get("commentsCount", 0),
+    #             "timestamp": p.get("timestamp"),
+    #         }
+    #         for p in items
+    #     ]
+    #     n = len(posts) or 1
+    #     return {
+    #         "posts": posts,
+    #         "totalPosts": len(posts),
+    #         "avgLikes": round(sum(p["likesCount"] for p in posts) / n),
+    #         "avgComments": round(sum(p["commentsCount"] for p in posts) / 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)
-        if not page:
+        # pages·posts 두 task 병렬 호출 (posts 실패해도 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"),
-            "phone": page.get("phone"),
-            "website": page.get("website"),
-            "address": page.get("address"),
+            "website": page.get("website") or page.get("websites"),
             "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]:
+        user = urlparse(url).path.strip("/").lstrip("@").split("/")[0] if "://" in url else url.lstrip("@")
+        return await self._run_actor(TIKTOK_ACTOR, {
+            "profiles": [user],
+            "resultsPerPage": 10,
+            "profileScrapeSections": ["videos"],
+            "profileSorting": "latest",
+            "shouldDownloadVideos": False,
+            "shouldDownloadCovers": False,
+            "shouldDownloadSubtitles": False,
+        })
+
+    async def get_tiktok_profile(self, url: str) -> dict | None:
+        items = await self.fetch_tiktok_profile(url)
+        if not items:
+            return None
+        author = (items[0] or {}).get("authorMeta") or {}
+        videos = [
+            {
+                "title": (v.get("text") or "")[:300],
+                "playCount": v.get("playCount", 0),
+                "diggCount": v.get("diggCount", 0),
+                "commentCount": v.get("commentCount", 0),
+                "shareCount": v.get("shareCount", 0),
+                "createTime": v.get("createTimeISO"),
+                "url": v.get("webVideoUrl"),
+            }
+            for v in items if isinstance(v, dict)
+        ]
+        return {
+            "handle": author.get("name"),
+            "profileImage": author.get("avatar"),
+            "nickname": author.get("nickName"),
+            "followers": author.get("fans", 0),
+            "following": author.get("following", 0),
+            "likes": author.get("heart", 0),
+            "videoCount": author.get("video", 0),
+            "verified": author.get("verified", False),
+            "bio": author.get("signature", ""),
+            "recentVideos": videos[:10],
         }

app/integrations/firecrawl.py

@@ -76,7 +76,7 @@ class FirecrawlClient:
                 "url": url,
                 "formats": ["json", "links"],
                 "jsonOptions": {
-                    "prompt": "Extract: clinic name (Korean), clinic name (English), address, phone with dash format, business hours, slogan, services offered, doctors with name/title/specialty, brand identity (primary/accent/background/text colors in hex, heading/body fonts, logo URL, favicon URL)",
+                    "prompt": "Extract: clinic name (Korean), clinic name (English), address, phone with dash format, business hours, slogan, services offered, doctors with name/title/specialty, brand identity (primary/accent/background/text colors in hex, heading/body fonts, logo URL from the actual header/main <img> src, og:image from <meta property='og:image'> content, favicon URL)",
                     "schema": {
                         "type": "object",
                         "properties": {
@@ -119,6 +119,7 @@ class FirecrawlClient:
                                     "headingFont":     {"type": "string"},
                                     "bodyFont":        {"type": "string"},
                                     "logoUrl":         {"type": "string"},
+                                    "ogImage":         {"type": "string"},
                                     "faviconUrl":      {"type": "string"},
                                 },
                             },

app/integrations/llm/gemini_vision.py

@@ -0,0 +1,334 @@
+"""Gemini Vision — 로고/브랜드 비주얼 자동 분석 (OpenAI 호환 모드).
+
+정확한 hex 색상은 color_extractor가 CSS에서 직접 뽑음 (Vision은 근사값밖에 못 냄).
+Vision은 사람이 봐야 알 수 있는 정성 정보 — 심볼 형태/워드마크/톤 — 를 담당.
+"""
+import asyncio
+import base64
+import json
+import logging
+import re
+import ssl
+import httpx
+import resvg_py
+from openai import AsyncOpenAI
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_MODEL = "gemini-2.5-flash"
+
+
+class VisionClient:
+    """Gemini Vision을 OpenAI 호환 endpoint로 호출. GEMINI_API_KEY만 필요."""
+
+    def __init__(self, api_key: str, model: str = DEFAULT_MODEL, timeout: float = 30.0, max_retries: int = 2):
+        self.client = AsyncOpenAI(
+            api_key=api_key,
+            base_url="https://generativelanguage.googleapis.com/v1beta/openai/",
+            timeout=timeout,
+            max_retries=max_retries,
+        )
+        self.model = model
+
+    @staticmethod
+    def _extract_json(text: str) -> dict | None:
+        if not text:
+            return None
+        m = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", text, re.DOTALL)
+        if m:
+            try:
+                return json.loads(m.group(1))
+            except json.JSONDecodeError:
+                pass
+        m = re.search(r"\{.*\}", text, re.DOTALL)
+        if m:
+            try:
+                return json.loads(m.group(0))
+            except json.JSONDecodeError:
+                return None
+        return None
+
+    @staticmethod
+    async def _fetch_as_data_url(url: str) -> str | None:
+        """Gemini는 URL 직접 fetch가 막힌 호스트가 많아 base64 인라인으로 변환.
+        + 'image does not exist' 같은 placeholder 이미지 거부 (작은 bytes / 잘못된 content-type).
+        + 한국 의료 사이트 중 SSL이 약해서 표준 검증에 실패하는 곳 대응 (3단 SSL fallback)."""
+        headers = {"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"}
+
+        def _weak_ctx() -> ssl.SSLContext:
+            ctx = ssl.create_default_context()
+            try:
+                ctx.set_ciphers("DEFAULT@SECLEVEL=1")
+            except ssl.SSLError:
+                pass
+            return ctx
+
+        last_err: Exception | None = None
+        for verify in (True, _weak_ctx(), False):
+            try:
+                async with httpx.AsyncClient(
+                    timeout=15.0, follow_redirects=True, headers=headers, verify=verify,
+                ) as c:
+                    resp = await c.get(url)
+                if resp.status_code != 200:
+                    logger.warning("[vision] fetch %s status=%s", url, resp.status_code)
+                    return None
+                mime = resp.headers.get("content-type", "").split(";")[0].strip()
+                # 실제 이미지가 아니면 거부 (HTML 페이지가 404 대신 200으로 리다이렉트 되는 경우)
+                if not mime.startswith("image/"):
+                    logger.warning("[vision] %s not an image (content-type=%s)", url, mime)
+                    return None
+                # SVG는 Gemini가 못 보므로 즉시 PNG로 래스터화 (resvg, in-memory ~1ms)
+                content = resp.content
+                if mime == "image/svg+xml" or url.lower().split("?")[0].endswith(".svg"):
+                    try:
+                        content = bytes(resvg_py.svg_to_bytes(svg_string=resp.text))
+                        mime = "image/png"
+                    except Exception as e:
+                        logger.warning("[vision] svg rasterize failed %s: %s", url, e)
+                        return None
+                size = len(content)
+                if size < 500:
+                    logger.warning("[vision] %s too small (%d bytes) — likely placeholder", url, size)
+                    return None
+                b64 = base64.b64encode(content).decode("ascii")
+                return f"data:{mime};base64,{b64}"
+            except (httpx.ConnectError, httpx.ReadError, ssl.SSLError) as e:
+                last_err = e
+                continue
+            except Exception as e:
+                logger.warning("[vision] fetch error %s: %s", url, e)
+                return None
+        logger.warning("[vision] fetch %s SSL fallback all failed: %s", url, last_err)
+        return None
+
+    async def _ask(self, image_urls: list[str], prompt: str, max_tokens: int = 4000) -> dict | None:
+        content: list[dict] = []
+        for u in image_urls:
+            if not u:
+                continue
+            data_url = await self._fetch_as_data_url(u)
+            if not data_url:
+                continue
+            content.append({"type": "image_url", "image_url": {"url": data_url}})
+        if not any(c.get("type") == "image_url" for c in content):
+            logger.warning("[vision] no images could be fetched")
+            return None
+        content.append({"type": "text", "text": prompt})
+
+        try:
+            resp = await self.client.chat.completions.create(
+                model=self.model,
+                messages=[{"role": "user", "content": content}],
+                max_tokens=max_tokens,
+            )
+            choice = resp.choices[0]
+            if choice.finish_reason != "stop":
+                logger.warning("[vision] unexpected finish_reason=%s", choice.finish_reason)
+            return self._extract_json(choice.message.content or "")
+        except Exception as e:
+            logger.warning("[vision] error: %s", e)
+            return None
+
+    async def describe_svg_text(self, svg_url: str) -> dict | None:
+        """SVG는 Gemini Vision이 못 보지만 XML 텍스트 자체는 LLM이 읽을 수 있음.
+        SVG 소스를 받아 그대로 text endpoint에 던지고 색·심볼·텍스트를 추론하게 함.
+        analyze_brand_assets와 동일한 스키마(logo_description/style/has_symbol/...) 반환."""
+        headers = {"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"}
+
+        def _weak_ctx() -> ssl.SSLContext:
+            ctx = ssl.create_default_context()
+            try:
+                ctx.set_ciphers("DEFAULT@SECLEVEL=1")
+            except ssl.SSLError:
+                pass
+            return ctx
+
+        svg_text: str | None = None
+        for verify in (True, _weak_ctx(), False):
+            try:
+                async with httpx.AsyncClient(
+                    timeout=15.0, follow_redirects=True, headers=headers, verify=verify,
+                ) as c:
+                    resp = await c.get(svg_url)
+                if resp.status_code == 200:
+                    svg_text = resp.text
+                break
+            except (httpx.ConnectError, httpx.ReadError, ssl.SSLError):
+                continue
+            except Exception as e:
+                logger.warning("[vision] svg fetch error %s: %s", svg_url, e)
+                return None
+        if not svg_text:
+            logger.warning("[vision] svg fetch failed %s", svg_url)
+            return None
+        # 페이로드 폭주 방지 — 평범한 로고 SVG는 수 KB 수준
+        if len(svg_text) > 60000:
+            svg_text = svg_text[:60000]
+
+        prompt = (
+            "아래는 병원 로고 SVG 소스 코드입니다. SVG 마크업(path/circle/text/fill/stroke 등)을 "
+            "읽고 로고의 시각적 특징을 추론해 아래 JSON 스키마로만 응답하세요. 코드펜스 없이 순수 JSON.\n"
+            "{\n"
+            '  "logo_description": "심볼 형태 + 워드마크 + 톤을 1~2문장 한국어로",\n'
+            '  "logo_style": "minimal | illustrative | typographic | abstract 중 하나",\n'
+            '  "has_symbol": "심볼/아이콘이 있으면 true, 글자만 있으면 false (boolean)",\n'
+            '  "logo_symbol": "심볼 묘사 (예: \'잎사귀\'). 없으면 빈 문자열",\n'
+            '  "logo_text": "워드마크 텍스트 그대로. <text> 태그 내용 우선",\n'
+            '  "logo_colors_desc": "쓰인 색감을 사람이 부르는 이름으로 (예: \'딥네이비 + 골드\'). hex 출력 금지"\n'
+            "}\n"
+            "주의: hex 값이나 URL은 출력하지 마세요 (별도 추출 로직 처리). 모든 텍스트는 한국어로.\n\n"
+            "SVG 소스:\n"
+            f"{svg_text}"
+        )
+        try:
+            resp = await self.client.chat.completions.create(
+                model=self.model,
+                messages=[{"role": "user", "content": prompt}],
+                max_tokens=8000,  # Gemini 2.5는 thinking 토큰을 max_tokens에서 차감하므로 여유 필요
+            )
+            choice = resp.choices[0]
+            if choice.finish_reason != "stop":
+                logger.warning("[vision] svg describe finish_reason=%s", choice.finish_reason)
+            result = self._extract_json(choice.message.content or "")
+        except Exception as e:
+            logger.warning("[vision] svg describe error: %s", e)
+            return None
+        if not result:
+            return None
+        result["logo_images"] = {"circle": None, "horizontal": svg_url, "korean": None}
+        return result
+
+    async def analyze_brand_assets(
+        self,
+        logo_url: str | None,
+        homepage_url: str | None,
+        additional_images: list[str] | None = None,
+    ) -> dict:
+        """로고 이미지를 보고 정성 분석. 정확한 hex는 color_extractor가 따로 처리하므로 여기선 안 뽑음."""
+        urls = [u for u in [logo_url] + list(additional_images or []) if u]
+        if not urls:
+            return {}
+        prompt = (
+            "당신은 브랜드 로고 시각 분석가입니다. 첨부된 이미지(첫 번째가 병원의 대표 로고)를 보고 "
+            "아래 JSON 스키마로만 응답하세요. 코드펜스 없이 순수 JSON만 출력.\n"
+            "{\n"
+            '  "logo_description": "로고를 1~2문장으로 설명 (심볼 형태 + 워드마크 + 전반적 톤). 예: \'둥근 잎사귀를 감싼 추상 심볼에 세리프 한글 워드마크, 차분하고 고급스러운 톤\'",\n'
+            '  "logo_style": "minimal | illustrative | typographic | abstract 중 하나",\n'
+            '  "has_symbol": "심볼/아이콘이 있으면 true, 글자만 있으면 false (boolean)",\n'
+            '  "logo_symbol": "심볼이 묘사하는 대상 (예: \'잎사귀\', \'추상 곡선\'). 없으면 빈 문자열",\n'
+            '  "logo_text": "로고에 보이는 워드마크 텍스트 그대로 (한글/영문). 없으면 빈 문자열",\n'
+            '  "logo_colors_desc": "로고에 쓰인 색감을 사람이 부르는 이름으로 서술 (예: \'딥네이비 + 골드\')",\n'
+            '  "logo_colors_hex": ["로고에서 시각적으로 두드러진 색 정확히 5개의 hex 근사값 배열. 예: [\'#1A2B3C\', \'#D4A017\', \'#FFFFFF\', \'#9E5C2A\', \'#1F1F1F\']. 강한 색이 5개 안 되면 음영/명도 차이로 5개 채울 것. 빈 배열 금지."]\n'
+            "}\n"
+            "주의: logo_colors_hex 는 시각 추정이라 정확도 떨어질 수 있음. CSS 추출이 우선이고 이건 fallback/보완 용.\n"
+            "모든 설명/텍스트 값은 반드시 한국어로 작성하세요 (영어 금지)."
+        )
+        result = await self._ask(urls, prompt)
+        if not result:
+            return {}
+        # logo_images는 우리가 직접 채움 (Vision은 묘사만)
+        result["logo_images"] = {"circle": None, "horizontal": logo_url, "korean": None}
+        # logo_colors_hex 5개 강제 정규화 — LLM 이 4개나 6개 줄 수도 있어서 길이 fallback.
+        hex_list = [h for h in (result.get("logo_colors_hex") or []) if isinstance(h, str) and h.startswith("#")]
+        if hex_list:
+            while len(hex_list) < 5:
+                hex_list.append(hex_list[-1])  # 마지막 색 복제로 패딩
+            result["logo_colors_hex"] = hex_list[:5]
+        else:
+            result["logo_colors_hex"] = []
+        return result
+
+    async def describe_channel_logos(
+        self,
+        official_logo_url: str | None,
+        channel_logos: list[dict],
+    ) -> dict | None:
+        """채널별 프로필 이미지(로고)를 보고 각각 설명 + 공식 로고와 일치 여부 평가.
+        channel_logos: [{"channel": "Instagram", "url": "..."}, ...]
+        반환: {"channel_logos": [{"channel","logo_description","is_official"}], "inconsistency_summary", "recommendation"}
+
+        **3채널씩 묶어 병렬 호출** (한 번에 다 묶으면 LLM이 채널-이미지 매칭 헷갈려 같은 묘사를
+        여러 채널에 복사하는 문제 — VIEW 한국페북·영문인스타가 둘 다 "공식 로고" 묘사로 잘못
+        박혔던 케이스 — 가 있어서 분리. 1채널씩 N번보다 가성비 좋음)."""
+        items = [c for c in channel_logos if c.get("url")]
+        if not items:
+            return None
+
+        CHUNK = 3
+
+        async def _chunk(batch: list[dict]) -> list[dict]:
+            urls = [official_logo_url] + [c["url"] for c in batch] if official_logo_url else [c["url"] for c in batch]
+            n = len(batch)
+            # 이미지 번호 ↔ 채널 매핑 명시
+            if official_logo_url:
+                mapping = "이미지 1 = 공식 로고\n" + "\n".join(
+                    f"이미지 {i+2} = {c.get('channel','?')} 채널 프로필" for i, c in enumerate(batch)
+                )
+                instruction = (
+                    f"{mapping}\n\n"
+                    f"이미지 2~{n+1}(채널 프로필 {n}개)을 각각 **그 이미지에 실제로 보이는 그대로** "
+                    "한국어 1문장으로 묘사하세요 (색·형태·텍스트·배경 그대로).\n"
+                    "❗ 공식 로고(이미지 1) 묘사를 절대 복사하지 마세요. 각 채널 이미지에 보이는 실제 특징만.\n"
+                    "각 채널이 공식 로고와 시각적으로 거의 동일하면 is_official=true, "
+                    "심볼/색/배경/텍스트가 다르거나 모델 사진이면 false.\n"
+                )
+            else:
+                mapping = "\n".join(f"이미지 {i+1} = {c.get('channel','?')} 채널 프로필" for i, c in enumerate(batch))
+                instruction = (
+                    f"{mapping}\n\n"
+                    f"각 이미지를 보이는 그대로 한국어 1문장으로 묘사 (색·형태·텍스트·배경).\n"
+                )
+            schema_lines = ",\n".join(
+                f'    {{"channel": "{c.get("channel","?")}", "logo_description": "...", "is_official": true}}'
+                for c in batch
+            )
+            p = (
+                instruction
+                + "\n아래 JSON으로만 응답 (코드펜스 없이, 순수 JSON):\n{\n"
+                + f'  "channel_logos": [\n{schema_lines}\n  ]\n'
+                + "}\n"
+                + f"channel 필드는 위 매핑 그대로 ({', '.join(c.get('channel','?') for c in batch)}). "
+                + "logo_description은 반드시 한국어 (영어 금지)."
+            )
+            r = await self._ask(urls, p)
+            if not r:
+                return []
+            out = []
+            for c in r.get("channel_logos", []):
+                out.append({
+                    "channel": c.get("channel", ""),
+                    "logo_description": c.get("logo_description", ""),
+                    "is_official": bool(c.get("is_official", False)) if official_logo_url else None,
+                })
+            return out
+
+        # 3개씩 청크 → 병렬
+        chunks = [items[i:i+CHUNK] for i in range(0, len(items), CHUNK)]
+        results = await asyncio.gather(*[_chunk(b) for b in chunks], return_exceptions=True)
+        channel_logos_out: list[dict] = []
+        for r in results:
+            if isinstance(r, Exception):
+                logger.warning("[vision] channel_logo chunk error: %s", r)
+                continue
+            channel_logos_out.extend(r)
+        if not channel_logos_out:
+            return None
+
+        # 일관성 요약 + 권고는 결정적 산출 (LLM 한번 더 안 부름)
+        if official_logo_url:
+            mismatches = [c["channel"] for c in channel_logos_out if not c.get("is_official")]
+            if not mismatches:
+                summary = "모든 채널이 공식 로고를 일관되게 사용하고 있습니다."
+                rec = "현재 일관성 유지."
+            else:
+                summary = f"{len(mismatches)}개 채널({', '.join(mismatches)})이 공식 로고와 다른 이미지를 사용해 브랜드 일관성이 부족합니다."
+                rec = "비공식 채널 프로필을 공식 로고로 통일 권고."
+        else:
+            summary, rec = "", ""
+
+        return {
+            "channel_logos": channel_logos_out,
+            "inconsistency_summary": summary,
+            "recommendation": rec,
+        }

app/integrations/llm/prompt.py

@@ -1,14 +1,7 @@
 import os
 from pydantic import BaseModel
 from common.utils import get_env
-from integrations.llm.schemas.report import (
-    ReportInput, ReportOutput,
-    CriticalIssuesInput, CriticalIssuesOutput,
-    YouTubeDiagnosisInput, YouTubeDiagnosisOutput,
-    BrandConsistencyInput, BrandConsistencyOutput,
-    TransformationInput, TransformationProposal,
-    RoadmapInput, RoadmapOutput,
-)
+from integrations.llm.schemas.report import ReportInput, ReportOutput, YouTubeDiagnosisInput, YouTubeDiagnosisOutput
 from integrations.llm.schemas.plan import PlanInput, PlanOutput
 from integrations.llm.schemas.market import (
     MarketCompetitorsInput, MarketCompetitorsOutput,
@@ -94,31 +87,3 @@ youtube_diagnosis_prompt = Prompt(
     input_class=YouTubeDiagnosisInput,
     output_class=YouTubeDiagnosisOutput,
 )
-
-brand_consistency_prompt = Prompt(
-    file_name="brand_consistency_prompt.txt",
-    prompt_model="REPORT_MODEL",
-    input_class=BrandConsistencyInput,
-    output_class=BrandConsistencyOutput,
-)
-
-critical_issues_prompt = Prompt(
-    file_name="critical_issues_prompt.txt",
-    prompt_model="REPORT_MODEL",
-    input_class=CriticalIssuesInput,
-    output_class=CriticalIssuesOutput,
-)
-
-transformation_prompt = Prompt(
-    file_name="transformation_prompt.txt",
-    prompt_model="REPORT_MODEL",
-    input_class=TransformationInput,
-    output_class=TransformationProposal,
-)
-
-roadmap_prompt = Prompt(
-    file_name="roadmap_prompt.txt",
-    prompt_model="REPORT_MODEL",
-    input_class=RoadmapInput,
-    output_class=RoadmapOutput,
-)

app/integrations/llm/schemas/plan.py

@@ -15,6 +15,14 @@ class PlanInput(BaseModel):
     market_keywords: str | None = None
     market_trend: str | None = None
     market_target_audience: str | None = None
+    tiktok: str | None = None
+    instagram_en: str | None = None
+    facebook_en: str | None = None
+    naver_blog: str | None = None
+    naver_cafe: str | None = None
+    kakao_talk: str | None = None
+    channel_logos: str | None = None
+    brand_assets: str | None = None
 
 
 # --- BrandGuide ---
@@ -51,7 +59,7 @@ class ChannelBrandingRule(BaseModel):
     profile_photo: str
     banner_spec: str
     bio_template: str
-    current_status: Literal["correct", "incorrect", "missing"]
+    current_status: Literal["correct", "incorrect", "N/A"]
 
 
 class BrandPlanInconsistencyValue(BaseModel):

app/integrations/llm/schemas/report.py

@@ -68,16 +68,18 @@ class RegistryData(BaseModel):
 
 
 class ClinicSnapshot(BaseModel):
-    name: str
-    name_en: str
-    staff_count: int
-    lead_doctor: LeadDoctor
-    overall_rating: float
-    total_reviews: int
-    certifications: list[str]
-    location: str
-    phone: str
-    domain: str
+    # _build_clinic_snapshot은 source 데이터 있을 때만 채움 (`if x:` 가드).
+    # required면 강남언니/홈페이지 누락 병원에서 ValidationError로 리포트 실패.
+    name: str | None = None
+    name_en: str | None = None
+    staff_count: int | None = None
+    lead_doctor: LeadDoctor | None = None
+    overall_rating: float | None = None
+    total_reviews: int | None = None
+    certifications: list[str] = []
+    location: str | None = None
+    phone: str | None = None
+    domain: str | None = None
     logo_images: LogoImages | None = None
     brand_colors: BrandColors | None = None
     source: DataSource | None = None
@@ -157,8 +159,8 @@ class InstagramAccount(BaseModel):
 
 
 class InstagramAudit(BaseModel):
-    accounts: list[InstagramAccount]
-    diagnosis: list[DiagnosisItem]
+    accounts: list[InstagramAccount] = []
+    diagnosis: list[DiagnosisItem] = []
 
 
 # --- Facebook ---
@@ -191,17 +193,17 @@ class FacebookPage(BaseModel):
     linked_domain: str
     reviews: int
     recent_post_age: str
-    has_whatsapp: bool
-    post_frequency: str | None = None
+    has_whatsapp: bool | None = None
+    post_frequency: str
     top_content_type: str | None = None
-    engagement: str | None = None
+    engagement: str
 
 
 class FacebookAudit(BaseModel):
-    pages: list[FacebookPage]
-    diagnosis: list[DiagnosisItem]
-    brand_inconsistencies: list[BrandInconsistency]
-    consolidation_recommendation: str
+    pages: list[FacebookPage] = []
+    diagnosis: list[DiagnosisItem] = []
+    brand_inconsistencies: list[BrandInconsistency] = []
+    consolidation_recommendation: str | None = None
 
 
 # --- 기타 채널 / 웹사이트 ---
@@ -314,6 +316,14 @@ class ReportInput(BaseModel):
     market_keywords: str | None = None
     market_trend: str | None = None
     market_target_audience: str | None = None
+    branding: str | None = None
+    brand_assets: str | None = None
+    tiktok: str | None = None
+    instagram_en: str | None = None
+    facebook_en: str | None = None
+    kakao_talk: str | None = None
+    naver_cafe: str | None = None
+    channel_logos: str | None = None
 
 
 # --- MarketingReport ---
@@ -355,47 +365,3 @@ class YouTubeDiagnosisInput(BaseModel):
 
 class YouTubeDiagnosisOutput(BaseModel):
     diagnosis: list[DiagnosisItem]
-
-
-# --- Diagnosis ---
-
-class CriticalIssuesInput(BaseModel):
-    clinic_name: str | None = None
-    data: str | None = None
-
-
-class CriticalIssuesOutput(BaseModel):
-    diagnosis: list[DiagnosisItem]
-
-
-# --- Roadmap ---
-
-class RoadmapInput(BaseModel):
-    clinic_name: str | None = None
-    data: str | None = None
-
-
-class RoadmapOutput(BaseModel):
-    roadmap: list[RoadmapMonth]
-
-
-# --- Transformation ---
-
-class TransformationInput(BaseModel):
-    clinic_name: str | None = None
-    data: str | None = None
-
-
-# --- BrandConsistency ---
-
-class BrandConsistencyInput(BaseModel):
-    clinic_name: str | None = None
-    mainpage: str | None = None
-    instagram: str | None = None
-    facebook: str | None = None
-    youtube: str | None = None
-    gangnam_unni: str | None = None
-
-
-class BrandConsistencyOutput(BaseModel):
-    brand_inconsistencies: list[BrandInconsistency]

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

@@ -1,18 +0,0 @@
-다음은 성형외과/피부과 {clinic_name} 의 채널별 브랜드 데이터입니다.
-
-공식 홈페이지: {mainpage}
-인스타그램: {instagram}
-페이스북: {facebook}
-유튜브: {youtube}
-강남언니: {gangnam_unni}
-
-위 채널들 간의 브랜드 불일치 항목을 분석해줘.
-비교 대상 필드 예시: 병원명(한글/영문), 전화번호, 주소, 로고, 슬로건, 소개 문구 등.
-
-각 항목은 다음 JSON 형식의 배열로 출력해줘:
-- field: 불일치 필드명
-- values: 채널별 실제 값 목록 (channel, value, is_correct)
-- impact: 불일치가 브랜드에 미치는 영향
-- recommendation: 개선 권고사항
-
-출처 번호([1], [2] 등)는 포함하지 마.

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

@@ -1,13 +0,0 @@
-다음은 성형외과/피부과 {clinic_name} 의 전 채널 수집 데이터입니다.
-
-{data}
-
-위 데이터를 바탕으로 이 병원의 마케팅 전반에 걸친 핵심 문제점과 개선사항을 진단해줘.
-각 항목은 category(진단 카테고리), detail(상세 설명), severity(critical/warning/info) 형식의 JSON 배열로 출력해줘.
-
-현재 주요 진단 카테고리는 3개야.
-브랜드 아이덴티티 파편화
-콘텐츠 전략 부재
-플랫폼 간 유입 단절
-
-출처 번호([1], [2] 등)는 포함하지 마.

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

@@ -32,19 +32,65 @@
 ## 분석 리포트
 {report}
 
+## 추가 채널 데이터 (네이버 블로그 / 틱톡 / 인스타그램 EN / 페이스북 EN / 네이버 카페 / 카카오톡)
+아래에 데이터가 있는 채널은 channelStrategies에 **반드시 포함**하세요 (네이버 블로그, 틱톡, 영문 인스타그램, 영문 페이스북, 네이버 카페, 카카오톡). channelBranding은 SNS·블로그·카페까지만 포함(카카오톡은 메신저라 제외). null이면 제외.
+
+### 네이버 블로그 (Naver Blog)
+{naver_blog}
+
+### 틱톡 (TikTok)
+{tiktok}
+
+### 인스타그램 (영문 계정)
+{instagram_en}
+
+### 페이스북 (영문 페이지)
+{facebook_en}
+
+### 네이버 카페 (공식 카페 운영 신호)
+{naver_cafe}
+- naver_cafe.cafeName: 카페명, naver_cafe.memberCount: 회원수
+- currentStatus는 "회원 N명" 형태로 간단하게. 게시글 수·최근 활동은 수집 불가 (추측 금지).
+- targetGoal은 회원 확보 목표 수치 + 운영 권장 (예: "회원 5,000명, 주 1~2회 공지 발행").
+
+### 카카오톡 채널 (URL only — 콘텐츠 수집 X, 존재 여부만)
+{kakao_talk}
+- channelStrategies 카드 하나로 포함. currentStatus는 "공식 카카오톡 채널 운영 중" 정도, targetGoal은 친구 추가 유도·상담 전환·자동응답 시나리오 구체화 등.
+
+## 채널별 로고 분석 (Gemini Vision) — 채널룰/일관성의 근거
+{channel_logos}
+- 위 channel_logos[]의 각 항목: channel(채널명), logo_description(프로필이 어떻게 생겼는지), is_official(공식 로고와 일치 여부).
+- **channelBranding[]은 "어떻게 해야 하는지 권장 가이드라인" 섹션입니다.** 채널 통일 전략 기준으로 권장값 박을 것:
+  - profilePhoto: **빈 문자열 ""로 두세요.** 시스템이 brand_assets.logo_description으로 직접 채우므로 LLM은 만들지 마세요.
+  - bannerSpec: 권장 배너 규격 (크기·디자인 가이드)
+  - bioTemplate: 권장 bio 템플릿 (구조·필수 요소·예약 링크 포함 여부)
+  - currentStatus: is_official=true면 "correct" / false면 "incorrect" (데이터 없는 채널은 "N/A") — 현재 상태 마커는 이 필드 하나로만.
+- 현재 채널 프로필 이미지의 실제 묘사(channel_logos.channel_logos[].logo_description)는 brandInconsistencies에서만 사용. channelBranding에서 채널별로 다른 묘사를 박지 마세요.
+- **brandInconsistencies[]에 "로고" 항목을 반드시 만드세요**: values[]에 채널마다 channel(채널명) / value(logo_description 그대로) / is_correct(is_official 값) 세 필드를 넣고, impact는 inconsistency_summary, recommendation은 channel_logos.recommendation 기반으로 작성 (공식 로고로 통일 권고 포함).
+
+## 브랜드 자산 (홈페이지 CSS에서 추출 — 결정적 데이터)
+{brand_assets}
+- brand_assets.color_palette[]의 hex와 brand_assets.brand_colors(primary/accent/text)는 **홈페이지 CSS에서 실제로 추출한 값**입니다.
+- **brandGuide.colors의 hex는 반드시 이 추출값을 그대로 사용하세요. hex를 새로 지어내거나 변형하지 마세요** (매 실행마다 동일해야 함). name/usage 설명은 의미있게 써도 되지만 hex 값 자체는 추출값으로 고정.
+
 ## 섹션별 작성 지침
 
 ### Section 1: brandGuide
-- colors: 병원 아이덴티티에 맞는 컬러 팔레트 3~5개 (hex + 사용 가이드)
+- colors: **brand_assets.color_palette / brand_colors의 hex를 그대로 사용** (홈페이지 CSS 추출값, 지어내기 금지). 3~5개, 각 hex에 name/usage 부여
 - fonts: 제목/본문/캡션용 폰트 시스템 (한글/영문 포함)
-- logoRules: DO/DON'T 형식의 로고 사용 규칙 4~6개
+- logoRules: 로고 사용 규칙 4~6개. 각 항목은 rule / description / correct 3개 필드로 구성:
+  - rule: 규칙을 요약한 **구체적인 제목**. "DO"·"DON'T" 같은 단어를 그대로 넣지 말 것. 실제 규칙 내용을 쓸 것 (예: "보라색+골드 깃털 로고 통일 사용", "모델 사진 프로필 금지", "비공식 변형 로고 사용 금지", "로고 주변 여백 확보").
+  - description: 해당 규칙의 상세 설명.
+  - correct: 권장 규칙(DO)이면 true, 금지/지양 규칙(DON'T)이면 false. 권장(true)과 금지(false)를 섞어서 작성.
 - toneOfVoice: 브랜드 성격 키워드, 커뮤니케이션 스타일, 권장/지양 표현 예시
 - channelBranding: 리포트에 존재하는 채널별 브랜딩 적용 규칙
 - brandInconsistencies: 채널 간 브랜딩 불일치 항목 및 개선 권고
 
 ### Section 2: channelStrategies
-- 리포트에 데이터가 있는 채널만 포함
-- 각 채널의 우선순위(P0/P1/P2), 목표, 콘텐츠 유형, 게시 빈도, 포맷 가이드라인 작성
+- 메인 SNS 채널(Instagram, Facebook, YouTube, TikTok, 네이버 블로그) + 영문 계정(Instagram EN, Facebook EN) + **네이버 카페 / 카카오톡** (URL 있을 때) 카드를 **모두 포함**. 데이터 없는 채널도 빠뜨리지 말 것.
+- **currentStatus**: 데이터 있는 채널은 실제 수치로 서술 (예: "14,047 팔로워, Reels 0개", "104K 구독자, 주 2~3회 업로드"). **데이터 없는 채널은 "계정 없음"** 으로 표시. `excellent`/`warning`/`good` 같은 등급·평가어 금지.
+- **targetGoal은 모든 채널에 반드시 채울 것** — 구체적 목표 수치(예: "50K 팔로워, Reels 주 5개"). 데이터 없는 채널도 시작 시 권장 목표를 작성하고 비우지 말 것.
+- 각 채널의 우선순위(P0/P1/P2), 콘텐츠 유형, 게시 빈도, 포맷 가이드라인 모두 권장값으로 작성 — 데이터 없어도 시작 권장값으로 채울 것.
 - customerJourneyStage는 해당 채널의 주요 기여 단계로 설정
 
 ### Section 3: contentStrategy

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

@@ -12,6 +12,17 @@
 - 시술: {services}
 - 의료진: {doctors}
 
+## 브랜드 자산 (홈페이지에서 자동 추출)
+
+### Firecrawl branding (로고 URL / 색상 / 폰트)
+{branding}
+
+### 추출된 브랜드 자산 (로고 묘사 + CSS 색상 팔레트)
+{brand_assets}
+
+⚠️ clinic_snapshot.logo_images / brand_colors는 위 추출값(brand_assets.logo_images, brand_assets.brand_colors)을 **그대로** 사용하세요. hex나 로고 URL을 절대 추측하지 마세요. 추출값이 null이면 해당 필드도 null로 두세요.
+로고에 대한 정성 평가(심볼/워드마크/톤)는 brand_assets.logo_description을 근거로 하고, 채널 프로필 이미지가 공식 로고와 일치하는지 판단할 때도 이 묘사를 기준으로 삼으세요.
+
 ## 시장 분석 데이터
 
 ### 경쟁 병원
@@ -43,10 +54,62 @@
 ### 강남언니
 {gangnam_unni}
 
+### 틱톡 (TikTok)
+{tiktok}
+
+### 인스타그램 (영문 계정)
+{instagram_en}
+
+### 페이스북 (영문 페이지)
+{facebook_en}
+
+### 카카오톡 채널 (URL only — 수집 데이터 없음, 존재 여부만 확인)
+{kakao_talk}
+
+### 네이버 카페 (공식 카페 운영 신호)
+{naver_cafe}
+- naver_cafe.cafeName: 카페명
+- naver_cafe.memberCount: 회원수
+- 게시글 총 수·최근 게시일은 로그인 필요라 수집 불가. 추측 금지. 위 두 값만 사용.
+
+### 채널별 로고 분석 (Gemini Vision)
+{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(설명문)을 넣으세요.
+- 위 값들은 channel_logos 데이터 기반으로만 작성하고 추측하지 마세요.
+- 채널 간 로고 불일치(is_official=false)는 brand 일관성 진단(problem_diagnosis/weaknesses)에 반영하세요.
+- **brand_inconsistencies[]에 "로고" 항목을 반드시 만드세요**: values[]에 channel_logos.channel_logos[] 각 채널마다 다음 3필드를 **그대로** 박을 것 — channel(채널명 그대로), value(해당 채널의 logo_description 문자열 그대로 복붙), is_correct(해당 채널의 is_official 값 그대로). ❗ **채널-묘사 매핑을 절대 swap·재해석·임의 변형 금지**. channel_logos에 적힌 그대로 사용. impact는 channel_logos.inconsistency_summary 사용, recommendation은 channel_logos.recommendation 사용.
+
+## 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 는 반드시 빈 배열 []로 두세요.** 계정 정보는 시스템이 수집 데이터로 직접 채우니 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) 작성 지침
+- other_channels에는 메인 audit(YouTube/Instagram/Facebook/Website)에 **포함되지 않은** 채널만 넣으세요.
+- 위 '채널 데이터'에 **실제 수집된 데이터가 있는 채널만** status=active와 실제 url로 일관되게 포함: 네이버 블로그, 강남언니, 틱톡, 영문 인스타그램({instagram_en}), 영문 페이스북({facebook_en}).
+- **영문 인스타그램·영문 페이스북은 KR 메인 audit(Instagram/Facebook)과 별개 계정이므로, 데이터가 있으면 반드시 other_channels에 "Instagram EN" / "Facebook EN"으로 각각 포함하세요 (절대 누락 금지).**
+- **카카오톡·네이버 카페**: {kakao_talk} 또는 {naver_cafe}에 url이 있으면 other_channels에 각각 "KakaoTalk" / "Naver Cafe"로 status=active + 해당 url로 포함. 수집된 콘텐츠 데이터는 없으므로 URL 존재 자체가 활성 채널 신호. **둘 다 null/빈 값이면 절대 만들지 마세요.**
+- **그 외 데이터 없는 채널(네이버플레이스/Threads 등)은 절대 임의로 만들지 마세요.** 데이터 없으면 그 채널은 생략 (랜덤 생성·추측 금지).
+- url은 수집 데이터의 실제 URL만 사용. 없으면 빈 문자열.
+- **URL에 'https://www.facebook.com/' 같은 prefix를 절대 직접 만들지 마세요.** 수집 데이터의 URL을 그대로 사용. 이미 'https://...' 가 붙은 URL에 또 prefix 붙이면 'https://www.facebook.com/https://facebook.com/X' 같이 깨집니다. 받은 URL = 출력 URL.
+
+## registry_data 작성 지침 (clinic_snapshot 안)
+- **registry_data.website_en / district / branches / brand_group / naver_place_url / gangnam_unni_url / google_maps_url 모두 제공된 데이터에 명시되지 않으면 반드시 null로 두세요.**
+- 영문 사이트 URL, 영문명, 지점 정보 같은 거 데이터에 없으면 **절대 추측하거나 그럴듯해 보이는 도메인을 지어내지 마세요** (예: 'thepsclinic.com', '*-eng.com' 같은 거).
+
 ## 분석 지침
 
 - 점수는 0~100 기준입니다.
+- **channel_scores(채널 종합도)에는 데이터가 있는 모든 채널을 각각 별도 항목으로 만드세요. 같은 플랫폼이라도 한국 계정과 영문 계정을 절대 하나로 합치지 마세요:**
+  - 인스타그램 KR → channel "Instagram", 영문 인스타그램({instagram_en}) 데이터가 있으면 → channel "Instagram EN" (별도 항목)
+  - 페이스북 KR → channel "Facebook", 영문 페이스북({facebook_en}) 데이터가 있으면 → channel "Facebook EN" (별도 항목)
+  - 틱톡({tiktok}) 데이터가 있으면 → channel "TikTok" (별도 항목)
+  - 데이터가 null인 계정은 항목을 만들지 마세요. icon은 instagram/facebook/video 등 플랫폼에 맞게 설정.
 - strengths와 weaknesses는 각 3개 이상 작성하세요.
 - roadmap은 우선순위 순으로 실행 가능한 액션으로 작성하세요.
-- kpis는 실제 수집된 수치 기반으로 현실적인 측정 가능 지표로 작성하세요.
+- kpi_dashboard는 코드가 결정적으로 산출해 후처리 강제 치환하므로 LLM 출력 무시됩니다. 빈 배열 또는 placeholder로 두세요.
 - conversion_strategy의 actions는 구체적인 실행 방안으로 작성하세요.

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

@@ -1,14 +0,0 @@
-다음은 성형외과/피부과 {clinic_name} 의 전 채널 수집 데이터입니다.
-
-{data}
-
-위 데이터를 바탕으로 이 병원의 3개월 마케팅 실행 로드맵을 수립해줘.
-month 1, 2, 3 각각 하나씩, 총 3개 항목을 포함한 roadmap JSON 배열로 출력해줘.
-
-각 항목은 아래 형식을 따라줘:
-- month: 월 번호 (1, 2, 3)
-- title: 해당 월의 핵심 테마 (예: "브랜드 정비")
-- subtitle: 한 줄 부제 (예: "기반 구축 — 로고·계정 통일")
-- tasks: 실행 과제 목록, 각 과제는 task(string)와 completed(false)로 구성
-
-출처 번호([1], [2] 등)는 포함하지 마.

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

@@ -1,14 +0,0 @@
-다음은 성형외과/피부과 {clinic_name} 의 전 채널 수집 데이터입니다.
-
-{data}
-
-위 데이터를 바탕으로 이 병원의 마케팅 전환 전략을 수립해줘.
-아래 5개 항목을 포함한 JSON을 출력해줘.
-
-1. brand_identity: 브랜드 아이덴티티 개선 항목 (area, as_is, to_be)
-2. content_strategy: 콘텐츠 전략 개선 항목 (area, as_is, to_be)
-3. platform_strategies: 플랫폼별 전략 (platform, icon, current_metric, target_metric, strategies[{strategy, detail}])
-4. website_improvements: 웹사이트 개선 항목 (area, as_is, to_be)
-5. new_channel_proposals: 신규 채널 제안 (channel, priority, rationale)
-
-출처 번호([1], [2] 등)는 포함하지 마.

app/integrations/naver.py

@@ -1,4 +1,5 @@
 import re
+import httpx
 from http import HTTPMethod
 from urllib.parse import urlparse
 from common.utils import http_request
@@ -64,6 +65,20 @@ class NaverClient:
             return None
         return resp.text
 
+    async def fetch_blog_total_count(self, handle: str) -> int | None:
+        """블로그 전체 글 수는 RSS에 없어서 PostList HTML에서 '554개의 글' 패턴 추출.
+        <h4 class="category_title pcol2">... 554개의 글</h4> 구조."""
+        resp = await http_request(
+            HTTPMethod.GET,
+            url=f"https://blog.naver.com/PostList.naver?blogId={handle}&from=postList&directAccess=true",
+            timeout=15,
+            label="naver-blog-postlist",
+        )
+        if not resp or not resp.is_success:
+            return None
+        m = re.search(r"(\d+)개의 글", resp.text)
+        return int(m.group(1)) if m else None
+
     async def get_blog_rss(self, url: str) -> dict | None:
         blog_handle = urlparse(url).path.strip("/").split("/")[0] if "://" in url else url
         xml = await self.fetch_blog_rss(blog_handle)
@@ -82,10 +97,71 @@ class NaverClient:
                 "postDate": date.group(1) if date else "",
                 "description": re.sub(r"<[^>]*>", "", desc.group(1) if desc else "").strip()[:150],
             })
+        # RSS의 totalCount 우선, 없으면 블로그 PostList 페이지에서 "N개의 글" 파싱, 그것도 없으면 RSS 글수
         total_match = re.search(r"<totalCount>(\d+)</totalCount>", xml)
+        if total_match:
+            total = int(total_match.group(1))
+        else:
+            total = await self.fetch_blog_total_count(blog_handle) or len(posts)
         return {
             "officialBlogUrl": f"https://blog.naver.com/{blog_handle}",
             "officialBlogHandle": blog_handle,
-            "totalResults": int(total_match.group(1)) if total_match else len(posts),
+            "totalResults": total,
             "posts": posts[:10],
         }
+
+    async def get_cafe_info(self, cafe_url: str, *_args, **_kwargs) -> dict | None:
+        """네이버 카페 운영 신호 수집. 2단계 fetch:
+        1) https://cafe.naver.com/{handle} → cafeId 추출
+        2) ArticleList.nhn?search.clubid={cafeId} → memberCount + cafeName 추출
+        본문/게시글은 로그인 필요라 못 가져옴. 회원수·카페명만 잡히면 충분.
+        common.http_request는 redirect 안 따라가서 카페 페이지에 안 맞아 httpx 직접 사용."""
+        handle = urlparse(cafe_url).path.strip("/").split("/")[0] if "://" in cafe_url else cafe_url.split("/")[-1]
+        if not handle:
+            return None
+
+        async with httpx.AsyncClient(
+            timeout=10, follow_redirects=True,
+            headers={"User-Agent": "Mozilla/5.0"},
+        ) as c:
+            # 1. cafeId 추출
+            try:
+                main = await c.get(f"https://cafe.naver.com/{handle}")
+            except Exception:
+                return {"url": f"https://cafe.naver.com/{handle}", "cafeHandle": handle, "accessible": False}
+            if main.status_code != 200:
+                return {"url": f"https://cafe.naver.com/{handle}", "cafeHandle": handle, "accessible": False}
+            cid_match = re.search(r'cafeId["\']?\s*[:=]\s*["\']?(\d+)', main.text)
+            cafe_id = cid_match.group(1) if cid_match else None
+
+            result: dict = {
+                "url": f"https://cafe.naver.com/{handle}",
+                "cafeHandle": handle,
+                "cafeId": cafe_id,
+                "accessible": True,
+                "cafeName": None,
+                "memberCount": None,
+            }
+            if not cafe_id:
+                return result
+
+            # 2. ArticleList 페이지에서 회원수 + 카페명 추출 (로그인 없이 접근 가능한 유일한 endpoint)
+            try:
+                listing = await c.get(
+                    f"https://cafe.naver.com/ArticleList.nhn?search.clubid={cafe_id}&search.menuid=&search.boardtype=L",
+                    headers={"Referer": f"https://cafe.naver.com/{handle}"},
+                )
+            except Exception:
+                return result
+            if listing.status_code != 200:
+                return result
+            mc = re.search(r'memberCount[^0-9]+(\d[\d,]*)', listing.text)
+            if mc:
+                result["memberCount"] = int(mc.group(1).replace(",", ""))
+            tm = re.search(r"<title>(.+?)\s*:\s*네이버 카페</title>", listing.text)
+            if tm:
+                name = re.sub(r"&amp;", "&", tm.group(1)).strip()
+                if "," in name:
+                    name = name.split(",", 1)[1].strip()
+                result["cafeName"] = name
+            return result

app/integrations/site_fetcher.py

@@ -0,0 +1,66 @@
+"""홈페이지 HTML + 외부 CSS 를 가져오는 fetch 전용 모듈.
+
+오래된 한국 의료 사이트들이 SSL DH_KEY_TOO_SMALL / cipher 약함 / host mismatch 등으로
+표준 fetch 에 차단되는 케이스가 많아 단계별 SSL fallback 으로 받는다.
+파싱·도메인 로직은 들어가지 않음 — 순수 HTTP 응답 본문 반환.
+"""
+import logging
+import re
+import ssl
+from urllib.parse import urljoin
+import httpx
+
+logger = logging.getLogger(__name__)
+
+CSS_LINK = re.compile(
+    r'<link[^>]+rel=["\']stylesheet["\'][^>]+href=["\']([^"\']+)["\']',
+    re.IGNORECASE,
+)
+
+
+def _make_ssl_context() -> ssl.SSLContext:
+    """보안 등급 1로 낮춤 + cert 검증 유지 (옛 한국 의료 사이트 cipher 약함 회피)."""
+    ctx = ssl.create_default_context()
+    try:
+        ctx.set_ciphers("DEFAULT@SECLEVEL=1")
+    except ssl.SSLError:
+        pass
+    return ctx
+
+
+async def fetch_html(url: str, timeout: float = 20.0) -> tuple[int, str]:
+    """SSL 검증 단계별 fallback 으로 HTML 본문 받기. 실패 시 (0, "")."""
+    headers = {"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"}
+    try:
+        async with httpx.AsyncClient(timeout=timeout, follow_redirects=True, headers=headers) as c:
+            r = await c.get(url)
+            return r.status_code, r.text
+    except (httpx.ConnectError, httpx.ReadError, ssl.SSLError) as e:
+        logger.info("[fetch] %s standard SSL failed: %s — fallback to weak cipher", url, e)
+    try:
+        async with httpx.AsyncClient(timeout=timeout, follow_redirects=True, headers=headers, verify=_make_ssl_context()) as c:
+            r = await c.get(url)
+            return r.status_code, r.text
+    except (httpx.ConnectError, httpx.ReadError, ssl.SSLError) as e:
+        logger.info("[fetch] %s weak cipher failed: %s — fallback to verify=False", url, e)
+    try:
+        async with httpx.AsyncClient(timeout=timeout, follow_redirects=True, headers=headers, verify=False) as c:
+            r = await c.get(url)
+            return r.status_code, r.text
+    except Exception as e:
+        logger.warning("[fetch] %s all fallbacks failed: %s", url, e)
+        return 0, ""
+
+
+async def fetch_html_and_css(homepage_url: str, max_css_files: int = 8) -> tuple[str, list[str]]:
+    """홈페이지 HTML + 외부 CSS(Top N) 한 번에 fetch. 실패 시 ("", [])."""
+    status, html = await fetch_html(homepage_url)
+    if status != 200 or not html:
+        logger.warning("[fetch] homepage fetch failed status=%s url=%s", status, homepage_url)
+        return "", []
+    css_texts: list[str] = []
+    for css_href in CSS_LINK.findall(html)[:max_css_files]:
+        cstatus, ctext = await fetch_html(urljoin(homepage_url, css_href), timeout=15.0)
+        if cstatus == 200 and ctext:
+            css_texts.append(ctext)
+    return html, css_texts

app/integrations/youtube.py

@@ -98,9 +98,11 @@ class YouTubeClient:
         ch = raw["channel"]
         stats = ch.get("statistics", {})
         snippet = ch.get("snippet", {})
+        thumbs = snippet.get("thumbnails", {})
         return {
             "channelId": raw["channelId"],
             "channelName": snippet.get("title"),
+            "profileImage": (thumbs.get("high") or thumbs.get("medium") or thumbs.get("default") or {}).get("url"),
             "handle": snippet.get("customUrl"),
             "description": snippet.get("description", ""),
             "publishedAt": snippet.get("publishedAt"),

app/mock/plan_viewclinic.json

@@ -153,7 +153,7 @@
         "profile_photo": "보라색+골드 깃털 로고",
         "banner_spec": "블로그 상단: 깃털 심볼 + 대표 이미지",
         "bio_template": "21년 무사고 VIEW 성형외과 공식 블로그\n가슴성형·안면윤곽·양악·눈코·리프팅",
-        "current_status": "missing"
+        "current_status": "N/A"
       },
       {
         "channel": "TikTok",
@@ -161,7 +161,7 @@
         "profile_photo": "보라색+골드 깃털 원형 로고",
         "banner_spec": "N/A",
         "bio_template": "VIEW 성형외과 — 안전이 예술이 되는 곳\n강남 신논현역 | 02-539-1177",
-        "current_status": "missing"
+        "current_status": "N/A"
       }
     ],
     "brand_inconsistencies": [

app/models/analysis.py

@@ -8,6 +8,11 @@ class Channels(BaseModel):
     facebook: str | None = None
     naver_blog: str | None = None
     gangnam_unni: str | None = None
+    tiktok: str | None = None
+    instagram_en: str | None = None
+    facebook_en: str | None = None
+    kakao_talk: str | None = None
+    naver_cafe: str | None = None
 
 
 class AnalysisOptions(BaseModel):

app/models/plan.py

@@ -49,7 +49,7 @@ class ChannelBrandingRule(CamelModel):
     profile_photo: str
     banner_spec: str
     bio_template: str
-    current_status: Literal["correct", "incorrect", "missing"]
+    current_status: Literal["correct", "incorrect", "N/A"]
 
 
 class BrandGuide(CamelModel):

app/models/report.py

@@ -66,16 +66,18 @@ class RegistryData(CamelModel):
 
 
 class ClinicSnapshot(CamelModel):
-    name: str
-    name_en: str
-    staff_count: int
-    lead_doctor: LeadDoctor
-    overall_rating: float
-    total_reviews: int
-    certifications: list[str]
-    location: str
-    phone: str
-    domain: str
+    # _build_clinic_snapshot은 source 데이터 있을 때만 필드 추가 (`if x:` 가드).
+    # 강남언니/홈페이지 수집 누락된 병원에서 required면 ValidationError로 리포트 전체 실패.
+    name: str | None = None
+    name_en: str | None = None
+    staff_count: int | None = None
+    lead_doctor: LeadDoctor | None = None
+    overall_rating: float | None = None
+    total_reviews: int | None = None
+    certifications: list[str] = []
+    location: str | None = None
+    phone: str | None = None
+    domain: str | None = None
     logo_images: LogoImages | None = None
     brand_colors: BrandColors | None = None
     source: DataSource | None = None
@@ -149,8 +151,8 @@ class InstagramAccount(CamelModel):
 
 
 class InstagramAudit(CamelModel):
-    accounts: list[InstagramAccount]
-    diagnosis: list[DiagnosisItem]
+    accounts: list[InstagramAccount] = []
+    diagnosis: list[DiagnosisItem] = []
 
 
 class BrandInconsistencyValue(CamelModel):
@@ -181,17 +183,17 @@ class FacebookPage(CamelModel):
     linked_domain: str
     reviews: int
     recent_post_age: str
-    has_whatsapp: bool
-    post_frequency: str | None = None
+    has_whatsapp: bool | None = None
+    post_frequency: str
     top_content_type: str | None = None
-    engagement: str | None = None
+    engagement: str
 
 
 class FacebookAudit(CamelModel):
-    pages: list[FacebookPage]
-    diagnosis: list[DiagnosisItem]
-    brand_inconsistencies: list[BrandInconsistency]
-    consolidation_recommendation: str
+    pages: list[FacebookPage] = []
+    diagnosis: list[DiagnosisItem] = []
+    brand_inconsistencies: list[BrandInconsistency] = []
+    consolidation_recommendation: str | None = None
 
 
 class OtherChannel(CamelModel):

app/models/status.py

@@ -46,6 +46,8 @@ class SourceType(StrEnum):
     GANGNAM_UNNI = "gangnam_unni"
     KAKAOTALK = "kakaotalk"
     NAVER_CAFE = "naver_cafe"
+    # 부가 수집/분석 (HTML/CSS 재크롤 + Vision 로고 매칭) — 한 raw_info entry 에 brandAssets/channelLogos 같이 보관.
+    BRANDING = "branding"
 
 
 class Language(StrEnum):

app/services/analysis.py

@@ -1,15 +1,18 @@
 import json
 import logging
-import os
 import re
 from datetime import datetime
 from urllib.parse import urlparse
-from common.db.run import select_run, update_run_report, update_run_plan
-from common.db.source import select_run_raw_data, select_run_mainpage_url
+from common.db.run import update_run_report, update_run_plan, select_run_report_data
+from common.db.source import select_run_raw_data, select_mainpage_logo_url
 from common.db.market import select_market
 from integrations.llm.llm_service import LLMService
-from integrations.llm.prompt import report_prompt, plan_prompt, youtube_diagnosis_prompt, brand_consistency_prompt, critical_issues_prompt, transformation_prompt, roadmap_prompt
-from integrations.llm.schemas.report import ReportOutput, ClinicSnapshot, YouTubeAudit, BrandConsistencyOutput, CriticalIssuesOutput, DiagnosisItem, TransformationProposal, RoadmapOutput, RoadmapMonth
+from integrations.llm.prompt import report_prompt, plan_prompt, youtube_diagnosis_prompt
+from integrations.llm.schemas.report import ReportOutput, ClinicSnapshot, YouTubeAudit
+from services.branding import analyze_branding
+from services.instagram_audit import build_instagram_audit
+from services.facebook_audit import build_facebook_audit
+from services.kpi_dashboard import build_kpi_dashboard
 from integrations.llm.schemas.plan import PlanOutput
 
 logger = logging.getLogger(__name__)
@@ -17,8 +20,9 @@ logger = logging.getLogger(__name__)
 
 async def generate_report(analysis_run_id: str) -> ReportOutput:
     raw = await select_run_raw_data(analysis_run_id)
-    clinic = raw.get("mainpage") or {}
-    market = await select_market(analysis_run_id)
+    clinic   = raw.get("mainpage") or {}
+    branding = raw.get("branding") or {}
+    market   = await select_market(analysis_run_id)
 
     def _json(v) -> str | None:
         return json.dumps(v, ensure_ascii=False) if v else None
@@ -35,21 +39,35 @@ async def generate_report(analysis_run_id: str) -> ReportOutput:
         "market_keywords":        _json(market.get("keywords")),
         "market_trend":           _json(market.get("trend")),
         "market_target_audience": _json(market.get("target_audience")),
+        # firecrawl 이 mainpage 에서 뽑은 branding 메타(logoUrl/ogImage/faviconUrl) + Vision/CSS 산출물
+        "branding":               _json(clinic.get("branding")),
+        "brand_assets":           _json(branding.get("brandAssets")),
+        "channel_logos":          _json(branding.get("channelLogos")),
+        # 부가 채널 (raw_info entry) — raw dict 의 한국식 key 그대로
+        "tiktok":                 _json(raw.get("tiktok")),
+        "instagram_en":           _json(raw.get("instagram_en")),
+        "facebook_en":            _json(raw.get("facebook_en")),
+        "kakao_talk":             _json(raw.get("kakaotalk")),
+        "naver_cafe":             _json(raw.get("naver_cafe")),
+        # 메인 5채널은 raw dict 그대로 펼쳐서 prompt placeholder 와 매칭
         **{
             source_type: _json(data)
             for source_type, data in raw.items()
-            if source_type != "mainpage"
+            if source_type not in {
+                "mainpage", "branding",
+                "tiktok", "instagram_en", "facebook_en", "kakaotalk", "naver_cafe",
+            }
         },
     }
 
     return await LLMService(provider="perplexity").generate(report_prompt, input_data)
 
+
 async def generate_plan(analysis_run_id: str) -> PlanOutput:
-    run = await select_run(analysis_run_id)
     raw = await select_run_raw_data(analysis_run_id)
-    clinic = raw.get("mainpage") or {}
-    report_data = run["report_data"]
-    report = json.loads(report_data) if isinstance(report_data, str) else report_data
+    clinic   = raw.get("mainpage") or {}
+    branding = raw.get("branding") or {}
+    report = await select_run_report_data(analysis_run_id)
     market = await select_market(analysis_run_id)
 
     def _json(v) -> str | None:
@@ -68,20 +86,28 @@ async def generate_plan(analysis_run_id: str) -> PlanOutput:
         "market_keywords":        _json(market.get("keywords")),
         "market_trend":           _json(market.get("trend")),
         "market_target_audience": _json(market.get("target_audience")),
+        "tiktok":                 _json(raw.get("tiktok")),
+        "instagram_en":           _json(raw.get("instagram_en")),
+        "facebook_en":            _json(raw.get("facebook_en")),
+        "naver_blog":             _json(_naver_blog_summary(raw.get("naver_blog"))),
+        "naver_cafe":             _json(raw.get("naver_cafe")),
+        "kakao_talk":             _json(raw.get("kakaotalk")),
+        "channel_logos":          _json(branding.get("channelLogos")),
+        "brand_assets":           _json(branding.get("brandAssets")),
     }
 
     return await LLMService(provider="perplexity").generate(plan_prompt, input_data)
 
 
-def _build_clinic_snapshot(gangnam_unni: dict, hospital: dict) -> dict:
+def _build_clinic_snapshot(gangnam_unni: dict, mainpage: dict, brand_assets: dict, logo_url: str | None) -> dict:
     snapshot: dict = {}
     doctors = gangnam_unni.get("doctors", [])
     lead = max(doctors, key=lambda d: d.get("reviews", 0)) if doctors else None
     if gangnam_unni.get("name"):             snapshot["name"]           = gangnam_unni["name"]
-    if hospital.get("clinicNameEn"):         snapshot["name_en"]        = hospital["clinicNameEn"]
-    if hospital.get("phone"):                snapshot["phone"]           = hospital["phone"]
-    domain = hospital.get("domain") or urlparse(hospital.get("sourceUrl") or "").netloc
-    if domain:                               snapshot["domain"]          = domain
+    if mainpage.get("clinicNameEn"):         snapshot["name_en"]        = mainpage["clinicNameEn"]
+    if mainpage.get("phone"):                snapshot["phone"]          = mainpage["phone"]
+    domain = mainpage.get("domain") or urlparse(mainpage.get("sourceUrl") or "").netloc
+    if domain:                               snapshot["domain"]         = domain
     if gangnam_unni.get("rating"):           snapshot["overall_rating"] = gangnam_unni["rating"]
     if gangnam_unni.get("totalReviews"):     snapshot["total_reviews"]  = gangnam_unni["totalReviews"]
     if gangnam_unni.get("address"):          snapshot["location"]       = gangnam_unni["address"]
@@ -94,9 +120,25 @@ def _build_clinic_snapshot(gangnam_unni: dict, hospital: dict) -> dict:
             "rating":       lead.get("rating"),
             "review_count": lead.get("reviews"),
         }
+    # logo URL 은 raw_info.logo_url 컬럼에서, brand_colors 는 JSON 에서 강제 주입. LLM 의 null 처리 차단.
+    if logo_url:
+        snapshot["logo_images"] = {"circle": None, "horizontal": logo_url, "korean": None}
+    if brand_assets.get("brand_colors"): snapshot["brand_colors"] = brand_assets["brand_colors"]
     return ClinicSnapshot.model_validate(snapshot).model_dump()
 
 
+def _naver_blog_summary(blog: dict | None) -> dict | None:
+    """plan 카드 한 장에 들어가는 건 전체 포스트 수와 최근 활동 시점뿐. 그 외(본문·링크·제목)는
+    던져봐야 토큰만 늘고 LLM 이 무관 정보로 hallucinate 함."""
+    if not blog:
+        return None
+    posts = blog.get("posts") or []
+    return {
+        "totalPosts":     blog.get("totalResults"),
+        "latestPostDate": posts[0].get("postDate") if posts else None,
+    }
+
+
 def _parse_iso_duration_seconds(iso: str) -> int:
     m = re.match(r"PT(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?", iso or "")
     if not m:
@@ -208,181 +250,86 @@ async def _build_youtube_audit(youtube: dict) -> dict:
     return YouTubeAudit.model_validate(yt_patch).model_dump()
 
 
-async def _build_roadmap(analysis_run_id: str, raw: dict) -> list[dict]:
-    result: RoadmapOutput = await LLMService(provider="perplexity").generate(
-        roadmap_prompt,
-        {
-            "clinic_name": (raw.get("mainpage") or {}).get("clinicName"),
-            "data":        json.dumps(raw, ensure_ascii=False),
-        },
-    )
-    return [RoadmapMonth.model_validate(item).model_dump() for item in result.roadmap]
-
-
-async def _build_transformation(analysis_run_id: str, raw: dict) -> dict:
-    result: TransformationProposal = await LLMService(provider="perplexity").generate(
-        transformation_prompt,
-        {
-            "clinic_name": (raw.get("mainpage") or {}).get("clinicName"),
-            "data":        json.dumps(raw, ensure_ascii=False),
-        },
-    )
-    return result.model_dump()
-
-
-async def _build_critical_issues(analysis_run_id: str, raw: dict) -> list[dict]:
-    result: CriticalIssuesOutput = await LLMService(provider="perplexity").generate(
-        critical_issues_prompt,
-        {
-            "clinic_name": (raw.get("mainpage") or {}).get("clinicName"),
-            "data":        json.dumps(raw, ensure_ascii=False),
-        },
-    )
-    return [DiagnosisItem.model_validate(item).model_dump() for item in result.diagnosis]
-
-
-async def _build_overrides(analysis_run_id: str) -> dict:
-    raw = await select_run_raw_data(analysis_run_id)
-    if not raw:
-        return {}
-
-    mainpage     = raw.get("mainpage", {}) or {}
-    instagram    = raw.get("instagram", {}) or {}
-    facebook     = raw.get("facebook", {}) or {}
-    youtube      = raw.get("youtube", {}) or {}
-    gangnam_unni = raw.get("gangnam_unni", {}) or {}
-
-    snapshot: dict = _build_clinic_snapshot(gangnam_unni, mainpage)
-    yt_patch: dict = await _build_youtube_audit(youtube)
-
-    # ── instagram ─────────────────────────────────────────────────────────────
-    ig_patch: dict = {}
-    if instagram.get("username"):  ig_patch["handle"]       = instagram["username"]
-    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 ──────────────────────────────────────────────────────────────
-    fb_pages: dict = {}
-    if facebook.get("pageUrl"):    fb_pages["url"]           = facebook["pageUrl"]
-    if facebook.get("pageUrl"):    fb_pages["link"]          = facebook["pageUrl"]
-    if facebook.get("pageName"):   fb_pages["page_name"]     = facebook["pageName"]
-    if facebook.get("followers"):  fb_pages["followers"]     = facebook["followers"]
-    if facebook.get("intro"):      fb_pages["bio"]           = facebook["intro"]
-    if facebook.get("categories"): fb_pages["category"]      = ", ".join(facebook["categories"])
-    if facebook.get("website"):    fb_pages["linked_domain"] = facebook["website"]
-
-    brand = await generate_brand_consistency(analysis_run_id)
-    brand_patch = brand.model_dump()["brand_inconsistencies"]
-    critical_issues = await _build_critical_issues(analysis_run_id, raw)
-    transformation = await _build_transformation(analysis_run_id, raw)
-    roadmap = await _build_roadmap(analysis_run_id, raw)
-
-    fb_patch: dict = {}
-    if fb_pages:
-        fb_patch["pages"] = [fb_pages]
-    if brand_patch:
-        fb_patch["brand_inconsistencies"] = brand_patch
-
-    overrides: dict = {}
-    if snapshot:
-        overrides["clinic_snapshot"] = snapshot
-    if ig_patch:
-        overrides["instagram_audit"] = {"accounts": [ig_patch]}
-    if fb_patch:
-        overrides["facebook_audit"] = fb_patch
-    if yt_patch:
-        overrides["youtube_audit"] = yt_patch
-    if critical_issues:
-        overrides["problem_diagnosis"] = critical_issues
-    if transformation:
-        overrides["transformation"] = transformation
-    if roadmap:
-        overrides["roadmap"] = roadmap
-    return overrides
-
-
 def _deep_merge(base: dict, overrides: dict) -> dict:
+    """dict 끼리 만나면 재귀로 안쪽까지 합치고, 그 외(list/scalar/None) 는 override 값으로 통째 치환."""
     for k, v in overrides.items():
         if isinstance(v, dict) and isinstance(base.get(k), dict):
             _deep_merge(base[k], v)
-        elif isinstance(v, list) and isinstance(base.get(k), list):
-            for i, item in enumerate(v):
-                if i < len(base[k]) and isinstance(item, dict) and isinstance(base[k][i], dict):
-                    _deep_merge(base[k][i], item)
         else:
             base[k] = v
     return base
 
-def _patch_report(result: ReportOutput, overrides: dict) -> ReportOutput:
-    base = result.model_dump()
-    for key in overrides:
-        base.pop(key, None)
-    return ReportOutput(**_deep_merge(base, overrides))
 
-
-_MOCK_DOMAINS = {"viewclinic.com"}
-_MOCK_REPORT_PATH = os.path.join(os.path.dirname(__file__), "../mock/report_viewclinic.json")
-
-
-async def _is_mock(analysis_run_id: str) -> bool:
-    url = await select_run_mainpage_url(analysis_run_id)
-    return any(domain in url for domain in _MOCK_DOMAINS)
-
-
-def _load_mock_report() -> ReportOutput:
-    with open(_MOCK_REPORT_PATH, encoding="utf-8") as f:
-        return ReportOutput(**json.load(f))
-
-
-_MOCK_PLAN_PATH = os.path.join(os.path.dirname(__file__), "../mock/plan_viewclinic.json")
-
-
-def _load_mock_plan() -> PlanOutput:
-    with open(_MOCK_PLAN_PATH, encoding="utf-8") as f:
-        return PlanOutput(**json.load(f))
-
-
-async def generate_brand_consistency(analysis_run_id: str) -> BrandConsistencyOutput:
+async def _build_overrides(analysis_run_id: str, result: ReportOutput) -> ReportOutput:
     raw = await select_run_raw_data(analysis_run_id)
+    if not raw:
+        return result
 
-    def _json(v) -> str | None:
-        return json.dumps(v, ensure_ascii=False) if v else None
+    mainpage     = raw.get("mainpage", {}) or {}
+    branding     = raw.get("branding", {}) or {}
+    instagram    = raw.get("instagram", {}) or {}
+    facebook     = raw.get("facebook", {}) or {}
+    youtube      = raw.get("youtube", {}) or {}
+    gangnam_unni = raw.get("gangnam_unni", {}) or {}
+    naver_blog   = raw.get("naver_blog", {}) or {}
+    instagram_en = raw.get("instagram_en", {}) or {}
+    facebook_en  = raw.get("facebook_en", {}) or {}
+    tiktok       = raw.get("tiktok", {}) or {}
+    naver_cafe   = raw.get("naver_cafe", {}) or {}
+    brand_assets  = branding.get("brandAssets") or {}
+    channel_logos = branding.get("channelLogos") or {}
+    logo_url      = await select_mainpage_logo_url(analysis_run_id)
 
-    mainpage = raw.get("mainpage") or {}
-    input_data = {
-        "clinic_name":  mainpage.get("clinicName"),
-        "mainpage":     _json(mainpage),
-        "instagram":    _json(raw.get("instagram")),
-        "facebook":     _json(raw.get("facebook")),
-        "youtube":      _json(raw.get("youtube")),
-        "gangnam_unni": _json(raw.get("gangnam_unni")),
+    llm_fb_pages = result.model_dump().get("facebook_audit", {}).get("pages", [])
+
+    snapshot: dict = _build_clinic_snapshot(gangnam_unni, mainpage, brand_assets, logo_url)
+    yt_patch: dict = await _build_youtube_audit(youtube)
+    ig_patch        = build_instagram_audit(instagram, instagram_en, channel_logos)
+    fb_patch        = build_facebook_audit(facebook, facebook_en, llm_fb_pages)
+    kpi_extras = {
+        "instagramEn": instagram_en,
+        "facebookEn":  facebook_en,
+        "tiktok":      tiktok,
+        "naverCafe":   naver_cafe,
     }
-    return await LLMService(provider="perplexity").generate(brand_consistency_prompt, input_data)
+    kpi = build_kpi_dashboard(instagram, facebook, youtube, gangnam_unni, kpi_extras, naver_blog)
+
+    overrides: dict = {}
+    if snapshot:  overrides["clinic_snapshot"]  = snapshot
+    if ig_patch:  overrides["instagram_audit"]  = ig_patch
+    if fb_patch:  overrides["facebook_audit"]   = fb_patch
+    if yt_patch:  overrides["youtube_audit"]    = yt_patch
+    if kpi:       overrides["kpi_dashboard"]    = kpi
+
+    merged = _deep_merge(result.model_dump(), overrides)
+    return ReportOutput(**merged)
 
 
 async def run_report_task(analysis_run_id: str) -> None:
     logger.info("[report] start run=%s", analysis_run_id)
-    if await _is_mock(analysis_run_id):
-        logger.info("[report] mock mode run=%s", analysis_run_id)
-        result = _load_mock_report()
-        result.youtube_audit.linked_urls = []
-    else:
-        result = await generate_report(analysis_run_id)
-
-    result = _patch_report(result, await _build_overrides(analysis_run_id))
+    await analyze_branding(analysis_run_id)
+    result = await generate_report(analysis_run_id)
+    result = await _build_overrides(analysis_run_id, result)
     await update_run_report(analysis_run_id, result.model_dump())
     logger.info("[report] done run=%s", analysis_run_id)
 
 
+def _patch_plan(result: PlanOutput, logo_desc: str) -> PlanOutput:
+    """brand_guide.channel_branding[].profile_photo 는 LLM 안 맡기고 코드가 박는다
+    (모든 채널 동일값 = brand_assets.logo_description). LLM 이 fallback 문구 hallucinate 방지."""
+    p = result.model_dump()
+    for ch in (p.get("brand_guide") or {}).get("channel_branding") or []:
+        ch["profile_photo"] = logo_desc
+    return PlanOutput(**p)
+
+
 async def run_plan_task(analysis_run_id: str) -> None:
     logger.info("[plan] start run=%s", analysis_run_id)
-    if await _is_mock(analysis_run_id):
-        logger.info("[plan] mock mode run=%s", analysis_run_id)
-        result = _load_mock_plan()
-    else:
-        result = await generate_plan(analysis_run_id)
+    result = await generate_plan(analysis_run_id)
+    # profile_photo 는 brand_assets.logo_description 으로 코드가 박음 (LLM "(가이드 미보유)" 같은 hallucination 차단).
+    raw = await select_run_raw_data(analysis_run_id)
+    branding = raw.get("branding") or {}
+    logo_desc = ((branding.get("brandAssets") or {}).get("logo_description")) or ""
+    result = _patch_plan(result, logo_desc)
     await update_run_plan(analysis_run_id, result.model_dump())
     logger.info("[plan] done run=%s", analysis_run_id)

app/services/brand_parser.py

@@ -0,0 +1,172 @@
+"""collect 단계 - HTML/CSS 텍스트에서 brand 로고 URL + 색상 추출"""
+import logging
+import re
+from collections import Counter
+from urllib.parse import urljoin
+
+logger = logging.getLogger(__name__)
+
+
+# ── 로고 URL 추출 ─────────────────────────────────────────────────────────────
+
+
+LOGO_IMG_PATTERNS = [
+    re.compile(r'<img[^>]*\bclass=["\'][^"\']*\blogo\b[^"\']*["\'][^>]*\bsrc=["\']([^"\']+)["\']', re.IGNORECASE),
+    re.compile(r'<img[^>]*\bsrc=["\']([^"\']+)["\'][^>]*\bclass=["\'][^"\']*\blogo\b[^"\']*["\']', re.IGNORECASE),
+    re.compile(r'<img[^>]*\bid=["\'][^"\']*\blogo\b[^"\']*["\'][^>]*\bsrc=["\']([^"\']+)["\']', re.IGNORECASE),
+    re.compile(r'<img[^>]*\balt=["\'][^"\']*\blogo\b[^"\']*["\'][^>]*\bsrc=["\']([^"\']+)["\']', re.IGNORECASE),
+    re.compile(r'<(?:a|h[1-6]|div|span)[^>]*\b(?:class|id)=["\'][^"\']*\blogo\b[^"\']*["\'][^>]*>(?:[^<]|<(?!img))*<img[^>]*\bsrc=["\']([^"\']+)["\']', re.IGNORECASE | re.DOTALL),
+    re.compile(r'<(?:a|div|span|h[1-6])[^>]*\b(?:class|id)=["\'][^"\']*\blogo\b[^"\']*["\'][^>]*\bstyle=["\'][^"\']*background(?:-image)?\s*:\s*url\(\s*["\']?([^"\')\s]+)', re.IGNORECASE),
+    re.compile(r'<(?:a|div|span|h[1-6])[^>]*\bstyle=["\'][^"\']*background(?:-image)?\s*:\s*url\(\s*["\']?([^"\')\s]+)[^"\']*["\'][^>]*\b(?:class|id)=["\'][^"\']*\blogo\b', re.IGNORECASE),
+    re.compile(r'<img[^>]*\bsrc=["\']([^"\']*\blogo\b[^"\']*\.(?:png|svg|jpe?g|webp)[^"\']*)["\']', re.IGNORECASE),
+    re.compile(r'<header\b[^>]*>(?:[^<]|<(?!img))*<img[^>]*\bsrc=["\']([^"\']+\.(?:png|svg|jpe?g|webp)[^"\']*)["\']', re.IGNORECASE | re.DOTALL),
+    re.compile(r'<nav\b[^>]*>(?:[^<]|<(?!img))*<img[^>]*\bsrc=["\']([^"\']+\.(?:png|svg|jpe?g|webp)[^"\']*)["\']', re.IGNORECASE | re.DOTALL),
+    re.compile(r'<meta[^>]*\bproperty=["\']og:image["\'][^>]*\bcontent=["\']([^"\']+)["\']', re.IGNORECASE),
+    re.compile(r'<meta[^>]*\bcontent=["\']([^"\']+)["\'][^>]*\bproperty=["\']og:image["\']', re.IGNORECASE),
+]
+
+LOGO_CSS_PATTERN = re.compile(
+    r'\.[\w-]*\blogo\b[\w-]*\s*(?:,\s*\.[\w-]+\s*)*\{[^}]*background(?:-image)?\s*:\s*url\(\s*["\']?([^"\')\s]+)',
+    re.IGNORECASE | re.DOTALL,
+)
+
+
+def find_logo_url_in_html(html: str, base_url: str, css_texts: list[str] | None = None) -> str | None:
+    """HTML 에서 logo URL 찾기. 우선순위: 1) class/id/alt 명시 img 2) 외부 CSS .logo bg 3) header/nav 첫 img."""
+
+    def _is_noise(src: str) -> bool:
+        if not src or src.startswith("data:"):
+            return True
+        if re.search(r"(blank|spacer|pixel|transparent|1x1)\b", src, re.IGNORECASE):
+            return True
+        if re.search(r"(lang[-_]?(kor|eng|chn|jpn|rus|jp|en|ko|cn|ar|in)|flag|country|icon-|btn-|arrow|prev|next|search)\b", src, re.IGNORECASE):
+            return True
+        return False
+
+    for pat in LOGO_IMG_PATTERNS[:8]:
+        for m in pat.finditer(html):
+            src = m.group(1)
+            if _is_noise(src):
+                continue
+            return urljoin(base_url, src)
+
+    for css in (css_texts or []):
+        m = LOGO_CSS_PATTERN.search(css)
+        if m:
+            src = m.group(1)
+            if not _is_noise(src):
+                return urljoin(base_url, src)
+
+    for pat in LOGO_IMG_PATTERNS[8:]:
+        for m in pat.finditer(html):
+            src = m.group(1)
+            if _is_noise(src):
+                continue
+            return urljoin(base_url, src)
+
+    return None
+
+
+# ── 색상 추출 ────────────────────────────────────────────────────────────────
+
+
+HEX6 = re.compile(r"#([0-9a-fA-F]{6})\b")
+HEX3 = re.compile(r"#([0-9a-fA-F]{3})\b(?![0-9a-fA-F])")
+RGB  = re.compile(r"rgba?\(\s*(\d{1,3})\s*,\s*(\d{1,3})\s*,\s*(\d{1,3})\s*(?:,\s*[\d.]+\s*)?\)")
+STYLE_BLOCK = re.compile(r"<style[^>]*>(.*?)</style>", re.IGNORECASE | re.DOTALL)
+
+NOISE = {
+    "#ffffff", "#000000", "#fff", "#000",
+    "#333", "#222", "#111", "#444", "#555", "#666", "#777", "#888", "#999",
+    "#aaa", "#bbb", "#ccc", "#ddd", "#eee", "#f0f0f0", "#f5f5f5", "#fafafa",
+}
+
+
+def _normalize(hex_str: str) -> str:
+    h = hex_str.lstrip("#").lower()
+    if len(h) == 3:
+        h = "".join(c * 2 for c in h)
+    if len(h) == 8:
+        h = h[:6]
+    return f"#{h}"
+
+
+def _rgb_to_hex(r: int, g: int, b: int) -> str:
+    return f"#{r:02x}{g:02x}{b:02x}"
+
+
+def _hex_to_rgb(h: str) -> tuple[int, int, int]:
+    h = h.lstrip("#")
+    return int(h[0:2], 16), int(h[2:4], 16), int(h[4:6], 16)
+
+
+def _distance(a: str, b: str) -> float:
+    ar, ag, ab = _hex_to_rgb(a)
+    br, bg, bb = _hex_to_rgb(b)
+    return ((ar - br) ** 2 + (ag - bg) ** 2 + (ab - bb) ** 2) ** 0.5
+
+
+def _is_grayscale(h: str, tol: int = 12) -> bool:
+    r, g, b = _hex_to_rgb(h)
+    return max(r, g, b) - min(r, g, b) < tol
+
+
+def _extract_hex(text: str) -> list[str]:
+    out: list[str] = []
+    out.extend(_normalize(m.group(0)) for m in HEX6.finditer(text))
+    out.extend(_normalize(m.group(0)) for m in HEX3.finditer(text))
+    for m in RGB.finditer(text):
+        r, g, b = int(m.group(1)), int(m.group(2)), int(m.group(3))
+        if 0 <= r <= 255 and 0 <= g <= 255 and 0 <= b <= 255:
+            out.append(_rgb_to_hex(r, g, b))
+    return out
+
+
+def _cluster(colors: Counter, threshold: float = 25.0) -> list[tuple[str, int]]:
+    ranked = colors.most_common()
+    clusters: list[tuple[str, int]] = []
+    for color, count in ranked:
+        merged = False
+        for i, (rep, rep_count) in enumerate(clusters):
+            if _distance(color, rep) < threshold:
+                clusters[i] = (rep, rep_count + count)
+                merged = True
+                break
+        if not merged:
+            clusters.append((color, count))
+    return clusters
+
+
+def extract_brand_colors_from_text(html: str, css_texts: list[str], source_url: str = "") -> dict:
+    """HTML + CSS 텍스트에서 hex 빈도 분석 → primary/accent/text + palette. (fetch 없음)"""
+    all_text_chunks: list[str] = list(STYLE_BLOCK.findall(html))
+    all_text_chunks.append(html)
+    all_text_chunks.extend(css_texts)
+
+    counter: Counter = Counter()
+    for text in all_text_chunks:
+        for color in _extract_hex(text):
+            if color in NOISE:
+                continue
+            counter[color] += 1
+
+    if not counter:
+        logger.info("[brand_parser] no colors extracted from %s", source_url)
+        return {}
+
+    clustered = _cluster(counter)
+    chromatic = [c for c, _ in clustered if not _is_grayscale(c)]
+    grayscale = [c for c, _ in clustered if _is_grayscale(c)]
+
+    palette_top = clustered[:8]
+    palette = [{"name": f"색상 {i+1}", "hex": h, "usage": f"빈도 {n}"} for i, (h, n) in enumerate(palette_top)]
+
+    return {
+        "brand_colors": {
+            "primary": chromatic[0] if chromatic else None,
+            "accent": chromatic[1] if len(chromatic) > 1 else None,
+            "text": grayscale[0] if grayscale else None,
+        },
+        "color_palette": palette,
+        "extracted_from": "html+css",
+    }

app/services/branding.py

@@ -0,0 +1,86 @@
+"""report 단계 - Gemini Vision 으로 로고 묘사 + 채널 로고 매칭."""
+import logging
+import os
+from urllib.parse import urlparse
+from common.db.source import (
+    select_run_raw_data, update_raw_info_merge,
+    select_branding_info_id, select_mainpage_logo_url,
+)
+from common.utils import _run_optional_step
+from integrations.llm.gemini_vision import VisionClient
+
+logger = logging.getLogger(__name__)
+
+
+async def _describe_logo(analysis_run_id: str, info_id: int, vc: VisionClient) -> None:
+    """공식 로고 정성 묘사. branding raw_info["brandAssets"] 머지.
+    호출 우선순위: raw_info.logo_url 컬럼 (HTML parser canonical) → firecrawl 메타 fallback."""
+    raw = await select_run_raw_data(analysis_run_id)
+    mainpage = raw.get("mainpage") or {}
+    homepage_url = mainpage.get("sourceUrl") or ""
+    branding_meta = mainpage.get("branding") or {}
+    column_logo = await select_mainpage_logo_url(analysis_run_id)
+    candidates = [u for u in [
+        column_logo,
+        branding_meta.get("logoUrl"),
+        branding_meta.get("faviconUrl"),
+    ] if u]
+    if homepage_url:
+        parsed = urlparse(homepage_url)
+        if parsed.scheme and parsed.netloc:
+            candidates.append(f"{parsed.scheme}://{parsed.netloc}/favicon.ico")
+    if not candidates:
+        logger.info("[brand_logo] skip — no candidates")
+        return
+    logger.info("[brand_logo] start run=%s candidates=%d", analysis_run_id, len(candidates))
+    result: dict = {}
+    for cand in candidates:
+        result = await vc.analyze_brand_assets(logo_url=cand, homepage_url=homepage_url)
+        if result:
+            break
+    if result:
+        await update_raw_info_merge(info_id, {"brandAssets": result})
+    logger.info("[brand_logo] done keys=%s", list(result.keys()) if result else None)
+
+
+async def _describe_channel_logos(analysis_run_id: str, info_id: int, vc: VisionClient) -> None:
+    """채널 프로필 로고를 공식 로고와 비교. branding raw_info["channelLogos"] 머지."""
+    raw = await select_run_raw_data(analysis_run_id)
+    official = await select_mainpage_logo_url(analysis_run_id)
+    _label = {
+        "instagram":    "Instagram",
+        "facebook":     "Facebook",
+        "youtube":      "YouTube",
+        "instagram_en": "Instagram EN",
+        "facebook_en":  "Facebook EN",
+        "tiktok":       "TikTok",
+    }
+    logos = [{"channel": label, "url": img}
+             for key, label in _label.items()
+             if (img := (raw.get(key) or {}).get("_logo_url"))]
+    if not logos:
+        logger.info("[channel_logos] skip — no channel profileImages")
+        return
+    logger.info("[channel_logos] start run=%s channels=%s official=%s",
+                analysis_run_id, [l["channel"] for l in logos], bool(official))
+    result = await vc.describe_channel_logos(official, logos)
+    if result:
+        await update_raw_info_merge(info_id, {"channelLogos": result})
+    logger.info("[channel_logos] done keys=%s", list(result.keys()) if result else None)
+
+
+async def analyze_branding(analysis_run_id: str) -> None:
+    """report build 직전 호출 — 로고 묘사 + 채널 로고 매칭 (Gemini). 둘 다 격리."""
+    api_key = os.getenv("GEMINI_API_KEY")
+    if not api_key:
+        logger.info("[branding] skip — GEMINI_API_KEY 없음")
+        return
+    branding_info_id = await select_branding_info_id(analysis_run_id)
+    if branding_info_id is None:
+        logger.info("[branding] skip — branding source 없음 run=%s", analysis_run_id)
+        return
+    vc = VisionClient(api_key)
+    logger.info("[branding] start run=%s", analysis_run_id)
+    await _run_optional_step(_describe_logo(analysis_run_id, branding_info_id, vc), "brand_logo")
+    await _run_optional_step(_describe_channel_logos(analysis_run_id, branding_info_id, vc), "channel_logos")
+    logger.info("[branding] done run=%s", analysis_run_id)

app/services/collect.py

@@ -2,16 +2,27 @@ import asyncio
 import logging
 from common.db.hospital import update_hospital_status, update_hospital
 from common.db.source import select_run_sources, update_raw_info_status, update_raw_info
-from common.utils import get_env
+from common.utils import get_env, _run_optional_step
 from integrations.apify import ApifyClient
 from integrations.naver import NaverClient
 from integrations.youtube import YouTubeClient
 from integrations.firecrawl import FirecrawlClient
 from models.status import SourceType
+from integrations.site_fetcher import fetch_html_and_css
+from services.brand_parser import find_logo_url_in_html, extract_brand_colors_from_text
+from common.db.source import update_raw_info_merge, update_raw_info_logo_url, select_run_raw_data
+from common.db.base import fetchone
+from services.facebook_audit import transform_for_storage as transform_facebook
 
 logger = logging.getLogger(__name__)
 
 
+async def _save_with_logo(info_id: int, data: dict) -> None:
+    await update_raw_info(info_id, data)
+    if data.get("profileImage"):
+        await update_raw_info_logo_url(info_id, data["profileImage"])
+
+
 async def collect_instagram(analysis_run_id: str, info_id: int, url: str) -> None:
     logger.info("[instagram] start run=%s url=%s", analysis_run_id, url)
     await update_raw_info_status(info_id, "processing")
@@ -20,7 +31,7 @@ async def collect_instagram(analysis_run_id: str, info_id: int, url: str) -> Non
         await update_raw_info_status(info_id, "failed")
         logger.warning("[instagram] failed run=%s", analysis_run_id)
         return
-    await update_raw_info(info_id, data)
+    await _save_with_logo(info_id, data)
     logger.info("[instagram] done run=%s", analysis_run_id)
 
 
@@ -32,7 +43,8 @@ async def collect_facebook(analysis_run_id: str, info_id: int, url: str) -> None
         await update_raw_info_status(info_id, "failed")
         logger.warning("[facebook] failed run=%s", analysis_run_id)
         return
-    await update_raw_info(info_id, data)
+    data = transform_facebook(data)
+    await _save_with_logo(info_id, data)
     logger.info("[facebook] done run=%s", analysis_run_id)
 
 
@@ -56,7 +68,7 @@ async def collect_youtube(analysis_run_id: str, info_id: int, url: str) -> None:
         await update_raw_info_status(info_id, "failed")
         logger.warning("[youtube] failed run=%s", analysis_run_id)
         return
-    await update_raw_info(info_id, data)
+    await _save_with_logo(info_id, data)
     logger.info("[youtube] done run=%s", analysis_run_id)
 
 
@@ -81,30 +93,107 @@ async def collect_mainpage(analysis_run_id: str, info_id: int, hospital_id: str,
         await update_raw_info_status(info_id, "failed")
         logger.warning("[mainpage] failed run=%s", analysis_run_id)
         return
+    # 홈페이지 URL 자체도 raw_data 에 박아둬야 brand_assets / 분석 단계에서 mainpage URL 재조회 없이 사용 가능.
+    data = {**data, "sourceUrl": url}
     await update_raw_info(info_id, data)
     await update_hospital(hospital_id, data, analysis_run_id=analysis_run_id)
     logger.info("[mainpage] done run=%s", analysis_run_id)
 
 
+async def collect_tiktok(analysis_run_id: str, info_id: int, url: str) -> None:
+    logger.info("[tiktok] start run=%s url=%s", analysis_run_id, url)
+    await update_raw_info_status(info_id, "processing")
+    data = await ApifyClient(get_env("APIFY_API_TOKEN")).get_tiktok_profile(url)
+    if data is None:
+        await update_raw_info_status(info_id, "failed")
+        logger.warning("[tiktok] failed run=%s", analysis_run_id)
+        return
+    await _save_with_logo(info_id, data)
+    logger.info("[tiktok] done run=%s", analysis_run_id)
+
+
+async def collect_naver_cafe(analysis_run_id: str, info_id: int, url: str) -> None:
+    """카페는 로그인 필요라 본문 못 봄. URL 활성·cafeId·이름 언급수만 신호로 수집."""
+    logger.info("[naver_cafe] start run=%s url=%s", analysis_run_id, url)
+    await update_raw_info_status(info_id, "processing")
+    data = await NaverClient(get_env("NAVER_CLIENT_ID"), get_env("NAVER_CLIENT_SECRET")).get_cafe_info(url)
+    if data is None:
+        await update_raw_info_status(info_id, "failed")
+        logger.warning("[naver_cafe] failed run=%s", analysis_run_id)
+        return
+    await update_raw_info(info_id, data)
+    logger.info("[naver_cafe] done run=%s", analysis_run_id)
+
+
+async def collect_kakaotalk(analysis_run_id: str, info_id: int, url: str) -> None:
+    """카카오톡은 수집 X — URL 보관만. LLM이 채널 존재 신호로만 사용."""
+    logger.info("[kakaotalk] url-only run=%s url=%s", analysis_run_id, url)
+    await update_raw_info(info_id, {"url": url})
+
+
+async def collect_brand_basics(analysis_run_id: str, info_id: int) -> None:
+    logger.info("[brand_basics] start run=%s info=%s", analysis_run_id, info_id)
+    raw = await select_run_raw_data(analysis_run_id)
+    mainpage = raw.get("mainpage") or {}
+    homepage_url = mainpage.get("sourceUrl") or ""
+    branding_meta = mainpage.get("branding") or {}
+
+    html, css_texts = await fetch_html_and_css(homepage_url) if homepage_url else ("", [])
+    html_logo_url = find_logo_url_in_html(html, homepage_url, css_texts) if html else None
+    css_colors    = extract_brand_colors_from_text(html, css_texts, homepage_url) if html else {}
+
+    logo_url = html_logo_url or branding_meta.get("logoUrl") or branding_meta.get("ogImage")
+    if logo_url:
+        mainpage_row = await fetchone(
+            "SELECT ri.info_id FROM raw_info ri JOIN remote_source rs USING (source_id)"
+            " WHERE ri.analysis_run_id = %s AND rs.source_type = 'mainpage' LIMIT 1",
+            (analysis_run_id,),
+        )
+        if mainpage_row:
+            await update_raw_info_logo_url(mainpage_row["info_id"], logo_url)
+
+    payload: dict = {}
+    if css_colors:
+        if css_colors.get("brand_colors"):  payload["brand_colors"]  = css_colors["brand_colors"]
+        if css_colors.get("color_palette"): payload["color_palette"] = css_colors["color_palette"]
+        payload["color_source"] = "html+css"
+    if payload:
+        await update_raw_info_merge(info_id, {"brandAssets": payload})
+    logger.info("[brand_basics] done logo_url=%s colors=%s", bool(logo_url), bool(payload))
+
+
 async def collect_all(analysis_run_id: str, hospital_id: str) -> None:
     rows = await select_run_sources(analysis_run_id)
 
+    # source_type → collector. KR/EN 구분은 collector 입장에서 동일, language 컬럼만 다름.
     _collectors = {
         SourceType.INSTAGRAM:    collect_instagram,
         SourceType.FACEBOOK:     collect_facebook,
         SourceType.NAVER_BLOG:   collect_naver_blog,
         SourceType.YOUTUBE:      collect_youtube,
         SourceType.GANGNAM_UNNI: collect_gangnam_unni,
+        SourceType.TIKTOK:       collect_tiktok,
+        SourceType.NAVER_CAFE:   collect_naver_cafe,
+        SourceType.KAKAOTALK:    collect_kakaotalk,
     }
 
     tasks = []
+    branding_info_id: int | None = None
     for row in rows:
         info_id     = row["info_id"]
         source_type = row["source_type"]
         url         = row["url"]
+        if source_type == SourceType.BRANDING:
+            branding_info_id = info_id  # mainpage·채널 수집 끝난 뒤 2단계에서 사용
+            continue
         if source_type == SourceType.MAINPAGE:
             tasks.append(collect_mainpage(analysis_run_id, info_id, hospital_id, url))
         elif source_type in _collectors:
             tasks.append(_collectors[source_type](analysis_run_id, info_id, url))
 
     await asyncio.gather(*tasks, return_exceptions=True)
+
+    # 2단계: branding (brandAssets → channelLogos 한 raw_info 안에 머지). mainpage·채널 raw_data 의존이라 순차.
+    # 부가 기능이라 실패해도 리포트는 나와야 하므로 _run_optional_step 으로 격리.
+    if branding_info_id is not None:
+        await _run_optional_step(collect_brand_basics(analysis_run_id, branding_info_id), "brand_basics")

app/services/facebook_audit.py

@@ -0,0 +1,106 @@
+"""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)

app/services/instagram_audit.py

@@ -0,0 +1,50 @@
+"""Instagram audit 계정(KR·EN)을 수집 데이터로 구성.
+fix 값(handle/followers/highlights/content_format 등)은 전부 코드에서 박는다 — LLM 출력 무시."""
+
+from integrations.llm.schemas.report import InstagramAudit
+
+_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_audit(instagram: dict, instagram_en: dict, channel_logos: dict) -> 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 InstagramAudit.model_validate({"accounts": accounts}).model_dump()

app/services/kpi_dashboard.py

@@ -0,0 +1,96 @@
+"""mockup 7개 역분석 — 채널 규모별 3개월/12개월 target 성장률 공식."""
+
+from integrations.llm.schemas.report import KPIMetric
+
+
+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 [KPIMetric.model_validate(k).model_dump() for k in kpis]

requirements.txt

@@ -10,3 +10,4 @@ passlib[bcrypt]==1.7.4
 python-multipart==0.0.26
 uuid6==2025.0.1
 aiomysql==0.3.2
+resvg-py==0.3.2