121 lines
5.7 KiB
Python
121 lines
5.7 KiB
Python
import logging
|
|
import os
|
|
from urllib.parse import urlparse
|
|
from common.db.source import select_run_raw_data, update_raw_info_merge
|
|
from integrations.vision import VisionClient
|
|
from integrations.color_extractor import extract_brand_assets_from_site
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def collect_brand_assets(analysis_run_id: str, info_id: int) -> None:
|
|
"""홈페이지에서 로고 URL + brand hex 색상 추출 → branding raw_info["brandAssets"] 머지.
|
|
- 로고 URL/hex: HTML·CSS 정규식 (color_extractor) — Vision 의존 X, 사이트 전체 컬러 시스템이 더 정확.
|
|
- 로고 정성 묘사(심볼/워드마크/톤): Gemini Vision (GEMINI_API_KEY 없으면 색상만 저장하고 skip).
|
|
"""
|
|
logger.info("[brand_assets] start run=%s info=%s", analysis_run_id, info_id)
|
|
raw = await select_run_raw_data(analysis_run_id)
|
|
mainpage = raw.get("mainpage") or {}
|
|
homepage_url = mainpage.get("sourceUrl") or ""
|
|
branding = mainpage.get("branding") or {}
|
|
|
|
# 0~1. 사이트 1회 fetch 로 logo URL + brand hex 동시 추출 (img/background-image/CSS .logo, Vision 의존 X)
|
|
site = await extract_brand_assets_from_site(homepage_url) if homepage_url else {}
|
|
html_logo_url = site.get("logo_url")
|
|
css_colors = site.get("colors") or {}
|
|
if html_logo_url:
|
|
logger.info("[brand_assets] HTML logo found: %s", html_logo_url)
|
|
if css_colors:
|
|
logger.info("[brand_assets] css colors: %s", css_colors.get("brand_colors"))
|
|
|
|
# 2. 로고/대표 이미지 후보 (logo → og:image → favicon 순)
|
|
logo_url = html_logo_url or branding.get("logoUrl")
|
|
og_image = branding.get("ogImage")
|
|
favicon = branding.get("faviconUrl")
|
|
candidates: list[tuple[str, str]] = []
|
|
if logo_url: candidates.append(("logo", logo_url))
|
|
if og_image: candidates.append(("og", og_image))
|
|
if favicon: candidates.append(("favicon", favicon))
|
|
if homepage_url:
|
|
parsed = urlparse(homepage_url)
|
|
if parsed.scheme and parsed.netloc:
|
|
candidates.append(("favicon", f"{parsed.scheme}://{parsed.netloc}/favicon.ico"))
|
|
|
|
if not candidates and not css_colors:
|
|
logger.info("[brand_assets] skip — no logo/og/favicon candidates and no CSS colors")
|
|
return
|
|
|
|
# 3. Vision 은 로고 정성 묘사만 (hex 는 CSS 추출이 더 정확). 키 없으면 색상만 저장.
|
|
# SVG 는 vision 내부에서 resvg 로 PNG 래스터화 후 Gemini 에 던지므로 분기 불필요.
|
|
result: dict = {}
|
|
used_kind: str | None = None
|
|
api_key = os.getenv("GEMINI_API_KEY")
|
|
if api_key and candidates:
|
|
vc = VisionClient(api_key)
|
|
for kind, cand in candidates:
|
|
result = await vc.analyze_brand_assets(logo_url=cand, homepage_url=homepage_url)
|
|
if result:
|
|
used_kind = kind
|
|
break
|
|
# favicon 으로만 분석된 경우 진짜 로고가 아니므로 logo URL 은 박지 않음 (묘사는 OK)
|
|
if result and used_kind == "favicon" and result.get("logo_images"):
|
|
result["logo_images"] = {"circle": None, "horizontal": None, "korean": None}
|
|
elif not api_key:
|
|
logger.info("[brand_assets] GEMINI_API_KEY not set — 색상만 저장, Vision 묘사 skip")
|
|
|
|
# 4. CSS 에서 추출한 brand_colors/palette 를 Vision 보다 우선 사용
|
|
if css_colors:
|
|
if css_colors.get("brand_colors"): result["brand_colors"] = css_colors["brand_colors"]
|
|
if css_colors.get("color_palette"): result["color_palette"] = css_colors["color_palette"]
|
|
result["color_source"] = "html+css"
|
|
elif result:
|
|
result["color_source"] = "vision"
|
|
|
|
if result:
|
|
result["logo_source"] = used_kind or "none"
|
|
await update_raw_info_merge(info_id, {"brandAssets": result})
|
|
logger.info("[brand_assets] done keys=%s", list(result.keys()) if result else None)
|
|
|
|
|
|
async def collect_channel_logos(analysis_run_id: str, info_id: int) -> None:
|
|
"""채널별 프로필 이미지(로고)를 모아 Gemini Vision 으로 설명 + 공식 로고 일치 여부 평가.
|
|
→ branding raw_info["channelLogos"] 머지. GEMINI_API_KEY 없으면 skip.
|
|
brand_assets(공식 로고) · 채널 raw_info(profileImage) 가 채워진 뒤 실행돼야 함."""
|
|
api_key = os.getenv("GEMINI_API_KEY")
|
|
if not api_key:
|
|
logger.info("[channel_logos] skip — GEMINI_API_KEY 없음")
|
|
return
|
|
|
|
raw = await select_run_raw_data(analysis_run_id)
|
|
branding = raw.get("branding") or {}
|
|
official = ((branding.get("brandAssets") or {}).get("logo_images") or {}).get("horizontal")
|
|
|
|
# KR 메인 채널 + EN/TikTok 부가 채널 profileImage 수집 (raw_info dict 키: instagram, instagram_en, ...)
|
|
_label = {
|
|
"instagram": "Instagram",
|
|
"facebook": "Facebook",
|
|
"youtube": "YouTube",
|
|
"instagram_en": "Instagram EN",
|
|
"facebook_en": "Facebook EN",
|
|
"tiktok": "TikTok",
|
|
}
|
|
logos: list[dict] = []
|
|
for key, label in _label.items():
|
|
img = (raw.get(key) or {}).get("profileImage")
|
|
if img:
|
|
logos.append({"channel": label, "url": img})
|
|
|
|
if not logos:
|
|
logger.info("[channel_logos] skip — 채널 프로필 이미지 없음")
|
|
return
|
|
|
|
logger.info("[channel_logos] start run=%s channels=%s official=%s",
|
|
analysis_run_id, [l["channel"] for l in logos], bool(official))
|
|
result = await VisionClient(api_key).describe_channel_logos(official, logos)
|
|
if result:
|
|
# Vision 이 못 본 채널도 url 은 채워둠 (프론트에서 이미지 표시용)
|
|
result["logos"] = logos
|
|
await update_raw_info_merge(info_id, {"channelLogos": result})
|
|
logger.info("[channel_logos] done run=%s keys=%s",
|
|
analysis_run_id, list(result.keys()) if result else None)
|