"""도구 레지스트리 — 에이전트가 할 수 있는 일의 **전부**가 여기 있다. ★★ 도구는 반드시 `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.media_crud import MediaCRUD 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.media_service import MediaService 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 = "" # ★ 이 도구가 바꾼 것은 **재발행해야 사이트에 반영된다.** 안내 문구는 도구가 아니라 # 런타임이 **한 번만** 붙인다 — 도구마다 문장에 박아 두면 한 발화로 셋을 고쳤을 때 # 같은 말이 세 번 나온다. republish: bool = False def _services(): """서비스는 매 호출 새로 만든다 — 라우터가 Depends 로 받는 것과 같은 수명이다. ★ Depends 기본값에 기대지 않고 의존을 손으로 넣는다. FastAPI 밖에서 부르면 기본값이 `Depends(...)` 객체 그대로라 서비스가 조용히 엉뚱한 것을 들고 돈다.""" place_crud = PlaceCRUD() return FactService(FactCRUD(), place_crud), SiteService(SiteCRUD(), place_crud, JobQueue()) def _media_service() -> MediaService: return MediaService(MediaCRUD(), PlaceCRUD()) # ── 읽기 ──────────────────────────────────────────────────────────────── 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 '껐습니다'}." 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} 옮겼습니다." # ── 사진 ───────────────────────────────────────────────────────────────── # # ★ 업로드·교체는 없다. 이미지 재게시 권리가 미결이라 저장 경로를 일부러 안 만들어 뒀다 # (docs/DECISIONS.md 1-2 · 5-3). 아래는 **이미 있는 사진의 노출과 순서**만 바꾼다. def _photo_name(row) -> str: """사장님이 부를 이름. Vision 이 만든 라벨·alt 가 유일한 단서다.""" return (row.label or "").strip() or (row.alt_text or "").strip() or "이름 없는 사진" async def _photos(ctx: ToolContext) -> list: res = await _media_service().list_media(ctx.user, ctx.place_id) return list(res.media or []) def _find_photo(rows: list, wanted: str): """★ 둘 이상 걸리면 고르지 않는다 — 추측으로 내리면 엉뚱한 사진이 사라지고, 사장님은 발행하고 나서야 안다(섹션과 같은 규칙).""" wanted = (wanted or "").strip() if not wanted: return None hits = [r for r in rows if wanted in _photo_name(r)] return hits[0] if len(hits) == 1 else None async def _list_photos(ctx: ToolContext, args: dict) -> str: rows = await _photos(ctx) if not rows: return "아직 등록된 사진이 없어요." lines = [] for i, r in enumerate(rows[:15]): where = " (객실·메뉴 전용)" if r.unit_id else "" mark = "" if r.publishable else " — 지금은 안 나감" head = "대표 " if i == 0 and not r.unit_id else "" lines.append(f"· {head}{_photo_name(r)}{where}{mark}") more = f"\n(그 밖에 {len(rows) - 15}장 더)" if len(rows) > 15 else "" return "홈페이지에 있는 사진입니다.\n" + "\n".join(lines) + more async def _hide_photo(ctx: ToolContext, args: dict) -> str: rows = await _photos(ctx) row = _find_photo(rows, args.get("name")) if row is None: raise ToolRejected("어느 사진을 말씀하시는지 못 찾았어요. '사진 목록' 이라고 하시면 보여드릴게요.") res = await _media_service().hide_media(ctx.user, ctx.place_id, str(row.media_id)) if not res.result.success: raise ToolRejected("그 사진을 내리지 못했습니다. 빌더 화면에서 확인해 주세요.") return f"'{_photo_name(row)}' 사진을 내렸습니다." async def _set_primary_photo(ctx: ToolContext, args: dict) -> str: rows = await _photos(ctx) row = _find_photo(rows, args.get("name")) if row is None: raise ToolRejected("어느 사진을 말씀하시는지 못 찾았어요. '사진 목록' 이라고 하시면 보여드릴게요.") if row.unit_id: raise ToolRejected(f"'{_photo_name(row)}' 은(는) 객실·메뉴 전용 사진이라 대표로 쓸 수 없어요.") res = await _media_service().set_primary(ctx.user, ctx.place_id, str(row.media_id)) if not res.result.success: raise ToolRejected("대표 사진을 바꾸지 못했습니다. 빌더 화면에서 확인해 주세요.") return f"대표 사진을 '{_photo_name(row)}' 으로 바꿨습니다." # ── 반쯤 되돌릴 수 있는 것 ─────────────────────────────────────────────── 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, republish=True, 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, republish=True, summary="홈페이지의 한 부분을 켜거나 끈다(예: 후기, 사진 갤러리, 예약 안내).", args={"name": "그 부분의 이름", "enabled": "켜면 true, 끄면 false"}, run=_toggle_section, ), Tool( name="move_section", grade=ToolGrade.REVERSIBLE, republish=True, summary="홈페이지에서 한 부분의 위치(순서)를 바꾼다.", args={"name": "옮길 부분의 이름", "to": "'맨 위' · '맨 아래' · 또는 그 뒤에 올 부분의 이름"}, run=_move_section, ), Tool( name="list_photos", grade=ToolGrade.READ, summary="홈페이지에 올라가 있는 사진 목록을 보여준다(맨 앞이 대표 사진).", run=_list_photos, ), Tool( name="hide_photo", grade=ToolGrade.REVERSIBLE, republish=True, summary="사진 한 장을 홈페이지에서 내린다. 새 사진을 올리는 것은 아직 못 한다.", args={"name": "그 사진의 이름(라벨)"}, run=_hide_photo, ), Tool( name="set_primary_photo", grade=ToolGrade.REVERSIBLE, republish=True, summary="대표 사진을 바꾼다(검색 결과와 목록 카드에 나오는 그림).", args={"name": "대표로 쓸 사진의 이름(라벨)"}, run=_set_primary_photo, ), Tool( name="publish", grade=ToolGrade.SEMI, summary="바뀐 내용을 홈페이지에 반영한다(재발행).", run=_publish, confirm="지금 홈페이지를 다시 발행할까요? 바뀐 내용이 손님에게 보이게 됩니다.", ), ] } async def sections_of(ctx: ToolContext): """런타임이 프롬프트에 실을 섹션 목록. 도구가 쓰는 것과 같은 함수여야 한다 — 다르면 모델이 본 이름과 도구가 찾는 이름이 갈린다.""" return await _sections_of(ctx) async def photo_names(ctx: ToolContext) -> list[str]: """런타임이 프롬프트에 실을 사진 이름. 도구가 찾는 이름과 **같은 함수**로 만든다 — 다르면 모델이 본 이름과 도구가 찾는 이름이 갈린다.""" return [_photo_name(r) for r in (await _photos(ctx))[:15]] 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" ]