Compare commits
8 Commits
652265cd19
...
56fa2c6238
| Author | SHA1 | Date |
|---|---|---|
|
|
56fa2c6238 | |
|
|
4bc7c9652c | |
|
|
bed5f0c274 | |
|
|
fa32109658 | |
|
|
dca0c78860 | |
|
|
db42805fdb | |
|
|
9da285e905 | |
|
|
8c1e513dc0 |
|
|
@ -27,6 +27,8 @@ 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 {}
|
||||
|
||||
|
|
@ -45,11 +47,12 @@ async def start_analysis(body: AnalysisCreate, background_tasks: BackgroundTasks
|
|||
if not hospital:
|
||||
raise HTTPException(status_code=409, detail="Clinic not found")
|
||||
|
||||
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
|
||||
# 사용자가 '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
|
||||
|
||||
analysis_run_id = await insert_analysis_run(
|
||||
analysis_run_id, hospital_id, hospital["owner_user_id"],
|
||||
|
|
@ -62,6 +65,8 @@ 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)
|
||||
|
|
|
|||
|
|
@ -61,6 +61,27 @@ 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."""
|
||||
return (u if "://" in u else "https://" + u) if u else 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
|
||||
|
|
|
|||
|
|
@ -13,6 +13,9 @@ 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("@")
|
||||
|
|
@ -65,6 +68,13 @@ 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"),
|
||||
|
|
@ -165,7 +175,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("clockworks~tiktok-scraper", {
|
||||
return await self._run_actor(TIKTOK_ACTOR, {
|
||||
"profiles": [user],
|
||||
"resultsPerPage": 10,
|
||||
"profileScrapeSections": ["videos"],
|
||||
|
|
|
|||
|
|
@ -84,22 +84,47 @@ 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 찾기. class/id/alt → 부모 + 중첩 img → background-image → src에 logo → header/nav → og:image 순."""
|
||||
for pat in LOGO_IMG_PATTERNS:
|
||||
"""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]:
|
||||
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):
|
||||
if _is_noise(src):
|
||||
continue
|
||||
return urljoin(base_url, src)
|
||||
# 외부 CSS에서 .logo background-image 추출
|
||||
|
||||
# 2) 외부 CSS의 .logo { background-image } — class-based 이므로 generic 패턴보다 우선
|
||||
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:"):
|
||||
if not _is_noise(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
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ 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
|
||||
|
||||
|
|
@ -56,7 +57,7 @@ class ChannelBrandingRule(BaseModel):
|
|||
profile_photo: str
|
||||
banner_spec: str
|
||||
bio_template: str
|
||||
current_status: Literal["correct", "incorrect", "missing"]
|
||||
current_status: Literal["correct", "incorrect", "N/A"]
|
||||
|
||||
|
||||
class BrandPlanInconsistencyValue(BaseModel):
|
||||
|
|
|
|||
|
|
@ -326,6 +326,8 @@ 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
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -32,8 +32,11 @@
|
|||
## 분석 리포트
|
||||
{report}
|
||||
|
||||
## 추가 채널 데이터 (틱톡 / 인스타그램 EN / 페이스북 EN)
|
||||
아래에 데이터가 있는 채널은 channelStrategies와 channelBranding에 **반드시 포함**하세요 (틱톡, 영문 인스타그램, 영문 페이스북). null이면 제외.
|
||||
## 추가 채널 데이터 (네이버 블로그 / 틱톡 / 인스타그램 EN / 페이스북 EN)
|
||||
아래에 데이터가 있는 채널은 channelStrategies와 channelBranding에 **반드시 포함**하세요 (네이버 블로그, 틱톡, 영문 인스타그램, 영문 페이스북). null이면 제외.
|
||||
|
||||
### 네이버 블로그 (Naver Blog)
|
||||
{naver_blog}
|
||||
|
||||
### 틱톡 (TikTok)
|
||||
{tiktok}
|
||||
|
|
@ -47,7 +50,12 @@
|
|||
## 채널별 로고 분석 (Gemini Vision) — 채널룰/일관성의 근거
|
||||
{channel_logos}
|
||||
- 위 channel_logos[]의 각 항목: channel(채널명), logo_description(프로필이 어떻게 생겼는지), is_official(공식 로고와 일치 여부).
|
||||
- **channelBranding[]를 이 데이터로 채우세요**: 채널별로 profilePhoto=해당 채널의 logo_description, currentStatus=is_official이 true면 "correct" / false면 "incorrect" (데이터 없는 채널은 "missing"). bannerSpec은 권장 배너 규격(크기/디자인)을 작성.
|
||||
- **channelBranding[]은 "어떻게 해야 하는지 권장 가이드라인" 섹션입니다.** 채널 통일 전략 기준으로 권장값 박을 것:
|
||||
- profilePhoto: **빈 문자열 ""로 두세요.** 시스템이 brand_assets.logo_description으로 직접 채우므로 LLM은 만들지 마세요.
|
||||
- bannerSpec: 권장 배너 규격 (크기·디자인 가이드)
|
||||
- bioTemplate: 권장 bio 템플릿 (구조·필수 요소·예약 링크 포함 여부)
|
||||
- currentStatus: is_official=true면 "correct" / false면 "incorrect" (데이터 없는 채널은 "N/A") — 현재 상태 마커는 이 필드 하나로만.
|
||||
- 현재 채널 프로필 이미지의 실제 묘사(channel_logos.channel_logos[].logo_description)는 brandInconsistencies에서만 사용. channelBranding에서 채널별로 다른 묘사를 박지 마세요.
|
||||
- **brandInconsistencies[]에 "로고" 항목을 반드시 만드세요**: values[]에 채널마다 channel(채널명) / value(logo_description 그대로) / is_correct(is_official 값) 세 필드를 넣고, impact는 inconsistency_summary, recommendation은 channel_logos.recommendation 기반으로 작성 (공식 로고로 통일 권고 포함).
|
||||
|
||||
## 브랜드 자산 (홈페이지 CSS에서 추출 — 결정적 데이터)
|
||||
|
|
@ -69,10 +77,10 @@
|
|||
- brandInconsistencies: 채널 간 브랜딩 불일치 항목 및 개선 권고
|
||||
|
||||
### Section 2: channelStrategies
|
||||
- 리포트에 데이터가 있는 채널만 포함
|
||||
- **currentStatus는 현재 채널 상태를 실제 수치로 서술** (예: "14,047 팔로워, Reels 0개", "104K 구독자, 주 2~3회 업로드"). `excellent`/`warning`/`good` 같은 등급·평가어를 절대 쓰지 마세요.
|
||||
- targetGoal은 구체적 목표 수치로 작성 (예: "50K 팔로워, Reels 주 5개")
|
||||
- 각 채널의 우선순위(P0/P1/P2), 콘텐츠 유형, 게시 빈도, 포맷 가이드라인 작성
|
||||
- 메인 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), 콘텐츠 유형, 게시 빈도, 포맷 가이드라인 모두 권장값으로 작성 — 데이터 없어도 시작 권장값으로 채울 것.
|
||||
- customerJourneyStage는 해당 채널의 주요 기여 단계로 설정
|
||||
|
||||
### Section 3: contentStrategy
|
||||
|
|
|
|||
|
|
@ -63,12 +63,19 @@
|
|||
### 페이스북 (영문 페이지)
|
||||
{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} 을 **그대로** 사용 (강남언니 표기명 '-본원' 등으로 바꾸지 말 것).
|
||||
|
|
@ -82,8 +89,14 @@
|
|||
- 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 등)은 절대 임의로 만들지 마세요.** 데이터 없으면 그 채널은 생략 (랜덤 생성·추측 금지).
|
||||
- **카카오톡·네이버 카페**: {kakao_talk} 또는 {naver_cafe}에 url이 있으면 other_channels에 각각 "KakaoTalk" / "Naver Cafe"로 status=active + 해당 url로 포함. 수집된 콘텐츠 데이터는 없으므로 URL 존재 자체가 활성 채널 신호. **둘 다 null/빈 값이면 절대 만들지 마세요.**
|
||||
- **그 외 데이터 없는 채널(네이버플레이스/Threads 등)은 절대 임의로 만들지 마세요.** 데이터 없으면 그 채널은 생략 (랜덤 생성·추측 금지).
|
||||
- url은 수집 데이터의 실제 URL만 사용. 없으면 빈 문자열.
|
||||
- **URL에 'https://www.facebook.com/' 같은 prefix를 절대 직접 만들지 마세요.** 수집 데이터의 URL을 그대로 사용. 이미 'https://...' 가 붙은 URL에 또 prefix 붙이면 'https://www.facebook.com/https://facebook.com/X' 같이 깨집니다. 받은 URL = 출력 URL.
|
||||
|
||||
## registry_data 작성 지침 (clinic_snapshot 안)
|
||||
- **registry_data.website_en / district / branches / brand_group / naver_place_url / gangnam_unni_url / google_maps_url 모두 제공된 데이터에 명시되지 않으면 반드시 null로 두세요.**
|
||||
- 영문 사이트 URL, 영문명, 지점 정보 같은 거 데이터에 없으면 **절대 추측하거나 그럴듯해 보이는 도메인을 지어내지 마세요** (예: 'thepsclinic.com', '*-eng.com' 같은 거).
|
||||
|
||||
## 분석 지침
|
||||
|
||||
|
|
|
|||
|
|
@ -64,6 +64,20 @@ class NaverClient:
|
|||
return None
|
||||
return resp.text
|
||||
|
||||
async def fetch_blog_total_count(self, handle: str) -> int | None:
|
||||
"""블로그 전체 글 수는 RSS에 없어서 PostList HTML에서 '554개의 글' 패턴 추출.
|
||||
<h4 class="category_title pcol2">... 554개의 글</h4> 구조."""
|
||||
resp = await http_request(
|
||||
HTTPMethod.GET,
|
||||
url=f"https://blog.naver.com/PostList.naver?blogId={handle}&from=postList&directAccess=true",
|
||||
timeout=15,
|
||||
label="naver-blog-postlist",
|
||||
)
|
||||
if not resp or not resp.is_success:
|
||||
return None
|
||||
m = re.search(r"(\d+)개의 글", resp.text)
|
||||
return int(m.group(1)) if m else None
|
||||
|
||||
async def get_blog_rss(self, url: str) -> dict | None:
|
||||
blog_handle = urlparse(url).path.strip("/").split("/")[0] if "://" in url else url
|
||||
xml = await self.fetch_blog_rss(blog_handle)
|
||||
|
|
@ -82,10 +96,15 @@ class NaverClient:
|
|||
"postDate": date.group(1) if date else "",
|
||||
"description": re.sub(r"<[^>]*>", "", desc.group(1) if desc else "").strip()[:150],
|
||||
})
|
||||
# RSS의 totalCount 우선, 없으면 블로그 PostList 페이지에서 "N개의 글" 파싱, 그것도 없으면 RSS 글수
|
||||
total_match = re.search(r"<totalCount>(\d+)</totalCount>", xml)
|
||||
if total_match:
|
||||
total = int(total_match.group(1))
|
||||
else:
|
||||
total = await self.fetch_blog_total_count(blog_handle) or len(posts)
|
||||
return {
|
||||
"officialBlogUrl": f"https://blog.naver.com/{blog_handle}",
|
||||
"officialBlogHandle": blog_handle,
|
||||
"totalResults": int(total_match.group(1)) if total_match else len(posts),
|
||||
"totalResults": total,
|
||||
"posts": posts[:10],
|
||||
}
|
||||
|
|
|
|||
|
|
@ -3,10 +3,12 @@
|
|||
정확한 hex 색상은 color_extractor가 CSS에서 직접 뽑음 (Vision은 근사값밖에 못 냄).
|
||||
Vision은 사람이 봐야 알 수 있는 정성 정보 — 심볼 형태/워드마크/톤 — 를 담당.
|
||||
"""
|
||||
import asyncio
|
||||
import base64
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
import ssl
|
||||
import httpx
|
||||
from openai import AsyncOpenAI
|
||||
|
||||
|
|
@ -48,10 +50,25 @@ 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)."""
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as c:
|
||||
resp = await c.get(url)
|
||||
+ 'image does not exist' 같은 placeholder 이미지 거부 (작은 bytes / 잘못된 content-type).
|
||||
+ 한국 의료 사이트 중 SSL이 약해서 표준 검증에 실패하는 곳 대응 (3단 SSL fallback)."""
|
||||
headers = {"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"}
|
||||
|
||||
def _weak_ctx() -> ssl.SSLContext:
|
||||
ctx = ssl.create_default_context()
|
||||
try:
|
||||
ctx.set_ciphers("DEFAULT@SECLEVEL=1")
|
||||
except ssl.SSLError:
|
||||
pass
|
||||
return ctx
|
||||
|
||||
last_err: Exception | None = None
|
||||
for verify in (True, _weak_ctx(), False):
|
||||
try:
|
||||
async with httpx.AsyncClient(
|
||||
timeout=15.0, follow_redirects=True, headers=headers, verify=verify,
|
||||
) as c:
|
||||
resp = await c.get(url)
|
||||
if resp.status_code != 200:
|
||||
logger.warning("[vision] fetch %s status=%s", url, resp.status_code)
|
||||
return None
|
||||
|
|
@ -66,9 +83,14 @@ class VisionClient:
|
|||
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
|
||||
except (httpx.ConnectError, httpx.ReadError, ssl.SSLError) as e:
|
||||
last_err = e
|
||||
continue
|
||||
except Exception as e:
|
||||
logger.warning("[vision] fetch error %s: %s", url, e)
|
||||
return None
|
||||
logger.warning("[vision] fetch %s SSL fallback all failed: %s", url, last_err)
|
||||
return None
|
||||
|
||||
async def _ask(self, image_urls: list[str], prompt: str, max_tokens: int = 4000) -> dict | None:
|
||||
content: list[dict] = []
|
||||
|
|
@ -136,38 +158,89 @@ class VisionClient:
|
|||
) -> dict | None:
|
||||
"""채널별 프로필 이미지(로고)를 보고 각각 설명 + 공식 로고와 일치 여부 평가.
|
||||
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")]
|
||||
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)
|
||||
CHUNK = 3
|
||||
|
||||
async def _chunk(batch: list[dict]) -> list[dict]:
|
||||
urls = [official_logo_url] + [c["url"] for c in batch] if official_logo_url else [c["url"] for c in batch]
|
||||
n = len(batch)
|
||||
# 이미지 번호 ↔ 채널 매핑 명시
|
||||
if official_logo_url:
|
||||
mapping = "이미지 1 = 공식 로고\n" + "\n".join(
|
||||
f"이미지 {i+2} = {c.get('channel','?')} 채널 프로필" for i, c in enumerate(batch)
|
||||
)
|
||||
instruction = (
|
||||
f"{mapping}\n\n"
|
||||
f"이미지 2~{n+1}(채널 프로필 {n}개)을 각각 **그 이미지에 실제로 보이는 그대로** "
|
||||
"한국어 1문장으로 묘사하세요 (색·형태·텍스트·배경 그대로).\n"
|
||||
"❗ 공식 로고(이미지 1) 묘사를 절대 복사하지 마세요. 각 채널 이미지에 보이는 실제 특징만.\n"
|
||||
"각 채널이 공식 로고와 시각적으로 거의 동일하면 is_official=true, "
|
||||
"심볼/색/배경/텍스트가 다르거나 모델 사진이면 false.\n"
|
||||
)
|
||||
else:
|
||||
mapping = "\n".join(f"이미지 {i+1} = {c.get('channel','?')} 채널 프로필" for i, c in enumerate(batch))
|
||||
instruction = (
|
||||
f"{mapping}\n\n"
|
||||
f"각 이미지를 보이는 그대로 한국어 1문장으로 묘사 (색·형태·텍스트·배경).\n"
|
||||
)
|
||||
schema_lines = ",\n".join(
|
||||
f' {{"channel": "{c.get("channel","?")}", "logo_description": "...", "is_official": true}}'
|
||||
for c in batch
|
||||
)
|
||||
p = (
|
||||
instruction
|
||||
+ "\n아래 JSON으로만 응답 (코드펜스 없이, 순수 JSON):\n{\n"
|
||||
+ f' "channel_logos": [\n{schema_lines}\n ]\n'
|
||||
+ "}\n"
|
||||
+ f"channel 필드는 위 매핑 그대로 ({', '.join(c.get('channel','?') for c in batch)}). "
|
||||
+ "logo_description은 반드시 한국어 (영어 금지)."
|
||||
)
|
||||
r = await self._ask(urls, p)
|
||||
if not r:
|
||||
return []
|
||||
out = []
|
||||
for c in r.get("channel_logos", []):
|
||||
out.append({
|
||||
"channel": c.get("channel", ""),
|
||||
"logo_description": c.get("logo_description", ""),
|
||||
"is_official": bool(c.get("is_official", False)) if official_logo_url else None,
|
||||
})
|
||||
return out
|
||||
|
||||
# 3개씩 청크 → 병렬
|
||||
chunks = [items[i:i+CHUNK] for i in range(0, len(items), CHUNK)]
|
||||
results = await asyncio.gather(*[_chunk(b) for b in chunks], return_exceptions=True)
|
||||
channel_logos_out: list[dict] = []
|
||||
for r in results:
|
||||
if isinstance(r, Exception):
|
||||
logger.warning("[vision] channel_logo chunk error: %s", r)
|
||||
continue
|
||||
channel_logos_out.extend(r)
|
||||
if not channel_logos_out:
|
||||
return None
|
||||
|
||||
# 일관성 요약 + 권고는 결정적 산출 (LLM 한번 더 안 부름)
|
||||
if official_logo_url:
|
||||
header = (
|
||||
"첨부 이미지 중 **첫 번째가 이 병원의 공식 로고**입니다. "
|
||||
f"이어지는 이미지들은 채널별 프로필 이미지이며 순서는: {channel_order}.\n"
|
||||
"각 채널 로고를 1문장으로 설명하고, 공식 로고(첫 번째)와 일치하면 is_official=true, "
|
||||
"비공식 변형/모델사진/다른 이미지면 false로 평가하세요.\n"
|
||||
)
|
||||
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:
|
||||
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)
|
||||
summary, rec = "", ""
|
||||
|
||||
return {
|
||||
"channel_logos": channel_logos_out,
|
||||
"inconsistency_summary": summary,
|
||||
"recommendation": rec,
|
||||
}
|
||||
|
|
|
|||
1233
app/mock_urls.py
1233
app/mock_urls.py
File diff suppressed because it is too large
Load Diff
|
|
@ -11,6 +11,8 @@ 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):
|
||||
|
|
|
|||
|
|
@ -49,7 +49,7 @@ class ChannelBrandingRule(CamelModel):
|
|||
profile_photo: str
|
||||
banner_spec: str
|
||||
bio_template: str
|
||||
current_status: Literal["correct", "incorrect", "missing"]
|
||||
current_status: Literal["correct", "incorrect", "N/A"]
|
||||
|
||||
|
||||
class BrandGuide(CamelModel):
|
||||
|
|
|
|||
|
|
@ -46,6 +46,8 @@ 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)
|
||||
|
|
@ -69,6 +71,7 @@ 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
|
||||
|
|
@ -89,6 +92,7 @@ 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")),
|
||||
}
|
||||
|
|
@ -96,6 +100,18 @@ 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,"
|
||||
|
|
@ -204,9 +220,25 @@ 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),
|
||||
|
|
|
|||
|
|
@ -80,6 +80,8 @@ 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,))
|
||||
|
|
@ -111,6 +113,7 @@ 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",
|
||||
)
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ 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
|
||||
|
|
@ -57,12 +58,19 @@ 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
|
||||
|
|
@ -95,9 +103,12 @@ 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 수집 → hospital raw_data에 저장 (별도 테이블 없이).
|
||||
인스타EN·페북EN은 기존 Apify 수집기 재사용, 틱톡은 신규 액터."""
|
||||
"""틱톡 / 인스타 EN / 페북 EN 수집 + 카카오톡/네이버 카페 URL만 보관 →
|
||||
모두 hospital raw_data에 저장. 인스타EN·페북EN은 기존 Apify 수집기 재사용, 틱톡은 신규 액터.
|
||||
카카오톡·네이버 카페는 콘텐츠 수집 안 함 (URL만 → LLM이 채널 존재 신호로 사용)."""
|
||||
apify = ApifyClient(get_env("APIFY_API_TOKEN"))
|
||||
jobs: dict = {}
|
||||
if instagram_en_url:
|
||||
|
|
@ -106,19 +117,25 @@ 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 = {}
|
||||
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 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}
|
||||
|
||||
if not results:
|
||||
logger.info("[extra_channels] 수집 결과 없음 run=%s", analysis_run_id)
|
||||
return
|
||||
|
|
|
|||
|
|
@ -30,6 +30,8 @@ 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 ────────────────────────────────────────────────────────────
|
||||
|
|
|
|||
Loading…
Reference in New Issue