app/api/analysis.py

@@ -8,28 +8,10 @@ from models.file import FileListItem, FileType, FileUploadResponse
 from models.status import AnalysisStatus
 from services.pipeline import run_pipeline
 from services.file import get_analysis_files_response, handle_analysis_file_upload, soft_delete_analysis_file
-from mock_urls import MOCK_CLINICS
-from common.utils import _normalize_homepage, _with_scheme
 
 router = APIRouter(prefix="/api/analysis", tags=["analysis"], dependencies=[Depends(verify_api_key)])
 logger = logging.getLogger(__name__)
 
-# 추후 DB에 클리닉별로 매핑할 채널(틱톡/영문 인스타·페북). 지금은 mock_urls에서 homepage 매칭으로 보충.
-def _extra_channels_from_mockurls(homepage_url: str) -> dict:
-    """homepage로 mock_urls에서 클리닉을 찾아 틱톡/영문 인스타·페북 URL 반환 (없으면 {})."""
-    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 {
-                "tiktok": _with_scheme(urls.get("tiktok")),
-                "instagram_en": _with_scheme(urls.get("instagramEn")),
-                "facebook_en": _with_scheme(urls.get("facebookEn")),
-            }
-    return {}
-
 
 @router.post("", status_code=status.HTTP_202_ACCEPTED, response_model=AnalysisStartResponse)
 async def start_analysis(body: AnalysisCreate, background_tasks: BackgroundTasks):
@@ -56,15 +38,7 @@ async def start_analysis(body: AnalysisCreate, background_tasks: BackgroundTasks
         ig_id, fb_id, nb_id, yt_id, gu_id,
     )
 
-    # 클라 값 우선, 없으면 보충 (추후 DB에서 클리닉별로 가져올 값)
-    mock_extra = _extra_channels_from_mockurls(hospital["url"])
-    extra_channels = {
-        "tiktok":       body.channels.tiktok or mock_extra.get("tiktok"),
-        "instagram_en": body.channels.instagram_en or mock_extra.get("instagram_en"),
-        "facebook_en":  body.channels.facebook_en or mock_extra.get("facebook_en"),
-    }
-    logger.info("[analysis] extra_channels=%s (mock_matched=%s)", extra_channels, bool(mock_extra))
-    background_tasks.add_task(run_pipeline, analysis_run_id, extra_channels)
+    background_tasks.add_task(run_pipeline, analysis_run_id)
 
     return AnalysisStartResponse(
         analysis_run_id=analysis_run_id,

app/common/db.py

@@ -263,19 +263,6 @@ async def save_hospital_raw_data(hospital_id: str, data: dict, analysis_run_id:
     await _insert_hospital_history(hospital_id, analysis_run_id)
 
 
-async def merge_hospital_raw_data(hospital_id: str, patch: dict) -> None:
-    """hospital_baseinfo.raw_data를 읽어 patch를 top-level 병합 후 저장 (read-modify-write).
-    부가 수집 단계들이 순차로 raw_data에 키를 덧붙일 때 사용."""
-    row = await fetchone("SELECT raw_data FROM hospital_baseinfo WHERE hospital_id = %s", (hospital_id,))
-    raw = row["raw_data"] if row else None
-    raw_data = json.loads(raw) if isinstance(raw, str) else (raw or {})
-    raw_data.update(patch)
-    await execute(
-        "UPDATE hospital_baseinfo SET raw_data = %s WHERE hospital_id = %s",
-        (json.dumps(raw_data, ensure_ascii=False), hospital_id),
-    )
-
-
 async def get_market_analysis(analysis_run_id: str) -> dict:
     rows = await fetchall(
         "SELECT analysis_type, data FROM market_analysis WHERE analysis_run_id = %s AND status = 'done'",

app/common/utils.py

@@ -1,11 +1,8 @@
 import os
 import asyncio
-import logging
 from http import HTTPMethod
 import httpx
 
-logger = logging.getLogger(__name__)
-
 REQUEST_TIMEOUT = 60
 
 
@@ -40,27 +37,3 @@ 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("/")
-
-
-def _with_scheme(u: str | None) -> str | None:
-    """scheme 없는 URL에 https:// 보정 (수집기 파싱용). 빈 값은 None."""
-    return (u if "://" in u else "https://" + u) if u else None

app/integrations/apify.py

@@ -44,7 +44,6 @@ class ApifyClient:
             return None
         return {
             "username": profile["username"],
-            "profileImage": profile.get("profilePicUrlHD") or profile.get("profilePicUrl"),
             "followers": profile.get("followersCount", 0),
             "following": profile.get("followsCount", 0),
             "posts": profile.get("postsCount", 0),
@@ -135,7 +134,6 @@ class ApifyClient:
             return None
         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),
@@ -147,45 +145,3 @@ class ApifyClient:
             "intro": page.get("intro"),
             "rating": page.get("rating"),
         }
-
-    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("clockworks~tiktok-scraper", {
-            "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/color_extractor.py

@@ -1,250 +0,0 @@
-"""홈페이지 HTML/CSS에서 hex 색상 직접 추출 + 빈도 기반 brand palette 산출.
-
-Vision LLM에 의존하지 않고 페이지의 실제 CSS 값을 정규식으로 잡음.
-로고만 분석하는 Vision보다 사이트 전체 컬러 시스템 (primary/secondary/background/text)을 더 정확히 추출.
-"""
-import logging
-import re
-import ssl
-from collections import Counter
-from urllib.parse import urljoin, urlparse
-import httpx
-
-logger = logging.getLogger(__name__)
-
-
-def _make_ssl_context() -> ssl.SSLContext:
-    """오래된 한국 의료 사이트들이 SSL DH_KEY_TOO_SMALL / cipher 약함 등으로 차단되는 문제 우회.
-    보안 등급 1로 낮춤 + cert 검증 유지."""
-    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 받기. 그랜드/톡스앤필 같은 oldsite 대응."""
-    headers = {"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"}
-    # 1차: 표준 검증
-    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)
-    # 2차: 약한 cipher 허용
-    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)
-    # 3차: SSL 검증 끔 (host mismatch 등)
-    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, ""
-
-LOGO_IMG_PATTERNS = [
-    # 1) <img class="...logo..." src="...">
-    re.compile(r'<img[^>]*\bclass=["\'][^"\']*\blogo\b[^"\']*["\'][^>]*\bsrc=["\']([^"\']+)["\']', re.IGNORECASE),
-    # 2) <img src="..." class="...logo...">
-    re.compile(r'<img[^>]*\bsrc=["\']([^"\']+)["\'][^>]*\bclass=["\'][^"\']*\blogo\b[^"\']*["\']', re.IGNORECASE),
-    # 3) <img id="...logo..." src="...">
-    re.compile(r'<img[^>]*\bid=["\'][^"\']*\blogo\b[^"\']*["\'][^>]*\bsrc=["\']([^"\']+)["\']', re.IGNORECASE),
-    # 4) <img alt="...logo..." src="...">
-    re.compile(r'<img[^>]*\balt=["\'][^"\']*\blogo\b[^"\']*["\'][^>]*\bsrc=["\']([^"\']+)["\']', re.IGNORECASE),
-    # 5) <a/h1 class="logo"><...nested...><img src="...">
-    re.compile(r'<(?:a|h[1-6]|div|span)[^>]*\b(?:class|id)=["\'][^"\']*\blogo\b[^"\']*["\'][^>]*>(?:[^<]|<(?!img))*<img[^>]*\bsrc=["\']([^"\']+)["\']', re.IGNORECASE | re.DOTALL),
-    # 6) inline background-image: <a/div class="logo" style="background-image: url(...)">
-    re.compile(r'<(?:a|div|span|h[1-6])[^>]*\b(?:class|id)=["\'][^"\']*\blogo\b[^"\']*["\'][^>]*\bstyle=["\'][^"\']*background(?:-image)?\s*:\s*url\(\s*["\']?([^"\')\s]+)', re.IGNORECASE),
-    # 7) inline background-image: <a/div style="background-image: url(...)" class="logo">  (속성 순서 반대)
-    re.compile(r'<(?:a|div|span|h[1-6])[^>]*\bstyle=["\'][^"\']*background(?:-image)?\s*:\s*url\(\s*["\']?([^"\')\s]+)[^"\']*["\'][^>]*\b(?:class|id)=["\'][^"\']*\blogo\b', re.IGNORECASE),
-    # 8) src 자체에 "logo" 포함 (header_logo.png, brand-logo.svg 등)
-    re.compile(r'<img[^>]*\bsrc=["\']([^"\']*\blogo\b[^"\']*\.(?:png|svg|jpe?g|webp)[^"\']*)["\']', re.IGNORECASE),
-    # 9) <header>...<img src="..."> (헤더 영역 첫 img)
-    re.compile(r'<header\b[^>]*>(?:[^<]|<(?!img))*<img[^>]*\bsrc=["\']([^"\']+\.(?:png|svg|jpe?g|webp)[^"\']*)["\']', re.IGNORECASE | re.DOTALL),
-    # 10) <nav>...<img src="..."> (nav 영역 첫 img)
-    re.compile(r'<nav\b[^>]*>(?:[^<]|<(?!img))*<img[^>]*\bsrc=["\']([^"\']+\.(?:png|svg|jpe?g|webp)[^"\']*)["\']', re.IGNORECASE | re.DOTALL),
-    # 11) Open Graph image (대표 이미지) - 최후 fallback
-    re.compile(r'<meta[^>]*\bproperty=["\']og:image["\'][^>]*\bcontent=["\']([^"\']+)["\']', re.IGNORECASE),
-    re.compile(r'<meta[^>]*\bcontent=["\']([^"\']+)["\'][^>]*\bproperty=["\']og:image["\']', re.IGNORECASE),
-]
-
-# CSS 파일에서 .logo { background-image: url(...) } 추출용
-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 찾기. class/id/alt → 부모 + 중첩 img → background-image → src에 logo → header/nav → og:image 순."""
-    for pat in LOGO_IMG_PATTERNS:
-        for m in pat.finditer(html):
-            src = m.group(1)
-            if not src or src.startswith("data:"):
-                continue
-            if re.search(r"(blank|spacer|pixel|transparent|1x1)\b", src, re.IGNORECASE):
-                continue
-            return urljoin(base_url, src)
-    # 외부 CSS에서 .logo background-image 추출
-    for css in (css_texts or []):
-        m = LOGO_CSS_PATTERN.search(css)
-        if m:
-            src = m.group(1)
-            if src and not src.startswith("data:"):
-                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*)?\)")
-CSS_VAR_HEX = re.compile(r"--[\w-]+\s*:\s*(#[0-9a-fA-F]{3,8})", re.IGNORECASE)
-CSS_LINK = re.compile(r'<link[^>]+rel=["\']stylesheet["\'][^>]+href=["\']([^"\']+)["\']', re.IGNORECASE)
-STYLE_BLOCK = re.compile(r"<style[^>]*>(.*?)</style>", re.IGNORECASE | re.DOTALL)
-
-# 무채색·아주 흔한 노이즈 컬러 (이런 건 brand color로 잡지 않음)
-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]:
-    """텍스트에서 모든 hex 색상 추출 (정규화)."""
-    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
-
-
-async def _fetch_html_and_css(homepage_url: str, max_css_files: int = 8) -> tuple[str, list[str]]:
-    """홈페이지 HTML + 외부 CSS(Top N)를 한 번에 fetch. 로고/색상 추출이 사이트를 중복으로 긁지 않도록 공유.
-    _fetch_html이 SSL 약함/host mismatch까지 fallback 처리. 실패 시 ("", [])."""
-    status, html = await _fetch_html(homepage_url)
-    if status != 200 or not html:
-        logger.warning("[color_extractor] 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
-
-
-def _colors_from_text(html: str, css_texts: list[str], source_url: str = "") -> dict:
-    """이미 받아온 HTML + CSS 텍스트에서 hex 빈도 분석 → primary/accent/text + palette. (fetch 없음, 순수 계산)"""
-    # 1. HTML 내 <style> 블록 + 통째(inline style="color:#...") + 외부 CSS
-    all_text_chunks: list[str] = list(STYLE_BLOCK.findall(html))
-    all_text_chunks.append(html)
-    all_text_chunks.extend(css_texts)
-
-    # 2. 모든 hex 추출 (NOISE 제외)
-    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("[color_extractor] no colors extracted from %s", source_url)
-        return {}
-
-    # 3. 비슷한 색 클러스터링
-    clustered = _cluster(counter)
-
-    # 4. primary = 빈도 높은 채도 있는 색 / accent = 두번째 채도 있는 색 / text = 빈도 높은 무채색
-    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",
-    }
-
-
-async def extract_brand_colors_from_site(homepage_url: str, max_css_files: int = 8) -> dict:
-    """홈페이지 HTML + 외부 CSS fetch → hex 색상 빈도 분석 → primary/accent/text + palette 5종."""
-    html, css_texts = await _fetch_html_and_css(homepage_url, max_css_files)
-    if not html:
-        return {}
-    return _colors_from_text(html, css_texts, homepage_url)
-
-
-async def extract_brand_assets_from_site(homepage_url: str, max_css_files: int = 8) -> dict:
-    """사이트를 한 번만 fetch해서 로고 URL과 brand 색상을 함께 추출.
-    반환: {"logo_url": str | None, "colors": {brand_colors, color_palette, ...} | {}}"""
-    html, css_texts = await _fetch_html_and_css(homepage_url, max_css_files)
-    if not html:
-        return {"logo_url": None, "colors": {}}
-    return {
-        "logo_url": find_logo_url_in_html(html, homepage_url, css_texts=css_texts),
-        "colors": _colors_from_text(html, css_texts, homepage_url),
-    }

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 from the actual header/main <img> src, og:image from <meta property='og:image'> content, 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, favicon URL)",
                     "schema": {
                         "type": "object",
                         "properties": {
@@ -119,7 +119,6 @@ class FirecrawlClient:
                                     "headingFont":     {"type": "string"},
                                     "bodyFont":        {"type": "string"},
                                     "logoUrl":         {"type": "string"},
-                                    "ogImage":         {"type": "string"},
                                     "faviconUrl":      {"type": "string"},
                                 },
                             },

app/integrations/llm/schemas/plan.py

@@ -15,11 +15,6 @@ 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
-    channel_logos: str | None = None
-    brand_assets: str | None = None
 
 
 # --- BrandGuide ---

app/integrations/llm/schemas/report.py

@@ -321,12 +321,6 @@ 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
-    channel_logos: str | None = None
 
 
 # --- MarketingReport ---

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

@@ -32,47 +32,19 @@
 ## 분석 리포트
 {report}
 
-## 추가 채널 데이터 (틱톡 / 인스타그램 EN / 페이스북 EN)
-아래에 데이터가 있는 채널은 channelStrategies와 channelBranding에 **반드시 포함**하세요 (틱톡, 영문 인스타그램, 영문 페이스북). null이면 제외.
-
-### 틱톡 (TikTok)
-{tiktok}
-
-### 인스타그램 (영문 계정)
-{instagram_en}
-
-### 페이스북 (영문 페이지)
-{facebook_en}
-
-## 채널별 로고 분석 (Gemini Vision) — 채널룰/일관성의 근거
-{channel_logos}
-- 위 channel_logos[]의 각 항목: channel(채널명), logo_description(프로필이 어떻게 생겼는지), is_official(공식 로고와 일치 여부).
-- **channelBranding[]를 이 데이터로 채우세요**: 채널별로 profilePhoto=해당 채널의 logo_description, currentStatus=is_official이 true면 "correct" / false면 "incorrect" (데이터 없는 채널은 "missing"). bannerSpec은 권장 배너 규격(크기/디자인)을 작성.
-- **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: **brand_assets.color_palette / brand_colors의 hex를 그대로 사용** (홈페이지 CSS 추출값, 지어내기 금지). 3~5개, 각 hex에 name/usage 부여
+- colors: 병원 아이덴티티에 맞는 컬러 팔레트 3~5개 (hex + 사용 가이드)
 - fonts: 제목/본문/캡션용 폰트 시스템 (한글/영문 포함)
-- logoRules: 로고 사용 규칙 4~6개. 각 항목은 rule / description / correct 3개 필드로 구성:
-  - rule: 규칙을 요약한 **구체적인 제목**. "DO"·"DON'T" 같은 단어를 그대로 넣지 말 것. 실제 규칙 내용을 쓸 것 (예: "보라색+골드 깃털 로고 통일 사용", "모델 사진 프로필 금지", "비공식 변형 로고 사용 금지", "로고 주변 여백 확보").
-  - description: 해당 규칙의 상세 설명.
-  - correct: 권장 규칙(DO)이면 true, 금지/지양 규칙(DON'T)이면 false. 권장(true)과 금지(false)를 섞어서 작성.
+- logoRules: DO/DON'T 형식의 로고 사용 규칙 4~6개
 - toneOfVoice: 브랜드 성격 키워드, 커뮤니케이션 스타일, 권장/지양 표현 예시
 - channelBranding: 리포트에 존재하는 채널별 브랜딩 적용 규칙
 - brandInconsistencies: 채널 간 브랜딩 불일치 항목 및 개선 권고
 
 ### Section 2: channelStrategies
 - 리포트에 데이터가 있는 채널만 포함
-- **currentStatus는 현재 채널 상태를 실제 수치로 서술** (예: "14,047 팔로워, Reels 0개", "104K 구독자, 주 2~3회 업로드"). `excellent`/`warning`/`good` 같은 등급·평가어를 절대 쓰지 마세요.
-- targetGoal은 구체적 목표 수치로 작성 (예: "50K 팔로워, Reels 주 5개")
-- 각 채널의 우선순위(P0/P1/P2), 콘텐츠 유형, 게시 빈도, 포맷 가이드라인 작성
+- 각 채널의 우선순위(P0/P1/P2), 목표, 콘텐츠 유형, 게시 빈도, 포맷 가이드라인 작성
 - customerJourneyStage는 해당 채널의 주요 기여 단계로 설정
 
 ### Section 3: contentStrategy

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

@@ -12,17 +12,6 @@
 - 시술: {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을 근거로 하고, 채널 프로필 이미지가 공식 로고와 일치하는지 판단할 때도 이 묘사를 기준으로 삼으세요.
-
 ## 시장 분석 데이터
 
 ### 경쟁 병원
@@ -54,38 +43,9 @@
 ### 강남언니
 {gangnam_unni}
 
-### 틱톡 (TikTok)
-{tiktok}
-
-### 인스타그램 (영문 계정)
-{instagram_en}
-
-### 페이스북 (영문 페이지)
-{facebook_en}
-
-### 채널별 로고 분석 (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(설명문)을 넣으세요.
-- **instagram_audit.accounts[].profile_photo** 는 해당 채널 로고를 짧게 서술 (예: `"모델 사진 (브랜드 로고 아님)"`, `"VIEW 골드 로고"`). 긴 문장 말고 짧게.
-- 위 값들은 channel_logos 데이터 기반으로만 작성하고 추측하지 마세요.
-- 채널 간 로고 불일치(is_official=false)는 brand 일관성 진단(problem_diagnosis/weaknesses)에 반영하세요.
-
-## 기타 채널 현황 (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"으로 각각 포함하세요 (절대 누락 금지).**
-- **수집 데이터에 없는 채널(카카오톡/네이버플레이스/네이버카페/Threads 등)은 절대 임의로 만들지 마세요.** 데이터 없으면 그 채널은 생략 (랜덤 생성·추측 금지).
-- url은 수집 데이터의 실제 URL만 사용. 없으면 빈 문자열.
-
 ## 분석 지침
 
 - 점수는 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는 실제 수집된 수치 기반으로 현실적인 측정 가능 지표로 작성하세요.

app/integrations/vision.py

@@ -1,171 +0,0 @@
-"""Gemini Vision — 로고/브랜드 비주얼 자동 분석 (OpenAI 호환 모드).
-
-정확한 hex 색상은 color_extractor가 CSS에서 직접 뽑음 (Vision은 근사값밖에 못 냄).
-Vision은 사람이 봐야 알 수 있는 정성 정보 — 심볼 형태/워드마크/톤 — 를 담당.
-"""
-import base64
-import json
-import logging
-import re
-import httpx
-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)."""
-        try:
-            async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) 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
-                size = len(resp.content)
-                if size < 500:
-                    logger.warning("[vision] %s too small (%d bytes) — likely placeholder", url, size)
-                    return None
-                b64 = base64.b64encode(resp.content).decode("ascii")
-                return f"data:{mime};base64,{b64}"
-        except Exception as e:
-            logger.warning("[vision] fetch error %s: %s", url, e)
-            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 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": "로고에 쓰인 색감을 사람이 부르는 이름으로 서술 (예: \'딥네이비 + 골드\'). 정확한 hex는 출력하지 말 것"\n'
-            "}\n"
-            "주의: 색상 hex 값이나 logo URL 같은 필드는 출력하지 마세요 (별도 추출 로직이 처리)."
-        )
-        result = await self._ask(urls, prompt)
-        if not result:
-            return {}
-        # logo_images는 우리가 직접 채움 (Vision은 묘사만)
-        result["logo_images"] = {"circle": None, "horizontal": logo_url, "korean": None}
-        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"}"""
-        items = [c for c in channel_logos if c.get("url")]
-        if not items:
-            return None
-
-        # 공식 로고가 있으면 맨 앞에 두고 기준으로 삼음
-        urls: list[str] = []
-        if official_logo_url:
-            urls.append(official_logo_url)
-        urls.extend(c["url"] for c in items)
-        channel_order = ", ".join(c.get("channel", "?") for c in items)
-
-        if official_logo_url:
-            header = (
-                "첨부 이미지 중 **첫 번째가 이 병원의 공식 로고**입니다. "
-                f"이어지는 이미지들은 채널별 프로필 이미지이며 순서는: {channel_order}.\n"
-                "각 채널 로고를 1문장으로 설명하고, 공식 로고(첫 번째)와 일치하면 is_official=true, "
-                "비공식 변형/모델사진/다른 이미지면 false로 평가하세요.\n"
-            )
-        else:
-            header = (
-                f"첨부 이미지는 한 병원의 채널별 프로필 이미지입니다. 순서: {channel_order}.\n"
-                "각 채널 로고를 1문장으로 설명하세요 (공식 로고 기준이 없으므로 is_official은 판단 가능하면만).\n"
-            )
-        prompt = (
-            header
-            + "아래 JSON으로만 응답 (코드펜스 없이 순수 JSON):\n"
-            "{\n"
-            '  "channel_logos": [{"channel": "...", "logo_description": "...", "is_official": true}],\n'
-            '  "inconsistency_summary": "채널 간 로고 일관성 1~2문장 요약",\n'
-            '  "recommendation": "통합 권고 1문장"\n'
-            "}"
-        )
-        return await self._ask(urls, prompt)

app/integrations/youtube.py

@@ -88,11 +88,9 @@ 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_urls.py

@@ -1,141 +0,0 @@
-# 프론트가 아직 안 보내는 채널(틱톡/영문 인스타·페북)을 homepage로 매칭해 보충하는 임시 mock 데이터.
-# 기존 mockUrls.json을 파이썬 모듈로 전환 — 런타임 파일 I/O 없이 직접 import.
-
-MOCK_CLINICS = [
-    {
-        "label": "뷰성형외과",
-        "urls": {
-            "homepage": "viewclinic.com",
-            "youtube": "youtube.com/channel/UCQqqH3Klj2HQSHNNSVug-CQ",
-            "instagram": "instagram.com/viewplastic",
-            "facebook": "facebook.com/viewps1",
-            "naverPlace": "https://naver.me/x9BxGXkK",
-            "naverBlog": "blog.naver.com/viewclinicps",
-            "gangnamUnni": "gangnamunni.com/hospitals/189",
-            "tiktok": "tiktok.com/@viewplastic",
-            "tiktokEn": "tiktok.com/@viewplasticsurgery",
-            "instagramEn": "instagram.com/view_plastic_surgery",
-            "facebookEn": "facebook.com/viewclinic"
-        }
-    },
-    {
-        "label": "바노바기 성형외과",
-        "urls": {
-            "homepage": "banobagi.com",
-            "youtube": "youtube.com/c/banobagips",
-            "instagram": "instagram.com/banobagi_ps",
-            "facebook": "facebook.com/BanobagiPlasticSurgery",
-            "naverPlace": "https://naver.me/xxY2yLr5",
-            "naverBlog": "blog.naver.com/banobagiprs",
-            "gangnamUnni": "gangnamunni.com/hospitals/23",
-            "tiktok": "",
-            "instagramEn": "instagram.com/english_banobagi",
-            "facebookEn": "facebook.com/englishbanobagi"
-        }
-    },
-    {
-        "label": "ID 성형외과",
-        "urls": {
-            "homepage": "idhospital.com",
-            "youtube": "youtube.com/user/IDhospital",
-            "instagram": "instagram.com/idhospital",
-            "facebook": "facebook.com/idhospital0050",
-            "naverPlace": "https://naver.me/GtURpCEn",
-            "naverBlog": "",
-            "gangnamUnni": "gangnamunni.com/hospitals/257",
-            "tiktok": "tiktok.com/@idhospitalkorea",
-            "instagramEn": "instagram.com/idhospitalkorea",
-            "facebookEn": "facebook.com/idhospital.eng"
-        }
-    },
-    {
-        "label": "JK 성형외과",
-        "urls": {
-            "homepage": "jkplastic.com",
-            "youtube": "youtube.com/channel/UC5F8dEt32hdp3cTeFyls4qg",
-            "instagram": "instagram.com/jkplasticsurgery_kr",
-            "facebook": "facebook.com/jkmedicalgroup",
-            "naverPlace": "https://naver.me/x67y6cAc",
-            "naverBlog": "blog.naver.com/jkstory1",
-            "gangnamUnni": "gangnamunni.com/hospitals/858",
-            "tiktok": "tiktok.com/@jkplastic",
-            "instagramEn": "instagram.com/jkplasticsurgery",
-            "facebookEn": "facebook.com/jkplastic"
-        }
-    },
-    {
-        "label": "그랜드 성형외과",
-        "urls": {
-            "homepage": "grandsurgery.com",
-            "youtube": "youtube.com/channel/UCU2o_aHqsNFuqwtdzVM3xbQ",
-            "instagram": "instagram.com/grand_korea",
-            "facebook": "facebook.com/grandps.korea",
-            "naverPlace": "https://naver.me/Fw7MYKWK",
-            "naverBlog": "blog.naver.com/grandprs",
-            "gangnamUnni": "gangnamunni.com/hospitals/62",
-            "tiktok": "",
-            "instagramEn": "instagram.com/grandps_eng",
-            "facebookEn": "facebook.com/grandplasticsurgery"
-        }
-    },
-    {
-        "label": "BK 성형외과",
-        "urls": {
-            "homepage": "bkhospital.com",
-            "youtube": "youtube.com/channel/UChJONft3hemy5DGbXUveTFg",
-            "instagram": "instagram.com/bkhospital_korea",
-            "facebook": "",
-            "naverPlace": "https://naver.me/517CTH3W",
-            "naverBlog": "",
-            "gangnamUnni": "",
-            "tiktok": "",
-            "instagramEn": "instagram.com/english_bkhospital",
-            "facebookEn": "facebook.com/BKPSKoreaE"
-        }
-    },
-    {
-        "label": "톡스앤필",
-        "urls": {
-            "homepage": "toxnfill.com",
-            "youtube": "youtube.com/channel/UCFpFZkm7mclD-z_-j7FTUag",
-            "instagram": "instagram.com/toxnfill_official",
-            "facebook": "facebook.com/toxnfill.official",
-            "naverPlace": "https://naver.me/FvEmJIHA",
-            "naverBlog": "blog.naver.com/toxnfill",
-            "gangnamUnni": "gangnamunni.com/hospitals/3702",
-            "tiktok": "tiktok.com/@toxnfillglobal",
-            "instagramEn": "instagram.com/toxnfill_global",
-            "facebookEn": "facebook.com/p/Toxnfill-Global-61557593068252"
-        }
-    },
-    {
-        "label": "더 압구정 성형외과",
-        "urls": {
-            "homepage": "theclinic.co.kr",
-            "youtube": "youtube.com/user/theplasticsurgery1",
-            "instagram": "instagram.com/the_plasticsurgery",
-            "facebook": "facebook.com/THEPS16445998",
-            "naverPlace": "",
-            "naverBlog": "blog.naver.com/with_theps",
-            "gangnamUnni": "gangnamunni.com/hospitals/30",
-            "tiktok": "",
-            "instagramEn": "instagram.com/the_plasticsurgery.en",
-            "facebookEn": "facebook.com/theps.english"
-        }
-    },
-    {
-        "label": "오라클 성형외과",
-        "urls": {
-            "homepage": "oracleclinic.com",
-            "youtube": "youtube.com/@oracle_medical_group",
-            "instagram": "instagram.com/oraclemedicalgroup",
-            "facebook": "facebook.com/oracleclinickr",
-            "naverPlace": "https://naver.me/GhbU3VtK",
-            "naverBlog": "",
-            "gangnamUnni": "gangnamunni.com/hospitals/125",
-            "tiktok": "tiktok.com/@oracleclinic_usa",
-            "instagramEn": "instagram.com/oracleclinic_global",
-            "facebookEn": "facebook.com/oracleclinicglobal"
-        }
-    }
-]

app/models/analysis.py

@@ -8,9 +8,6 @@ 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
 
 
 class AnalysisOptions(BaseModel):

app/services/analysis.py

@@ -39,12 +39,6 @@ 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")),
-        "branding":               _json(clinic.get("branding")),
-        "brand_assets":           _json(clinic.get("brandAssets")),
-        "tiktok":                 _json(clinic.get("tiktok")),
-        "instagram_en":           _json(clinic.get("instagramEn")),
-        "facebook_en":            _json(clinic.get("facebookEn")),
-        "channel_logos":          _json(clinic.get("channelLogos")),
         **{
             channel: _json(data)
             for channel, data in raw.items()
@@ -84,110 +78,11 @@ 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(clinic.get("tiktok")),
-        "instagram_en":           _json(clinic.get("instagramEn")),
-        "facebook_en":            _json(clinic.get("facebookEn")),
-        "channel_logos":          _json(clinic.get("channelLogos")),
-        "brand_assets":           _json(clinic.get("brandAssets")),
     }
 
     return await LLMService(provider="perplexity").generate(plan_prompt, input_data)
 
 
-def _en_instagram_account(d: dict) -> dict:
-    """영문 인스타 raw_data → InstagramAccount dict (factual 값 + 빈 정성필드). audit 보강용."""
-    return {
-        "handle": d["username"], "language": "EN", "label": "인스타그램 EN",
-        "posts": d.get("posts") or 0, "followers": d.get("followers") or 0,
-        "following": d.get("following") or 0, "category": "",
-        "profile_link": f"https://www.instagram.com/{d['username']}/",
-        "highlights": [], "reels_count": 0, "content_format": "", "profile_photo": "",
-        "bio": d.get("bio") or "",
-    }
-
-
-def _en_facebook_page(d: dict) -> dict:
-    """영문 페북 raw_data → FacebookPage dict (factual 값 + 빈 정성필드). audit 보강용."""
-    url = d.get("pageUrl") or ""
-    return {
-        "url": url, "page_name": d.get("pageName") or "", "language": "EN", "label": "페이스북 EN",
-        "followers": d.get("followers") or 0, "following": 0,
-        "category": ", ".join(d.get("categories") or []), "bio": d.get("intro") or "",
-        "logo": "", "logo_description": "", "link": url, "linked_domain": d.get("website") or "",
-        "reviews": 0, "recent_post_age": "", "has_whatsapp": False,
-    }
-
-
-def _clinic_snapshot(brand_assets: dict, g: dict) -> dict:
-    """brandAssets(색·로고) + 강남언니(평점/리뷰/대표의) → clinic_snapshot 정확값."""
-    snap: dict = {}
-    if brand_assets.get("brand_colors"): snap["brand_colors"] = brand_assets["brand_colors"]
-    if brand_assets.get("logo_images"):  snap["logo_images"]  = brand_assets["logo_images"]
-    if g.get("name"):             snap["name"]           = g["name"]
-    if g.get("rating"):           snap["overall_rating"] = g["rating"]
-    if g.get("totalReviews"):     snap["total_reviews"]  = g["totalReviews"]
-    if g.get("address"):          snap["location"]       = g["address"]
-    if g.get("badges"):           snap["certifications"] = g["badges"]
-    if g.get("totalMajorStaffs"): snap["staff_count"]    = g["totalMajorStaffs"]
-    doctors = g.get("doctors", [])
-    if doctors:
-        lead = max(doctors, key=lambda d: d.get("reviews", 0))
-        snap["lead_doctor"] = {
-            "name": lead.get("name"), "credentials": lead.get("specialty"),
-            "rating": lead.get("rating"), "review_count": lead.get("reviews"),
-        }
-    return snap
-
-
-def _instagram_patch(ig: dict) -> dict:
-    """instagram_data(KR) → instagram_audit.accounts factual 덮어쓰기 값."""
-    p: dict = {}
-    if ig.get("username"):
-        p["handle"] = ig["username"]
-        p["profile_link"] = f"https://www.instagram.com/{ig['username']}/"
-    if ig.get("posts"):     p["posts"] = ig["posts"]
-    if ig.get("followers"): p["followers"] = ig["followers"]
-    if ig.get("following"): p["following"] = ig["following"]
-    if ig.get("bio"):       p["bio"] = ig["bio"]
-    return p
-
-
-def _facebook_patch(fb: dict) -> dict:
-    """facebook_data(KR) → facebook_audit.pages factual 덮어쓰기 값."""
-    p: dict = {}
-    if fb.get("pageUrl"):
-        p["url"] = fb["pageUrl"]
-        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"]
-    return p
-
-
-def _youtube_patch(yt: dict) -> dict:
-    """youtube_data → youtube_audit factual 덮어쓰기 값."""
-    p: dict = {}
-    if yt.get("channelName"): p["channel_name"] = yt["channelName"]
-    if yt.get("handle"):      p["handle"] = yt["handle"]
-    if yt.get("subscribers"): p["subscribers"] = yt["subscribers"]
-    if yt.get("totalVideos"): p["total_videos"] = yt["totalVideos"]
-    if yt.get("totalViews"):  p["total_views"] = yt["totalViews"]
-    if yt.get("publishedAt"): p["channel_created_date"] = yt["publishedAt"][:10]
-    if yt.get("description"): p["channel_description"] = yt["description"]
-    if yt.get("videos"):
-        p["top_videos"] = [
-            {
-                "title": v["title"], "views": v["views"], "duration": v.get("duration"),
-                "type": "Short" if "M" not in v.get("duration", "") else "Long",
-                "uploaded_ago": v.get("date", "")[:10],
-            }
-            for v in yt["videos"]
-        ]
-    return p
-
-
 async def _build_overrides(analysis_run_id: str) -> dict:
     run = await fetchone(
         "SELECT hospital_id, instagram_data_id, facebook_data_id,"
@@ -205,15 +100,68 @@ async def _build_overrides(analysis_run_id: str) -> dict:
     hospital     = json.loads(hospital_row["raw_data"]) if hospital_row and isinstance(hospital_row.get("raw_data"), str) else (hospital_row or {}).get("raw_data") or {}
     instagram    = await fetch_raw("instagram_data",    run["instagram_data_id"])    or {}
     facebook     = await fetch_raw("facebook_data",     run["facebook_data_id"])     or {}
+    naver_blog   = await fetch_raw("naver_blog_data",   run["naver_blog_data_id"])   or {}
     youtube      = await fetch_raw("youtube_data",      run["youtube_data_id"])      or {}
     gangnam_unni = await fetch_raw("gangnam_unni_data", run["gangnam_unni_data_id"]) or {}
 
-    snapshot = _clinic_snapshot(hospital.get("brandAssets") or {}, gangnam_unni)
-    ig_patch = _instagram_patch(instagram)
-    fb_patch = _facebook_patch(facebook)
-    yt_patch = _youtube_patch(youtube)
-    ig_en = hospital.get("instagramEn") or {}
-    fb_en = hospital.get("facebookEn") or {}
+    snapshot: dict = {}
+
+    # ── gangnam_unni ──────────────────────────────────────────────────────────
+    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 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"]
+    if gangnam_unni.get("badges"):       snapshot["certifications"] = gangnam_unni["badges"]
+    if gangnam_unni.get("totalMajorStaffs"): snapshot["staff_count"]    = gangnam_unni["totalMajorStaffs"]
+    if lead:
+        snapshot["lead_doctor"] = {
+            "name":         lead.get("name"),
+            "credentials":  lead.get("specialty"),
+            "rating":       lead.get("rating"),
+            "review_count": lead.get("reviews"),
+        }
+
+    # ── 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_patch: dict = {}
+    if facebook.get("pageUrl"):    fb_patch["url"]           = facebook["pageUrl"]
+    if facebook.get("pageUrl"):    fb_patch["link"]          = facebook["pageUrl"]
+    if facebook.get("pageName"):   fb_patch["page_name"]     = facebook["pageName"]
+    if facebook.get("followers"):  fb_patch["followers"]     = facebook["followers"]
+    if facebook.get("intro"):      fb_patch["bio"]           = facebook["intro"]
+    if facebook.get("categories"): fb_patch["category"]     = ", ".join(facebook["categories"])
+    if facebook.get("website"):    fb_patch["linked_domain"] = facebook["website"]
+
+    # ── youtube ───────────────────────────────────────────────────────────────
+    yt_patch: dict = {}
+    if youtube.get("channelName"):  yt_patch["channel_name"]        = youtube["channelName"]
+    if youtube.get("handle"):       yt_patch["handle"]              = youtube["handle"]
+    if youtube.get("subscribers"):  yt_patch["subscribers"]         = youtube["subscribers"]
+    if youtube.get("totalVideos"):  yt_patch["total_videos"]        = youtube["totalVideos"]
+    if youtube.get("totalViews"):   yt_patch["total_views"]         = youtube["totalViews"]
+    if youtube.get("publishedAt"):  yt_patch["channel_created_date"] = youtube["publishedAt"][:10]
+    if youtube.get("description"):  yt_patch["channel_description"] = youtube["description"]
+    if youtube.get("videos"):
+        yt_patch["top_videos"] = [
+            {
+                "title":        v["title"],
+                "views":        v["views"],
+                "duration":     v.get("duration"),
+                "type":         "Short" if "M" not in v.get("duration", "") else "Long",
+                "uploaded_ago": v.get("date", "")[:10],
+            }
+            for v in youtube["videos"]
+        ]
 
     overrides: dict = {}
     if snapshot:
@@ -224,10 +172,6 @@ async def _build_overrides(analysis_run_id: str) -> dict:
         overrides["facebook_audit"] = {"pages": [fb_patch]}
     if yt_patch:
         overrides["youtube_audit"] = yt_patch
-    if ig_en.get("username"):
-        overrides["_en_ig_account"] = _en_instagram_account(ig_en)
-    if fb_en.get("pageUrl") or fb_en.get("pageName"):
-        overrides["_en_fb_page"] = _en_facebook_page(fb_en)
     return overrides
 
 
@@ -243,22 +187,8 @@ def _deep_merge(base: dict, overrides: dict) -> dict:
             base[k] = v
     return base
 
-def _ensure_en_entry(audit: dict, list_key: str, en_entry: dict | None) -> None:
-    """audit 리스트(accounts/pages)에 EN 항목이 없으면 추가 — LLM 누락 대비, 중복 방지."""
-    if not en_entry:
-        return
-    items = audit.setdefault(list_key, [])
-    if not any(it.get("language") == "EN" for it in items):
-        items.append(en_entry)
-
-
 def _patch_report(result: ReportOutput, overrides: dict) -> ReportOutput:
-    en_ig = overrides.pop("_en_ig_account", None)
-    en_fb = overrides.pop("_en_fb_page", None)
     merged = _deep_merge(result.model_dump(), overrides)
-    # LLM이 audit에 영문 계정을 빠뜨려도 항상 KR+EN 둘 다 보장.
-    _ensure_en_entry(merged.setdefault("instagram_audit", {}), "accounts", en_ig)
-    _ensure_en_entry(merged.setdefault("facebook_audit", {}), "pages", en_fb)
     return ReportOutput(**merged)
 
 

app/services/collect.py

@@ -1,7 +1,7 @@
 import asyncio
 import logging
+from common.db import fetchone
 from common.db import (
-    fetchone,
     set_instagram_status, save_instagram_raw_data,
     set_facebook_status, save_facebook_raw_data,
     set_naver_blog_status, save_naver_blog_raw_data,
@@ -9,12 +9,11 @@ from common.db import (
     set_gangnam_unni_status, save_gangnam_unni_raw_data,
     execute, save_hospital_raw_data,
 )
-from common.utils import get_env, _run_optional_step
+from common.utils import get_env
 from integrations.apify import ApifyClient
 from integrations.naver import NaverClient
 from integrations.youtube import YouTubeClient
 from integrations.firecrawl import FirecrawlClient
-from services.enrichment import collect_brand_assets, collect_extra_channels, collect_channel_logos
 
 logger = logging.getLogger(__name__)
 
@@ -75,9 +74,6 @@ async def collect_all(
     naver_blog_id: int | None = None,
     youtube_id: int | None = None,
     gangnam_unni_id: int | None = None,
-    tiktok_url: str | None = None,
-    instagram_en_url: str | None = None,
-    facebook_en_url: str | None = None,
 ) -> None:
     async def _url(table: str, row_id: int) -> str:
         row = await fetchone(f"SELECT url FROM {table} WHERE id = %s", (row_id,))
@@ -98,18 +94,3 @@ async def collect_all(
         tasks.append(collect_gangnam_unni(analysis_run_id, gangnam_unni_id, await _url("gangnam_unni_data", gangnam_unni_id)))
 
     await asyncio.gather(*tasks, return_exceptions=True)
-
-    # 아래 3단계는 모두 hospital raw_data를 read-modify-write 하므로 race 방지 위해 순차.
-    #   brand_assets  : clinic_info가 채운 branding.logoUrl로 공식 로고/hex 추출
-    #   extra_channels: 틱톡/인스타EN/페북EN 수집
-    #   channel_logos : 공식 로고(brand_assets)+채널 profileImage(extra_channels) 채워진 뒤 Vision 비교
-    # 부가 기능이라 실패해도 리포트는 나와야 하므로 _run_optional_step으로 각각 격리.
-    await _run_optional_step(collect_brand_assets(analysis_run_id, hospital_id), "brand_assets")
-    await _run_optional_step(
-        collect_extra_channels(
-            analysis_run_id, hospital_id,
-            tiktok_url=tiktok_url, instagram_en_url=instagram_en_url, facebook_en_url=facebook_en_url,
-        ),
-        "extra_channels",
-    )
-    await _run_optional_step(collect_channel_logos(analysis_run_id, hospital_id), "channel_logos")

app/services/enrichment.py

@@ -1,175 +0,0 @@
-import asyncio
-import json
-import logging
-import os
-from urllib.parse import urlparse
-from common.db import fetchone, fetch_raw, merge_hospital_raw_data
-from common.utils import get_env
-from integrations.apify import ApifyClient
-from integrations.vision import VisionClient
-from integrations.color_extractor import extract_brand_assets_from_site
-
-logger = logging.getLogger(__name__)
-
-
-async def collect_brand_assets(analysis_run_id: str, hospital_id: str) -> None:
-    """홈페이지에서 로고 URL + brand hex 색상을 뽑아 raw_data["brandAssets"]에 저장.
-    - 로고 URL/hex: HTML·CSS 정규식 (color_extractor) — Vision 의존 X, 사이트 전체 컬러 시스템이 더 정확.
-    - 로고 정성 묘사(심볼/워드마크/톤): Gemini Vision (GEMINI_API_KEY 없으면 색상만 저장하고 skip).
-    """
-    logger.info("[brand_assets] start run=%s", analysis_run_id)
-    row = await fetchone(
-        "SELECT raw_data, url FROM hospital_baseinfo WHERE hospital_id = %s",
-        (hospital_id,),
-    )
-    if not row:
-        return
-    raw = row["raw_data"]
-    raw_data = json.loads(raw) if isinstance(raw, str) else (raw or {})
-    branding = raw_data.get("branding") or {}
-    homepage_url = row["url"]
-
-    # 0~1. 사이트 1회 fetch로 logo URL + brand hex 동시 추출 (img/background-image/CSS .logo, Vision 의존 X)
-    site = await extract_brand_assets_from_site(homepage_url) if homepage_url else {}
-    html_logo_url = site.get("logo_url")
-    css_colors = site.get("colors") or {}
-    if html_logo_url:
-        logger.info("[brand_assets] HTML logo found: %s", html_logo_url)
-    if css_colors:
-        logger.info("[brand_assets] css colors: %s", css_colors.get("brand_colors"))
-
-    # 2. 로고/대표 이미지 후보 (logo → og:image → favicon 순)
-    logo_url = html_logo_url or branding.get("logoUrl")
-    og_image = branding.get("ogImage")
-    favicon = branding.get("faviconUrl")
-    candidates: list[tuple[str, str]] = []
-    if logo_url: candidates.append(("logo", logo_url))
-    if og_image: candidates.append(("og", og_image))
-    if favicon:  candidates.append(("favicon", favicon))
-    if homepage_url:
-        parsed = urlparse(homepage_url)
-        if parsed.scheme and parsed.netloc:
-            candidates.append(("favicon", f"{parsed.scheme}://{parsed.netloc}/favicon.ico"))
-
-    if not candidates and not css_colors:
-        logger.info("[brand_assets] skip — no logo/og/favicon candidates and no CSS colors")
-        return
-
-    # 3. Vision은 로고 정성 묘사만 (hex는 CSS 추출이 더 정확). 키 없으면 색상만 저장.
-    result: dict = {}
-    used_kind: str | None = None
-    api_key = os.getenv("GEMINI_API_KEY")
-    if api_key and candidates:
-        vc = VisionClient(api_key)
-        for kind, cand in candidates:
-            result = await vc.analyze_brand_assets(logo_url=cand, homepage_url=homepage_url)
-            if result:
-                used_kind = kind
-                break
-        # favicon으로만 분석된 경우 진짜 로고가 아니므로 logo URL은 박지 않음 (묘사는 OK)
-        if result and used_kind == "favicon" and result.get("logo_images"):
-            result["logo_images"] = {"circle": None, "horizontal": None, "korean": None}
-    elif not api_key:
-        logger.info("[brand_assets] GEMINI_API_KEY not set — 색상만 저장, Vision 묘사 skip")
-
-    # 4. CSS에서 추출한 brand_colors/palette를 Vision보다 우선 사용
-    if css_colors:
-        if css_colors.get("brand_colors"):
-            result["brand_colors"] = css_colors["brand_colors"]
-        if css_colors.get("color_palette"):
-            result["color_palette"] = css_colors["color_palette"]
-        result["color_source"] = "html+css"
-    elif result:
-        result["color_source"] = "vision"
-
-    if result:
-        result["logo_source"] = used_kind or "none"
-        await merge_hospital_raw_data(hospital_id, {"brandAssets": result})
-    logger.info("[brand_assets] done keys=%s", list(result.keys()) if result else None)
-
-
-async def collect_extra_channels(
-    analysis_run_id: str,
-    hospital_id: str,
-    tiktok_url: str | None = None,
-    instagram_en_url: str | None = None,
-    facebook_en_url: str | None = None,
-) -> None:
-    """틱톡 / 인스타 EN / 페북 EN 수집 → hospital raw_data에 저장 (별도 테이블 없이).
-    인스타EN·페북EN은 기존 Apify 수집기 재사용, 틱톡은 신규 액터."""
-    apify = ApifyClient(get_env("APIFY_API_TOKEN"))
-    jobs: dict = {}
-    if instagram_en_url:
-        jobs["instagramEn"] = apify.get_instagram_profile(instagram_en_url)
-    if facebook_en_url:
-        jobs["facebookEn"] = apify.get_facebook_page(facebook_en_url)
-    if tiktok_url:
-        jobs["tiktok"] = apify.get_tiktok_profile(tiktok_url)
-    if not jobs:
-        return
-
-    logger.info("[extra_channels] start run=%s channels=%s", analysis_run_id, list(jobs))
-    done = await asyncio.gather(*jobs.values(), return_exceptions=True)
-    results: dict = {}
-    for key, res in zip(jobs.keys(), done):
-        if isinstance(res, Exception):
-            logger.warning("[extra_channels] %s 수집 실패: %s", key, res)
-        elif res:
-            results[key] = res
-    if not results:
-        logger.info("[extra_channels] 수집 결과 없음 run=%s", analysis_run_id)
-        return
-
-    await merge_hospital_raw_data(hospital_id, results)
-    logger.info("[extra_channels] done run=%s keys=%s", analysis_run_id, list(results))
-
-
-async def collect_channel_logos(analysis_run_id: str, hospital_id: str) -> None:
-    """채널별 프로필 이미지(로고)를 모아 Gemini Vision으로 설명 + 공식 로고 일치 여부 평가.
-    → hospital raw_data["channelLogos"]에 저장. GEMINI_API_KEY 없으면 skip.
-    brand_assets(공식 로고)·extra_channels(틱톡/EN profileImage) 다음에 실행돼야 함."""
-    api_key = os.getenv("GEMINI_API_KEY")
-    if not api_key:
-        logger.info("[channel_logos] skip — GEMINI_API_KEY 없음")
-        return
-
-    hrow = await fetchone("SELECT raw_data FROM hospital_baseinfo WHERE hospital_id = %s", (hospital_id,))
-    raw = hrow["raw_data"] if hrow else None
-    raw_data = json.loads(raw) if isinstance(raw, str) else (raw or {})
-    official = ((raw_data.get("brandAssets") or {}).get("logo_images") or {}).get("horizontal")
-
-    run = await fetchone(
-        "SELECT instagram_data_id, facebook_data_id, youtube_data_id"
-        " FROM analysis_runs WHERE analysis_run_id = %s",
-        (analysis_run_id,),
-    )
-    logos: list[dict] = []
-    # 전용 테이블 채널 (KR)
-    for ch, table, col in [
-        ("Instagram", "instagram_data", "instagram_data_id"),
-        ("Facebook", "facebook_data", "facebook_data_id"),
-        ("YouTube", "youtube_data", "youtube_data_id"),
-    ]:
-        rid = (run or {}).get(col)
-        if rid:
-            d = await fetch_raw(table, rid) or {}
-            if d.get("profileImage"):
-                logos.append({"channel": ch, "url": d["profileImage"]})
-    # 추가 채널 (hospital raw_data)
-    for ch, key in [("Instagram EN", "instagramEn"), ("Facebook EN", "facebookEn"), ("TikTok", "tiktok")]:
-        img = (raw_data.get(key) or {}).get("profileImage")
-        if img:
-            logos.append({"channel": ch, "url": img})
-
-    if not logos:
-        logger.info("[channel_logos] skip — 채널 프로필 이미지 없음")
-        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 VisionClient(api_key).describe_channel_logos(official, logos)
-    if result:
-        # Vision이 못 본 채널도 url은 채워둠 (프론트에서 이미지 표시용)
-        result["logos"] = logos
-        await merge_hospital_raw_data(hospital_id, {"channelLogos": result})
-    logger.info("[channel_logos] done run=%s keys=%s", analysis_run_id, list(result.keys()) if result else None)

app/services/pipeline.py

@@ -8,9 +8,8 @@ from services.analysis import run_report_task, run_plan_task
 logger = logging.getLogger(__name__)
 
 
-async def run_pipeline(analysis_run_id: str, extra_channels: dict | None = None) -> None:
+async def run_pipeline(analysis_run_id: str) -> None:
     logger.info("[pipeline] start run=%s", analysis_run_id)
-    extra_channels = extra_channels or {}
 
     # ── 1. Collect ──────────────────────────────────────────────────────────
     run = await fetchone(
@@ -27,9 +26,6 @@ async def run_pipeline(analysis_run_id: str, extra_channels: dict | None = None)
         naver_blog_id=run["naver_blog_data_id"],
         youtube_id=run["youtube_data_id"],
         gangnam_unni_id=run["gangnam_unni_data_id"],
-        tiktok_url=extra_channels.get("tiktok"),
-        instagram_en_url=extra_channels.get("instagram_en"),
-        facebook_en_url=extra_channels.get("facebook_en"),
     )
 
     # ── 2. Market ────────────────────────────────────────────────────────────