openClaw_agent_dm/skills/google-calendar/google_calendar.py

#!/usr/bin/env python3
"""
Google Calendar Skill for OpenClaw
提供日历读取和写入功能
"""

import json
import os
from datetime import datetime, timedelta
from typing import Optional, Dict, List, Any

try:
    from google.oauth2 import service_account
    from googleapiclient.discovery import build
    from googleapiclient.errors import HttpError
    GOOGLE_LIBS_AVAILABLE = True
except ImportError:
    GOOGLE_LIBS_AVAILABLE = False


class GoogleCalendarClient:
    """Google Calendar 客户端"""
    
    SCOPES = ['https://www.googleapis.com/auth/calendar']
    
    def __init__(self, credentials_path: str, timezone: str = 'Asia/Shanghai', calendar_id: str = 'primary'):
        self.credentials_path = credentials_path
        self.timezone = timezone
        self.calendar_id = calendar_id
        self.service = None
        self._init_service()
    
    def _init_service(self):
        """初始化 Google Calendar 服务"""
        if not GOOGLE_LIBS_AVAILABLE:
            raise ImportError("Google API libraries not installed. Run: pip install google-auth google-api-python-client")
        
        if not os.path.exists(self.credentials_path):
            raise FileNotFoundError(f"Credentials file not found: {self.credentials_path}")
        
        try:
            credentials = service_account.Credentials.from_service_account_file(
                self.credentials_path, scopes=self.SCOPES
            )
            self.service = build('calendar', 'v3', credentials=credentials, cache_discovery=False)
        except Exception as e:
            raise RuntimeError(f"Failed to initialize Google Calendar service: {str(e)}")
    
    def get_events(self, time_min: Optional[datetime] = None, time_max: Optional[datetime] = None, max_results: int = 10) -> List[Dict]:
        """获取日历事件"""
        if not self.service:
            return []
        
        try:
            now = datetime.utcnow().isoformat() + 'Z'
            
            events_result = self.service.events().list(
                calendarId=self.calendar_id,
                timeMin=time_min.isoformat() + 'Z' if time_min else now,
                timeMax=time_max.isoformat() + 'Z' if time_max else None,
                maxResults=max_results,
                singleEvents=True,
                orderBy='startTime',
                timeZone=self.timezone
            ).execute()
            
            events = events_result.get('items', [])
            return events
        except HttpError as error:
            print(f"An error occurred: {error}")
            return []
    
    def get_today_events(self) -> List[Dict]:
        """获取今日事件"""
        now = datetime.now()
        start_of_day = datetime(now.year, now.month, now.day)
        end_of_day = start_of_day + timedelta(days=1)
        return self.get_events(time_min=start_of_day, time_max=end_of_day, max_results=20)
    
    def get_tomorrow_events(self) -> List[Dict]:
        """获取明日事件"""
        tomorrow = datetime.now() + timedelta(days=1)
        start_of_day = datetime(tomorrow.year, tomorrow.month, tomorrow.day)
        end_of_day = start_of_day + timedelta(days=1)
        return self.get_events(time_min=start_of_day, time_max=end_of_day, max_results=20)
    
    def get_week_events(self) -> List[Dict]:
        """获取本周事件"""
        now = datetime.now()
        start_of_week = now - timedelta(days=now.weekday())
        start_of_week = datetime(start_of_week.year, start_of_week.month, start_of_week.day)
        end_of_week = start_of_week + timedelta(days=7)
        return self.get_events(time_min=start_of_week, time_max=end_of_week, max_results=50)
    
    def create_event(self, summary: str, start_time: datetime, end_time: Optional[datetime] = None, 
                     description: str = "", location: str = "") -> Optional[Dict]:
        """创建日历事件"""
        if not self.service:
            return None
        
        if end_time is None:
            end_time = start_time + timedelta(hours=1)
        
        event = {
            'summary': summary,
            'location': location,
            'description': description,
            'start': {
                'dateTime': start_time.isoformat(),
                'timeZone': self.timezone,
            },
            'end': {
                'dateTime': end_time.isoformat(),
                'timeZone': self.timezone,
            },
        }
        
        try:
            event = self.service.events().insert(
                calendarId=self.calendar_id, 
                body=event
            ).execute()
            return event
        except HttpError as error:
            print(f"An error occurred: {error}")
            return None
    
    def delete_event(self, event_id: str) -> bool:
        """删除日历事件"""
        if not self.service:
            return False
        
        try:
            self.service.events().delete(
                calendarId=self.calendar_id,
                eventId=event_id
            ).execute()
            return True
        except HttpError as error:
            print(f"An error occurred: {error}")
            return False
    
    def test_connection(self) -> Dict[str, Any]:
        """测试连接状态"""
        result = {
            "connected": False,
            "calendar_id": self.calendar_id,
            "timezone": self.timezone,
            "error": None
        }
        
        try:
            calendar = self.service.calendars().get(calendarId=self.calendar_id).execute()
            result["connected"] = True
            result["calendar_name"] = calendar.get('summary', 'Unknown')
            result["calendar_description"] = calendar.get('description', '')
        except Exception as e:
            result["error"] = str(e)
        
        return result


def format_events(events: List[Dict], title: str = "日历事件") -> str:
    """格式化事件列表为可读文本"""
    if not events:
        return f"📅 {title}: 暂无事件"
    
    lines = [f"📅 {title}:"]
    for event in events:
        summary = event.get('summary', '无标题')
        start = event.get('start', {})
        start_time = start.get('dateTime', start.get('date', '未知时间'))
        
        # 格式化时间
        try:
            dt = datetime.fromisoformat(start_time.replace('Z', '+00:00'))
            time_str = dt.strftime('%m/%d %H:%M')
        except:
            time_str = start_time
        
        location = event.get('location', '')
        location_str = f" @ {location}" if location else ""
        
        lines.append(f"  • {time_str} {summary}{location_str}")
    
    return '\n'.join(lines)


# 命令处理函数 (供 OpenClaw 调用)
def handle_calendar_command(command: str, args: List[str], config: Dict) -> str:
    """处理日历命令"""
    try:
        client = GoogleCalendarClient(
            credentials_path=config.get('credentials_path', '/root/.openclaw/credentials/google-calendar-life.json'),
            timezone=config.get('timezone', 'Asia/Shanghai'),
            calendar_id=config.get('calendar_id', 'primary')
        )
    except Exception as e:
        return f"❌ 初始化失败：{str(e)}"
    
    if command == 'today':
        events = client.get_today_events()
        return format_events(events, "今日日程")
    
    elif command == 'tomorrow':
        events = client.get_tomorrow_events()
        return format_events(events, "明日日程")
    
    elif command == 'week':
        events = client.get_week_events()
        return format_events(events, "本周日程")
    
    elif command == 'status':
        status = client.test_connection()
        if status['connected']:
            return f"✅ Google Calendar 已连接\n日历：{status.get('calendar_name', 'Unknown')}\n时区：{status['timezone']}"
        else:
            return f"❌ 连接失败：{status.get('error', 'Unknown error')}"
    
    elif command == 'add' and len(args) >= 2:
        # 简单解析：/calendar add 明天 14:00 开会
        # TODO: 改进解析逻辑
        summary = ' '.join(args[2:]) if len(args) > 2 else '新事件'
        start_time = datetime.now() + timedelta(hours=1)
        event = client.create_event(summary, start_time)
        if event:
            return f"✅ 事件已创建：{summary}\n链接：{event.get('htmlLink', '')}"
        else:
            return "❌ 创建事件失败"
    
    elif command == 'help':
        return """📅 Google Calendar 命令帮助:
  /calendar today - 查看今日日程
  /calendar tomorrow - 查看明日日程
  /calendar week - 查看本周日程
  /calendar status - 检查连接状态
  /calendar add <时间> <事件> - 添加新事件
  /calendar help - 显示帮助"""
    
    else:
        return f"❌ 未知命令：{command}\n使用 /calendar help 查看帮助"


if __name__ == '__main__':
    # 测试
    import sys
    if len(sys.argv) > 1:
        cmd = sys.argv[1]
        args = sys.argv[2:]
        config = {
            'credentials_path': '/root/.openclaw/credentials/google-calendar-life.json',
            'timezone': 'Asia/Shanghai'
        }
        result = handle_calendar_command(cmd, args, config)
        print(result)
    else:
        print("Usage: python google_calendar.py <command> [args]")
        print("Commands: today, tomorrow, week, status, add, help")