hbyang
ea9ae1ac2a
"문구 변경밖에 안 된다" 는 지적에서 시작했다. 페이지 구성은 sites.theme.sections 배열 하나이고 배열 순서가 곧 발행본의 순서라, 그 JSON 을 만지는 도구 셋을 붙였다. - list_sections(READ) · toggle_section(REVERSIBLE) · move_section(REVERSIBLE) - 목록은 site_payload._sections 를 그대로 쓴다 — 발행본이 쓰는 그 함수다. 표를 따로 만들면 에디터·발행본·대화 셋이 갈라지고 "껐는데 나온다" 가 된다 - 잠긴 섹션은 못 끈다. _sections 가 어차피 켜서 내보내므로 끌 수 있게 두면 화면만 거짓말한다 - 이름이 둘 이상 걸리면 고르지 않는다 — 추측으로 고르면 발행하고 나서야 안다 - sections 만 갈아끼운다 — theme 을 통째로 쓰면 고른 색·서체가 말없이 사라진다 ★ 템플릿·색은 넣지 않았다. templatesFor() 가 색·look·기본 섹션·배리에이션을 함께 계산해서, 백엔드가 template_id 만 바꾸면 "레이아웃은 새것, 색은 옛것" 이 된다 — 레지스트리를 공유 단일 출처로 옮기는 작업이 먼저다(docs/AGENT.md). test_agent_runtime 26 passed(구성 7건 추가). 전체 864 passed / 53 failed — 53 은 이번 변경 전과 동일. npm run lint 통과 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
318 lines
15 KiB
Python
318 lines
15 KiB
Python
"""도구 레지스트리 — 에이전트가 할 수 있는 일의 **전부**가 여기 있다.
|
|
|
|
★★ 도구는 반드시 `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_SiteTheme, 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 _sections_of(ctx: ToolContext) -> tuple[list, dict]:
|
|
"""지금 발행본에 서는 섹션 목록(해석된 결과)과 저장된 theme.
|
|
|
|
★ `site_payload._sections` 를 그대로 쓴다 — 발행본이 쓰는 바로 그 함수다. 표를 따로
|
|
만들면 에디터·발행본·대화 셋이 갈라지고, 사장님은 "껐는데 나온다" 를 겪는다.
|
|
★ 저장값이 없어도 업종 기본이 선다. 그래서 아직 한 번도 디자인을 만지지 않은
|
|
사업장에서도 대화가 바로 통한다."""
|
|
_fact, site_service = _services()
|
|
res = await site_service.get_site(ctx.user, ctx.place_id)
|
|
theme = dict((res.site.theme if res.site and res.site.theme else {}) or {})
|
|
spec = site_payload._DEFAULT_THEME[PlaceCategory(ctx.place.category).value]["sections"]
|
|
return site_payload._sections(theme.get("sections"), spec), theme
|
|
|
|
|
|
def _find_section(rows: list, wanted: str):
|
|
"""이름이나 id 로 찾는다. 사장님은 '후기' 처럼 줄여 말한다 — 부분 일치도 받는다.
|
|
|
|
★ 둘 이상 걸리면 **고르지 않는다**(None). 추측으로 고르면 엉뚱한 섹션을 끄고,
|
|
사장님은 그 사실을 발행하고 나서야 안다."""
|
|
wanted = (wanted or "").strip()
|
|
if not wanted:
|
|
return None
|
|
exact = [r for r in rows if r["id"] == wanted or r["name"] == wanted]
|
|
if len(exact) == 1:
|
|
return exact[0]
|
|
partial = [r for r in rows if wanted in r["name"]]
|
|
return partial[0] if len(partial) == 1 else None
|
|
|
|
|
|
async def _save_sections(ctx: ToolContext, theme: dict, rows: list) -> None:
|
|
"""★ theme 의 나머지 칸(colors·fontStyle·look…)을 그대로 들고 간다. sections 만 갈아끼운다 —
|
|
통째로 새로 쓰면 사장님이 고른 색과 서체가 말없이 사라진다."""
|
|
_fact, site_service = _services()
|
|
theme["sections"] = rows
|
|
res = await site_service.set_theme(ctx.user, ctx.place_id, Req_SiteTheme(theme=theme))
|
|
if not res.result.success:
|
|
raise ToolRejected("화면 구성을 저장하지 못했습니다. 빌더 화면에서 확인해 주세요.")
|
|
|
|
|
|
def _layout_line(row: dict) -> str:
|
|
mark = "켜짐" if row["enabled"] else "꺼짐"
|
|
lock = " (항상 켜짐)" if row["locked"] else ""
|
|
return f"· {row['name']} — {mark}{lock}"
|
|
|
|
|
|
async def _list_sections(ctx: ToolContext, args: dict) -> str:
|
|
rows, _theme = await _sections_of(ctx)
|
|
body = "\n".join(_layout_line(r) for r in rows)
|
|
return f"지금 홈페이지는 위에서부터 이 순서입니다.\n{body}"
|
|
|
|
|
|
async def _toggle_section(ctx: ToolContext, args: dict) -> str:
|
|
rows, theme = await _sections_of(ctx)
|
|
row = _find_section(rows, args.get("name"))
|
|
if row is None:
|
|
raise ToolRejected("어느 부분을 말씀하시는지 못 찾았어요. '목록' 이라고 하시면 보여드릴게요.")
|
|
|
|
on = (args.get("enabled") or "").strip().lower() in ("true", "1", "켜", "켜기", "on", "yes")
|
|
# ★ 잠긴 섹션은 끌 수 없다. SEO·필수 마크업 때문에 잠긴 것이라, 끄면 발행 게이트에 걸린다
|
|
# (site_payload._sections 가 어차피 켜서 내보낸다 — 화면만 거짓말하게 된다).
|
|
if row["locked"] and not on:
|
|
raise ToolRejected(f"{row['name']} 은(는) 홈페이지에 꼭 있어야 하는 부분이라 끌 수 없어요.")
|
|
if row["enabled"] == on:
|
|
return f"{row['name']} 은(는) 이미 {'켜져' if on else '꺼져'} 있어요."
|
|
|
|
row["enabled"] = on
|
|
await _save_sections(ctx, theme, rows)
|
|
return (f"{row['name']} 을(를) {'켰습니다' if on else '껐습니다'}. "
|
|
f"사이트에 반영하려면 다시 발행해야 해요 — 지금 할까요?")
|
|
|
|
|
|
async def _move_section(ctx: ToolContext, args: dict) -> str:
|
|
"""★ 배열 순서가 곧 발행본의 섹션 순서다(site_payload._sections)."""
|
|
rows, theme = await _sections_of(ctx)
|
|
row = _find_section(rows, args.get("name"))
|
|
if row is None:
|
|
raise ToolRejected("어느 부분을 말씀하시는지 못 찾았어요. '목록' 이라고 하시면 보여드릴게요.")
|
|
|
|
to = (args.get("to") or "").strip()
|
|
rest = [r for r in rows if r["id"] != row["id"]]
|
|
if to in ("맨 위", "처음", "위", "top", "first"):
|
|
moved, where = [row] + rest, "맨 위로"
|
|
elif to in ("맨 아래", "마지막", "아래", "bottom", "last"):
|
|
moved, where = rest + [row], "맨 아래로"
|
|
else:
|
|
anchor = _find_section(rest, to)
|
|
if anchor is None:
|
|
raise ToolRejected("어디로 옮길지 못 찾았어요. '소개 다음으로' 처럼 말씀해 주세요.")
|
|
at = rest.index(anchor) + 1
|
|
moved, where = rest[:at] + [row] + rest[at:], f"{anchor['name']} 다음으로"
|
|
|
|
await _save_sections(ctx, theme, moved)
|
|
return (f"{row['name']} 을(를) {where} 옮겼습니다. "
|
|
f"사이트에 반영하려면 다시 발행해야 해요 — 지금 할까요?")
|
|
|
|
|
|
# ── 반쯤 되돌릴 수 있는 것 ───────────────────────────────────────────────
|
|
|
|
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="list_sections",
|
|
grade=ToolGrade.READ,
|
|
summary="홈페이지가 어떤 부분들로 어떤 순서로 되어 있는지 보여준다.",
|
|
run=_list_sections,
|
|
),
|
|
Tool(
|
|
name="toggle_section",
|
|
grade=ToolGrade.REVERSIBLE,
|
|
summary="홈페이지의 한 부분을 켜거나 끈다(예: 후기, 사진 갤러리, 예약 안내).",
|
|
args={"name": "그 부분의 이름", "enabled": "켜면 true, 끄면 false"},
|
|
run=_toggle_section,
|
|
),
|
|
Tool(
|
|
name="move_section",
|
|
grade=ToolGrade.REVERSIBLE,
|
|
summary="홈페이지에서 한 부분의 위치(순서)를 바꾼다.",
|
|
args={"name": "옮길 부분의 이름", "to": "'맨 위' · '맨 아래' · 또는 그 뒤에 올 부분의 이름"},
|
|
run=_move_section,
|
|
),
|
|
Tool(
|
|
name="publish",
|
|
grade=ToolGrade.SEMI,
|
|
summary="바뀐 내용을 홈페이지에 반영한다(재발행).",
|
|
run=_publish,
|
|
confirm="지금 홈페이지를 다시 발행할까요? 바뀐 내용이 손님에게 보이게 됩니다.",
|
|
),
|
|
]
|
|
}
|
|
|
|
|
|
async def sections_of(ctx: ToolContext):
|
|
"""런타임이 프롬프트에 실을 섹션 목록. 도구가 쓰는 것과 같은 함수여야 한다 —
|
|
다르면 모델이 본 이름과 도구가 찾는 이름이 갈린다."""
|
|
return await _sections_of(ctx)
|
|
|
|
|
|
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"
|
|
]
|