Compare commits

..

No commits in common. "56fa2c6238f8ad526e5c99f52dd432b5d370fe93" and "652265cd19cbc7c8c6d79206b3ab3d502388bfb5" have entirely different histories.

17 changed files with 96 additions and 1520 deletions

View File

@ -27,8 +27,6 @@ def _extra_channels_from_mockurls(homepage_url: str) -> dict:
"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 {}
@ -47,12 +45,11 @@ async def start_analysis(body: AnalysisCreate, background_tasks: BackgroundTasks
if not hospital:
raise HTTPException(status_code=409, detail="Clinic not found")
# 사용자가 'gangnamunni.com/...' 같이 scheme/www 없이 줘도 _with_scheme이 https://www. 보강.
ig_id = await insert_instagram_row(hospital_id, _with_scheme(body.channels.instagram)) if body.channels.instagram else None
fb_id = await insert_facebook_row(hospital_id, _with_scheme(body.channels.facebook)) if body.channels.facebook else None
nb_id = await insert_naver_blog_row(hospital_id, _with_scheme(body.channels.naver_blog)) if body.channels.naver_blog else None
yt_id = await insert_youtube_row(hospital_id, _with_scheme(body.channels.youtube)) if body.channels.youtube else None
gu_id = await insert_gangnam_unni_row(hospital_id, _with_scheme(body.channels.gangnam_unni)) if body.channels.gangnam_unni else None
ig_id = await insert_instagram_row(hospital_id, body.channels.instagram) if body.channels.instagram else None
fb_id = await insert_facebook_row(hospital_id, body.channels.facebook) if body.channels.facebook else None
nb_id = await insert_naver_blog_row(hospital_id, body.channels.naver_blog) if body.channels.naver_blog else None
yt_id = await insert_youtube_row(hospital_id, body.channels.youtube) if body.channels.youtube else None
gu_id = await insert_gangnam_unni_row(hospital_id, body.channels.gangnam_unni) if body.channels.gangnam_unni else None
analysis_run_id = await insert_analysis_run(
analysis_run_id, hospital_id, hospital["owner_user_id"],
@ -65,8 +62,6 @@ async def start_analysis(body: AnalysisCreate, background_tasks: BackgroundTasks
"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"),
"kakao_talk": body.channels.kakao_talk or mock_extra.get("kakao_talk"),
"naver_cafe": body.channels.naver_cafe or mock_extra.get("naver_cafe"),
}
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)

View File

@ -61,27 +61,6 @@ def _normalize_homepage(url: str) -> str:
return u.rstrip("/")
# SSL 인증서가 www.* 에만 유효한 도메인 — bare 도메인이면 사용자 클릭 시 브라우저 SSL warning 뜸.
_WWW_REQUIRED = ("gangnamunni.com", "facebook.com", "instagram.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
"""scheme 없는 URL에 https:// 보정 (수집기 파싱용). 빈 값은 None."""
return (u if "://" in u else "https://" + u) if u else None

View File

@ -13,9 +13,6 @@ IG_HIGHLIGHTS_ACTOR = "igview-owner~instagram-highlights-scraper"
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("@")
@ -68,13 +65,6 @@ class ApifyClient:
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"),
@ -175,7 +165,7 @@ class ApifyClient:
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, {
return await self._run_actor("clockworks~tiktok-scraper", {
"profiles": [user],
"resultsPerPage": 10,
"profileScrapeSections": ["videos"],

View File

@ -84,47 +84,22 @@ LOGO_CSS_PATTERN = re.compile(
def find_logo_url_in_html(html: str, base_url: str, css_texts: list[str] | None = None) -> str | None:
"""HTML에서 logo URL 찾기. 우선순위:
1) 패턴 1~8 (class/id/alt/src에 'logo' 명시된 img 가장 specific)
2) 외부 CSS의 .logo background-image (class-based, specific)
3) 패턴 9~10 (<header>/<nav> img 가장 generic, 잘못 잡힐 위험 )
"""
def _is_noise(src: str) -> bool:
"""logo로 잘못 잡힐 가능성 높은 URL 패턴 — lang/flag/icon/arrow/spacer 등."""
if not src or src.startswith("data:"):
return True
if re.search(r"(blank|spacer|pixel|transparent|1x1)\b", src, re.IGNORECASE):
return True
# 헤더 첫 img가 lang flag / 검색 아이콘 / 네비 화살표인 경우 (JK plastic 한국어 깃발이 잡히던 케이스)
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
# 1) class/id/alt/src/inline-bg/src-with-logo 패턴 (1~8)
for pat in LOGO_IMG_PATTERNS[:8]:
"""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 _is_noise(src):
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)
# 2) 외부 CSS의 .logo { background-image } — class-based 이므로 generic 패턴보다 우선
# 외부 CSS에서 .logo background-image 추출
for css in (css_texts or []):
m = LOGO_CSS_PATTERN.search(css)
if m:
src = m.group(1)
if not _is_noise(src):
if src and not src.startswith("data:"):
return urljoin(base_url, src)
# 3) header/nav 첫 img — 가장 generic, lang flag 등 noise 필터 강화 적용
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

View File

@ -18,7 +18,6 @@ class PlanInput(BaseModel):
tiktok: str | None = None
instagram_en: str | None = None
facebook_en: str | None = None
naver_blog: str | None = None
channel_logos: str | None = None
brand_assets: str | None = None
@ -57,7 +56,7 @@ class ChannelBrandingRule(BaseModel):
profile_photo: str
banner_spec: str
bio_template: str
current_status: Literal["correct", "incorrect", "N/A"]
current_status: Literal["correct", "incorrect", "missing"]
class BrandPlanInconsistencyValue(BaseModel):

View File

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

View File

@ -32,11 +32,8 @@
## 분석 리포트
{report}
## 추가 채널 데이터 (네이버 블로그 / 틱톡 / 인스타그램 EN / 페이스북 EN)
아래에 데이터가 있는 채널은 channelStrategies와 channelBranding에 **반드시 포함**하세요 (네이버 블로그, 틱톡, 영문 인스타그램, 영문 페이스북). null이면 제외.
### 네이버 블로그 (Naver Blog)
{naver_blog}
## 추가 채널 데이터 (틱톡 / 인스타그램 EN / 페이스북 EN)
아래에 데이터가 있는 채널은 channelStrategies와 channelBranding에 **반드시 포함**하세요 (틱톡, 영문 인스타그램, 영문 페이스북). null이면 제외.
### 틱톡 (TikTok)
{tiktok}
@ -50,12 +47,7 @@
## 채널별 로고 분석 (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에서 채널별로 다른 묘사를 박지 마세요.
- **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에서 추출 — 결정적 데이터)
@ -77,10 +69,10 @@
- brandInconsistencies: 채널 간 브랜딩 불일치 항목 및 개선 권고
### Section 2: channelStrategies
- 메인 SNS 채널(Instagram, Facebook, YouTube, TikTok, 네이버 블로그) + 영문 계정(Instagram EN, Facebook EN) 카드를 **모두 포함**. 데이터 없는 채널도 빠뜨리지 말 것.
- **currentStatus**: 데이터 있는 채널은 실제 수치로 서술 (예: "14,047 팔로워, Reels 0개", "104K 구독자, 주 2~3회 업로드"). **데이터 없는 채널은 "계정 없음"** 으로 표시. `excellent`/`warning`/`good` 같은 등급·평가어 금지.
- **targetGoal은 모든 채널에 반드시 채울 것** — 구체적 목표 수치(예: "50K 팔로워, Reels 주 5개"). 데이터 없는 채널도 시작 시 권장 목표를 작성하고 비우지 말 것.
- 각 채널의 우선순위(P0/P1/P2), 콘텐츠 유형, 게시 빈도, 포맷 가이드라인 모두 권장값으로 작성 — 데이터 없어도 시작 권장값으로 채울 것.
- 리포트에 데이터가 있는 채널만 포함
- **currentStatus는 현재 채널 상태를 실제 수치로 서술** (예: "14,047 팔로워, Reels 0개", "104K 구독자, 주 2~3회 업로드"). `excellent`/`warning`/`good` 같은 등급·평가어를 절대 쓰지 마세요.
- targetGoal은 구체적 목표 수치로 작성 (예: "50K 팔로워, Reels 주 5개")
- 각 채널의 우선순위(P0/P1/P2), 콘텐츠 유형, 게시 빈도, 포맷 가이드라인 작성
- customerJourneyStage는 해당 채널의 주요 기여 단계로 설정
### Section 3: contentStrategy

View File

@ -63,19 +63,12 @@
### 페이스북 (영문 페이지)
{facebook_en}
### 카카오톡 채널 (URL only — 수집 데이터 없음, 존재 여부만 확인)
{kakao_talk}
### 네이버 카페 (URL only — 수집 데이터 없음, 존재 여부만 확인)
{naver_cafe}
### 채널별 로고 분석 (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} 을 **그대로** 사용 (강남언니 표기명 '-본원' 등으로 바꾸지 말 것).
@ -89,14 +82,8 @@
- 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 등)은 절대 임의로 만들지 마세요.** 데이터 없으면 그 채널은 생략 (랜덤 생성·추측 금지).
- **수집 데이터에 없는 채널(카카오톡/네이버플레이스/네이버카페/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' 같은 거).
## 분석 지침

View File

@ -64,20 +64,6 @@ 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)
@ -96,15 +82,10 @@ 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": total,
"totalResults": int(total_match.group(1)) if total_match else len(posts),
"posts": posts[:10],
}

View File

@ -3,12 +3,10 @@
정확한 hex 색상은 color_extractor가 CSS에서 직접 뽑음 (Vision은 근사값밖에 ).
Vision은 사람이 봐야 있는 정성 정보 심볼 형태/워드마크/ 담당.
"""
import asyncio
import base64
import json
import logging
import re
import ssl
import httpx
from openai import AsyncOpenAI
@ -50,25 +48,10 @@ class VisionClient:
@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)
+ '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
@ -83,14 +66,9 @@ class VisionClient:
return None
b64 = base64.b64encode(resp.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
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] = []
@ -158,89 +136,38 @@ class VisionClient:
) -> dict | None:
"""채널별 프로필 이미지(로고)를 보고 각각 설명 + 공식 로고와 일치 여부 평가.
channel_logos: [{"channel": "Instagram", "url": "..."}, ...]
반환: {"channel_logos": [{"channel","logo_description","is_official"}], "inconsistency_summary", "recommendation"}
**3채널씩 묶어 병렬 호출** ( 번에 묶으면 LLM이 채널-이미지 매칭 헷갈려 같은 묘사를
여러 채널에 복사하는 문제 VIEW 한국페북·영문인스타가 "공식 로고" 묘사로 잘못
박혔던 케이스 있어서 분리. 1채널씩 N번보다 가성비 좋음)."""
반환: {"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
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 한번 더 안 부름)
# 공식 로고가 있으면 맨 앞에 두고 기준으로 삼음
urls: list[str] = []
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 = "", ""
urls.append(official_logo_url)
urls.extend(c["url"] for c in items)
channel_order = ", ".join(c.get("channel", "?") for c in items)
return {
"channel_logos": channel_logos_out,
"inconsistency_summary": summary,
"recommendation": rec,
}
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'
"}\n"
"모든 logo_description·inconsistency_summary·recommendation은 반드시 한국어로 작성하세요 (영어 금지)."
)
return await self._ask(urls, prompt)

File diff suppressed because it is too large Load Diff

View File

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

View File

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

View File

@ -46,8 +46,6 @@ async def generate_report(analysis_run_id: str) -> ReportOutput:
"tiktok": _json(clinic.get("tiktok")),
"instagram_en": _json(clinic.get("instagramEn")),
"facebook_en": _json(clinic.get("facebookEn")),
"kakao_talk": _json(clinic.get("kakaoTalk")),
"naver_cafe": _json(clinic.get("naverCafe")),
"channel_logos": _json(clinic.get("channelLogos")),
**{
channel: _json(data)
@ -71,7 +69,6 @@ async def generate_plan(analysis_run_id: str) -> PlanOutput:
report_data = run["report_data"]
report = json.loads(report_data) if isinstance(report_data, str) else report_data
market = await get_market_analysis(analysis_run_id)
raw = await get_analysis_raw_data(analysis_run_id)
def _json(v) -> str | None:
return json.dumps(v, ensure_ascii=False) if v else None
@ -92,7 +89,6 @@ async def generate_plan(analysis_run_id: str) -> PlanOutput:
"tiktok": _json(clinic.get("tiktok")),
"instagram_en": _json(clinic.get("instagramEn")),
"facebook_en": _json(clinic.get("facebookEn")),
"naver_blog": _json(_naver_blog_summary(raw.get("naver_blog"))),
"channel_logos": _json(clinic.get("channelLogos")),
"brand_assets": _json(clinic.get("brandAssets")),
}
@ -100,18 +96,6 @@ async def generate_plan(analysis_run_id: str) -> PlanOutput:
return await LLMService(provider="perplexity").generate(plan_prompt, input_data)
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,
}
async def _build_overrides(analysis_run_id: str) -> dict:
run = await fetchone(
"SELECT hospital_id, instagram_data_id, facebook_data_id,"
@ -220,25 +204,9 @@ async def run_report_task(analysis_run_id: str) -> None:
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)
result = await generate_plan(analysis_run_id)
# profile_photo 는 brand_assets.logo_description 으로 코드가 박음 (LLM "(가이드 미보유)" 같은 hallucination 차단)
run = await fetchone("SELECT hospital_id FROM analysis_runs WHERE analysis_run_id = %s", (analysis_run_id,))
if run:
hr = await fetchone("SELECT raw_data FROM hospital_baseinfo WHERE hospital_id = %s", (run["hospital_id"],))
h = json.loads(hr["raw_data"]) if hr and isinstance(hr.get("raw_data"), str) else (hr or {}).get("raw_data") or {}
logo_desc = ((h.get("brandAssets") or {}).get("logo_description")) or ""
result = _patch_plan(result, logo_desc)
await execute(
"UPDATE analysis_runs SET plan_data = %s WHERE analysis_run_id = %s",
(json.dumps(result.model_dump(), ensure_ascii=False), analysis_run_id),

View File

@ -80,8 +80,6 @@ async def collect_all(
tiktok_url: str | None = None,
instagram_en_url: str | None = None,
facebook_en_url: str | None = None,
kakao_talk_url: str | None = None,
naver_cafe_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,))
@ -113,7 +111,6 @@ async def collect_all(
collect_extra_channels(
analysis_run_id, hospital_id,
tiktok_url=tiktok_url, instagram_en_url=instagram_en_url, facebook_en_url=facebook_en_url,
kakao_talk_url=kakao_talk_url, naver_cafe_url=naver_cafe_url,
),
"extra_channels",
)

View File

@ -2,7 +2,6 @@ import asyncio
import json
import logging
import os
import re
from urllib.parse import urlparse
from common.db import fetchone, fetch_raw, merge_hospital_raw_data
from common.utils import get_env
@ -58,19 +57,12 @@ async def collect_brand_assets(analysis_run_id: str, hospital_id: str) -> None:
return
# 3. Vision은 로고 정성 묘사만 (hex는 CSS 추출이 더 정확). 키 없으면 색상만 저장.
# Gemini Vision은 SVG 미지원 → SVG URL이 후보로 들어오면 Vision skip하고 URL만 그대로 박음 (묘사 없음).
SVG_URL = re.compile(r"\.svg(?:\?|#|$)", re.I)
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:
if SVG_URL.search(cand):
logger.info("[brand_assets] %s URL is SVG — Vision 분석 skip, URL만 보관: %s", kind, cand)
result = {"logo_images": {"circle": None, "horizontal": cand, "korean": None}}
used_kind = kind
break
result = await vc.analyze_brand_assets(logo_url=cand, homepage_url=homepage_url)
if result:
used_kind = kind
@ -103,12 +95,9 @@ async def collect_extra_channels(
tiktok_url: str | None = None,
instagram_en_url: str | None = None,
facebook_en_url: str | None = None,
kakao_talk_url: str | None = None,
naver_cafe_url: str | None = None,
) -> None:
"""틱톡 / 인스타 EN / 페북 EN 수집 + 카카오톡/네이버 카페 URL만 보관 →
모두 hospital raw_data에 저장. 인스타EN·페북EN은 기존 Apify 수집기 재사용, 틱톡은 신규 액터.
카카오톡·네이버 카페는 콘텐츠 수집 (URL만 LLM이 채널 존재 신호로 사용)."""
"""틱톡 / 인스타 EN / 페북 EN 수집 → hospital raw_data에 저장 (별도 테이블 없이).
인스타EN·페북EN은 기존 Apify 수집기 재사용, 틱톡은 신규 액터."""
apify = ApifyClient(get_env("APIFY_API_TOKEN"))
jobs: dict = {}
if instagram_en_url:
@ -117,25 +106,19 @@ async def collect_extra_channels(
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 = {}
if jobs:
logger.info("[extra_channels] start run=%s channels=%s", analysis_run_id, list(jobs))
done = await asyncio.gather(*jobs.values(), return_exceptions=True)
for key, res in zip(jobs.keys(), done):
if isinstance(res, Exception):
logger.warning("[extra_channels] %s 수집 실패: %s", key, res)
elif res:
if key == "facebookEn":
res = transform_facebook(res)
results[key] = res
# URL-only 채널 (수집 X, 존재 여부만)
if kakao_talk_url:
results["kakaoTalk"] = {"url": kakao_talk_url}
if naver_cafe_url:
results["naverCafe"] = {"url": naver_cafe_url}
for key, res in zip(jobs.keys(), done):
if isinstance(res, Exception):
logger.warning("[extra_channels] %s 수집 실패: %s", key, res)
elif res:
if key == "facebookEn":
res = transform_facebook(res)
results[key] = res
if not results:
logger.info("[extra_channels] 수집 결과 없음 run=%s", analysis_run_id)
return

View File

@ -30,8 +30,6 @@ async def run_pipeline(analysis_run_id: str, extra_channels: dict | None = None)
tiktok_url=extra_channels.get("tiktok"),
instagram_en_url=extra_channels.get("instagram_en"),
facebook_en_url=extra_channels.get("facebook_en"),
kakao_talk_url=extra_channels.get("kakao_talk"),
naver_cafe_url=extra_channels.get("naver_cafe"),
)
# ── 2. Market ────────────────────────────────────────────────────────────