247 lines
12 KiB
Python
247 lines
12 KiB
Python
"""Gemini Vision — 로고/브랜드 비주얼 자동 분석 (OpenAI 호환 모드).
|
|
|
|
정확한 hex 색상은 color_extractor가 CSS에서 직접 뽑음 (Vision은 근사값밖에 못 냄).
|
|
Vision은 사람이 봐야 알 수 있는 정성 정보 — 심볼 형태/워드마크/톤 — 를 담당.
|
|
"""
|
|
import asyncio
|
|
import base64
|
|
import json
|
|
import logging
|
|
import re
|
|
import ssl
|
|
import httpx
|
|
from openai import AsyncOpenAI
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
DEFAULT_MODEL = "gemini-2.5-flash"
|
|
|
|
|
|
class VisionClient:
|
|
"""Gemini Vision을 OpenAI 호환 endpoint로 호출. GEMINI_API_KEY만 필요."""
|
|
|
|
def __init__(self, api_key: str, model: str = DEFAULT_MODEL, timeout: float = 30.0, max_retries: int = 2):
|
|
self.client = AsyncOpenAI(
|
|
api_key=api_key,
|
|
base_url="https://generativelanguage.googleapis.com/v1beta/openai/",
|
|
timeout=timeout,
|
|
max_retries=max_retries,
|
|
)
|
|
self.model = model
|
|
|
|
@staticmethod
|
|
def _extract_json(text: str) -> dict | None:
|
|
if not text:
|
|
return None
|
|
m = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", text, re.DOTALL)
|
|
if m:
|
|
try:
|
|
return json.loads(m.group(1))
|
|
except json.JSONDecodeError:
|
|
pass
|
|
m = re.search(r"\{.*\}", text, re.DOTALL)
|
|
if m:
|
|
try:
|
|
return json.loads(m.group(0))
|
|
except json.JSONDecodeError:
|
|
return None
|
|
return None
|
|
|
|
@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)
|
|
if resp.status_code != 200:
|
|
logger.warning("[vision] fetch %s status=%s", url, resp.status_code)
|
|
return None
|
|
mime = resp.headers.get("content-type", "").split(";")[0].strip()
|
|
# 실제 이미지가 아니면 거부 (HTML 페이지가 404 대신 200으로 리다이렉트 되는 경우)
|
|
if not mime.startswith("image/"):
|
|
logger.warning("[vision] %s not an image (content-type=%s)", url, mime)
|
|
return None
|
|
size = len(resp.content)
|
|
if size < 500:
|
|
logger.warning("[vision] %s too small (%d bytes) — likely placeholder", url, size)
|
|
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
|
|
|
|
async def _ask(self, image_urls: list[str], prompt: str, max_tokens: int = 4000) -> dict | None:
|
|
content: list[dict] = []
|
|
for u in image_urls:
|
|
if not u:
|
|
continue
|
|
data_url = await self._fetch_as_data_url(u)
|
|
if not data_url:
|
|
continue
|
|
content.append({"type": "image_url", "image_url": {"url": data_url}})
|
|
if not any(c.get("type") == "image_url" for c in content):
|
|
logger.warning("[vision] no images could be fetched")
|
|
return None
|
|
content.append({"type": "text", "text": prompt})
|
|
|
|
try:
|
|
resp = await self.client.chat.completions.create(
|
|
model=self.model,
|
|
messages=[{"role": "user", "content": content}],
|
|
max_tokens=max_tokens,
|
|
)
|
|
choice = resp.choices[0]
|
|
if choice.finish_reason != "stop":
|
|
logger.warning("[vision] unexpected finish_reason=%s", choice.finish_reason)
|
|
return self._extract_json(choice.message.content or "")
|
|
except Exception as e:
|
|
logger.warning("[vision] error: %s", e)
|
|
return None
|
|
|
|
async def analyze_brand_assets(
|
|
self,
|
|
logo_url: str | None,
|
|
homepage_url: str | None,
|
|
additional_images: list[str] | None = None,
|
|
) -> dict:
|
|
"""로고 이미지를 보고 정성 분석. 정확한 hex는 color_extractor가 따로 처리하므로 여기선 안 뽑음."""
|
|
urls = [u for u in [logo_url] + list(additional_images or []) if u]
|
|
if not urls:
|
|
return {}
|
|
prompt = (
|
|
"당신은 브랜드 로고 시각 분석가입니다. 첨부된 이미지(첫 번째가 병원의 대표 로고)를 보고 "
|
|
"아래 JSON 스키마로만 응답하세요. 코드펜스 없이 순수 JSON만 출력.\n"
|
|
"{\n"
|
|
' "logo_description": "로고를 1~2문장으로 설명 (심볼 형태 + 워드마크 + 전반적 톤). 예: \'둥근 잎사귀를 감싼 추상 심볼에 세리프 한글 워드마크, 차분하고 고급스러운 톤\'",\n'
|
|
' "logo_style": "minimal | illustrative | typographic | abstract 중 하나",\n'
|
|
' "has_symbol": "심볼/아이콘이 있으면 true, 글자만 있으면 false (boolean)",\n'
|
|
' "logo_symbol": "심볼이 묘사하는 대상 (예: \'잎사귀\', \'추상 곡선\'). 없으면 빈 문자열",\n'
|
|
' "logo_text": "로고에 보이는 워드마크 텍스트 그대로 (한글/영문). 없으면 빈 문자열",\n'
|
|
' "logo_colors_desc": "로고에 쓰인 색감을 사람이 부르는 이름으로 서술 (예: \'딥네이비 + 골드\'). 정확한 hex는 출력하지 말 것"\n'
|
|
"}\n"
|
|
"주의: 색상 hex 값이나 logo URL 같은 필드는 출력하지 마세요 (별도 추출 로직이 처리).\n"
|
|
"모든 설명/텍스트 값은 반드시 한국어로 작성하세요 (영어 금지)."
|
|
)
|
|
result = await self._ask(urls, prompt)
|
|
if not result:
|
|
return {}
|
|
# logo_images는 우리가 직접 채움 (Vision은 묘사만)
|
|
result["logo_images"] = {"circle": None, "horizontal": logo_url, "korean": None}
|
|
return result
|
|
|
|
async def describe_channel_logos(
|
|
self,
|
|
official_logo_url: str | None,
|
|
channel_logos: list[dict],
|
|
) -> dict | None:
|
|
"""채널별 프로필 이미지(로고)를 보고 각각 설명 + 공식 로고와 일치 여부 평가.
|
|
channel_logos: [{"channel": "Instagram", "url": "..."}, ...]
|
|
반환: {"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
|
|
|
|
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:
|
|
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 = "", ""
|
|
|
|
return {
|
|
"channel_logos": channel_logos_out,
|
|
"inconsistency_summary": summary,
|
|
"recommendation": rec,
|
|
}
|