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

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

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 로 받는 것과 같은 수명이다."""
    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 의 함수로 만든다.
    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 를 그대로 통과시킨다.
    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"
    ]