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

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

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

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

한때 이 네 가지가 한 파일 500줄에 뭉쳐 있었다. "FAQ 답이 이상하다" 를 고치러 와도
어디를 봐야 할지가 파일 안에서 갈리지 않았다.
"""
import hashlib
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 import provider
from services.llm.errors import LlmError
from services.llm.errors import LlmInvalidOutput as GeminiInvalidOutput
from services.llm.errors import LlmNotConfigured as GeminiNotConfigured
from services.prompts.copy import RESPONSE_SCHEMA, build_prompt


def is_configured() -> bool:
    """호출측(copy_service.py, place_service.py 등)은 이 겹만 안다 — 어느 공급자가 활성인지는 몰라도 된다."""
    return provider.active().is_configured()


@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)
    source: str = ""  # ★ "openai:gpt-5.6-luna" 형식 — copy_steps.py 가 fact 출처 표기에 쓴다


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: Optional[str] = None,
    max_retries: int = 2,
    client: Optional[httpx.AsyncClient] = None,
) -> GeneratedCopy:
    """확보된 fact 만으로 소개문·메타설명·FAQ 를 만든다.

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

    prompt = build_prompt(place_name, category, facts, max_faqs, unit_grounding, records, suggested_questions)

    owns_client = client is None
    client = client or httpx.AsyncClient(timeout=httpx.Timeout(120.0, connect=10.0))
    try:
        llm_result = await llm.generate(
            client, model, prompt=prompt, response_schema=RESPONSE_SCHEMA, temperature=0.2, max_retries=max_retries,
        )
    finally:
        if owns_client:
            await client.aclose()

    parsed = llm_result.json
    usage = llm_result.usage

    result = GeneratedCopy(source=f"{llm.__name__.rsplit('.', 1)[-1]}:{model}")

    # ── 소개문 ──
    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"[llm-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} · 약 ${llm.price(model, usage)}"
    )
    return result


# ── 요약(summarize_text) ──────────────────────────────────────────────────
# ★ generate_copy 와 다르다: 여기서 압축하는 문장은 **이미 승인된 값**이다(fact 로 저장된 intro·
#   room_intro). 새 사실을 만드는 게 아니라 같은 내용을 짧게 쓰는 것뿐이라 ground_check 를 다시
#   걸지 않는다 — "사실을 더하지 마라"는 프롬프트 지시로 충분하다.
_SUMMARY_CACHE: dict[str, str] = {}
_SUMMARY_CACHE_MAX = 500
_SUMMARY_PROMPT = (
    "다음 숙소 소개에서 핵심 특징 1~2개만 골라 한국어 한 문장, 공백 포함 60~80자로 요약해줘. "
    "원문에 없는 사실이나 과장 표현을 추가하지 말고, 선택한 사실의 조건과 부정 표현을 유지해. "
    "반복되는 상호명, 인사말, 홍보 수식어는 생략해. "
    "요약문만 출력하고 다른 말은 붙이지 마.\n\n"
)


async def summarize_text(
    text: str,
    *,
    model: Optional[str] = None,
    max_retries: int = 2,
    client: Optional[httpx.AsyncClient] = None,
) -> Optional[str]:
    """캔버스 미리보기용 축약문. 실패해도 예외를 올리지 않는다 — 호출측은 None 이면 원문을 쓴다.

    ★ DB 에 남기지 않는다. 같은 원문은 프로세스 메모리 캐시(sha256 키)로 재호출을 막는다
      (서버 재시작하면 비워진다 — 요구사항: "DB 저장은 생략하고 프론트 응답에만 실어준다").
    """
    stripped = text.strip()
    if not stripped:
        return None
    llm = provider.active()
    if not llm.is_configured():
        return None
    model = model or llm.DEFAULT_MODEL

    # 길이 기준을 바꾼 뒤 이전 길이의 요약을 재사용하지 않도록 프롬프트도 키에 넣는다.
    cache_key = hashlib.sha256((_SUMMARY_PROMPT + stripped).encode("utf-8")).hexdigest()
    cached = _SUMMARY_CACHE.get(cache_key)
    if cached is not None:
        return cached

    owns_client = client is None
    client = client or httpx.AsyncClient(timeout=httpx.Timeout(60.0, connect=10.0))
    try:
        result = await llm.generate(client, model, prompt=_SUMMARY_PROMPT + stripped, temperature=0.2, max_retries=max_retries)
        summary = result.text.strip()
    except LlmError as ex:
        LOG.w(f"[llm-text] 요약 실패: {ex}")
        return None
    finally:
        if owns_client:
            await client.aclose()

    if not summary:
        return None
    if len(_SUMMARY_CACHE) >= _SUMMARY_CACHE_MAX:
        _SUMMARY_CACHE.clear()  # 간단한 캐시 상한 — 관리 도구 트래픽 규모에는 LRU 가 과하다.
    _SUMMARY_CACHE[cache_key] = summary
    return summary


@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: Optional[str] = None,
    max_retries: int = 2,
    client: Optional[httpx.AsyncClient] = None,
) -> GeneratedSong:
    """이 업소의 노래 가사를 쓴다.

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

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

    prompt = build_song_prompt(place_name, get_schema(category).label, region, grounding, intro)

    owns_client = client is None
    client = client or httpx.AsyncClient(timeout=httpx.Timeout(120.0, connect=10.0))
    try:
        llm_result = await llm.generate(
            client, model, prompt=prompt, response_schema=SONG_SCHEMA, temperature=0.9, max_retries=max_retries,
        )
    finally:
        if owns_client:
            await client.aclose()

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

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