gpu-services/nanoclaude/tools/calendar_tool.py

"""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. vobject 있으면 사용, 없으면 icalendar fallback."""
    try:
        # vobject 방식
        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:
        pass

    # fallback: icalendar 라이브러리로 raw data 파싱
    try:
        from icalendar import Calendar
        cal = Calendar.from_ical(event.data)
        for comp in cal.walk():
            if comp.name == "VEVENT":
                summary = str(comp.get("summary", "(제목 없음)"))
                dtstart = comp.get("dtstart")
                dtend = comp.get("dtend")
                start_str = dtstart.dt.strftime("%Y-%m-%d %H:%M") if dtstart else ""
                end_str = dtend.dt.strftime("%Y-%m-%d %H:%M") if dtend else ""
                return {"summary": summary, "start": start_str, "end": end_str}
    except Exception:
        pass

    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 = "", end_time: str = "") -> dict:
    """일정 생성 초안 (실제 생성 <20><><EFBFBD> 함)."""
    title = title or "일정"
    time_display = f"{time} ~ {end_time}" if end_time else time
    draft = {
        "title": title,
        "date": date,
        "time": time,
        "end_time": end_time,
        "description": description,
        "display": f"📅 제목: {title}\n날짜: {date} {time_display}\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")
        end_time_str = draft_data.get("end_time", "")
        description = draft_data.get("description", "")

        dt_start = datetime.strptime(f"{date_str} {time_str}", "%Y-%m-%d %H:%M")
        if end_time_str:
            dt_end = datetime.strptime(f"{date_str} {end_time_str}", "%Y-%m-%d %H:%M")
        else:
            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))