Files
gpu-services/nanoclaude/tools/calendar_tool.py
Hyungi Ahn 6e24da56a4 feat: 이드 도구 확장 — 캘린더/메일/문서 연동 (read-only + 캘린더 생성 확인)
- tools/calendar_tool.py: CalDAV search/today/create_draft/create_confirmed
- tools/email_tool.py: IMAP search/read (전송 비활성화)
- tools/document_tool.py: Document Server search/read (read-only)
- tools/registry.py: 도구 디스패처 + WRITE_OPS 안전장치 + 에러 표준화
- 분류기: "tools" 액션 추가, 도구 목록/파라미터 스키마/규칙 명시
- Worker: tools 분기 + tool timeout 10초 + payload 2000자 제한
- conversation: pending_draft (TTL 5분) + create 확인 플로우
- 현재 시간을 분류기에 전달 (날짜 질문 대응)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-06 13:39:15 +09:00

141 lines
4.7 KiB
Python

"""Calendar 도구 — CalDAV를 통한 일정 조회/생성."""
from __future__ import annotations
import logging
import uuid
from datetime import datetime, timedelta
import caldav
from config import settings
logger = logging.getLogger(__name__)
TOOL_NAME = "calendar"
def _make_result(ok: bool, operation: str, data=None, summary: str = "", error: str | None = None) -> dict:
return {"ok": ok, "tool": TOOL_NAME, "operation": operation, "data": data or [], "summary": summary, "error": error}
def _get_client():
if not settings.caldav_url or not settings.caldav_user:
return None
return caldav.DAVClient(
url=settings.caldav_url,
username=settings.caldav_user,
password=settings.caldav_pass,
ssl_verify_cert=False,
)
def _parse_event(event) -> dict:
"""VEVENT → dict."""
try:
vevent = event.vobject_instance.vevent
summary = str(vevent.summary.value) if hasattr(vevent, "summary") else "(제목 없음)"
dtstart = vevent.dtstart.value
dtend = vevent.dtend.value if hasattr(vevent, "dtend") else None
start_str = dtstart.strftime("%Y-%m-%d %H:%M") if hasattr(dtstart, "strftime") else str(dtstart)
end_str = dtend.strftime("%Y-%m-%d %H:%M") if dtend and hasattr(dtend, "strftime") else ""
return {"summary": summary, "start": start_str, "end": end_str}
except Exception:
return {"summary": "(파싱 실패)", "start": "", "end": ""}
async def today() -> dict:
"""오늘 일정 조회."""
now = datetime.now()
start = now.replace(hour=0, minute=0, second=0)
end = start + timedelta(days=1)
return await search(start.strftime("%Y-%m-%d"), end.strftime("%Y-%m-%d"))
async def search(date_from: str, date_to: str) -> dict:
"""기간 내 일정 검색."""
try:
client = _get_client()
if not client:
return _make_result(False, "search", error="CalDAV 설정이 없습니다.")
principal = client.principal()
calendars = principal.calendars()
if not calendars:
return _make_result(True, "search", data=[], summary="캘린더가 없습니다.")
start = datetime.strptime(date_from, "%Y-%m-%d")
end = datetime.strptime(date_to, "%Y-%m-%d") + timedelta(days=1)
events = []
for cal in calendars:
try:
results = cal.date_search(start=start, end=end, expand=True)
for ev in results:
events.append(_parse_event(ev))
except Exception:
continue
summary = f"{date_from}~{date_to} 기간에 {len(events)}개의 일정이 있습니다."
return _make_result(True, "search", data=events, summary=summary)
except Exception as e:
logger.exception("Calendar search failed")
return _make_result(False, "search", error=str(e))
async def create_draft(title: str, date: str, time: str, description: str = "") -> dict:
"""일정 생성 초안 (실제 생성 안 함)."""
title = title or "일정"
draft = {
"title": title,
"date": date,
"time": time,
"description": description,
"display": f"📅 제목: {title}\n날짜: {date} {time}\n설명: {description or '없음'}",
}
return _make_result(True, "create_draft", data=draft, summary=draft["display"])
async def create_confirmed(draft_data: dict) -> dict:
"""사용자 확인 후 실제 CalDAV 이벤트 생성."""
try:
client = _get_client()
if not client:
return _make_result(False, "create_confirmed", error="CalDAV 설정이 없습니다.")
title = draft_data.get("title", "일정")
date_str = draft_data.get("date", "")
time_str = draft_data.get("time", "00:00")
description = draft_data.get("description", "")
dt_start = datetime.strptime(f"{date_str} {time_str}", "%Y-%m-%d %H:%M")
dt_end = dt_start + timedelta(hours=1)
uid = str(uuid.uuid4())
vcal = f"""BEGIN:VCALENDAR
VERSION:2.0
PRODID:-//NanoClaude//이드//KO
BEGIN:VEVENT
UID:{uid}
DTSTART:{dt_start.strftime('%Y%m%dT%H%M%S')}
DTEND:{dt_end.strftime('%Y%m%dT%H%M%S')}
SUMMARY:{title}
DESCRIPTION:{description}
END:VEVENT
END:VCALENDAR"""
principal = client.principal()
calendars = principal.calendars()
if not calendars:
return _make_result(False, "create_confirmed", error="캘린더를 찾을 수 없습니다.")
calendars[0].save_event(vcal)
return _make_result(True, "create_confirmed", data={"uid": uid, "title": title}, summary=f"'{title}' 일정이 등록되었습니다!")
except Exception as e:
logger.exception("Calendar create failed")
return _make_result(False, "create_confirmed", error=str(e))