o2o-site-AEO/solution/backend/services/external/gemini.py

"""사진 라벨·접근성 alt 생성 — 겹들을 엮어 결과를 만드는 자리.

이 파일이 하는 일은 **엮는 것뿐**이다. 고칠 것이 생기면 해당 겹으로 바로 간다:

    무엇을 묻는가      services/prompts/vision.py   프롬프트·응답 스키마
    어떻게 부르는가    services/llm/gemini.py       HTTP·재시도·토큰·비용
    무엇을 돌려주는가  여기                          배치 나누기 → 호출 → ref 매칭 → 조립

결과는 순서가 아니라 `ref` 로 매칭하고, 신뢰도가 낮으면 사람 확인 대상으로 남긴다.
(문장 생성과 달리 여기엔 grounding 겹이 없다 — 사진 설명은 대조할 fact 가 없고,
 대신 신뢰도 임계값과 사람 확인 큐가 그 몫을 한다.)
"""
import base64
import json
from dataclasses import dataclass, field
from typing import Optional

import httpx

from common.enums import PlaceCategory
from common.logger import LOG
from services.llm.gemini import (
    DEFAULT_MODEL,
    GeminiError,
    GeminiInvalidOutput,
    GeminiNotConfigured,
    Usage,
    call,
    extract_text,
    is_configured,
    price,
    read_usage,
)
from services.prompts.vision import RESPONSE_SCHEMA, build_prompt

# URL 확장자 대신 파일 시그니처로 MIME을 판별한다.
_MAGIC = (
    (b"\x89PNG\r\n\x1a\n", "image/png"),
    (b"\xff\xd8\xff", "image/jpeg"),
    (b"GIF87a", "image/gif"),
    (b"GIF89a", "image/gif"),
)


@dataclass
class ImageInput:
    """분석할 사진 1장. origin_url 이 결과 매칭 키다(place.media.origin_url 과 같은 값)."""

    origin_url: str
    data: Optional[bytes] = None       # 이미 받아둔 바이트. 없으면 fetch_url 에서 내려받는다
    fetch_url: Optional[str] = None    # 내려받을 주소. 비면 origin_url 을 쓴다
    mime_type: Optional[str] = None    # 없으면 시그니처로 판별
    unit_name_hint: Optional[str] = None  # "A동 스탠다드" 같은 힌트가 있으면 라벨에 반영


@dataclass
class VisionResult:
    """사진 1장의 분석 결과. 입력과 1:1 로 대응하며 실패해도 자리를 지킨다."""

    origin_url: str
    label: Optional[str] = None
    alt_text: Optional[str] = None
    confidence: float = 0.0
    needs_review: bool = True   # ★ 기본이 '사람 확인 필요'다. 확신이 있을 때만 내려간다
    ok: bool = False
    error: Optional[str] = None


@dataclass
class _Usage:
    input_tokens: int = 0
    output_tokens: int = 0
    batches: int = 0
    failed_batches: int = 0
    by_model: dict = field(default_factory=dict)


def _sniff_mime(data: bytes) -> str:
    for magic, mime in _MAGIC:
        if data.startswith(magic):
            return mime
    if data[:4] == b"RIFF" and data[8:12] == b"WEBP":
        return "image/webp"
    return "image/jpeg"   # 판별 실패 시 가장 흔한 형식으로 시도한다


async def _load_bytes(client: httpx.AsyncClient, image: ImageInput) -> bytes:
    """이미지 바이트 확보. 이미 있으면 그대로, 없으면 내려받는다.

    직접 내려받는 이유: 타임아웃을 우리가 통제하고, 사진별 실패를 개별로 보고하기 위해서다."""
    if image.data:
        return image.data
    url = image.fetch_url or image.origin_url
    resp = await client.get(url, follow_redirects=True)
    if resp.status_code != 200:
        raise GeminiError(f"이미지 내려받기 실패 status={resp.status_code}")
    return resp.content


async def _run_batch(
    client: httpx.AsyncClient,
    batch: list[ImageInput],
    *,
    model: str,
    category: Optional[PlaceCategory],
    unit_names: Optional[list[str]],
    confidence_threshold: float,
    max_retries: int,
    usage: _Usage,
) -> dict[str, VisionResult]:
    """배치 1개 처리. 반환 {origin_url: VisionResult}.

    ★ 배치가 통째로 실패해도 예외를 밖으로 던지지 않는다 — 호출측이 나머지 배치를 계속 돌려야 한다."""
    out: dict[str, VisionResult] = {}
    ref_map: dict[str, ImageInput] = {}
    parts: list[dict] = [{"text": ""}]   # 자리를 잡아두고 프롬프트는 아래에서 채운다

    for i, image in enumerate(batch):
        ref = f"img-{i}"
        try:
            data = await _load_bytes(client, image)
        except Exception as ex:
            # 이 사진만 실패. 배치의 나머지는 그대로 보낸다.
            out[image.origin_url] = VisionResult(
                origin_url=image.origin_url, ok=False, needs_review=True,
                error=f"이미지 로드 실패: {type(ex).__name__}: {ex}",
            )
            continue
        ref_map[ref] = image
        parts.append({"text": f"[{ref}]" + (f" (힌트: {image.unit_name_hint})" if image.unit_name_hint else "")})
        parts.append({
            "inline_data": {"mime_type": image.mime_type or _sniff_mime(data), "data": base64.b64encode(data).decode()}
        })

    if not ref_map:
        return out

    parts[0] = {"text": build_prompt(category, unit_names, sorted(ref_map))}
    body = {
        "contents": [{"role": "user", "parts": parts}],
        "generationConfig": {
            "responseMimeType": "application/json",
            "responseSchema": RESPONSE_SCHEMA,
            "temperature": 0,
        },
    }

    usage.batches += 1
    try:
        payload = await call(client, model, body, max_retries)
        parsed = json.loads(extract_text(payload))
    except GeminiNotConfigured:
        raise                       # 키 문제는 전체를 중단시킨다 — 나머지 배치도 어차피 실패한다
    except Exception as ex:
        usage.failed_batches += 1
        LOG.w(f"[gemini] 배치 실패(계속) {len(ref_map)}장: {type(ex).__name__}: {ex}")
        for image in ref_map.values():
            out[image.origin_url] = VisionResult(
                origin_url=image.origin_url, ok=False, needs_review=True,
                error=f"{type(ex).__name__}: {str(ex)[:200]}",
            )
        return out

    batch_usage = read_usage(payload)
    usage.input_tokens += batch_usage.input_tokens
    usage.output_tokens += batch_usage.output_tokens

    # ★ 순서가 아니라 ref 로 매칭한다.
    seen: set[str] = set()
    for item in parsed.get("items") or []:
        ref = str(item.get("ref", "")).strip()
        image = ref_map.get(ref)
        if image is None:
            continue            # 모델이 없는 ref 를 지어냈다 — 버린다
        seen.add(ref)
        try:
            confidence = float(item.get("confidence") or 0.0)
        except (TypeError, ValueError):
            confidence = 0.0
        confidence = max(0.0, min(1.0, confidence))
        label = (item.get("label") or "").strip() or None
        alt = (item.get("alt_text") or "").strip() or None
        out[image.origin_url] = VisionResult(
            origin_url=image.origin_url,
            label=label,
            alt_text=alt,
            confidence=confidence,
            # ★ 신뢰도 미달이거나 라벨/alt 가 비면 자동 반영하지 않는다.
            needs_review=confidence < confidence_threshold or not label or not alt,
            ok=True,
        )

    # 응답에서 빠진 사진 — 조용히 사라지지 않게 자리를 채운다.
    for ref, image in ref_map.items():
        if ref not in seen:
            out[image.origin_url] = VisionResult(
                origin_url=image.origin_url, ok=False, needs_review=True,
                error="응답에 해당 ref 가 없다",
            )
    return out


async def analyze_images(
    images: list[ImageInput],
    *,
    category: Optional[PlaceCategory] = None,
    unit_names: Optional[list[str]] = None,
    model: str = DEFAULT_MODEL,
    batch_size: int = 10,
    confidence_threshold: float = 0.7,
    max_retries: int = 2,
    client: Optional[httpx.AsyncClient] = None,
) -> list[VisionResult]:
    """사진들을 분류하고 alt 를 만든다.

    ★ 반환 길이는 항상 입력과 같다. 실패분도 ok=False 로 자리를 지킨다 —
      호출측이 길이나 순서로 매칭하다 어긋나면 엉뚱한 사진에 alt 가 붙는다.
    ★ needs_review=True 인 항목은 자동 반영하지 말고 사람 확인 큐(MediaStatus.PENDING_REVIEW)로 보낸다.
    """
    if not is_configured():
        raise GeminiNotConfigured("GEMINI_API_KEY 가 설정되지 않았다")
    if not images:
        return []

    owns_client = client is None
    # 사진 여러 장을 배치로 보내므로 타임아웃을 넉넉히 잡는다.
    client = client or httpx.AsyncClient(timeout=httpx.Timeout(180.0, connect=10.0))
    usage = _Usage()
    merged: dict[str, VisionResult] = {}

    try:
        for start in range(0, len(images), max(1, batch_size)):
            batch = images[start:start + max(1, batch_size)]
            merged.update(await _run_batch(
                client, batch,
                model=model, category=category, unit_names=unit_names,
                confidence_threshold=confidence_threshold, max_retries=max_retries, usage=usage,
            ))
    finally:
        if owns_client:
            await client.aclose()

    results = [
        merged.get(img.origin_url)
        or VisionResult(origin_url=img.origin_url, ok=False, needs_review=True, error="결과 누락")
        for img in images
    ]

    ok = sum(1 for r in results if r.ok)
    review = sum(1 for r in results if r.needs_review)
    LOG.i(
        f"[gemini] 사진분석 {len(results)}장 (성공 {ok} · 확인필요 {review}) · "
        f"배치 {usage.batches}(실패 {usage.failed_batches}) · model={model} · "
        f"tokens in={usage.input_tokens} out={usage.output_tokens} · 약 ${price(model, Usage(usage.input_tokens, usage.output_tokens))}"
    )
    return results