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")), "tiktok": _with_scheme(urls.get("tiktok")),
"instagram_en": _with_scheme(urls.get("instagramEn")), "instagram_en": _with_scheme(urls.get("instagramEn")),
"facebook_en": _with_scheme(urls.get("facebookEn")), "facebook_en": _with_scheme(urls.get("facebookEn")),
"kakao_talk": _with_scheme(urls.get("kakaoTalk")),
"naver_cafe": _with_scheme(urls.get("naverCafe")),
} }
return {} return {}
@ -47,12 +45,11 @@ async def start_analysis(body: AnalysisCreate, background_tasks: BackgroundTasks
if not hospital: if not hospital:
raise HTTPException(status_code=409, detail="Clinic not found") 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, body.channels.instagram) if body.channels.instagram else None
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, body.channels.facebook) if body.channels.facebook 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, body.channels.naver_blog) if body.channels.naver_blog 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, body.channels.youtube) if body.channels.youtube 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, body.channels.gangnam_unni) if body.channels.gangnam_unni else None
gu_id = await insert_gangnam_unni_row(hospital_id, _with_scheme(body.channels.gangnam_unni)) if body.channels.gangnam_unni else None
analysis_run_id = await insert_analysis_run( analysis_run_id = await insert_analysis_run(
analysis_run_id, hospital_id, hospital["owner_user_id"], 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"), "tiktok": body.channels.tiktok or mock_extra.get("tiktok"),
"instagram_en": body.channels.instagram_en or mock_extra.get("instagram_en"), "instagram_en": body.channels.instagram_en or mock_extra.get("instagram_en"),
"facebook_en": body.channels.facebook_en or mock_extra.get("facebook_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)) 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, extra_channels)

View File

@ -61,27 +61,6 @@ def _normalize_homepage(url: str) -> str:
return u.rstrip("/") return u.rstrip("/")
# SSL 인증서가 www.* 에만 유효한 도메인 — bare 도메인이면 사용자 클릭 시 브라우저 SSL warning 뜸.
_WWW_REQUIRED = ("gangnamunni.com", "facebook.com", "instagram.com")
def _with_scheme(u: str | None) -> str | None: def _with_scheme(u: str | None) -> str | None:
"""scheme 없는 URL에 https:// 보정 (수집기/링크 표시용). 빈 값은 None. """scheme 없는 URL에 https:// 보정 (수집기 파싱용). 빈 값은 None."""
+ 중첩된 https:// 끼어있으면 마지막 URL만 추출 (LLM이 가끔 'https://www.X/https://Y' 같이 만듦). return (u if "://" in u else "https://" + u) if u else None
+ 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

View File

@ -13,9 +13,6 @@ IG_HIGHLIGHTS_ACTOR = "igview-owner~instagram-highlights-scraper"
FB_PAGES_ACTOR = "apify~facebook-pages-scraper" FB_PAGES_ACTOR = "apify~facebook-pages-scraper"
FB_POSTS_ACTOR = "apify~facebook-posts-scraper" FB_POSTS_ACTOR = "apify~facebook-posts-scraper"
# TikTok
TIKTOK_ACTOR = "clockworks~tiktok-scraper"
def _ig_username(url: str) -> str: def _ig_username(url: str) -> str:
return urlparse(url).path.strip("/").split("/")[0] if "://" in url else url.lstrip("@") return urlparse(url).path.strip("/").split("/")[0] if "://" in url else url.lstrip("@")
@ -68,13 +65,6 @@ class ApifyClient:
return None return None
if isinstance(highlights, Exception): if isinstance(highlights, Exception):
highlights = [] 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 { return {
"username": profile["username"], "username": profile["username"],
"profileImage": profile.get("hdProfilePicUrl") or profile.get("profilePicUrl"), "profileImage": profile.get("hdProfilePicUrl") or profile.get("profilePicUrl"),
@ -175,7 +165,7 @@ class ApifyClient:
async def fetch_tiktok_profile(self, url: str) -> list[dict]: async def fetch_tiktok_profile(self, url: str) -> list[dict]:
user = urlparse(url).path.strip("/").lstrip("@").split("/")[0] if "://" in url else url.lstrip("@") 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], "profiles": [user],
"resultsPerPage": 10, "resultsPerPage": 10,
"profileScrapeSections": ["videos"], "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: def find_logo_url_in_html(html: str, base_url: str, css_texts: list[str] | None = None) -> str | None:
"""HTML에서 logo URL 찾기. 우선순위: """HTML에서 logo URL 찾기. class/id/alt → 부모 + 중첩 img → background-image → src에 logo → header/nav → og:image 순."""
1) 패턴 1~8 (class/id/alt/src에 'logo' 명시된 img 가장 specific) for pat in LOGO_IMG_PATTERNS:
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]:
for m in pat.finditer(html): for m in pat.finditer(html):
src = m.group(1) 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 continue
return urljoin(base_url, src) return urljoin(base_url, src)
# 외부 CSS에서 .logo background-image 추출
# 2) 외부 CSS의 .logo { background-image } — class-based 이므로 generic 패턴보다 우선
for css in (css_texts or []): for css in (css_texts or []):
m = LOGO_CSS_PATTERN.search(css) m = LOGO_CSS_PATTERN.search(css)
if m: if m:
src = m.group(1) src = m.group(1)
if not _is_noise(src): if src and not src.startswith("data:"):
return urljoin(base_url, src) 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 return None

View File

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

View File

@ -326,8 +326,6 @@ class ReportInput(BaseModel):
tiktok: str | None = None tiktok: str | None = None
instagram_en: str | None = None instagram_en: str | None = None
facebook_en: str | None = None facebook_en: str | None = None
kakao_talk: str | None = None
naver_cafe: str | None = None
channel_logos: str | None = None channel_logos: str | None = None

View File

@ -32,11 +32,8 @@
## 분석 리포트 ## 분석 리포트
{report} {report}
## 추가 채널 데이터 (네이버 블로그 / 틱톡 / 인스타그램 EN / 페이스북 EN) ## 추가 채널 데이터 (틱톡 / 인스타그램 EN / 페이스북 EN)
아래에 데이터가 있는 채널은 channelStrategies와 channelBranding에 **반드시 포함**하세요 (네이버 블로그, 틱톡, 영문 인스타그램, 영문 페이스북). null이면 제외. 아래에 데이터가 있는 채널은 channelStrategies와 channelBranding에 **반드시 포함**하세요 (틱톡, 영문 인스타그램, 영문 페이스북). null이면 제외.
### 네이버 블로그 (Naver Blog)
{naver_blog}
### 틱톡 (TikTok) ### 틱톡 (TikTok)
{tiktok} {tiktok}
@ -50,12 +47,7 @@
## 채널별 로고 분석 (Gemini Vision) — 채널룰/일관성의 근거 ## 채널별 로고 분석 (Gemini Vision) — 채널룰/일관성의 근거
{channel_logos} {channel_logos}
- 위 channel_logos[]의 각 항목: channel(채널명), logo_description(프로필이 어떻게 생겼는지), is_official(공식 로고와 일치 여부). - 위 channel_logos[]의 각 항목: channel(채널명), logo_description(프로필이 어떻게 생겼는지), is_official(공식 로고와 일치 여부).
- **channelBranding[]은 "어떻게 해야 하는지 권장 가이드라인" 섹션입니다.** 채널 통일 전략 기준으로 권장값 박을 것: - **channelBranding[]를 이 데이터로 채우세요**: 채널별로 profilePhoto=해당 채널의 logo_description, currentStatus=is_official이 true면 "correct" / false면 "incorrect" (데이터 없는 채널은 "missing"). bannerSpec은 권장 배너 규격(크기/디자인)을 작성.
- profilePhoto: **빈 문자열 ""로 두세요.** 시스템이 brand_assets.logo_description으로 직접 채우므로 LLM은 만들지 마세요.
- bannerSpec: 권장 배너 규격 (크기·디자인 가이드)
- bioTemplate: 권장 bio 템플릿 (구조·필수 요소·예약 링크 포함 여부)
- currentStatus: is_official=true면 "correct" / false면 "incorrect" (데이터 없는 채널은 "N/A") — 현재 상태 마커는 이 필드 하나로만.
- 현재 채널 프로필 이미지의 실제 묘사(channel_logos.channel_logos[].logo_description)는 brandInconsistencies에서만 사용. channelBranding에서 채널별로 다른 묘사를 박지 마세요.
- **brandInconsistencies[]에 "로고" 항목을 반드시 만드세요**: values[]에 채널마다 channel(채널명) / value(logo_description 그대로) / is_correct(is_official 값) 세 필드를 넣고, impact는 inconsistency_summary, recommendation은 channel_logos.recommendation 기반으로 작성 (공식 로고로 통일 권고 포함). - **brandInconsistencies[]에 "로고" 항목을 반드시 만드세요**: values[]에 채널마다 channel(채널명) / value(logo_description 그대로) / is_correct(is_official 값) 세 필드를 넣고, impact는 inconsistency_summary, recommendation은 channel_logos.recommendation 기반으로 작성 (공식 로고로 통일 권고 포함).
## 브랜드 자산 (홈페이지 CSS에서 추출 — 결정적 데이터) ## 브랜드 자산 (홈페이지 CSS에서 추출 — 결정적 데이터)
@ -77,10 +69,10 @@
- brandInconsistencies: 채널 간 브랜딩 불일치 항목 및 개선 권고 - brandInconsistencies: 채널 간 브랜딩 불일치 항목 및 개선 권고
### Section 2: channelStrategies ### Section 2: channelStrategies
- 메인 SNS 채널(Instagram, Facebook, YouTube, TikTok, 네이버 블로그) + 영문 계정(Instagram EN, Facebook EN) 카드를 **모두 포함**. 데이터 없는 채널도 빠뜨리지 말 것. - 리포트에 데이터가 있는 채널만 포함
- **currentStatus**: 데이터 있는 채널은 실제 수치로 서술 (예: "14,047 팔로워, Reels 0개", "104K 구독자, 주 2~3회 업로드"). **데이터 없는 채널은 "계정 없음"** 으로 표시. `excellent`/`warning`/`good` 같은 등급·평가어 금지. - **currentStatus는 현재 채널 상태를 실제 수치로 서술** (예: "14,047 팔로워, Reels 0개", "104K 구독자, 주 2~3회 업로드"). `excellent`/`warning`/`good` 같은 등급·평가어를 절대 쓰지 마세요.
- **targetGoal은 모든 채널에 반드시 채울 것** — 구체적 목표 수치(예: "50K 팔로워, Reels 주 5개"). 데이터 없는 채널도 시작 시 권장 목표를 작성하고 비우지 말 것. - targetGoal은 구체적 목표 수치로 작성 (예: "50K 팔로워, Reels 주 5개")
- 각 채널의 우선순위(P0/P1/P2), 콘텐츠 유형, 게시 빈도, 포맷 가이드라인 모두 권장값으로 작성 — 데이터 없어도 시작 권장값으로 채울 것. - 각 채널의 우선순위(P0/P1/P2), 콘텐츠 유형, 게시 빈도, 포맷 가이드라인 작성
- customerJourneyStage는 해당 채널의 주요 기여 단계로 설정 - customerJourneyStage는 해당 채널의 주요 기여 단계로 설정
### Section 3: contentStrategy ### Section 3: contentStrategy

View File

@ -63,19 +63,12 @@
### 페이스북 (영문 페이지) ### 페이스북 (영문 페이지)
{facebook_en} {facebook_en}
### 카카오톡 채널 (URL only — 수집 데이터 없음, 존재 여부만 확인)
{kakao_talk}
### 네이버 카페 (URL only — 수집 데이터 없음, 존재 여부만 확인)
{naver_cafe}
### 채널별 로고 분석 (Gemini Vision) ### 채널별 로고 분석 (Gemini Vision)
{channel_logos} {channel_logos}
- channel_logos.channel_logos[]에 각 채널의 로고 설명(logo_description)과 공식 로고 일치 여부(is_official)가 있습니다. - channel_logos.channel_logos[]에 각 채널의 로고 설명(logo_description)과 공식 로고 일치 여부(is_official)가 있습니다.
- **facebook_audit.pages[].logo** 는 짧은 판정 타이틀로: is_official=true면 `"일치 (공식 로고)"`, false면 `"불일치 (비공식 변형)"`. 그리고 **facebook_audit.pages[].logo_description** 에 해당 채널의 logo_description(설명문)을 넣으세요. - **facebook_audit.pages[].logo** 는 짧은 판정 타이틀로: is_official=true면 `"일치 (공식 로고)"`, false면 `"불일치 (비공식 변형)"`. 그리고 **facebook_audit.pages[].logo_description** 에 해당 채널의 logo_description(설명문)을 넣으세요.
- 위 값들은 channel_logos 데이터 기반으로만 작성하고 추측하지 마세요. - 위 값들은 channel_logos 데이터 기반으로만 작성하고 추측하지 마세요.
- 채널 간 로고 불일치(is_official=false)는 brand 일관성 진단(problem_diagnosis/weaknesses)에 반영하세요. - 채널 간 로고 불일치(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 / 채널 audit 작성 지침 (수집 데이터 그대로, 추측 금지)
- clinic_snapshot.name 은 {clinic_name} 을 **그대로** 사용 (강남언니 표기명 '-본원' 등으로 바꾸지 말 것). - clinic_snapshot.name 은 {clinic_name} 을 **그대로** 사용 (강남언니 표기명 '-본원' 등으로 바꾸지 말 것).
@ -89,14 +82,8 @@
- other_channels에는 메인 audit(YouTube/Instagram/Facebook/Website)에 **포함되지 않은** 채널만 넣으세요. - other_channels에는 메인 audit(YouTube/Instagram/Facebook/Website)에 **포함되지 않은** 채널만 넣으세요.
- 위 '채널 데이터'에 **실제 수집된 데이터가 있는 채널만** status=active와 실제 url로 일관되게 포함: 네이버 블로그, 강남언니, 틱톡, 영문 인스타그램({instagram_en}), 영문 페이스북({facebook_en}). - 위 '채널 데이터'에 **실제 수집된 데이터가 있는 채널만** status=active와 실제 url로 일관되게 포함: 네이버 블로그, 강남언니, 틱톡, 영문 인스타그램({instagram_en}), 영문 페이스북({facebook_en}).
- **영문 인스타그램·영문 페이스북은 KR 메인 audit(Instagram/Facebook)과 별개 계정이므로, 데이터가 있으면 반드시 other_channels에 "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은 수집 데이터의 실제 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 None
return resp.text 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: async def get_blog_rss(self, url: str) -> dict | None:
blog_handle = urlparse(url).path.strip("/").split("/")[0] if "://" in url else url blog_handle = urlparse(url).path.strip("/").split("/")[0] if "://" in url else url
xml = await self.fetch_blog_rss(blog_handle) xml = await self.fetch_blog_rss(blog_handle)
@ -96,15 +82,10 @@ class NaverClient:
"postDate": date.group(1) if date else "", "postDate": date.group(1) if date else "",
"description": re.sub(r"<[^>]*>", "", desc.group(1) if desc else "").strip()[:150], "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) 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 { return {
"officialBlogUrl": f"https://blog.naver.com/{blog_handle}", "officialBlogUrl": f"https://blog.naver.com/{blog_handle}",
"officialBlogHandle": blog_handle, "officialBlogHandle": blog_handle,
"totalResults": total, "totalResults": int(total_match.group(1)) if total_match else len(posts),
"posts": posts[:10], "posts": posts[:10],
} }

View File

@ -3,12 +3,10 @@
정확한 hex 색상은 color_extractor가 CSS에서 직접 뽑음 (Vision은 근사값밖에 ). 정확한 hex 색상은 color_extractor가 CSS에서 직접 뽑음 (Vision은 근사값밖에 ).
Vision은 사람이 봐야 있는 정성 정보 심볼 형태/워드마크/ 담당. Vision은 사람이 봐야 있는 정성 정보 심볼 형태/워드마크/ 담당.
""" """
import asyncio
import base64 import base64
import json import json
import logging import logging
import re import re
import ssl
import httpx import httpx
from openai import AsyncOpenAI from openai import AsyncOpenAI
@ -50,25 +48,10 @@ class VisionClient:
@staticmethod @staticmethod
async def _fetch_as_data_url(url: str) -> str | None: async def _fetch_as_data_url(url: str) -> str | None:
"""Gemini는 URL 직접 fetch가 막힌 호스트가 많아 base64 인라인으로 변환. """Gemini는 URL 직접 fetch가 막힌 호스트가 많아 base64 인라인으로 변환.
+ 'image does not exist' 같은 placeholder 이미지 거부 (작은 bytes / 잘못된 content-type). + 'image does not exist' 같은 placeholder 이미지 거부 (작은 bytes / 잘못된 content-type)."""
+ 한국 의료 사이트 SSL이 약해서 표준 검증에 실패하는 대응 (3 SSL fallback).""" try:
headers = {"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"} async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as c:
resp = await c.get(url)
def _weak_ctx() -> ssl.SSLContext:
ctx = ssl.create_default_context()
try:
ctx.set_ciphers("DEFAULT@SECLEVEL=1")
except ssl.SSLError:
pass
return ctx
last_err: Exception | None = None
for verify in (True, _weak_ctx(), False):
try:
async with httpx.AsyncClient(
timeout=15.0, follow_redirects=True, headers=headers, verify=verify,
) as c:
resp = await c.get(url)
if resp.status_code != 200: if resp.status_code != 200:
logger.warning("[vision] fetch %s status=%s", url, resp.status_code) logger.warning("[vision] fetch %s status=%s", url, resp.status_code)
return None return None
@ -83,14 +66,9 @@ class VisionClient:
return None return None
b64 = base64.b64encode(resp.content).decode("ascii") b64 = base64.b64encode(resp.content).decode("ascii")
return f"data:{mime};base64,{b64}" return f"data:{mime};base64,{b64}"
except (httpx.ConnectError, httpx.ReadError, ssl.SSLError) as e: except Exception as e:
last_err = e logger.warning("[vision] fetch error %s: %s", url, e)
continue return None
except Exception as e:
logger.warning("[vision] fetch error %s: %s", url, e)
return None
logger.warning("[vision] fetch %s SSL fallback all failed: %s", url, last_err)
return None
async def _ask(self, image_urls: list[str], prompt: str, max_tokens: int = 4000) -> dict | None: async def _ask(self, image_urls: list[str], prompt: str, max_tokens: int = 4000) -> dict | None:
content: list[dict] = [] content: list[dict] = []
@ -158,89 +136,38 @@ class VisionClient:
) -> dict | None: ) -> dict | None:
"""채널별 프로필 이미지(로고)를 보고 각각 설명 + 공식 로고와 일치 여부 평가. """채널별 프로필 이미지(로고)를 보고 각각 설명 + 공식 로고와 일치 여부 평가.
channel_logos: [{"channel": "Instagram", "url": "..."}, ...] channel_logos: [{"channel": "Instagram", "url": "..."}, ...]
반환: {"channel_logos": [{"channel","logo_description","is_official"}], "inconsistency_summary", "recommendation"} 반환: {"channel_logos": [{"channel","logo_description","is_official"}], "inconsistency_summary", "recommendation"}"""
**3채널씩 묶어 병렬 호출** ( 번에 묶으면 LLM이 채널-이미지 매칭 헷갈려 같은 묘사를
여러 채널에 복사하는 문제 VIEW 한국페북·영문인스타가 "공식 로고" 묘사로 잘못
박혔던 케이스 있어서 분리. 1채널씩 N번보다 가성비 좋음)."""
items = [c for c in channel_logos if c.get("url")] items = [c for c in channel_logos if c.get("url")]
if not items: if not items:
return None return None
CHUNK = 3 # 공식 로고가 있으면 맨 앞에 두고 기준으로 삼음
urls: list[str] = []
async def _chunk(batch: list[dict]) -> list[dict]:
urls = [official_logo_url] + [c["url"] for c in batch] if official_logo_url else [c["url"] for c in batch]
n = len(batch)
# 이미지 번호 ↔ 채널 매핑 명시
if official_logo_url:
mapping = "이미지 1 = 공식 로고\n" + "\n".join(
f"이미지 {i+2} = {c.get('channel','?')} 채널 프로필" for i, c in enumerate(batch)
)
instruction = (
f"{mapping}\n\n"
f"이미지 2~{n+1}(채널 프로필 {n}개)을 각각 **그 이미지에 실제로 보이는 그대로** "
"한국어 1문장으로 묘사하세요 (색·형태·텍스트·배경 그대로).\n"
"❗ 공식 로고(이미지 1) 묘사를 절대 복사하지 마세요. 각 채널 이미지에 보이는 실제 특징만.\n"
"각 채널이 공식 로고와 시각적으로 거의 동일하면 is_official=true, "
"심볼/색/배경/텍스트가 다르거나 모델 사진이면 false.\n"
)
else:
mapping = "\n".join(f"이미지 {i+1} = {c.get('channel','?')} 채널 프로필" for i, c in enumerate(batch))
instruction = (
f"{mapping}\n\n"
f"각 이미지를 보이는 그대로 한국어 1문장으로 묘사 (색·형태·텍스트·배경).\n"
)
schema_lines = ",\n".join(
f' {{"channel": "{c.get("channel","?")}", "logo_description": "...", "is_official": true}}'
for c in batch
)
p = (
instruction
+ "\n아래 JSON으로만 응답 (코드펜스 없이, 순수 JSON):\n{\n"
+ f' "channel_logos": [\n{schema_lines}\n ]\n'
+ "}\n"
+ f"channel 필드는 위 매핑 그대로 ({', '.join(c.get('channel','?') for c in batch)}). "
+ "logo_description은 반드시 한국어 (영어 금지)."
)
r = await self._ask(urls, p)
if not r:
return []
out = []
for c in r.get("channel_logos", []):
out.append({
"channel": c.get("channel", ""),
"logo_description": c.get("logo_description", ""),
"is_official": bool(c.get("is_official", False)) if official_logo_url else None,
})
return out
# 3개씩 청크 → 병렬
chunks = [items[i:i+CHUNK] for i in range(0, len(items), CHUNK)]
results = await asyncio.gather(*[_chunk(b) for b in chunks], return_exceptions=True)
channel_logos_out: list[dict] = []
for r in results:
if isinstance(r, Exception):
logger.warning("[vision] channel_logo chunk error: %s", r)
continue
channel_logos_out.extend(r)
if not channel_logos_out:
return None
# 일관성 요약 + 권고는 결정적 산출 (LLM 한번 더 안 부름)
if official_logo_url: if official_logo_url:
mismatches = [c["channel"] for c in channel_logos_out if not c.get("is_official")] urls.append(official_logo_url)
if not mismatches: urls.extend(c["url"] for c in items)
summary = "모든 채널이 공식 로고를 일관되게 사용하고 있습니다." channel_order = ", ".join(c.get("channel", "?") for c in items)
rec = "현재 일관성 유지."
else:
summary = f"{len(mismatches)}개 채널({', '.join(mismatches)})이 공식 로고와 다른 이미지를 사용해 브랜드 일관성이 부족합니다."
rec = "비공식 채널 프로필을 공식 로고로 통일 권고."
else:
summary, rec = "", ""
return { if official_logo_url:
"channel_logos": channel_logos_out, header = (
"inconsistency_summary": summary, "첨부 이미지 중 **첫 번째가 이 병원의 공식 로고**입니다. "
"recommendation": rec, 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 tiktok: str | None = None
instagram_en: str | None = None instagram_en: str | None = None
facebook_en: str | None = None facebook_en: str | None = None
kakao_talk: str | None = None
naver_cafe: str | None = None
class AnalysisOptions(BaseModel): class AnalysisOptions(BaseModel):

View File

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

View File

@ -46,8 +46,6 @@ async def generate_report(analysis_run_id: str) -> ReportOutput:
"tiktok": _json(clinic.get("tiktok")), "tiktok": _json(clinic.get("tiktok")),
"instagram_en": _json(clinic.get("instagramEn")), "instagram_en": _json(clinic.get("instagramEn")),
"facebook_en": _json(clinic.get("facebookEn")), "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_logos": _json(clinic.get("channelLogos")),
**{ **{
channel: _json(data) channel: _json(data)
@ -71,7 +69,6 @@ async def generate_plan(analysis_run_id: str) -> PlanOutput:
report_data = run["report_data"] report_data = run["report_data"]
report = json.loads(report_data) if isinstance(report_data, str) else report_data report = json.loads(report_data) if isinstance(report_data, str) else report_data
market = await get_market_analysis(analysis_run_id) market = await get_market_analysis(analysis_run_id)
raw = await get_analysis_raw_data(analysis_run_id)
def _json(v) -> str | None: def _json(v) -> str | None:
return json.dumps(v, ensure_ascii=False) if v else 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")), "tiktok": _json(clinic.get("tiktok")),
"instagram_en": _json(clinic.get("instagramEn")), "instagram_en": _json(clinic.get("instagramEn")),
"facebook_en": _json(clinic.get("facebookEn")), "facebook_en": _json(clinic.get("facebookEn")),
"naver_blog": _json(_naver_blog_summary(raw.get("naver_blog"))),
"channel_logos": _json(clinic.get("channelLogos")), "channel_logos": _json(clinic.get("channelLogos")),
"brand_assets": _json(clinic.get("brandAssets")), "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) 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: async def _build_overrides(analysis_run_id: str) -> dict:
run = await fetchone( run = await fetchone(
"SELECT hospital_id, instagram_data_id, facebook_data_id," "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) 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: async def run_plan_task(analysis_run_id: str) -> None:
logger.info("[plan] start run=%s", analysis_run_id) logger.info("[plan] start run=%s", analysis_run_id)
result = await generate_plan(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( await execute(
"UPDATE analysis_runs SET plan_data = %s WHERE analysis_run_id = %s", "UPDATE analysis_runs SET plan_data = %s WHERE analysis_run_id = %s",
(json.dumps(result.model_dump(), ensure_ascii=False), analysis_run_id), (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, tiktok_url: str | None = None,
instagram_en_url: str | None = None, instagram_en_url: str | None = None,
facebook_en_url: str | None = None, facebook_en_url: str | None = None,
kakao_talk_url: str | None = None,
naver_cafe_url: str | None = None,
) -> None: ) -> None:
async def _url(table: str, row_id: int) -> str: async def _url(table: str, row_id: int) -> str:
row = await fetchone(f"SELECT url FROM {table} WHERE id = %s", (row_id,)) row = await fetchone(f"SELECT url FROM {table} WHERE id = %s", (row_id,))
@ -113,7 +111,6 @@ async def collect_all(
collect_extra_channels( collect_extra_channels(
analysis_run_id, hospital_id, analysis_run_id, hospital_id,
tiktok_url=tiktok_url, instagram_en_url=instagram_en_url, facebook_en_url=facebook_en_url, 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", "extra_channels",
) )

View File

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