Eason (陈医生)
1 month ago
parent
fe762f2b2a
commit
c3b41bc1d0
5 changed files with 421 additions and 0 deletions
@ -0,0 +1,258 @@ |
||||
#!/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") |
||||
@ -0,0 +1,2 @@ |
||||
google-auth>=2.0.0 |
||||
google-api-python-client>=2.0.0 |
||||
@ -0,0 +1,29 @@ |
||||
{ |
||||
"name": "google-calendar", |
||||
"version": "1.0.0", |
||||
"description": "Google Calendar 集成 - 读取和写入用户日程", |
||||
"author": "OpenClaw Team", |
||||
"enabled": true, |
||||
"commands": [ |
||||
{ |
||||
"name": "calendar", |
||||
"description": "日历管理命令", |
||||
"handler": "google_calendar.handle_calendar_command", |
||||
"usage": "/calendar <today|tomorrow|week|add|delete> [参数]", |
||||
"examples": [ |
||||
"/calendar today", |
||||
"/calendar tomorrow", |
||||
"/calendar add 明天 14:00 开会" |
||||
] |
||||
} |
||||
], |
||||
"config": { |
||||
"credentials_path": "/root/.openclaw/credentials/google-calendar-life.json", |
||||
"timezone": "Asia/Shanghai", |
||||
"calendar_id": "primary" |
||||
}, |
||||
"dependencies": [ |
||||
"google-auth", |
||||
"google-api-python-client" |
||||
] |
||||
} |
||||
Loading…
Reference in new issue