o2o-site-AEO/solution/backend/services/agent/tools.py

"""도구 레지스트리 — 에이전트가 할 수 있는 일의 **전부**가 여기 있다.

★★ 도구는 반드시 `services/*` 를 통과한다. `crud`·`models` 를 직접 부르면 업종 스키마
   검증 · 출처 필수 · 정정본 보호 · 소유자 범위가 통째로 사라지는데, **아무 증상이 없다** —
   값은 들어가고 빌드는 성공하고 화면도 뜬다. `collect_service.store_facts` 가
   "크롤러가 우회할 수 있는 뒷문을 만들지 않는다" 로 막아 둔 그 문이고, 에이전트에게만
   열어 줄 이유가 없다.

★ 결과 문구는 도구가 만든다. LLM 이 쓰게 두면 **하지 않은 일을 했다고 말할 수 있고**,
  사장님에게는 그 말이 사실로 보인다.

★ 등급은 여기서 못 박는다. LLM 이 정하게 두면 프롬프트에 끼어든 한 줄이 확인 절차를
  건너뛴다 — 되돌릴 수 없는 행위일수록 그 값을 모델에 맡기면 안 된다.
"""

import uuid
from dataclasses import dataclass, field
from enum import Enum
from typing import Awaitable, Callable

from common.category_schema.loader import get_schema
from common.enums import ErrorType, PlaceCategory, SourceType
from common.models.gmodel import UserInfo
from crud.fact_crud import FactCRUD
from crud.job_crud import JobQueue
from crud.place_crud import PlaceCRUD
from crud.site_crud import SiteCRUD
from router.v1.fact.protocol import Req_UpsertFact
from router.v1.site.protocol import Req_StartBuild
from services import site_payload
from services.fact_service import FactService
from services.site_service import SiteService


class ToolGrade(str, Enum):
    """되돌릴 수 있느냐가 승인 강도를 정한다 — 분류가 아니라 동작을 가르는 값이다."""

    READ = "READ"              # 승인 없음
    REVERSIBLE = "REVERSIBLE"  # 실행하고 알린다. 사장님이 다시 고치면 된다
    SEMI = "SEMI"              # 실행 전에 한 번 묻는다(되돌릴 수는 있으나 그 사이 밖에서 읽힌다)


@dataclass
class ToolContext:
    user: UserInfo
    place_id: str
    place: object


@dataclass
class Tool:
    name: str
    grade: ToolGrade
    summary: str
    args: dict = field(default_factory=dict)
    run: Callable[[ToolContext, dict], Awaitable[str]] = None
    # SEMI 도구가 실행 전에 사장님께 보일 문장.
    confirm: str = ""


def _services():
    """서비스는 매 호출 새로 만든다 — 라우터가 Depends 로 받는 것과 같은 수명이다.

    ★ Depends 기본값에 기대지 않고 의존을 손으로 넣는다. FastAPI 밖에서 부르면
      기본값이 `Depends(...)` 객체 그대로라 서비스가 조용히 엉뚱한 것을 들고 돈다."""
    place_crud = PlaceCRUD()
    return FactService(FactCRUD(), place_crud), SiteService(SiteCRUD(), place_crud, JobQueue())


# ── 읽기 ────────────────────────────────────────────────────────────────

async def _get_site_status(ctx: ToolContext, args: dict) -> str:
    _fact, site_service = _services()
    res = await site_service.get_site(ctx.user, ctx.place_id)
    site = res.site
    if site is None or site.published_at is None:
        return "아직 발행 전입니다. 준비가 되면 발행해 드릴게요."
    # ★ 주소는 site_payload 의 함수로 만든다. 문자열로 조립하면 canonical 과 갈린다
    #   (CLAUDE.md '슬러그 규칙은 두 곳에 있고 같아야 한다').
    url = f"{site_payload.publish_origin()}/s/{site_payload.publish_slug(ctx.place, site)}"
    when = site.published_at.strftime("%Y-%m-%d %H:%M")
    return f"발행되어 있습니다.\n주소: {url}\n마지막 발행: {when}"


async def _list_facts(ctx: ToolContext, args: dict) -> str:
    fact_service, _site = _services()
    res = await fact_service.list_facts(ctx.user, ctx.place_id, publishable_only=True)
    rows = [f for f in (res.facts or []) if (f.value or "").strip()]
    schema = get_schema(PlaceCategory(ctx.place.category))
    keyword = (args.get("keyword") or "").strip()
    if keyword:
        rows = [f for f in rows if keyword in f.key or keyword in ((schema.get(f.key).label if schema.get(f.key) else ""))]
    if not rows:
        return "저장된 가게 정보가 아직 없습니다." if not keyword else f"'{keyword}' 로 찾은 정보가 없습니다."
    lines = []
    for f in rows[:20]:
        spec = schema.get(f.key)
        lines.append(f"· {spec.label if spec else f.key}: {f.value}")
    more = f"\n(그 밖에 {len(rows) - 20}개 더 있습니다)" if len(rows) > 20 else ""
    return "지금 저장된 정보입니다.\n" + "\n".join(lines) + more


# ── 되돌릴 수 있는 쓰기 ──────────────────────────────────────────────────

async def _set_fact(ctx: ToolContext, args: dict) -> str:
    key, value = (args.get("key") or "").strip(), (args.get("value") or "").strip()
    if not key or not value:
        raise ToolRejected("무엇을 어떤 값으로 바꿀지 알려 주세요.")

    schema = get_schema(PlaceCategory(ctx.place.category))
    spec = schema.get(key)
    # ★ LLM 이 없는 key 를 지어낼 수 있다. 스키마가 최종 판정이다.
    if spec is None:
        raise ToolRejected("그 항목은 이 가게에서 쓰지 않는 정보라 고칠 수 없어요.")
    if spec.scope != "place":
        raise ToolRejected(f"{spec.label} 은 객실·메뉴마다 다른 값이라 대화로는 아직 고칠 수 없어요.")

    fact_service, _site = _services()
    # ★ FactService 를 그대로 통과시킨다. source_type=OWNER 라 노출값을 즉시 교체하고,
    #   정정본 잠금·업종 스키마 검증이 전부 거기서 걸린다.
    res = await fact_service.upsert_fact(
        ctx.user, ctx.place_id, Req_UpsertFact(key=key, value=value, source_type=SourceType.OWNER)
    )
    if not res.result.success:
        raise ToolRejected("그 값을 저장하지 못했습니다. 형식을 확인해 주세요.")

    # ★ fact 는 바뀌었지만 사이트는 안 바뀐다. 이 한 줄이 빠지면 사장님은 반영된 줄 알고
    #   확인하러 갔다가 옛 값을 보고 "고장났네" 가 된다.
    return f"{spec.label} 을(를) {value} 로 바꿨습니다. 사이트에 반영하려면 다시 발행해야 해요 — 지금 할까요?"


# ── 반쯤 되돌릴 수 있는 것 ───────────────────────────────────────────────

async def _publish(ctx: ToolContext, args: dict) -> str:
    _fact, site_service = _services()
    res = await site_service.start_build(ctx.user, ctx.place_id, Req_StartBuild(publish=True))
    if not res.result.success:
        if res.result.code == ErrorType.PLACE_NOT_VERIFIED.value:
            raise ToolRejected("가게 확인이 끝나지 않아 발행할 수 없어요. 빌더 화면에서 가게 정보를 먼저 확인해 주세요.")
        raise ToolRejected("발행을 시작하지 못했습니다. 빌더 화면에서 확인해 주세요.")
    return "발행을 시작했습니다. 1분쯤 걸리고, 끝나면 사이트에 반영됩니다."


class ToolRejected(RuntimeError):
    """도구가 실행을 거절했다 — 사장님께 그대로 보여 줄 한국어 문장을 담는다."""


REGISTRY: dict[str, Tool] = {
    t.name: t
    for t in [
        Tool(
            name="get_site_status",
            grade=ToolGrade.READ,
            summary="홈페이지가 발행됐는지, 주소와 마지막 발행 시각을 알려준다.",
            run=_get_site_status,
        ),
        Tool(
            name="list_facts",
            grade=ToolGrade.READ,
            summary="지금 저장된 가게 정보를 보여준다.",
            args={"keyword": "찾고 싶은 항목이 있으면 그 말(선택)"},
            run=_list_facts,
        ),
        Tool(
            name="set_fact",
            grade=ToolGrade.REVERSIBLE,
            summary="가게 정보 한 항목을 고친다. 사이트에 반영되려면 발행이 따로 필요하다.",
            args={"key": "아래 항목 목록의 key", "value": "바꿀 값"},
            run=_set_fact,
        ),
        Tool(
            name="publish",
            grade=ToolGrade.SEMI,
            summary="바뀐 내용을 홈페이지에 반영한다(재발행).",
            run=_publish,
            confirm="지금 홈페이지를 다시 발행할까요? 바뀐 내용이 손님에게 보이게 됩니다.",
        ),
    ]
}


def describe() -> list[dict]:
    """프롬프트에 실을 도구 목록. ★ 등급은 싣지 않는다 — 모델이 알 필요도, 정할 이유도 없다."""
    return [{"name": t.name, "설명": t.summary, "args": t.args} for t in REGISTRY.values()]


def fields_of(place) -> list[dict]:
    schema = get_schema(PlaceCategory(place.category))
    return [
        {"key": k, "label": spec.label, "type": spec.type}
        for k, spec in schema.fields.items()
        if spec.scope == "place"
    ]