website audit, other channels 분리 추가

app/integrations/firecrawl.py

@@ -74,7 +74,7 @@ class FirecrawlClient:
             headers=self._headers(),
             json_body={
                 "url": url,
-                "formats": ["json", "links"],
+                "formats": ["json", "links", "html"],
                 "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)",
                     "schema": {
@@ -147,6 +147,7 @@ class FirecrawlClient:
             # "socialMedia": info.get("socialMedia", {}),
             "branding":      info.get("branding", {}),
             "siteLinks":     data.get("links", []),
+            "html":          data.get("html", ""),  # raw HTML — collect 단계에서 tracking 추출 후 raw_data 에는 저장 안 함.
             "sourceUrl":     url,
         }
 

app/integrations/llm/prompt.py

@@ -1,7 +1,7 @@
 import os
 from pydantic import BaseModel
 from common.utils import get_env
-from integrations.llm.schemas.report import ReportInput, ReportOutput, YouTubeDiagnosisInput, YouTubeDiagnosisOutput
+from integrations.llm.schemas.report import ReportInput, ReportOutput, YouTubeDiagnosisInput, YouTubeDiagnosisOutput, OtherChannelsInput, OtherChannelsOutput
 from integrations.llm.schemas.plan import PlanInput, PlanOutput
 from integrations.llm.schemas.market import (
     MarketCompetitorsInput, MarketCompetitorsOutput,
@@ -87,3 +87,10 @@ youtube_diagnosis_prompt = Prompt(
     input_class=YouTubeDiagnosisInput,
     output_class=YouTubeDiagnosisOutput,
 )
+
+other_channels_prompt = Prompt(
+    file_name="other_channels_prompt.txt",
+    prompt_model="REPORT_MODEL",
+    input_class=OtherChannelsInput,
+    output_class=OtherChannelsOutput,
+)

app/integrations/llm/schemas/report.py

@@ -215,6 +215,19 @@ class OtherChannel(BaseModel):
     url: str | None = None
 
 
+class OtherChannelsInput(BaseModel):
+    clinic_name:  str
+    tiktok:       str | None = None
+    kakao_talk:   str | None = None
+    naver_cafe:   str | None = None
+    naver_blog:   str | None = None
+    gangnam_unni: str | None = None
+
+
+class OtherChannelsOutput(BaseModel):
+    other_channels: list[OtherChannel]
+
+
 class TrackingPixel(BaseModel):
     name: str
     installed: bool

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

@@ -0,0 +1,45 @@
+당신은 의료 마케팅 분석가입니다. 아래 부가 채널 데이터를 보고 `other_channels` 리스트만 JSON 으로 생성하세요.
+결과물은 한국어로 작성하세요.
+
+## 병원
+- 병원명: {clinic_name}
+
+## 부가 채널 raw 데이터
+
+### TikTok
+{tiktok}
+
+### KakaoTalk
+{kakao_talk}
+
+### Naver Cafe
+{naver_cafe}
+
+### Naver Blog
+{naver_blog}
+
+### Gangnam Unni
+{gangnam_unni}
+
+## 작성 지침
+
+- 메인 audit(YouTube/Instagram KR/Facebook KR/Website)에 **포함되지 않은** 채널만 넣으세요.
+- 위 채널 데이터에 **실제 값이 있는 채널만** status=active 와 실제 URL 로 일관되게 포함:
+- **카카오톡·네이버 카페**: {kakao_talk} 또는 {naver_cafe} 에 url 이 있으면 each "KakaoTalk" / "Naver Cafe" 로 status=active + 해당 url 로 포함. 수집된 콘텐츠 데이터는 없으므로 URL 존재 자체가 활성 채널 신호. **둘 다 null/빈 값이면 절대 만들지 마세요.**
+- **그 외 데이터 없는 채널(네이버플레이스/Threads 등)은 절대 임의로 만들지 마세요.** 데이터 없으면 그 채널은 생략 (랜덤 생성·추측 금지).
+- url 은 위 raw 데이터의 **실제 URL 만** 사용. 없으면 빈 문자열.
+- **URL 에 'https://www.facebook.com/' 같은 prefix 를 절대 직접 만들지 마세요.** 받은 데이터 URL = 출력 URL. 이미 'https://...' 가 붙은 URL 에 또 prefix 붙이면 깨집니다.
+- `details` 는 해당 채널의 짧은 1줄 묘사 (예: "팔로워 1.2만, 주 2회 게시", "오픈채팅 상담", "환자 후기 게시판"). 데이터 없으면 빈 문자열.
+
+## 출력 형식
+
+```json
+{{
+  "other_channels": [
+    {{"name": "Naver Blog", "status": "active", "details": "...", "url": "https://..."}},
+    ...
+  ]
+}}
+```
+
+`status` 는 active / inactive / unknown / not_found 중 하나. 데이터·URL 모두 없으면 그 채널 자체를 생략하세요 (null 항목 만들지 말 것).

app/services/collect.py

@@ -1,5 +1,6 @@
 import asyncio
 import logging
+from urllib.parse import urlparse
 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, _run_optional_step
@@ -10,6 +11,7 @@ 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 services.website_parser import extract_tracking_pixels, extract_sns_links, extract_additional_domains, extract_main_cta
 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
@@ -84,6 +86,22 @@ async def collect_gangnam_unni(analysis_run_id: str, info_id: int, url: str) ->
     logger.info("[gangnam_unni] done run=%s", analysis_run_id)
 
 
+def _extract_website_audit(html: str, site_links: list[str], url: str) -> dict:
+    """raw HTML + siteLinks 에서 main CTA / tracking pixels / SNS / additional domains 정규식 추출."""
+    if not html:
+        return {}
+    primary_host = urlparse(url).netloc
+    result: dict = {
+        "trackingPixels":    extract_tracking_pixels(html),
+        "snsLinks":          extract_sns_links(site_links, html),
+        "additionalDomains": extract_additional_domains(site_links, primary_host),
+    }
+    html_cta = extract_main_cta(html)
+    if html_cta:
+        result["mainCta"] = html_cta
+    return result
+
+
 async def collect_mainpage(analysis_run_id: str, info_id: int, hospital_id: str, url: str) -> None:
     logger.info("[mainpage] start run=%s url=%s", analysis_run_id, url)
     await update_raw_info_status(info_id, "processing")
@@ -93,8 +111,12 @@ 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
+
+    html = data.pop("html", "") or ""  # raw_data 에는 저장 안 함 — 여기서만 사용하고 버림.
     # 홈페이지 URL 자체도 raw_data 에 박아둬야 brand_assets / 분석 단계에서 mainpage URL 재조회 없이 사용 가능.
     data = {**data, "sourceUrl": url}
+    data.update(_extract_website_audit(html, data.get("siteLinks", []), 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)

app/services/website_parser.py

@@ -0,0 +1,178 @@
+"""collect 단계 - HTML / siteLinks 에서 tracking pixels / SNS links / additional domains 추출.
+모두 정규식·도메인 매칭 기반 deterministic 추출 (LLM 미경유)."""
+import re
+from urllib.parse import urlparse
+
+
+# ── tracking pixels ──────────────────────────────────────────────────────────
+
+# 픽셀별 시그니처: 이름 → 정규식 리스트. installed 판정은 OR (하나라도 매치되면 설치된 것).
+# ID 캡처는 모든 패턴 시도 후 첫 non-empty 그룹 사용 (signature-only 패턴과 ID-capture 패턴 혼재 OK).
+_TRACKING_PIXEL_PATTERNS: dict[str, list[re.Pattern]] = {
+    "Google Analytics": [
+        re.compile(r"googletagmanager\.com/gtag/js\?id=(G-[A-Z0-9]+)", re.IGNORECASE),
+        re.compile(r"google-analytics\.com/analytics\.js", re.IGNORECASE),
+        re.compile(r"\bgtag\(\s*['\"]config['\"]\s*,\s*['\"](G-[A-Z0-9]+|UA-\d+-\d+)['\"]", re.IGNORECASE),
+        re.compile(r"\b(UA-\d+-\d+)\b"),
+    ],
+    "Google Tag Manager": [
+        re.compile(r"googletagmanager\.com/gtm\.js\?id=(GTM-[A-Z0-9]+)", re.IGNORECASE),
+        re.compile(r"\b(GTM-[A-Z0-9]+)\b"),
+    ],
+    "Facebook Pixel": [
+        re.compile(r"connect\.facebook\.net/[^/]+/fbevents\.js", re.IGNORECASE),
+        re.compile(r"fbq\(\s*['\"]init['\"]\s*,\s*['\"](\d{10,20})['\"]", re.IGNORECASE),
+        re.compile(r"facebook\.com/tr/?\?id=(\d{10,20})", re.IGNORECASE),  # noscript image pixel fallback
+        re.compile(r"_fbq\.push\(\s*\[\s*['\"]init['\"]\s*,\s*['\"](\d{10,20})['\"]", re.IGNORECASE),  # legacy
+    ],
+    "Naver Analytics": [
+        re.compile(r"wcs\.naver\.net/wcslog\.js", re.IGNORECASE),
+        re.compile(r"\bwcs_add\b", re.IGNORECASE),
+        re.compile(r"\bwcs\.inflow\(", re.IGNORECASE),
+        re.compile(r"wcs_add\s*\[\s*['\"]wa['\"]\s*\]\s*=\s*['\"]([a-zA-Z0-9_]+)['\"]", re.IGNORECASE),  # wa ID
+    ],
+    "Kakao Pixel": [
+        re.compile(r"t1\.daumcdn\.net/kas/static/kp\.js", re.IGNORECASE),
+        re.compile(r"kakaoPixel\s*\(\s*['\"]?(\d+)['\"]?", re.IGNORECASE),
+    ],
+}
+
+
+def extract_tracking_pixels(html: str) -> list[dict]:
+    """HTML 에서 트래킹 픽셀 설치 여부 + ID 추출. 환각 0 — 정규식 매치만 신뢰.
+    모든 패턴 시도 → 하나라도 매치되면 installed. ID 는 캡처된 첫 non-empty 그룹 사용."""
+    if not html:
+        return []
+    pixels: list[dict] = []
+    for name, patterns in _TRACKING_PIXEL_PATTERNS.items():
+        any_match = False
+        captured_id: str | None = None
+        for pat in patterns:
+            m = pat.search(html)
+            if not m:
+                continue
+            any_match = True
+            if m.groups() and m.group(1) and not captured_id:
+                captured_id = m.group(1)
+        if any_match:
+            pixels.append({
+                "name":      name,
+                "installed": True,
+                "details":   captured_id,
+            })
+    return pixels
+
+
+# ── main CTA ─────────────────────────────────────────────────────────────────
+
+_CTA_KEYWORDS = re.compile(
+    r"(상담|예약|문의|신청|등록|진료|Book\s*Now|Consult|Reservation|Contact|Apply)",
+    re.IGNORECASE,
+)
+_BUTTON_TAG = re.compile(r"<button\b[^>]*>(.*?)</button>", re.IGNORECASE | re.DOTALL)
+_ANCHOR_TAG = re.compile(r"<a\b[^>]*>(.*?)</a>", re.IGNORECASE | re.DOTALL)
+
+
+def _clean_text(inner: str) -> str:
+    text = re.sub(r"<[^>]+>", "", inner)
+    return re.sub(r"\s+", " ", text).strip()
+
+
+def extract_main_cta(html: str) -> str:
+    """HTML 에서 primary CTA 텍스트 추출. 우선순위: <button> 첫 매치 → <a> 첫 매치.
+    CTA 키워드 매칭 + 길이 1~20자 (버튼 라벨 추정)."""
+    if not html:
+        return ""
+    for pat in (_BUTTON_TAG, _ANCHOR_TAG):
+        for m in pat.finditer(html):
+            text = _clean_text(m.group(1))
+            if text and 1 <= len(text) <= 20 and _CTA_KEYWORDS.search(text):
+                return text
+    return ""
+
+
+# ── SNS links ────────────────────────────────────────────────────────────────
+
+# SNS 도메인 → 표준 platform 이름. siteLinks 필터링 + DOM 위치 판정에 공유.
+_SNS_DOMAINS: dict[str, str] = {
+    "facebook.com":   "Facebook",
+    "instagram.com":  "Instagram",
+    "youtube.com":    "YouTube",
+    "youtu.be":       "YouTube",
+    "tiktok.com":     "TikTok",
+    "pf.kakao.com":   "KakaoTalk",
+    "open.kakao.com": "KakaoTalk",
+    "blog.naver.com": "Naver Blog",
+    "cafe.naver.com": "Naver Cafe",
+    "x.com":          "X",
+    "twitter.com":    "X",
+}
+
+_FOOTER_BLOCK = re.compile(r"<footer\b[^>]*>(.*?)</footer>", re.IGNORECASE | re.DOTALL)
+_HEADER_BLOCK = re.compile(r"<header\b[^>]*>(.*?)</header>", re.IGNORECASE | re.DOTALL)
+
+
+def _platform_for(url: str) -> str | None:
+    try:
+        host = urlparse(url).netloc.lower().lstrip("www.")
+    except Exception:
+        return None
+    for domain, name in _SNS_DOMAINS.items():
+        if host == domain or host.endswith("." + domain):
+            return name
+    return None
+
+
+def _location_for(url: str, html: str) -> str:
+    """url 이 HTML 의 <header>/<footer> 안에 들어있는지로 위치 판정. 둘 다 아니면 'body'."""
+    if not html:
+        return ""
+    needle = re.escape(url)
+    footer = _FOOTER_BLOCK.search(html)
+    if footer and re.search(needle, footer.group(1)):
+        return "footer"
+    header = _HEADER_BLOCK.search(html)
+    if header and re.search(needle, header.group(1)):
+        return "header"
+    return "body"
+
+
+def extract_sns_links(site_links: list[str], html: str = "") -> list[dict]:
+    """siteLinks 에서 SNS 도메인 매칭. location 은 HTML 의 footer/header tag 블록으로 판정.
+    중복 platform 은 첫 URL 만 유지."""
+    seen: dict[str, dict] = {}
+    for url in site_links or []:
+        platform = _platform_for(url)
+        if not platform or platform in seen:
+            continue
+        seen[platform] = {
+            "platform": platform,
+            "url":      url,
+            "location": _location_for(url, html),
+        }
+    return list(seen.values())
+
+
+# ── additional domains ───────────────────────────────────────────────────────
+
+def extract_additional_domains(site_links: list[str], primary_host: str) -> list[dict]:
+    """siteLinks 의 host 중 primary 도메인이 아닌 외부 도메인 모으기. SNS·맵 등 utility 도메인 제외.
+    purpose 분류는 LLM 필요 — 여기선 빈 문자열."""
+    skip_domains = set(_SNS_DOMAINS) | {
+        "map.kakao.com", "map.naver.com", "maps.google.com", "goo.gl",
+    }
+    primary = (primary_host or "").lower().lstrip("www.")
+    seen: dict[str, dict] = {}
+    for url in site_links or []:
+        if not url or not url.startswith("http"):
+            continue
+        try:
+            host = urlparse(url).netloc.lower().lstrip("www.")
+        except Exception:
+            continue
+        if not host or host == primary or host in seen:
+            continue
+        if any(host == d or host.endswith("." + d) for d in skip_domains):
+            continue
+        seen[host] = {"domain": host, "purpose": ""}
+    return list(seen.values())