o2o-site-AEO/solution/backend/services/copy_service.py

"""소개문·FAQ 생성 — COPY 잡이 하는 일.

★ LLM 은 사실을 만들지 않는다. 문장만 쓴다.
  - 입력은 **확보된 fact(노출 가능한 것)만**. 미검증 값으로 문장을 쓰면 그 문장도 미검증이다.
  - 생성물은 `ground_check` 를 통과한 것만 저장한다(클라이언트가 이미 걸러 보내지만 근거를 다시 요구한다).
  - 소개문은 fact 로 들어가되 **UNVERIFIED 후보**다. 사람이 승인해야 사이트에 나간다.
  - FAQ 도 같은 검증 흐름을 탄다. 근거 fact 가 없는 FAQ 는 저장하지 않는다.
"""
import uuid

from common.category_schema import CategorySchemaError, get_schema
from common.database.db_session_manager import DB_SESSION_MNG
from common.database.model.models import facts, faqs, place_links, places, units
from common.enums import (
    PUBLISHABLE_FACT_STATUSES,
    DBWRType,
    ErrorType,
    FactStatus,
    PlaceCategory,
    SourceType,
)
from common.logger import LOG
from common.models.gmodel import UserInfo
from common.utils.gtime import GTime
from config.server_configs import external_api_config
from crud.fact_crud import FactCRUD
from crud.faq_crud import FaqCRUD
from crud.place_crud import PlaceCRUD
from router.v1.fact.protocol import Req_UpsertFact
from services.external import gemini_text
from services.fact_service import FactService

_fact_crud = FactCRUD()
_faq_crud = FaqCRUD()
_place_crud = PlaceCRUD()


class CopyAborted(RuntimeError):
    """재시도해도 소용없는 중단 — 잡의 last_error 로 남는다."""


async def run_copy(job: dict) -> dict:
    """COPY 잡 핸들러. payload: {place_id, company_id, requested_by?}"""
    payload = job["payload"]
    place_id = payload["place_id"]
    company_id = payload["company_id"]

    if not gemini_text.is_configured():
        raise CopyAborted("GEMINI_API_KEY 미설정 — 소개문·FAQ 를 생성할 수 없다")

    err, place = await DB_SESSION_MNG.execute_lambda(
        places.DBType(),
        DBWRType.DB_READ.value,
        lambda s: _place_crud.get_place(s, uuid.UUID(company_id), uuid.UUID(place_id)),
    )
    if err != ErrorType.SUCCESS or place is None:
        raise CopyAborted(f"사업장을 찾을 수 없다: {place_id}")

    try:
        schema = get_schema(PlaceCategory(place.category))
    except (CategorySchemaError, ValueError) as ex:
        raise CopyAborted(f"지원하지 않는 업종: {place.category}") from ex

    # ★ 노출 가능한 fact 만 근거로 준다. 미검증 값으로 쓴 문장은 그 자체가 미검증이다.
    pid = uuid.UUID(place_id)
    f_err, fact_rows = await DB_SESSION_MNG.execute_lambda(
        facts.DBType(),
        DBWRType.DB_READ.value,
        lambda s: _fact_crud.list_facts(s, pid, None, None, True, True),
    )
    if f_err != ErrorType.SUCCESS:
        raise CopyAborted(f"fact 조회 실패: {f_err.name}")

    grounded = [
        gemini_text.FactInput(
            key=r.key,
            label=(schema.get(r.key).label if schema.get(r.key) else r.key),
            value=r.value,
            unit=r.unit,
        )
        for r in fact_rows
        if r.unit_id is None and (r.value or "").strip()
    ]
    # ★ 수집 원문도 근거로 넘긴다 — fact 가 아니라 place_links.raw 에 박제된 글이다.
    #
    #   왜 필요한가: 소개 원문(TourAPI overview·네이버 description)에만 있는 정보가 있다.
    #   '전면 통창 실내 온수풀', '판교역에서 3분' 같은 것들인데, 이게 근거에 없으면
    #   ground_check 가 그 문장을 전부 반려해 소개문·FAQ 가 앙상해진다.
    #
    #   왜 fact 로 넣지 않는가: `intro` 는 allow_llm=True 라 LLM 의 출력 칸이다.
    #   원문을 그 칸에 넣었더니 457자 원문이 발행본의 '숙소 소개' 를 차지했다(2026-08-31).
    #   근거로만 쓰고 저장은 하지 않는다 — 원문은 화면에 나가지 않는다.
    l_err, link_rows = await DB_SESSION_MNG.execute_lambda(
        place_links.DBType(),
        DBWRType.DB_READ.value,
        lambda s: _place_crud.list_links(s, pid, True),
    )
    if l_err == ErrorType.SUCCESS:
        for link in (link_rows or []):
            text = ((link.raw or {}).get("text") or "").strip() if isinstance(link.raw, dict) else ""
            if text:
                grounded.append(gemini_text.FactInput(
                    key=f"source:{link.link_id}", label="수집 원문", value=text[:4000],
                ))

    # 객실·메뉴 요약도 근거로 넘긴다 — "최대 4명" 같은 수치가 통과하려면 근거에 있어야 한다.
    #
    # ★ 근거 없음 판정보다 **먼저** 읽는다.
    #   예전에는 사업장 fact 가 0건이면 여기까지 오지 못하고 되돌아갔다. 그런데 네이버에
    #   요금표만 올라온 모텔은 사업장 fact 가 0건이고 객실 fact 만 있다 — 쓸 근거가 있는데도
    #   "근거 없음"으로 끝나 소개문·FAQ 가 영구히 생기지 않았다.
    u_err, unit_rows = await DB_SESSION_MNG.execute_lambda(
        units.DBType(),
        DBWRType.DB_READ.value,
        lambda s: _place_crud.list_units(s, pid),
    )
    unit_summaries = []
    if u_err == ErrorType.SUCCESS:
        by_unit: dict = {}
        for r in fact_rows:
            if r.unit_id and (r.value or "").strip():
                by_unit.setdefault(str(r.unit_id), {})[r.key] = r.value
        unit_summaries = [
            {
                "name": u.name,
                "facts": by_unit.get(str(u.unit_id), {}),
                # 스키마 라벨·단위를 같이 넘긴다 — 이게 없으면 프롬프트에 'weekday_price' 라는
                # 날 key 가 그대로 실려 모델이 그 낱말로 문장을 쓴다.
                "labels": {
                    key: {
                        "label": schema.get(key).label if schema.get(key) else key,
                        "unit": schema.get(key).unit if schema.get(key) else None,
                    }
                    for key in by_unit.get(str(u.unit_id), {})
                },
            }
            for u in unit_rows
            if by_unit.get(str(u.unit_id))
        ]

    if not grounded and not unit_summaries:
        # 근거가 없으면 아무것도 쓰지 않는다 — 유료 호출조차 하지 않는다.
        return {"place_id": place_id, "intro": False, "faqs": 0, "note": "근거로 쓸 확인된 fact 가 없다"}

    try:
        copy = await gemini_text.generate_copy(
            place.name,
            PlaceCategory(place.category),
            grounded,
            unit_summaries=unit_summaries or None,
            model=external_api_config.gemini_text_model,
        )
    except gemini_text.GeminiNotConfigured as ex:
        raise CopyAborted(str(ex)) from ex

    now = GTime.UTC()
    stat = {
        "place_id": place_id,
        "grounded_facts": len(grounded),
        "intro": False,
        "meta": False,
        "faqs": 0,
        # ★ 반려된 문장을 그대로 남긴다 — 소개문이 왜 안 나왔는지 운영자가 알아야 한다.
        "rejected": [list(r) for r in (copy.rejected or [])][:20],
    }

    actor = UserInfo(
        user_id=payload.get("requested_by") or str(place.verified_by or uuid.uuid4()),
        id="generator",
        company_id=company_id,
        role=1,
    )
    service = FactService(_fact_crud, _place_crud)

    # 소개문·메타는 fact 로 들어간다 — FactService 가 allow_llm 을 다시 확인한다(뒷문 없음).
    for key, text_value in (("intro", copy.intro), ("meta_description", copy.meta_description)):
        if not (text_value or "").strip():
            continue
        if not (schema.get(key) and schema.get(key).allow_llm):
            # 이 업종 스키마가 LLM 작성을 허용하지 않는 필드다. 조용히 건너뛴다.
            continue
        res = await service.upsert_fact(
            actor, place_id,
            Req_UpsertFact(
                key=key, value=text_value.strip(),
                source_type=SourceType.LLM, source_url=f"gemini:{external_api_config.gemini_text_model}",
            ),
        )
        if res.result.success:
            stat["intro" if key == "intro" else "meta"] = True
        else:
            stat["rejected"].append([key, res.result.desc])

    # 확인 안 된 기존 생성 FAQ 는 내리고 새로 넣는다. 사람이 확인한 FAQ 는 건드리지 않는다.
    await DB_SESSION_MNG.execute_lambda_claim(
        faqs.DBType(),
        lambda s: _faq_crud.expire_generated(s, pid, now),
    )
    for order, faq in enumerate(copy.faqs or []):
        if not faq.fact_keys:
            # ★ 근거 없는 FAQ 는 저장하지 않는다.
            stat["rejected"].append([faq.question, "근거 fact 없음"])
            continue
        row = faqs(
            place_id=pid,
            question=faq.question,
            answer=faq.answer,
            source_fact_ids=list(faq.fact_keys),
            generated_by=SourceType.LLM.value,
            status=FactStatus.UNVERIFIED.value,
            sort_order=order,
        )
        run_err = await DB_SESSION_MNG.execute_lambda_run(
            [faqs.DBType()],
            [lambda s, r=row: _faq_crud.add_faq(s, r)],
        )
        if run_err == ErrorType.SUCCESS:
            stat["faqs"] += 1

    LOG.i(f"[copy] place={place_id} 소개문 {'O' if stat['intro'] else 'X'} · FAQ {stat['faqs']}건 · "
          f"반려 {len(stat['rejected'])}건 (근거 fact {len(grounded)}개)")
    return stat