MCP Calendar Server

""" MCP 캘린더 서비스 레이어 """ from datetime import datetime, timedelta from typing import Dict, List, Optional from ..models import ( CalendarEvent, CalendarEventRequest, CalendarEventResponse, EventStatus, McpResult ) from ..exceptions import EventNotFound, InvalidEventData, UnauthorizedAccess, CalendarException class CalendarService: """캘린더 이벤트 관리 서비스""" def __init__(self): self.events_db: Dict[int, CalendarEvent] = {} self.next_id = 1 # 샘플 데이터 추가 self._initialize_sample_data() def _initialize_sample_data(self): """샘플 데이터 초기화""" sample_events = [ CalendarEvent( id=1, user_id=1, title="팀 미팅", description="주간 팀 미팅", location="회의실 A", start_time=datetime(2025, 8, 2, 10, 0), duration=60, category="WORK", stamina_cost=20, status=EventStatus.PLANNED, created_at=datetime.now() ), CalendarEvent( id=2, user_id=1, title="영어 공부", description="토익 공부", start_time=datetime(2025, 8, 2, 14, 0), duration=120, category="STUDY", stamina_cost=30, status=EventStatus.PLANNED, created_at=datetime.now() ) ] for event in sample_events: self.events_db[event.id] = event self.next_id = 3 def fetch_events(self, user_id: int) -> McpResult: """사용자의 모든 이벤트 조회""" try: user_events = [ event for event in self.events_db.values() if event.user_id == user_id ] return McpResult(success=True, data=user_events) except Exception as e: return McpResult(success=False, error=str(e)) def create_event(self, request: CalendarEventRequest, user_id: int) -> McpResult: """새 이벤트 생성""" try: # 시간 충돌 검사 if self._has_time_conflict(request.start_time, request.duration, user_id): return McpResult( success=False, error="해당 시간에 이미 다른 일정이 있습니다" ) event = CalendarEvent( id=self.next_id, user_id=user_id, title=request.title, description=request.description, location=request.location, start_time=request.start_time, duration=request.duration, category=request.category, stamina_cost=request.stamina_cost, status=request.status or EventStatus.PLANNED, created_at=request.created_at or datetime.now() ) self.events_db[self.next_id] = event self.next_id += 1 return McpResult(success=True, data=event) except Exception as e: return McpResult(success=False, error=str(e)) def update_event(self, event_id: int, request: CalendarEventRequest, user_id: int) -> McpResult: """이벤트 수정""" try: if event_id not in self.events_db: raise EventNotFound(event_id) existing_event = self.events_db[event_id] if existing_event.user_id != user_id: raise UnauthorizedAccess(user_id, event_id) # 시간 충돌 검사 (자기 자신 제외) if self._has_time_conflict(request.start_time, request.duration, user_id, exclude_id=event_id): return McpResult( success=False, error="해당 시간에 이미 다른 일정이 있습니다" ) updated_event = CalendarEvent( id=event_id, user_id=user_id, title=request.title, description=request.description, location=request.location, start_time=request.start_time, duration=request.duration, category=request.category, stamina_cost=request.stamina_cost, status=request.status or existing_event.status, stamina_after_completion=existing_event.stamina_after_completion, created_at=existing_event.created_at ) self.events_db[event_id] = updated_event return McpResult(success=True, data=updated_event) except CalendarException: raise except Exception as e: return McpResult(success=False, error=str(e)) def delete_event(self, event_id: int, user_id: int) -> McpResult: """이벤트 삭제""" try: if event_id not in self.events_db: raise EventNotFound(event_id) existing_event = self.events_db[event_id] if existing_event.user_id != user_id: raise UnauthorizedAccess(user_id, event_id) del self.events_db[event_id] return McpResult(success=True, data=None) except CalendarException: raise except Exception as e: return McpResult(success=False, error=str(e)) def get_event_by_id(self, event_id: int, user_id: int) -> McpResult: """ID로 특정 이벤트 조회""" try: if event_id not in self.events_db: raise EventNotFound(event_id) event = self.events_db[event_id] if event.user_id != user_id: raise UnauthorizedAccess(user_id, event_id) return McpResult(success=True, data=event) except CalendarException: raise except Exception as e: return McpResult(success=False, error=str(e)) def complete_event(self, event_id: int, user_id: int, stamina_after: int) -> McpResult: """이벤트 완료 처리""" try: if event_id not in self.events_db: raise EventNotFound(event_id) event = self.events_db[event_id] if event.user_id != user_id: raise UnauthorizedAccess(user_id, event_id) event.status = EventStatus.COMPLETED event.stamina_after_completion = stamina_after return McpResult(success=True, data=event) except CalendarException: raise except Exception as e: return McpResult(success=False, error=str(e)) def _has_time_conflict(self, start_time: datetime, duration: int, user_id: int, exclude_id: Optional[int] = None) -> bool: """시간 충돌 검사""" end_time = start_time + timedelta(minutes=duration) for event_id, event in self.events_db.items(): if exclude_id and event_id == exclude_id: continue if event.user_id != user_id or event.status == EventStatus.CANCELED: continue event_end_time = event.start_time + timedelta(minutes=event.duration) # 시간 겹침 확인 if (start_time < event_end_time and end_time > event.start_time): return True return False def to_response(self, event: CalendarEvent) -> CalendarEventResponse: """Entity를 Response로 변환""" return CalendarEventResponse( id=event.id, title=event.title, description=event.description, location=event.location, start_time=event.start_time, duration=event.duration, category=event.category, stamina_cost=event.stamina_cost, status=event.status, stamina_after_completion=event.stamina_after_completion, created_at=event.created_at )