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

"""소개문·메타설명·FAQ 생성 — 겹들을 엮어 결과를 만드는 자리.

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

    무엇을 묻는가      services/prompts/copy.py        프롬프트·응답 스키마
    어떻게 부르는가    services/llm/gemini.py          HTTP·재시도·토큰·비용
    답을 믿을 것인가   services/grounding/copy.py      ground_check · faq_polarity_ok
    무엇을 돌려주는가  여기                            근거 모으기 → 호출 → 검증 → 조립

한때 이 네 가지가 한 파일 500줄에 뭉쳐 있었다. "FAQ 답이 이상하다" 를 고치러 와도
어디를 봐야 할지가 파일 안에서 갈리지 않았다.
"""
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.grounding.copy import FactInput, faq_polarity_ok, ground_check
from services.llm.gemini import (
    DEFAULT_MODEL as DEFAULT_TEXT_MODEL,
    GeminiError,
    GeminiInvalidOutput,
    GeminiNotConfigured,
    Usage,
    call,
    extract_text,
    is_configured,
    price,
    read_usage,
)
from services.prompts.copy import RESPONSE_SCHEMA, build_prompt


@dataclass
class GeneratedFaq:
    question: str
    answer: str
    fact_keys: list[str] = field(default_factory=list)


@dataclass
class GeneratedCopy:
    """생성 결과. 검증을 통과한 것만 담긴다.

    rejected 에는 (버린 내용, 사유) 가 들어간다 — 조용히 버리지 않는다.
    운영자가 "왜 소개문이 안 나왔나" 를 이 목록으로 읽는다."""

    intro: Optional[str] = None
    intro_fact_keys: list[str] = field(default_factory=list)
    meta_description: Optional[str] = None
    faqs: list[GeneratedFaq] = field(default_factory=list)
    rejected: list[tuple[str, str]] = field(default_factory=list)


def _unit_facts(unit_summaries: Optional[list[dict]]) -> list[FactInput]:
    """객실·프로그램 요약을 근거 fact 로 펼친다.

    {"name": "A동", "facts": {"max_capacity": "4"}} → FactInput("A동:max_capacity", …)
    이렇게 해야 "최대 4명" 같은 문장이 근거 있는 것으로 통과한다.

    ★ `labels` 가 함께 오면 스키마 라벨·단위를 쓴다({key: {"label","unit"}}).
      이 목록은 프롬프트에도 그대로 실리므로, 라벨이 없으면 모델이 'weekday_price' 라는
      날 key 를 보고 글을 쓴다 — "weekday_price는 20000입니다" 같은 문장이 나온다.
      없으면 지금까지처럼 key 를 라벨 자리에 둔다(호출측이 스키마를 모를 수 있다).
    """
    out: list[FactInput] = []
    for unit in unit_summaries or []:
        name = str(unit.get("name") or "").strip()
        labels = unit.get("labels") or {}
        if name:
            out.append(FactInput(key=f"unit:{name}", label="객실·프로그램명", value=name))
        for key, value in (unit.get("facts") or {}).items():
            if value is None or str(value).strip() == "":
                continue
            spec = labels.get(key) or {}
            out.append(FactInput(
                key=f"{name}:{key}" if name else key,
                label=spec.get("label") or key,
                value=str(value),
                unit=spec.get("unit"),
            ))
    return out


def _valid_keys(claimed: list, allowed: set[str]) -> list[str]:
    """모델이 적어준 근거 key 중 실제로 존재하는 것만 남긴다(없는 key 를 지어내기도 한다)."""
    return [k for k in (claimed or []) if isinstance(k, str) and k in allowed]


async def generate_copy(
    place_name: str,
    category: PlaceCategory,
    facts: list[FactInput],
    *,
    unit_summaries: Optional[list[dict]] = None,
    records: Optional[list[str]] = None,
    suggested_questions: Optional[list[str]] = None,
    max_faqs: int = 8,
    model: str = DEFAULT_TEXT_MODEL,
    max_retries: int = 2,
    client: Optional[httpx.AsyncClient] = None,
) -> GeneratedCopy:
    """확보된 fact 만으로 소개문·메타설명·FAQ 를 만든다.

    ★ facts 가 비면 **API 를 호출하지 않고** 빈 결과를 돌려준다 —
      근거 없이 문장을 쓰면 그게 곧 환각이다.
    ★ 생성 결과는 전부 ground_check 를 통과한 것만 담긴다. 통과 못 한 항목은 rejected 로 간다.
    ★ 생성 대상 필드는 업종 스키마의 allow_llm=True 인 것뿐이다(호출측이 필터링해서 넘긴다).
    """
    if not is_configured():
        raise GeminiNotConfigured("GEMINI_API_KEY 가 설정되지 않았다")
    # ★ 사업장 fact 가 없어도 객실·메뉴 근거가 있으면 쓴다. 요금표만 있는 모텔이 그 경우다 —
    #   "대실 20,000원" 은 근거 있는 사실이고, 손님이 가장 먼저 묻는 것이기도 하다.
    unit_grounding = _unit_facts(unit_summaries)
    if not facts and not unit_grounding:
        LOG.i(f"[gemini-text] '{place_name}' 근거 fact 0건 — 생성하지 않는다(호출 없음)")
        return GeneratedCopy(rejected=[("(전체)", "근거 fact 가 없다 — 생성하지 않았다")])

    # 검증에 쓸 근거 = 넘겨받은 fact + 객실 요약 + 상호명(상호에 숫자가 있어도 근거로 본다)
    grounding = list(facts) + unit_grounding
    grounding.append(FactInput(key="place_name", label="상호명", value=place_name))
    allowed_keys = {f.key for f in facts} | {f.key for f in grounding}

    body = {
        "contents": [{"role": "user", "parts": [{
            "text": build_prompt(place_name, category, facts, max_faqs, unit_grounding, records, suggested_questions)
        }]}],
        "generationConfig": {
            "responseMimeType": "application/json",
            "responseSchema": RESPONSE_SCHEMA,
            "temperature": 0.2,
        },
    }

    owns_client = client is None
    client = client or httpx.AsyncClient(timeout=httpx.Timeout(120.0, connect=10.0))
    try:
        payload = await call(client, model, body, max_retries)
        parsed = json.loads(extract_text(payload))
    except json.JSONDecodeError as ex:
        raise GeminiInvalidOutput(f"구조화 출력 파싱 실패: {ex}") from ex
    finally:
        if owns_client:
            await client.aclose()

    usage = read_usage(payload)

    result = GeneratedCopy()

    # ── 소개문 ──
    intro = (parsed.get("intro") or "").strip()
    if intro:
        ok, reasons = ground_check(intro, grounding)
        if ok:
            result.intro = intro
            result.intro_fact_keys = _valid_keys(parsed.get("intro_fact_keys"), allowed_keys)
        else:
            result.rejected.append((intro, " / ".join(reasons)))

    # ── 메타 설명 ──
    meta_desc = (parsed.get("meta_description") or "").strip()
    if meta_desc:
        ok, reasons = ground_check(meta_desc, grounding)
        if ok:
            result.meta_description = meta_desc
        else:
            result.rejected.append((meta_desc, " / ".join(reasons)))

    # ── FAQ ── 항목마다 따로 검사한다. 하나가 걸려도 나머지는 산다.
    for item in (parsed.get("faqs") or [])[:max_faqs]:
        question = (item.get("question") or "").strip()
        answer = (item.get("answer") or "").strip()
        if not question or not answer:
            continue
        keys = _valid_keys(item.get("fact_keys"), allowed_keys)
        if not keys:
            # ★ 근거를 못 대는 FAQ 는 버린다 — 사실인지 확인할 방법이 없다.
            result.rejected.append((question, "근거 fact_keys 가 없다"))
            continue
        ok, reasons = ground_check(f"{question} {answer}", grounding)
        # 질문은 주장이 아니라 값-반대 판정에서 빠진다. 그 빈틈은 답변 쪽에서 따로 막는다.
        polar_ok, polar_reasons = faq_polarity_ok(question, answer, grounding)
        if not ok or not polar_ok:
            result.rejected.append((question, " / ".join(reasons + polar_reasons)))
            continue
        result.faqs.append(GeneratedFaq(question=question, answer=answer, fact_keys=keys))

    LOG.i(
        f"[gemini-text] '{place_name}' 생성 — 소개문 {'O' if result.intro else 'X'} · "
        f"메타 {'O' if result.meta_description else 'X'} · FAQ {len(result.faqs)}건 · "
        f"반려 {len(result.rejected)}건 · model={model} · "
        f"tokens in={usage.input_tokens} out={usage.output_tokens} · 약 ${price(model, usage)}"
    )
    return result


@dataclass
class GeneratedSong:
    """가사 생성 결과. 곡은 여기서 만들지 않는다 — 작곡은 services/external/suno 다."""

    title: str
    lyrics: str
    style: str


async def generate_song(
    place_name: str,
    category: PlaceCategory,
    *,
    region: str,
    grounding: list[str],
    intro: str = "",
    model: str = DEFAULT_TEXT_MODEL,
    max_retries: int = 2,
    client: Optional[httpx.AsyncClient] = None,
) -> GeneratedSong:
    """이 업소의 노래 가사를 쓴다.

    ★ `ground_check` 를 걸지 않는다. 가사는 사실 진술이 아니라 정서라 문장 단위로 근거를
      맞추면 전부 반려된다("밤이 깊어도 불이 켜져 있다" 에 대응하는 fact 는 없다).
      대신 프롬프트가 **없는 시설·숫자를 말하지 말라**고 못 박는다(services/prompts/song 머리주석).
    ★ 재료가 하나도 없으면 부르지 않는다 — 소개문과 같은 규칙이다. 상호와 지역만으로 쓴 노래는
      어느 숙소에 붙여도 말이 되는 노래이고, 그건 이 기능이 하려던 일이 아니다.
    """
    if not is_configured():
        raise GeminiNotConfigured("GEMINI_API_KEY 가 설정되지 않았다")
    if not grounding and not (intro or "").strip():
        raise GeminiInvalidOutput("가사를 쓸 재료가 없다 — 확인된 fact 도 소개문도 없다")

    from common.category_schema import get_schema
    from services.prompts.song import RESPONSE_SCHEMA as SONG_SCHEMA, build_prompt as build_song_prompt

    body = {
        "contents": [{"role": "user", "parts": [{
            "text": build_song_prompt(place_name, get_schema(category).label, region, grounding, intro)
        }]}],
        "generationConfig": {
            "responseMimeType": "application/json",
            "responseSchema": SONG_SCHEMA,
            # 소개문(0.2)보다 높다 — 노래는 정확해야 하는 글이 아니라 흥얼거릴 글이다.
            "temperature": 0.9,
        },
    }

    owns_client = client is None
    client = client or httpx.AsyncClient(timeout=httpx.Timeout(120.0, connect=10.0))
    try:
        payload = await call(client, model, body, max_retries)
        parsed = json.loads(extract_text(payload))
    except json.JSONDecodeError as ex:
        raise GeminiInvalidOutput(f"가사 파싱 실패: {ex}") from ex
    finally:
        if owns_client:
            await client.aclose()

    title = (parsed.get("title") or "").strip()
    lyrics = (parsed.get("lyrics") or "").strip()
    style = (parsed.get("style") or "").strip()
    if not lyrics:
        raise GeminiInvalidOutput("가사가 비어 있다")

    usage = read_usage(payload)
    LOG.i(
        f"[gemini-text] '{place_name}' 가사 — '{title}' ({style}) · {len(lyrics)}자 · "
        f"tokens in={usage.input_tokens} out={usage.output_tokens} · 약 ${price(model, usage)}"
    )
    # 제목이 비면 상호를 쓴다 — 빈 제목은 플레이어에서 빈 줄로 보인다.
    return GeneratedSong(title=title or place_name, lyrics=lyrics, style=style or "acoustic ballad")