"""예약 플랫폼 통합 관리 MCP 서버"""
import os
import asyncio
import threading
from datetime import datetime, date, timedelta
from typing import List, Optional
from fastmcp import FastMCP
from dotenv import load_dotenv
from .models import Reservation, Customer, Platform, RevenueData
from .storage import Storage
from .ical_parser import ICalParser
# 환경 변수 로드
load_dotenv()
# MCP 서버 초기화
mcp = FastMCP("reservation-platform-mcp")
# 저장소 초기화
storage = Storage()
# 동기화 설정
ICAL_SYNC_ENABLED = os.getenv("ICAL_SYNC_ENABLED", "true").lower() == "true"
ICAL_SYNC_INTERVAL = int(os.getenv("ICAL_SYNC_INTERVAL", "3600")) # 기본 1시간
# ===== 예약 조회 Tools =====
@mcp.tool()
def get_today_reservations() -> str:
"""오늘 예약 현황을 조회합니다.
Returns:
오늘의 모든 예약 목록 (시간순 정렬)
"""
today = date.today()
reservations = storage.get_reservations_by_date(today)
if not reservations:
return "오늘 예약이 없습니다."
# 시간순 정렬
reservations.sort(key=lambda r: r.start_time)
result = f"📅 오늘({today.strftime('%Y-%m-%d')}) 예약 현황 ({len(reservations)}건)\n\n"
for r in reservations:
result += f"• [{r.platform}] {r.start_time.strftime('%H:%M')}-{r.end_time.strftime('%H:%M')} | "
result += f"{r.customer_name} | {r.price:,.0f}원\n"
return result
@mcp.tool()
def get_reservations_by_date(target_date: str) -> str:
"""특정 날짜의 예약 현황을 조회합니다.
Args:
target_date: 조회할 날짜 (YYYY-MM-DD 형식)
Returns:
해당 날짜의 모든 예약 목록
"""
try:
target = datetime.strptime(target_date, "%Y-%m-%d").date()
except ValueError:
return "날짜 형식이 올바르지 않습니다. YYYY-MM-DD 형식으로 입력해주세요."
reservations = storage.get_reservations_by_date(target)
if not reservations:
return f"{target_date}에는 예약이 없습니다."
reservations.sort(key=lambda r: r.start_time)
result = f"📅 {target_date} 예약 현황 ({len(reservations)}건)\n\n"
for r in reservations:
result += f"• [{r.platform}] {r.start_time.strftime('%H:%M')}-{r.end_time.strftime('%H:%M')} | "
result += f"{r.customer_name} | {r.price:,.0f}원\n"
return result
@mcp.tool()
def get_reservations_by_platform(platform: str) -> str:
"""특정 플랫폼의 모든 예약을 조회합니다.
Args:
platform: 플랫폼 이름 (airbnb, spacecloud, naver, yanolja, kakao)
Returns:
해당 플랫폼의 모든 예약 목록
"""
if platform not in ["airbnb", "spacecloud", "naver", "yanolja", "kakao"]:
return "올바른 플랫폼 이름을 입력해주세요. (airbnb, spacecloud, naver, yanolja, kakao)"
reservations = storage.get_reservations_by_platform(platform)
if not reservations:
return f"{platform}에는 예약이 없습니다."
reservations.sort(key=lambda r: r.start_time, reverse=True)
result = f"🏢 {platform.upper()} 예약 목록 ({len(reservations)}건)\n\n"
for r in reservations[:20]: # 최근 20건만
result += f"• {r.start_time.strftime('%Y-%m-%d %H:%M')} | "
result += f"{r.customer_name} | {r.price:,.0f}원\n"
if len(reservations) > 20:
result += f"\n... 외 {len(reservations) - 20}건"
return result
@mcp.tool()
def get_week_reservations() -> str:
"""이번 주 예약 현황을 조회합니다.
Returns:
이번 주(월~일)의 모든 예약 목록
"""
today = date.today()
# 이번 주 월요일
monday = today - timedelta(days=today.weekday())
# 이번 주 일요일
sunday = monday + timedelta(days=6)
all_reservations = storage.get_all_reservations()
week_reservations = [
r for r in all_reservations
if monday <= r.start_time.date() <= sunday
]
if not week_reservations:
return "이번 주에는 예약이 없습니다."
week_reservations.sort(key=lambda r: r.start_time)
result = f"📅 이번 주({monday} ~ {sunday}) 예약 현황 ({len(week_reservations)}건)\n\n"
current_date = None
for r in week_reservations:
r_date = r.start_time.date()
if r_date != current_date:
current_date = r_date
weekday = ['월', '화', '수', '목', '금', '토', '일'][r_date.weekday()]
result += f"\n{r_date.strftime('%m/%d')}({weekday})\n"
result += f" • [{r.platform}] {r.start_time.strftime('%H:%M')}-{r.end_time.strftime('%H:%M')} | "
result += f"{r.customer_name} | {r.price:,.0f}원\n"
return result
# ===== 중복 예약 체크 Tools =====
@mcp.tool()
def check_duplicate_reservations() -> str:
"""현재 등록된 예약 중 시간이 겹치는 중복 예약이 있는지 확인합니다.
Returns:
중복 예약 목록 (없으면 "중복 없음" 메시지)
"""
all_reservations = storage.get_all_reservations()
duplicates = []
for i, r1 in enumerate(all_reservations):
for r2 in all_reservations[i+1:]:
if r1.is_overlapping(r2):
duplicates.append((r1, r2))
if not duplicates:
return "✅ 중복 예약이 없습니다."
result = f"⚠️ 중복 예약 발견! ({len(duplicates)}건)\n\n"
for r1, r2 in duplicates:
result += f"• {r1.start_time.strftime('%Y-%m-%d %H:%M')} - {r1.end_time.strftime('%H:%M')}\n"
result += f" - [{r1.platform}] {r1.customer_name}\n"
result += f" - [{r2.platform}] {r2.customer_name}\n\n"
return result
@mcp.tool()
def check_time_slot(start_time: str, end_time: str) -> str:
"""특정 시간대에 예약 가능한지 확인합니다.
Args:
start_time: 시작 시간 (YYYY-MM-DD HH:MM 형식)
end_time: 종료 시간 (YYYY-MM-DD HH:MM 형식)
Returns:
예약 가능 여부 및 충돌하는 예약 정보
"""
try:
start = datetime.strptime(start_time, "%Y-%m-%d %H:%M")
end = datetime.strptime(end_time, "%Y-%m-%d %H:%M")
except ValueError:
return "시간 형식이 올바르지 않습니다. YYYY-MM-DD HH:MM 형식으로 입력해주세요."
# 임시 예약 객체 생성
temp_reservation = Reservation(
id="temp",
platform="airbnb",
customer_name="temp",
start_time=start,
end_time=end,
price=0
)
conflicts = storage.find_overlapping_reservations(temp_reservation)
if not conflicts:
return f"✅ {start_time} ~ {end_time} 예약 가능합니다!"
result = f"❌ {start_time} ~ {end_time} 예약 불가 (충돌 {len(conflicts)}건)\n\n"
result += "충돌하는 예약:\n"
for c in conflicts:
result += f"• [{c.platform}] {c.start_time.strftime('%H:%M')}-{c.end_time.strftime('%H:%M')} | {c.customer_name}\n"
return result
# ===== 매출/정산 Tools =====
@mcp.tool()
def get_revenue_summary(period: str = "month") -> str:
"""매출 요약 정보를 조회합니다.
Args:
period: 기간 (today, week, month)
Returns:
기간별 매출 요약 (총액, 건수, 플랫폼별)
"""
today = date.today()
if period == "today":
start_date = today
end_date = today
period_name = "오늘"
elif period == "week":
start_date = today - timedelta(days=today.weekday())
end_date = start_date + timedelta(days=6)
period_name = "이번 주"
else: # month
start_date = today.replace(day=1)
next_month = today.replace(day=28) + timedelta(days=4)
end_date = next_month.replace(day=1) - timedelta(days=1)
period_name = f"{today.year}년 {today.month}월"
all_reservations = storage.get_all_reservations()
period_reservations = [
r for r in all_reservations
if start_date <= r.start_time.date() <= end_date
]
if not period_reservations:
return f"{period_name}에는 예약이 없습니다."
# 플랫폼별 매출 계산
platform_revenue = {}
total_amount = 0
for r in period_reservations:
if r.platform not in platform_revenue:
platform_revenue[r.platform] = {"amount": 0, "count": 0}
platform_revenue[r.platform]["amount"] += r.price
platform_revenue[r.platform]["count"] += 1
total_amount += r.price
result = f"💰 {period_name} 매출 요약\n\n"
result += f"총 매출: {total_amount:,.0f}원\n"
result += f"총 예약: {len(period_reservations)}건\n"
result += f"평균 금액: {total_amount / len(period_reservations):,.0f}원\n\n"
result += "플랫폼별 매출:\n"
for platform, data in sorted(platform_revenue.items(), key=lambda x: x[1]["amount"], reverse=True):
result += f"• {platform}: {data['amount']:,.0f}원 ({data['count']}건)\n"
return result
@mcp.tool()
def calculate_platform_fees() -> str:
"""각 플랫폼의 수수료를 계산합니다.
Returns:
플랫폼별 총 매출, 수수료, 순수익
"""
all_reservations = storage.get_all_reservations()
platforms = storage.get_all_platforms()
# 플랫폼별 설정을 딕셔너리로 변환
platform_config = {p.platform: p for p in platforms}
# 플랫폼별 매출 집계
platform_stats = {}
for r in all_reservations:
if r.platform not in platform_stats:
platform_stats[r.platform] = {
"total": 0,
"count": 0,
"fee_rate": platform_config.get(r.platform, None).fee_rate if r.platform in platform_config else 0
}
platform_stats[r.platform]["total"] += r.price
platform_stats[r.platform]["count"] += 1
if not platform_stats:
return "매출 데이터가 없습니다."
result = "💳 플랫폼 수수료 계산\n\n"
total_revenue = 0
total_fees = 0
total_net = 0
for platform, stats in sorted(platform_stats.items()):
revenue = stats["total"]
fee_rate = stats["fee_rate"]
fee = revenue * fee_rate
net = revenue - fee
total_revenue += revenue
total_fees += fee
total_net += net
result += f"[{platform.upper()}]\n"
result += f" 매출: {revenue:,.0f}원 ({stats['count']}건)\n"
result += f" 수수료({fee_rate*100:.1f}%): {fee:,.0f}원\n"
result += f" 순수익: {net:,.0f}원\n\n"
result += "─" * 40 + "\n"
result += f"총 매출: {total_revenue:,.0f}원\n"
result += f"총 수수료: {total_fees:,.0f}원\n"
result += f"총 순수익: {total_net:,.0f}원\n"
return result
# ===== 고객 관리 Tools =====
@mcp.tool()
def get_blacklist() -> str:
"""블랙리스트에 등록된 고객 목록을 조회합니다.
Returns:
블랙리스트 고객 목록
"""
blacklist = storage.get_blacklist()
if not blacklist:
return "블랙리스트에 등록된 고객이 없습니다."
result = f"🚫 블랙리스트 ({len(blacklist)}명)\n\n"
for customer in blacklist:
result += f"• {customer.name} ({customer.phone})\n"
result += f" 노쇼 {customer.noshow_count}회 | 총 예약 {customer.total_reservations}회\n"
if customer.notes:
result += f" 메모: {customer.notes}\n"
result += "\n"
return result
@mcp.tool()
def add_to_blacklist(phone: str, name: str, reason: str = "") -> str:
"""고객을 블랙리스트에 추가합니다.
Args:
phone: 고객 연락처
name: 고객 이름
reason: 블랙리스트 등록 사유
Returns:
등록 완료 메시지
"""
customer = storage.get_customer(phone)
if customer:
customer.is_blacklisted = True
if reason:
customer.notes = reason
else:
customer = Customer(
phone=phone,
name=name,
is_blacklisted=True,
notes=reason
)
storage.save_customer(customer)
return f"✅ {name}({phone})님을 블랙리스트에 추가했습니다."
@mcp.tool()
def get_noshow_customers(min_count: int = 1) -> str:
"""노쇼 이력이 있는 고객 목록을 조회합니다.
Args:
min_count: 최소 노쇼 횟수 (기본값: 1)
Returns:
노쇼 고객 목록
"""
noshow_customers = storage.get_noshow_customers(min_count)
if not noshow_customers:
return f"노쇼 {min_count}회 이상인 고객이 없습니다."
result = f"⚠️ 노쇼 {min_count}회 이상 고객 ({len(noshow_customers)}명)\n\n"
for customer in sorted(noshow_customers, key=lambda c: c.noshow_count, reverse=True):
status = "🚫 블랙리스트" if customer.is_blacklisted else ""
result += f"• {customer.name} ({customer.phone}) {status}\n"
result += f" 노쇼 {customer.noshow_count}회 | 총 예약 {customer.total_reservations}회\n\n"
return result
# ===== 메시지 생성 Tools =====
@mcp.tool()
def generate_reminder_message(reservation_date: str) -> str:
"""특정 날짜 예약 고객에게 보낼 안내 메시지를 생성합니다.
Args:
reservation_date: 예약 날짜 (YYYY-MM-DD 형식)
Returns:
고객별 안내 메시지
"""
try:
target = datetime.strptime(reservation_date, "%Y-%m-%d").date()
except ValueError:
return "날짜 형식이 올바르지 않습니다. YYYY-MM-DD 형식으로 입력해주세요."
reservations = storage.get_reservations_by_date(target)
if not reservations:
return f"{reservation_date}에는 예약이 없습니다."
result = f"📨 {reservation_date} 예약 안내 메시지\n\n"
for r in sorted(reservations, key=lambda x: x.start_time):
message = f"[{r.customer_name}님 - {r.start_time.strftime('%H:%M')}]\n"
message += f"안녕하세요, {r.customer_name}님!\n"
message += f"{r.start_time.strftime('%Y년 %m월 %d일 %H:%M')} 예약 안내드립니다.\n\n"
message += f"예약 시간: {r.start_time.strftime('%H:%M')} ~ {r.end_time.strftime('%H:%M')}\n"
message += f"예약 금액: {r.price:,.0f}원\n\n"
message += "즐거운 시간 되시길 바랍니다!\n"
result += message + "\n" + "─" * 40 + "\n\n"
return result
# ===== 플랫폼 동기화 Tools =====
@mcp.tool()
def sync_platform_ical(platform: str, ical_url: str) -> str:
"""플랫폼의 iCal URL에서 예약 데이터를 동기화합니다.
Args:
platform: 플랫폼 이름 (airbnb, spacecloud, naver, yanolja, kakao)
ical_url: iCal 캘린더 URL
Returns:
동기화 결과 메시지
"""
if platform not in ["airbnb", "spacecloud", "naver", "yanolja", "kakao"]:
return "올바른 플랫폼 이름을 입력해주세요."
try:
reservations = ICalParser.sync_platform(platform, ical_url)
# 기존 플랫폼 예약을 삭제하고 새로운 예약으로 교체
storage.sync_platform_reservations(platform, reservations)
# 플랫폼 설정 업데이트
config = storage.get_platform_config(platform)
if config:
config.ical_url = ical_url
storage.update_platform_config(config)
return f"✅ {platform}에서 {len(reservations)}건의 예약을 동기화했습니다."
except Exception as e:
return f"❌ 동기화 실패: {str(e)}"
# ===== 자동 동기화 함수 =====
def sync_all_platforms():
"""모든 플랫폼의 iCal을 동기화합니다"""
platforms = {
"airbnb": os.getenv("AIRBNB_ICAL_URL"),
"spacecloud": os.getenv("SPACECLOUD_ICAL_URL"),
"naver": os.getenv("NAVER_ICAL_URL"),
"yanolja": os.getenv("YANOLJA_ICAL_URL"),
"kakao": os.getenv("KAKAO_ICAL_URL"),
}
synced_count = 0
total_reservations = 0
for platform, ical_url in platforms.items():
if ical_url and ical_url.strip():
try:
print(f"🔄 Syncing {platform}...")
reservations = ICalParser.sync_platform(platform, ical_url)
storage.sync_platform_reservations(platform, reservations)
# 플랫폼 설정 업데이트
config = storage.get_platform_config(platform)
if config:
config.ical_url = ical_url
storage.update_platform_config(config)
print(f"✅ {platform}: {len(reservations)}건 동기화 완료")
synced_count += 1
total_reservations += len(reservations)
except Exception as e:
print(f"❌ {platform} 동기화 실패: {e}")
if synced_count > 0:
print(f"\n✅ 총 {synced_count}개 플랫폼에서 {total_reservations}건의 예약을 동기화했습니다.")
else:
print("\n⚠️ iCal URL이 설정된 플랫폼이 없습니다. .env 파일을 확인해주세요.")
return synced_count, total_reservations
def background_sync_task():
"""백그라운드에서 정기적으로 동기화를 수행합니다"""
import time
while True:
try:
print(f"\n{'='*60}")
print(f"🔄 자동 동기화 시작 ({datetime.now().strftime('%Y-%m-%d %H:%M:%S')})")
print(f"{'='*60}")
sync_all_platforms()
print(f"\n⏰ 다음 동기화까지 {ICAL_SYNC_INTERVAL}초 대기...")
print(f"{'='*60}\n")
except Exception as e:
print(f"❌ 자동 동기화 오류: {e}")
time.sleep(ICAL_SYNC_INTERVAL)
def start_background_sync():
"""백그라운드 동기화 스레드 시작"""
if ICAL_SYNC_ENABLED:
print("\n🚀 백그라운드 자동 동기화 활성화")
print(f" 동기화 주기: {ICAL_SYNC_INTERVAL}초 ({ICAL_SYNC_INTERVAL // 60}분)")
# 초기 동기화
print("\n📥 초기 동기화 수행 중...")
sync_all_platforms()
# 백그라운드 스레드 시작
sync_thread = threading.Thread(target=background_sync_task, daemon=True)
sync_thread.start()
print("✅ 백그라운드 동기화 스레드 시작됨\n")
else:
print("\n⚠️ 자동 동기화가 비활성화되어 있습니다.")
print(" 활성화하려면 .env에서 ICAL_SYNC_ENABLED=true로 설정하세요.\n")
if __name__ == "__main__":
# 환경 변수에서 포트 읽기 (기본값: 8000)
port = int(os.environ.get("PORT", 8000))
# Transport 선택 (기본값: stdio, 환경 변수로 변경 가능)
transport = os.environ.get("TRANSPORT", "stdio")
print("=" * 70)
print("🎯 예약 플랫폼 통합 관리 MCP 서버")
print("=" * 70)
# 백그라운드 동기화 시작
start_background_sync()
if transport == "http":
# HTTP transport (Remote MCP 서버용)
print(f"\n🚀 Starting Remote MCP Server on http://0.0.0.0:{port}/mcp")
print("=" * 70 + "\n")
mcp.run(
transport="http",
host="0.0.0.0",
port=port,
path="/mcp"
)
elif transport == "sse":
# SSE transport (레거시 지원)
print(f"\n🚀 Starting SSE MCP Server on http://0.0.0.0:{port}/sse")
print("=" * 70 + "\n")
mcp.run(transport="sse", host="0.0.0.0", port=port)
else:
# STDIO transport (로컬 개발용, 기본값)
print("\n🚀 Starting Local MCP Server (STDIO)")
print("=" * 70 + "\n")
mcp.run()
