diff --git a/app.py b/app.py index 0a6a5a1..68511f6 100644 --- a/app.py +++ b/app.py @@ -1,8 +1,10 @@ -from flask import Flask, render_template, request, jsonify +from flask import Flask, render_template, request, jsonify, session +import os from calendar_agent_deepseek import CalendarAgentDeepSeek app = Flask(__name__) +app.secret_key = os.getenv("FLASK_SECRET_KEY", "dev-secret-key") # Initialize the calendar agent try: @@ -35,7 +37,21 @@ def process_command(): }) try: - response = agent.process_command(user_input, selected_calendar) + # 获取历史消息（最多10条） + history = session.get('chat_history', []) + if not isinstance(history, list): + history = [] + + # 调用代理，传入历史消息 + response = agent.process_command(user_input, selected_calendar, history=history) + + # 追加当前轮对话到历史并裁剪长度 + history.append({'role': 'user', 'content': user_input}) + history.append({'role': 'assistant', 'content': response}) + if len(history) > 10: + history = history[-10:] + session['chat_history'] = history + return jsonify({ 'success': True, 'message': response diff --git a/caldav_client.py b/caldav_client.py index 5eee64b..3d0a1fc 100644 --- a/caldav_client.py +++ b/caldav_client.py @@ -1,9 +1,10 @@ import caldav -from icalendar import Calendar, Event, vCalAddress, vText +from icalendar import Calendar, Event, Todo, Alarm, vCalAddress, vText from datetime import datetime, timedelta import pytz import os from typing import List, Dict, Optional +from datetime import datetime as _dt class AppleCalendarClient: def __init__(self, server_url: str, username: str, password: str): @@ -63,7 +64,8 @@ def create_event(self, end_time: datetime, description: str = "", location: str = "", - calendar_name: str = None) -> str: + calendar_name: str = None, + priority: str = None) -> str: """Create a new calendar event""" calendar = self.get_calendar_by_name(calendar_name) if calendar_name else self.get_default_calendar() @@ -72,20 +74,89 @@ def create_event(self, event = Event() event.add('summary', title) - event.add('dtstart', start_time) - event.add('dtend', end_time) + event.add('dtstart', self._to_utc(start_time)) + event.add('dtend', self._to_utc(end_time)) if description: event.add('description', description) if location: event.add('location', location) + if priority is not None: + event.add('priority', priority) # Add creation timestamp event.add('dtstamp', datetime.now(pytz.UTC)) # Save event calendar_event = calendar.save_event(event.to_ical()) - return calendar_event.url + print(f"event的形式是{event}") # 调试 + return event + + def create_event_with_alarm(self, + title: str, + start_time: datetime, + end_time: datetime, + alarm: Dict = None, + description: str = "", + location: str = "", + calendar_name: str = None, + priority: str = None) -> str: + """Create a new calendar event with VALARM subcomponent""" + calendar = self.get_calendar_by_name(calendar_name) if calendar_name else self.get_default_calendar() + + if not calendar: + raise ValueError("No calendar available") + + event = Event() + event.add('summary', title) + event.add('dtstart', self._to_utc(start_time)) + event.add('dtend', self._to_utc(end_time)) + + if description: + event.add('description', description) + if location: + event.add('location', location) + if priority is not None: + event.add('priority', priority) + + # Add creation timestamp + event.add('dtstamp', datetime.now(pytz.UTC)) + + alarms = alarm if isinstance(alarm, list) else ([alarm] if isinstance(alarm, dict) else []) + for a in alarms: + try: + cal_alarm = Alarm() + cal_alarm.add('action', (a.get('action') or 'DISPLAY')) + trig_value, trig_params = self._parse_trigger(a.get('trigger')) + from datetime import datetime as dt + if isinstance(trig_value, dt) and trig_value.tzinfo is None: + try: + local_tz = datetime.now().astimezone().tzinfo + tz_to_use = start_time.tzinfo or local_tz + trig_value = trig_value.replace(tzinfo=tz_to_use) + except Exception: + pass + if trig_value is not None: + if trig_params: + cal_alarm.add('trigger', trig_value, parameters={'related': trig_params.get('related')}) + else: + cal_alarm.add('trigger', trig_value) + if a.get('description'): + cal_alarm.add('description', a.get('description')) + if a.get('repeat') is not None: + cal_alarm.add('repeat', a.get('repeat')) + if a.get('duration'): + cal_alarm.add('duration', a.get('duration')) + if a.get('attach'): + cal_alarm.add('attach', a.get('attach')) + event.add_component(cal_alarm) + print(f"infomation of event: {event}") + except Exception as e: + print(f"附加 VALARM 失败: {e}") + + # Save event + calendar_event = calendar.save_event(event.to_ical()) + return event def get_events(self, start_date: datetime = None, @@ -95,7 +166,7 @@ def get_events(self, calendar = self.get_calendar_by_name(calendar_name) if calendar_name else self.get_default_calendar() if not calendar: - return [] + return [], [] # Default to today if no dates provided if not start_date: @@ -104,30 +175,36 @@ def get_events(self, end_date = start_date + timedelta(days=1) events = calendar.date_search(start=start_date, end=end_date) - + original_ical_data = [] parsed_events = [] for event in events: ical_data = event.icalendar_component + original_ical_data.append(ical_data) + print(f'原始ical数据: {ical_data}') event_data = { 'id': event.url, 'title': str(ical_data.get('summary', '')), 'start': ical_data.get('dtstart').dt if ical_data.get('dtstart') else None, 'end': ical_data.get('dtend').dt if ical_data.get('dtend') else None, - 'description': str(ical_data.get('description', '')), - 'location': str(ical_data.get('location', '')) + 'description': str(ical_data.get('description', '')) if ical_data.get('description') else None, + 'location': str(ical_data.get('location', '')) if ical_data.get('location') else None, + 'priority': str(ical_data.get('priority', '')) if ical_data.get('priority') else None, + 'alarms': list(ical_data.walk('valarm')) } parsed_events.append(event_data) - return parsed_events + return original_ical_data, parsed_events def update_event(self, event_id: str, title: str = None, start_time: datetime = None, end_time: datetime = None, + alarm: Optional[List[Dict]] = None, description: str = None, location: str = None, - calendar_name: str = None) -> bool: + calendar_name: str = None, + priority: Optional[int] = None) -> bool: """Update an existing event""" try: # If calendar_name is specified, only search in that calendar @@ -141,23 +218,68 @@ def update_event(self, for event in events: if event.url == event_id: ical_data = event.icalendar_component - # Update fields if title: ical_data['summary'] = title if start_time: - ical_data['dtstart'] = start_time + ical_data['dtstart'] = self._to_utc(start_time) if end_time: - ical_data['dtend'] = end_time + ical_data['dtend'] = self._to_utc(end_time) if description is not None: ical_data['description'] = description if location is not None: ical_data['location'] = location - - # Save updated event - event.data = ical_data.to_ical() + if priority is not None: + ical_data['priority'] = priority + if alarm: + alarms = alarm + elif ical_data.walk('valarm'): + alarms = [] + for al in ical_data.walk('valarm'): + new_alarm = {} + new_alarm['action'] = str(al.get('action')) if al.get('action') else 'DISPLAY' + trig = al.get('trigger') + new_alarm['trigger'] = trig.dt if getattr(trig, 'dt', None) is not None else trig + if al.get('description'): + new_alarm['description'] = str(al.get('description')) + else: + new_alarm['description'] = None + alarms.append(new_alarm) + else: + alarms = None + + try: + event.delete() + success = True + except Exception: + success = False + + if not success: + return False + print(f'更新后的ical数据: {ical_data}') + if not alarms: + self.create_event( + title = str(ical_data.get('summary', '')), + start_time = (ical_data.get('dtstart') if ical_data.get('dtstart') else None), + end_time = (ical_data.get('dtend') if ical_data.get('dtend') else None), + description = (str(ical_data.get('description', '')) if ical_data.get('description') else None), + location = (str(ical_data.get('location', '')) if ical_data.get('location') else None), + priority = (int(str(ical_data.get('priority'))) if ical_data.get('priority') else None), + calendar_name = (calendar.name if hasattr(calendar, 'name') else None), + ) + else: + self.create_event_with_alarm( + title = str(ical_data.get('summary', '')), + start_time = (ical_data.get('dtstart') if ical_data.get('dtstart') else None), + end_time = (ical_data.get('dtend') if ical_data.get('dtend') else None), + alarm = alarms, + description = (str(ical_data.get('description', '')) if ical_data.get('description') else None), + location = (str(ical_data.get('location', '')) if ical_data.get('location') else None), + priority = (int(str(ical_data.get('priority'))) if ical_data.get('priority') else None), + calendar_name = (calendar.name if hasattr(calendar, 'name') else None), + ) return True - + print(f"Event not found: {event_id}") return False except Exception as e: @@ -190,44 +312,239 @@ def delete_event(self, event_id: str, calendar_name: str = None) -> bool: print(f"Error deleting event: {e}") return False - def search_events(self, query: str, calendar_name: str = None) -> List[Dict]: - """Search events by title or description, only return future and today's events""" + def _filter_event(self, event, search_info: Dict, start, end, match_mode, exclude_all_day): + """ + 过滤单个事件，根据搜索条件判断是否匹配 + """ + ical_data = event.icalendar_component + title = str(ical_data.get('summary', '')).lower() + desc = str(ical_data.get('description', '')).lower() + + # 关键词过滤 + keyword = search_info.get('keyword') if search_info else None + if keyword and not (str(keyword).lower() in title or str(keyword).lower() in desc): + return None + + # 暂时不对比时间 + # # 提取事件时间 + # ev_start_prop = ical_data.get('dtstart') + # ev_end_prop = ical_data.get('dtend') + # ev_start = ev_start_prop.dt if ev_start_prop is not None else None + # ev_end = ev_end_prop.dt if ev_end_prop is not None else None + + # # 时区对齐 + # ev_start, start_cmp = self._align_timezone(ev_start, start) + # ev_end, end_cmp = self._align_timezone(ev_end, end) + + # # 时间范围匹配 + # if not self._is_event_in_range(ev_start, ev_end, start_cmp, end_cmp, match_mode): + # return None + # print(f"事件数据：{event.data}") + # 构建返回数据 + return ical_data + + def _is_all_day_event(self, ev_start, ev_end): + """检测全天事件（支持date类型和datetime型全天事件）""" + from datetime import datetime, date, time + + # 传统检测：date类型 + if ev_start is not None and not isinstance(ev_start, datetime): + return True + if ev_end is not None and not isinstance(ev_end, datetime): + return True + + # iCloud风格检测：datetime类型但时间范围为00:00-23:59 + if (isinstance(ev_start, datetime) and isinstance(ev_end, datetime) and + ev_start.time() == time(0, 0) and ev_end.time() == time(23, 59)): + return True + + # 持续时间检测：接近24小时的事件 + if (isinstance(ev_start, datetime) and isinstance(ev_end, datetime) and + (ev_end - ev_start).total_seconds() >= 86340): # 23小时59分钟 + return True + + return False + + def _align_timezone(self, event_time, reference_time): + """时区对齐，避免aware/naive比较错误""" + from datetime import datetime + + if not isinstance(event_time, datetime) or not isinstance(reference_time, datetime): + return event_time, reference_time + + if event_time.tzinfo is not None and reference_time.tzinfo is None: + return event_time, reference_time.replace(tzinfo=event_time.tzinfo) + if event_time.tzinfo is None and reference_time.tzinfo is not None: + return event_time.replace(tzinfo=reference_time.tzinfo), reference_time + + return event_time, reference_time + + def _is_event_in_range(self, ev_start, ev_end, search_start, search_end, match_mode): + """判断事件是否在搜索时间范围内""" + print(f"事件开始时间是 {ev_start}, 结束时间是 {ev_end}") + print(f"搜索开始时间是 {search_start}, 搜索结束时间是 {search_end}") + if match_mode == 'precise': + # 精确匹配：事件完全在搜索范围内 + if ev_start is not None and ev_end is not None: + return (ev_start >= search_start) and (ev_end <= search_end) + return False + else: + # 重叠匹配：事件与搜索范围有重叠 + if ev_start is not None and ev_end is not None: + return (ev_start < search_end) and (ev_end > search_start) + elif ev_start is not None and ev_end is None: + return ev_start < search_end + elif ev_end is not None and ev_start is None: + return ev_end > search_start + else: + return False + + def search_events(self, search_info: Dict, calendar_name: str = None) -> tuple[List, List]: + """ + 根据关键词和时间范围搜索事件 + + Args: + search_info: 搜索参数，包含keyword、start_time、end_time、match_mode、exclude_all_day + calendar_name: 日历名称，为空时使用默认日历 + + Returns: + List[Dict]: 匹配的事件列表 + """ from datetime import datetime calendar = self.get_calendar_by_name(calendar_name) if calendar_name else self.get_default_calendar() - if not calendar: - return [] - - # 只搜索近3个月及未来的事件，避免加载所有历史事件 - current_time = datetime.now() - three_months_ago = current_time.replace(day=1) - timedelta(days=90) - + return [], [] + + # 解析时间范围 + info = search_info or {} + + if search_info.get('start_time'): + start_date = search_info['start_time'] + if isinstance(start_date, str): + start_date = datetime.fromisoformat(start_date) + + # Use provided end_time if available, otherwise default to end of day + if search_info.get('end_time'): + end_date = search_info['end_time'] + if isinstance(end_date, str): + end_date = datetime.fromisoformat(end_date) + else: + end_date = start_date.replace(hour=23, minute=59, second=59) + else: + # Default to today + start_date = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) + end_date = start_date.replace(hour=23, minute=59, second=59) + + + print(f"搜索时间范围: {start_date} 到 {end_date}") # 调试信息 + try: # 使用日期范围搜索，避免加载所有事件 - events = calendar.date_search(start=three_months_ago, end=current_time + timedelta(days=365)) - matching_events = [] - + events = calendar.date_search(start=start_date, end=end_date) + original_ical_data = [] + parsed_events = [] + + # # 解析匹配模式和全天事件过滤设置 + # from datetime import datetime as _dt, timedelta as _td + # match_mode = (info.get('match_mode') or 'overlap').lower() + # exclude_all_day = info.get('exclude_all_day', False) + + # # 智能默认设置：短时间范围自动使用精确匹配和排除全天事件 + # if isinstance(start, _dt) and isinstance(end, _dt): + # duration = end - start + # if info.get('match_mode') is None and duration <= _td(hours=6): + # match_mode = 'precise' + # if info.get('exclude_all_day') is None and duration <= _td(hours=12): + # exclude_all_day = True + + # 过滤事件 + match_mode = search_info.get('match_mode') or 'overlap' + exclude_all_day = search_info.get('exclude_all_day', False) + for event in events: - ical_data = event.icalendar_component - title = str(ical_data.get('summary', '')).lower() - desc = str(ical_data.get('description', '')).lower() - - if query.lower() in title or query.lower() in desc: - event_data = { + print(event.data) + ical_data = self._filter_event(event, search_info, start_date, end_date, match_mode, exclude_all_day) + if ical_data: + parsed_data = { 'id': event.url, 'title': str(ical_data.get('summary', '')), 'start': ical_data.get('dtstart').dt if ical_data.get('dtstart') else None, 'end': ical_data.get('dtend').dt if ical_data.get('dtend') else None, - 'description': str(ical_data.get('description', '')), - 'location': str(ical_data.get('location', '')) + 'description': str(ical_data.get('description', '')) if ical_data.get('description') else None, + 'location': str(ical_data.get('location', '')) if ical_data.get('location') else None, + 'priority': str(ical_data.get('priority', '')) if ical_data.get('priority') else None, + 'alarm': list(ical_data.walk('VALARM')) if ical_data.walk('VALARM') else None, } - matching_events.append(event_data) + parsed_events.append(parsed_data) + original_ical_data.append(ical_data) - return matching_events + return original_ical_data, parsed_events except Exception as e: print(f"搜索事件时出错: {e}") - return [] + return [], [] + + def add_alarm_to_item(self, event_id: str, alarm: Dict, calendar_name: str = None) -> bool: + """Attach a VALARM to an existing VEVENT identified by URL.""" + try: + calendars_to_search = [self.get_calendar_by_name(calendar_name)] if calendar_name else self.calendars + for calendar in calendars_to_search: + if not calendar: + continue + items = [] + # Search both events and todos + if hasattr(calendar, 'events'): + try: + items.extend(calendar.events()) + except Exception: + pass + + for item in items: + if item.url == event_id: + ical_data = item.icalendar_component + try: + cal_alarm = Alarm() + cal_alarm.add('action', alarm.get('action') or 'DISPLAY') + trig_value, trig_params = self._parse_trigger(alarm.get('trigger')) + from datetime import datetime as dt + if isinstance(trig_value, dt) and trig_value.tzinfo is None: + try: + ev_start = None + try: + ev_start_prop = ical_data.get('dtstart') + ev_start = ev_start_prop.dt if ev_start_prop is not None else None + except Exception: + ev_start = None + local_tz = datetime.now().astimezone().tzinfo + tz_to_use = (ev_start.tzinfo if isinstance(ev_start, dt) and ev_start.tzinfo is not None else local_tz) + trig_value = trig_value.replace(tzinfo=tz_to_use) + except Exception: + pass + if trig_value is not None: + if trig_params: + cal_alarm.add('trigger', trig_value, parameters={'related': trig_params.get('related')}) + else: + cal_alarm.add('trigger', trig_value) + if alarm.get('description'): + cal_alarm.add('description', alarm.get('description')) + if alarm.get('repeat') is not None: + cal_alarm.add('repeat', alarm.get('repeat')) + if alarm.get('duration'): + cal_alarm.add('duration', alarm.get('duration')) + if alarm.get('attach'): + cal_alarm.add('attach', alarm.get('attach')) + + ical_data.add_component(cal_alarm) + item.data = ical_data.to_ical() + return True + except Exception as e: + print(f"添加 VALARM 失败: {e}") + return False + print(f"未找到可添加提醒的项目: {event_id}") + return False + except Exception as e: + print(f"Error adding VALARM: {e}") + return False def get_calendar_by_name(self, name: str): """Get calendar by name""" @@ -236,6 +553,63 @@ def get_calendar_by_name(self, name: str): return cal return None + def _to_utc(self, dt): + from datetime import datetime as dtcls, date as dcls + if isinstance(dt, dtcls): + if dt.tzinfo is None: + try: + local_tz = datetime.now().astimezone().tzinfo + except Exception: + local_tz = pytz.UTC + dt = dt.replace(tzinfo=local_tz) + return dt.astimezone(pytz.UTC) + return dt + + def _parse_trigger(self, value): + try: + from datetime import timedelta + from datetime import datetime as dt + import re + if value is None: + return None, {} + if isinstance(value, timedelta): + return value, {'related': 'START'} + if isinstance(value, dt): + return value, {} + if isinstance(value, str): + s = value.strip() + sign = -1 if s.startswith('-') else 1 + if s.startswith('-'): + s = s[1:] + if s.startswith('P'): + days = 0 + hours = 0 + minutes = 0 + seconds = 0 + m = re.search(r'P(\d+)D', s) + if m: + days = int(m.group(1)) + if 'T' in s: + t = s.split('T')[1] + mh = re.search(r'(\d+)H', t) + mm = re.search(r'(\d+)M', t) + ms = re.search(r'(\d+)S', t) + hours = int(mh.group(1)) if mh else 0 + minutes = int(mm.group(1)) if mm else 0 + seconds = int(ms.group(1)) if ms else 0 + td = timedelta(days=days, hours=hours, minutes=minutes, seconds=seconds) + td = timedelta(seconds=sign * td.total_seconds()) + return td, {'related': 'START'} + try: + from dateutil.parser import parse + dtv = parse(s) + return dtv, {} + except Exception: + return None, {} + return None, {} + except Exception: + return None, {} + # Helper functions for date parsing def parse_natural_date(date_str: str) -> datetime: """Parse natural language dates like 'tomorrow 3pm', 'next Monday', etc.""" diff --git a/calendar_agent_deepseek.py b/calendar_agent_deepseek.py index e92da68..af8cfc7 100644 --- a/calendar_agent_deepseek.py +++ b/calendar_agent_deepseek.py @@ -1,8 +1,11 @@ import os import json -from typing import Dict, List, Optional +from typing import Dict, List, Optional, Tuple from datetime import datetime + +from urllib3 import response + from caldav_client import AppleCalendarClient from deepseek_parser import DeepSeekCalendarParser @@ -73,7 +76,7 @@ def _load_config(self) -> Dict: except Exception as e: raise ValueError(f"读取配置文件失败: {e}") - def process_command(self, user_input: str, selected_calendar: str = None) -> str: + def process_command(self, user_input: str, selected_calendar: str = None, history: Optional[list] = None) -> str: """ Process natural language command using DeepSeek and execute calendar operation @@ -86,42 +89,80 @@ def process_command(self, user_input: str, selected_calendar: str = None) -> str try: print(f"🔍 Processing command: {user_input}") # Parse user intent and extract details using DeepSeek - parsed_intent = self.nlp_parser.parse_command(user_input) - print(f"🔍 Parsed intent: {parsed_intent}") - - if not parsed_intent.get('intent'): - print(f"❌ No intent detected for: {user_input}") - return "抱歉，我没有理解您的指令。请尝试使用更清晰的表达，比如：'创建明天下午3点的会议' 或 '查看今天的日程'" - - intent = parsed_intent['intent'] - - if intent == 'create': - return self._handle_create_event(parsed_intent, selected_calendar) - elif intent == 'read': - return self._handle_read_events(parsed_intent, selected_calendar) - elif intent == 'update': - return self._handle_update_event(parsed_intent, selected_calendar) - elif intent == 'delete': - return self._handle_delete_event(parsed_intent, selected_calendar) - else: - return f"暂不支持的操作: {intent}" + now = datetime.now().astimezone() + current_time_zone = now.tzinfo + UTC_offset = now.utcoffset() + # user_input = user_input + f"当前时区为{current_time_zone}，UTC偏移为{UTC_offset}" + parsed = self.nlp_parser.parse_command(user_input, history=history) + print(f"🔍 Parsed: {parsed}") + + assistant_message = parsed.get('assistant_message') if isinstance(parsed, dict) else None + payload = parsed.get('payload') if isinstance(parsed, dict) else parsed + # normalize search_info from top-level or payload, and parse if stringified + search_info = None + if isinstance(parsed, dict) and parsed.get('search_info') is not None: + search_info = parsed.get('search_info') + elif isinstance(payload, dict) and payload.get('search_info') is not None: + search_info = payload.get('search_info') + try: + if isinstance(search_info, str): + import json as _json + search_info = _json.loads(search_info) + except Exception: + pass + + # 多轮对话：当 intent 为空时，不执行任何增删查改，仅返回 assistant_message + if not payload or payload.get('intent') is None: + return assistant_message or "" + + operation_result = None + + if payload.get('intent') is not None: + + intent = payload['intent'] + + if intent == 'create': + needComment, origin, operation_result = self._handle_create_event(payload, selected_calendar) + elif intent == 'read': + needComment, origin, operation_result = self._handle_read_events(payload, selected_calendar, search_info) + elif intent == 'update': + needComment, origin, operation_result = self._handle_update_event(payload, selected_calendar, search_info) + elif intent == 'delete': + needComment, origin, operation_result = self._handle_delete_event(payload, selected_calendar, search_info) + + # Build final response in order: KB -> assistant -> operation + # final_parts = [] + # if kb_msg: + # final_parts.append(kb_msg) + # if assistant_message: + # final_parts.append(assistant_message) + # if operation_result: + # final_parts.append(operation_result) + # 生成日程评价 - 添加一些温馨提醒 + if needComment: + comment = self.nlp_parser.generate_comment(operation_result, assistant_message) + # if comment: + # final_parts.append(comment) + # return "\n\n".join(final_parts) + return comment except Exception as e: return f"处理指令时出现错误: {str(e)}" - - def _handle_create_event(self, parsed_intent: Dict, selected_calendar: str = None) -> str: + + def _handle_create_event(self, parsed_intent: Dict, selected_calendar: str = None) -> Tuple[bool, List, str]: """Handle event creation""" - # Validate required fields + from datetime import datetime, timedelta + needComment = True + # Validate title if not parsed_intent.get('title'): - return "请提供事件的标题，例如：'创建和张三的会议'" + return needComment, [], "请提供标题，例如：'创建和张三的会议' 或 '添加提交作业的待办'" - if not parsed_intent.get('start_time'): - return "请提供事件的时间，例如：'明天下午3点'" + attrs = parsed_intent.get('component_attributes') or {} - # Set default end time if not provided - from datetime import datetime + # Default VEVENT creation + if not parsed_intent.get('start_time'): + return needComment, [], "请提供事件的时间，例如：'明天下午3点'" - # Handle both string and datetime objects start_time = parsed_intent['start_time'] if isinstance(start_time, str): start_time = datetime.fromisoformat(start_time) @@ -131,37 +172,55 @@ def _handle_create_event(self, parsed_intent: Dict, selected_calendar: str = Non if isinstance(end_time, str): end_time = datetime.fromisoformat(end_time) else: - # Default: 1 hour duration + # by default, set end time to 1 hour after start time end_time = start_time.replace(hour=start_time.hour + 1) - # Create the event - event_id = self.calendar_client.create_event( - title=parsed_intent['title'], - start_time=start_time, - end_time=end_time, - description=parsed_intent.get('description', ''), - location=parsed_intent.get('location', ''), - calendar_name=selected_calendar if selected_calendar is not None else None - ) + alarms = attrs.get('alarms') if isinstance(attrs.get('alarms'), list) else (attrs.get('alarms') and [attrs.get('alarms')]) + if alarms: + ical_data = self.calendar_client.create_event_with_alarm( + title=parsed_intent['title'], + start_time=start_time, + end_time=end_time, + alarm=alarms, + description=parsed_intent.get('description', ''), + location=parsed_intent.get('location', ''), + calendar_name=selected_calendar if selected_calendar is not None else None, + priority=attrs.get('priority') + ) + alarms = list(ical_data.walk('VALARM')) + else: + ical_data = self.calendar_client.create_event( + title=parsed_intent['title'], + start_time=start_time, + end_time=end_time, + description=parsed_intent.get('description', ''), + location=parsed_intent.get('location', ''), + calendar_name=selected_calendar if selected_calendar is not None else None, + priority=attrs.get('priority'), + ) + print(f"the created event data is {ical_data}") + response = f"✅ 已成功创建事件: {parsed_intent['title']}\n" \ + f"📅 时间: {start_time.strftime('%Y-%m-%d %H:%M')} - {end_time.strftime('%Y-%m-%d %H:%M')}\n" \ + f"📍 地点: {parsed_intent.get('location', '未指定')}\n" \ + f"📝 描述: {parsed_intent.get('description', '无')} \n " \ + f"🔔 提醒: {alarms if alarms else '无'}\n" \ + f"优先级: {attrs.get('priority', '无')}\n" \ - return f"✅ 已成功创建事件: {parsed_intent['title']}\n" \ - f"📅 时间: {start_time.strftime('%Y-%m-%d %H:%M')} - {end_time.strftime('%H:%M')}\n" \ - f"📍 地点: {parsed_intent.get('location', '未指定')}\n" \ - f"📝 描述: {parsed_intent.get('description', '无')}" + return needComment, ical_data, response - def _handle_read_events(self, parsed_intent: Dict, selected_calendar: str = None) -> str: + def _handle_read_events(self, parsed_intent: Dict, selected_calendar: str = None, search_info: Dict = None) -> Tuple[bool, List, str]: """Handle event reading/listing""" print(f"🔍 Reading events with parsed intent: {parsed_intent}") - + needComment = True # Determine time range for search - if parsed_intent.get('start_time'): - start_date = parsed_intent['start_time'] + if search_info.get('start_time'): + start_date = search_info['start_time'] if isinstance(start_date, str): start_date = datetime.fromisoformat(start_date) # Use provided end_time if available, otherwise default to end of day - if parsed_intent.get('end_time'): - end_date = parsed_intent['end_time'] + if search_info.get('end_time'): + end_date = search_info['end_time'] if isinstance(end_date, str): end_date = datetime.fromisoformat(end_date) else: @@ -175,142 +234,183 @@ def _handle_read_events(self, parsed_intent: Dict, selected_calendar: str = None # If searching for specific event if parsed_intent.get('title'): - events = self.calendar_client.search_events( - parsed_intent['title'], + ical_data, parse_events = self.calendar_client.search_events( + search_info, calendar_name=selected_calendar if selected_calendar is not None else None ) else: - events = self.calendar_client.get_events( + ical_data, parse_events = self.calendar_client.get_events( start_date=start_date, end_date=end_date, calendar_name=selected_calendar if selected_calendar is not None else None ) - - if not events: + + if not parse_events: # Determine which day we're querying for the message query_date = start_date.date() today = datetime.now().date() - - if query_date == today: - return "📅 今天没有安排任何事件" - elif query_date == today.replace(day=today.day + 1): - return "📅 明天没有安排任何事件" - else: - return f"📅 {query_date.strftime('%Y年%m月%d日')} 没有安排任何事件" - + return needComment, [], "📅 查询时间范围内没有安排任何事件或日程" + + needComment = True response = "📅 您的日程安排:\n\n" - for i, event in enumerate(events, 1): - start_str = event['start'].strftime('%H:%M') if event['start'] else '未知时间' - end_str = event['end'].strftime('%H:%M') if event['end'] else '未知时间' + for i, event in enumerate(parse_events): + start_str = event['start'].strftime('%Y-%m-%d %H:%M') if event['start'] else '未知时间' + end_str = event['end'].strftime('%Y-%m-%d %H:%M') if event['end'] else '未知时间' - response += f"{i}. {event['title']}\n" + response += f"{i+1}. {event['title']}\n" response += f" 时间: {start_str} - {end_str}\n" if event.get('location'): response += f" 地点: {event['location']}\n" if event.get('description'): response += f" 描述: {event['description']}\n" + if event.get('priority'): + response += f" 优先级: {event['priority']}\n" + if event.get('alarm'): + response += f" 提醒: {event['alarm']}\n" response += "\n" - return response.strip() + return needComment, ical_data, response.strip() - def _handle_update_event(self, parsed_intent: Dict, selected_calendar: str = None) -> str: + def _handle_update_event(self, parsed_intent: Dict, selected_calendar: str = None, search_info: Dict = None) -> Tuple[bool, List, str]: """Handle event updates""" - if not parsed_intent.get('target_event'): - # Try to find event by title - if parsed_intent.get('title'): - events = self.calendar_client.search_events(parsed_intent['title']) - if events: - parsed_intent['target_event'] = events[0]['id'] - else: - return "找不到指定的事件，请提供更具体的信息" - else: - return "请指定要更新的事件，例如：'修改和张三的会议时间'" - - # Update the event - success = self.calendar_client.update_event( - event_id=parsed_intent['target_event'], - title=parsed_intent.get('title'), - start_time=parsed_intent.get('start_time'), - end_time=parsed_intent.get('end_time'), - description=parsed_intent.get('description'), - location=parsed_intent.get('location'), - calendar_name=selected_calendar if selected_calendar is not None else None - ) + needComment = True + from datetime import datetime + if parsed_intent.get('title'): + original_ical_data, parse_events = self.calendar_client.search_events( + search_info, + calendar_name=selected_calendar if selected_calendar is not None else None + ) - if success: - return "✅ 事件已成功更新" + print(f"🔍 搜索到的事件: {parse_events}") # 调试信息 + if len(original_ical_data) > 1: + response = f"查询到的可能的事件如下：\n" + for i, event in enumerate(parse_events): + start_str = event['start'].strftime('%Y-%m-%d %H:%M') if event['start'] else '未知时间' + end_str = event['end'].strftime('%Y-%m-%d %H:%M') if event['end'] else '未知时间' + + response += f"{i+1}. {event['title']}\n" + response += f" 时间: {start_str} - {end_str}\n" + if event.get('location'): + response += f" 地点: {event['location']}\n" + if event.get('description'): + response += f" 描述: {event['description']}\n" + if event.get('priority'): + response += f" 优先级: {event['priority']}\n" + if event.get('alarm'): + response += f" 提醒: {event['alarm']}\n" + response += "\n" + response += "请提供更具体的事件信息。" + print(f"response: {response}") + return needComment, [], response.strip() + + if parse_events and original_ical_data: + success = False + # 优先使用事件对象（包含url） + first_obj = parse_events[0] if parse_events else None + success = False + if first_obj is not None: + event_id = first_obj.get('id') + print(f"🔄 目标事件URL: {event_id}") + st = parsed_intent.get('start_time') + en = parsed_intent.get('end_time') + if isinstance(st, datetime): + pass + elif isinstance(st, str): + try: + st = datetime.fromisoformat(st) + except Exception: + st = parsed_intent.get('start_time') + if isinstance(en, datetime): + pass + elif isinstance(en, str): + try: + en = datetime.fromisoformat(en) + except Exception: + en = parsed_intent.get('end_time') + + success = self.calendar_client.update_event( + event_id=event_id, + title=parsed_intent.get('title'), + start_time=st, + end_time=en, + alarm=parsed_intent.get('component_attributes').get('alarms'), + description=parsed_intent.get('description'), + location=parsed_intent.get('location'), + calendar_name=selected_calendar if selected_calendar is not None else None, + priority=(parsed_intent.get('priority') or (parsed_intent.get('component_attributes') or {}).get('priority')), + ) + + if success: + new_title = parsed_intent.get('title') or first_obj.get('title') if first_obj else None + start_val = st if st else (first_obj.get('start') if first_obj else None) + end_val = en if en else (first_obj.get('end') if first_obj else None) + desc_val = parsed_intent.get('description') or (first_obj.get('description') if first_obj else None) + loc_val = parsed_intent.get('location') or (first_obj.get('location') if first_obj else None) + pri_val = parsed_intent.get('priority') or (first_obj.get('priority') if first_obj else None) + response = f"✅ 事件已成功更新 \n更新内容:\n" + if new_title is not None: + response += f" 标题: {str(new_title)}\n" + if start_val is not None: + try: + response += f" 开始时间: {start_val.strftime('%Y-%m-%d %H:%M')}\n" + except Exception: + response += f" 开始时间: {str(start_val)}\n" + if end_val is not None: + try: + response += f" 结束时间: {end_val.strftime('%Y-%m-%d %H:%M')}\n" + except Exception: + response += f" 结束时间: {str(end_val)}\n" + if desc_val is not None: + response += f" 描述: {str(desc_val)}\n" + if loc_val is not None: + response += f" 地点: {str(loc_val)}\n" + if pri_val is not None: + response += f" 优先级: {str(pri_val)}\n" + return needComment, [], response.strip() + else: + return needComment, [], "找不到指定的事件，请提供更具体的信息" else: - return "❌ 更新事件失败，请检查事件ID是否正确" - - def _handle_delete_event(self, parsed_intent: Dict, selected_calendar: str = None) -> str: + return needComment, [], "请指定要更新的事件，例如：'修改和张三的会议时间'" + + def _handle_delete_event(self, parsed_intent: Dict, selected_calendar: str = None, search_info: Dict = None) -> Tuple[bool, List, str]: """Handle event deletion""" print(f"尝试删除事件: {parsed_intent.get('target_event')}") - - # Handle "all" target_event (delete all matching events) - if parsed_intent.get('target_event') == 'all': - if parsed_intent.get('title'): - # Delete all events with matching title - events = self.calendar_client.search_events( - parsed_intent['title'], - calendar_name=selected_calendar if selected_calendar is not None else None - ) - if events: - deleted_count = 0 - for event in events: - print(f"找到事件，开始删除: {event['id']}") - if self.calendar_client.delete_event( - event['id'], - calendar_name=selected_calendar if selected_calendar is not None else None - ): - deleted_count += 1 - - if deleted_count > 0: - return f"✅ 已成功删除 {deleted_count} 个事件" - else: - return "❌ 删除事件失败" + needComment = True + # Delete all events with matching title + original_ical_data, parsed_events = self.calendar_client.search_events( + search_info, + calendar_name=selected_calendar if selected_calendar is not None else None + ) + events = parsed_events + if events: + mode = ((search_info or {}).get('match_mode') or '').lower() + if mode == 'all': + deleted_count = 0 + for event in events: + eid = event.get('id') + print(f"找到事件，开始删除: {eid}") + if eid and self.calendar_client.delete_event( + eid, + calendar_name=selected_calendar if selected_calendar is not None else None + ): + deleted_count += 1 + + if deleted_count > 0: + return needComment, [], f"✅ 已成功删除 {deleted_count} 个事件" else: - return "找不到指定的事件" + return needComment, [], "❌ 删除事件失败" else: - return "请指定要删除的事件标题，例如：'删除所有会议'" - - # Handle specific event deletion - if not parsed_intent.get('target_event'): - # Try to find event by title - if parsed_intent.get('title'): - events = self.calendar_client.search_events( - parsed_intent['title'], - calendar_name=selected_calendar if selected_calendar is not None else None - ) - if events: - print(f"找到事件: {events}") - deleted_count = 0 - for event in events: - print(f"找到事件，开始删除: {event['id']}") - if self.calendar_client.delete_event( - event['id'], + first = events[0] + eid = first.get('id') + if eid and self.calendar_client.delete_event( + eid, calendar_name=selected_calendar if selected_calendar is not None else None ): - deleted_count += 1 - - if deleted_count > 0: - return f"✅ 已成功删除 {deleted_count} 个事件" - else: - return "❌ 删除事件失败" + return needComment, [], "✅ 已成功删除事件" else: - return "找不到指定的事件，请提供更具体的信息" - else: - return "请指定要删除的事件，例如：'删除和张三的会议'" - - # Delete specific event by ID - success = self.calendar_client.delete_event( - parsed_intent['target_event'], - calendar_name=selected_calendar if selected_calendar is not None else None - ) - - if success: - return "✅ 事件已成功删除" + return needComment, [], "❌ 删除事件失败" else: - return "❌ 删除事件失败，请检查事件ID是否正确" + return needComment, [], "找不到指定的事件，请提供更具体的信息" def get_calendar_list(self) -> List[str]: """Get list of available calendars""" diff --git a/deepseek_parser.py b/deepseek_parser.py index 469cdc2..48857e7 100644 --- a/deepseek_parser.py +++ b/deepseek_parser.py @@ -24,41 +24,90 @@ def __init__(self, api_key: str = None): self.api_url = "https://api.deepseek.com/v1/chat/completions" - # System prompt for calendar parsing - self.system_prompt = """你是一个专业的日历助理，专门解析用户对日历事件的指令。 - -请将用户的自然语言指令解析为结构化的JSON格式，包含以下字段： -- intent: 操作意图 (create, read, update, delete) -- title: 事件标题 -- start_time: 开始时间 (ISO格式: YYYY-MM-DDTHH:MM:SS) -- end_time: 结束时间 (ISO格式: YYYY-MM-DDTHH:MM:SS) -- description: 事件描述 -- location: 事件地点 -- target_event: 目标事件ID (用于更新/删除) - -时间解析规则： -- 使用当前时间作为参考：{current_time} -- 基于人类作息习惯合理判断"今天"和"明天"的含义 -- 凌晨时段（0-6点）的特殊处理： - - 如果用户在凌晨说"今天"，通常指的是已经开始的这一天 - - 如果用户在凌晨说"明天"，通常指的是即将到来的白天（即今天白天） - - 如果用户在凌晨说"后天"，通常指的是24小时后的白天 -- 对于查询指令（如"明天有什么事"）： - - 必须提供准确的start_time和end_time - - "明天"：start_time = 明天00:00:00，end_time = 明天23:59:59 - - "今天"：start_time = 今天00:00:00，end_time = 今天23:59:59 -- "下周" = 当前日期 + 7天 -- 对于创建指令，如果没有指定时间，默认为当前时间+1小时开始，持续1小时 - -意图识别： -- create: 创建、添加、安排、预定、新建 -- read: 查看、显示、列出、检查、看看、有什么事、日程安排 -- update: 更新、修改、改变、调整、重新安排 -- delete: 删除、取消、移除 - -重要：对于查询类指令（如"明天有什么事"），必须提供准确的时间范围，不能留空。 - -返回格式必须是纯JSON，不要有其他文本。""" + self.system_prompt = """你是名为 Alice 的顶级日历管理助手。目标：主动、可靠地帮助用户创建/查询/更新/删除日程与提醒，并尽量完整保留用户提供的信息。只输出一个纯 JSON，顶层包含 assistant_message 与 payload。 + +【输出格式】 +- 顶层字段：assistant_message（字符串，自然语言回复），payload（对象），search_info（对象或 null，用于事件搜索）。 +- payload 必含字段（缺失请用 null）：intent, title, start_time, end_time, description, location, component_type, component_attributes。 +- 所有时间使用 ISO 格式：YYYY-MM-DDTHH:MM:SS（适用于 start_time/end_time 以及 dtstart/dtend/due）。 +- 无法确认的信息一律使用 null，不要臆造值。 + +【行为准则】 +1) 模糊指令：时间或细节不明确（如“过几天”“下周左右”“复习”）时，intent=null；在 payload.description 列出待澄清要点，并用 assistant_message 简洁提示用户。 +2) 重要时间：若出现 DDL 或关键时间（00:00/12:00/23:59），在 payload.description 添加“重要提醒：…”，assistant_message 也需提示。 +3) 删除/修改日程：在检测到删除delete或者修改update意图的时候，payload里面尽量携带用户需要删除或者修改的行程的信息，比如”这周日两点开始的presentation时间改到下午三点钟“，这时候title就是presentation，开始时间是下午三点，可以依照默认规则，如果没有提到结束事件，就按照一个小时行程计算。 +4) 上下文一致：参考历史消息，避免冲突。若与历史内容冲突，assistant_message 提示用户确认；若是历史事件，不新增，只修改。 +5) 搜索关键词：若用户搜索或修改事件（如“明天的下午两点的会议改到三点”），请输出 search_info 为一个 JSON 对象（不要输出被转义的字符串）。示例：{"keyword": "会议", "start_time": "2025-01-01T14:00:00", "end_time": "2025-01-01T23:59:59", "match_mode": "precise", "exclude_all_day": true}。其中 keyword 为查询关键词；start_time/end_time 为可选的 ISO 时间范围，根据用户提供的信息决定，未指定则由代理默认查询近三个月；match_mode 在intent是"update"的时候可为 "overlap" 或 "precise"：overlap 表示只要与窗口有交集即可，precise 表示事件需完全落在窗口内， 在intent是"delete"的时候可以是"all" or "first", all代表删除所有匹配的事件，first代表第一个事件，考虑安全性，默认是first；exclude_all_day 为布尔值，true 时排除全天事件。 + +【类型判定（component_type）】 +- VEVENT：会议/约会/活动/事情/行程/日程/工作等一般场景。 +- UNKNOWN：如果无法用已知的信息来判定用户意图，则判定为 UNKNOWN。 + +【类型特例】 +- VEVENT：通常需要 start_time；若用户未给出时间，intent=null 并在 assistant_message 中提出澄清。 + +【属性提取（component_attributes）】缺失请使用 null： +- VEVENT: { summary, dtstart, dtend, location, description, rrule, attendees, organizer, uid, status, categories, priority, url, transp, sequence, created, dtstamp, last_modified, recurrence_id, x_apple_structured_location, x_apple_creator_identity, x_apple_creator_team_identity, x_apple_travel_advisory_behavior, alarms } + +【字段规范】 +- 时间类（dtstart/dtend/due/trigger）尽量使用 ISO 字符串；相对触发使用 iCalendar TRIGGER 语法（如 "-PT15M"、"-P1D"）。 +- 文本类（summary/location/description/status/categories）使用简明中英文。 +- 列表（attendees）返回人员姓名、邮箱或电话数组。 +- 未知或不可解析字段用 null。 + +【Apple/ICS 扩展（尽量完整保留用户信息）】 +- attendees：支持 "mailto:xxx@example.com" 或 "tel:+"；只给姓名时保留原文本；多位参与者返回数组。 +- organizer：同 attendees，支持 "mailto:" 或 "tel:"；不可解析则保留原文本。 +- url：原样存为字符串（含 URL 编码也保留）。 +- transp：透明度，常用 "OPAQUE" 或 "TRANSPARENT"。 +- sequence：整数，默认 0；当用户说明“更新版本/重新安排”时可自增。 +- created/dtstamp/last_modified：时间戳（ISO 字符串）；未给出则为 null（服务器可能自动填充 dtstamp/last-modified）。 +- recurrence_id：重复事件实例标识；仅在用户明确指定某个实例时填写，否则 null。 +- rrule：重复规则字符串（如 "FREQ=WEEKLY;INTERVAL=1;BYDAY=MO,WE,FR"）；用户给出“每周一三五/每月最后一天”等时尽量映射；无法确定则 null。 +- categories：字符串数组（如 ["工作", "学习"]）；只给一个分类则返回单元素数组。 +- priority：整数 0-9（或 1-9），数字越小优先级越高；无法识别则 null。 +- x_apple_structured_location：字符串 "geo:,"（如 "geo:22.284009,114.137831"）。 +- x_apple_creator_identity / x_apple_creator_team_identity：保留为字符串；未提供则 null。 +- x_apple_travel_advisory_behavior：保留为字符串（如 "DISABLED"）。 +- alarms：VALARM 列表；每个提醒包含 { action, trigger, description, repeat, duration, attach, uid, x_wr_alarmuid, related }；触发器支持相对（如 "-PT10M"、"-PT1H"、"-P1D"）或绝对时间（ISO）。当 assistant_message 承诺会提醒（例如“提前十五分钟提醒”），必须在 payload.component_attributes.alarms 中给出对应提醒，默认 action="DISPLAY"，并设置 related="START"；相对触发统一使用 RFC5545 时长字符串（如 "-PT15M"）。 + +【提醒生成规则】 +- 用户出现“提醒/闹钟/通知/提前X分钟/提前X小时/前一天”等表达时，payload.component_attributes.alarms 至少包含一个提醒，trigger 使用 RFC5545 时长字符串并与表达一致（如“提前十五分钟”→"-PT15M"，“提前一小时”→"-PT1H"，“提前一天”→"-P1D"），related="START"。 +- 若给出具体提醒时间点（如“当天下午2:00提醒”），trigger 使用绝对时间 ISO 字符串，例如 "2025-11-25T14:00:00"。 +- 未说明 action 时默认 action="DISPLAY"；未说明 description 时使用事件的简要描述或标题。 + +【优先级推断规则】 +- 当用户使用明确的紧急/重要语义但未给出数字优先级时，自动设定 priority： + * 出现“非常紧急/马上/立刻/火急/不得延误/必须马上”→ priority=5 + * 出现“重要会议/不能迟到/关键事项/DDL/截止/考试/面试/面谈/关键节点”→ priority=4 + * 出现“较重要/尽量/尽快/优先处理”→ priority=3 +- 未出现明显紧急/重要语义时，priority=null。 + +【时区与 TZID】 +- 当文本出现地区/城市（如“香港”）或显式时区（如 "Asia/Hong_Kong"），在时间语义中尽量保留该时区信息；无法明确则按本地时区推断。 + +【时间解析规则】 +- 使用当前时间作为参考：{current_time}。 +- 查询范围： + * “明天”：start_time=明天 00:00:00；end_time=明天 23:59:59。 + * “今天”：start_time=今天 00:00:00；end_time=今天 23:59:59。 +- 凌晨（0-6点）语义： + * “今天”指当前这一天；“明天”指即将到来的白天；“后天”约为+24小时后的白天。 +- “下周”=当前日期+7天。 +- 创建指令未指定时间：默认 start_time=当前时间+1小时，end_time=start_time+1小时（持续1小时）。 + +【意图识别词汇映射】 +- create：创建、添加、安排、预定、新建，以及一些与创建相关的词汇，比如“新建事件”、“添加会议”等。 +- read：查看、显示、列出、检查、看看、有什么事、日程安排，以及一些与查询相关的词汇，比如“查询会议”、“查看事件”等。 +- update：更新、修改、改变、调整、重新安排，以及一些与更新相关的词汇，比如“修改会议时间”、“调整事件”等。 +- delete：删除、取消、移除，以及一些与删除相关的词汇，比如“删除会议”、“取消事件”等。 +如果没有出现上面的关键词，可以根据查询语句分析一下用户意图，比如用户询问“最近有什么考试吗“，可以判断意图为read,且因为包含了“考试”这个关键词，所以可以确定用户是想查询考试日程，可以在title里面放入”考试“这个关键词。 +如果同时出现时间和事件关键词，比如“明天的会议改成6点”，意图分析可以判断为update。可以先分析时间，通过时间查询日程，再定位到具体的事件进行修改；如果查询不到，可以再试试先查询事件，时间范围可以设计成近两周，看看是否有相关事件。 + +【完整性原则与重要说明】 +- 创建 VEVENT 时，务必尽可能包含用户明确提供的每个信息；无法解析或不适用的字段使用 null。 +- 当指令过于模糊或无法确定具体事件/时间，payload.intent 必须为 null。 +- 输出必须是纯 JSON，不要有其他文本。""" def _load_config(self) -> Dict: """Load configuration with priority: config_private.json > config.json""" @@ -142,7 +191,7 @@ def _should_enable_reasoning(self, user_input: str) -> bool: return False - def parse_command(self, user_input: str) -> Dict: + def parse_command(self, user_input: str, history: Optional[list] = None) -> Dict: """ Parse natural language command using DeepSeek API @@ -152,9 +201,11 @@ def parse_command(self, user_input: str) -> Dict: Returns: Parsed command as dictionary """ + # print(f'位置：deepseek_parser.py, 用户原始指令: {user_input}，历史消息: {history}') + current_time = datetime.now().isoformat() - system_prompt = self.system_prompt.format(current_time=current_time) - print(f"🧠 Parsing command: {user_input}") + system_prompt = self.system_prompt.replace("{current_time}", current_time) + # print(f"🧠 Parsing command: {user_input}") try: headers = { @@ -162,14 +213,30 @@ def parse_command(self, user_input: str) -> Dict: "Authorization": f"Bearer {self.api_key}" } + # 构造消息列表，包含系统提示与历史消息 + messages = [ + {"role": "system", "content": system_prompt} + ] + + # # 注入会话历史（若提供），只保留 role/content 字段 + # if history and isinstance(history, list): + # for msg in history[-10:]: # 额外保护，最多携带最近10条 + # if isinstance(msg, dict) and msg.get('role') and msg.get('content'): + # messages.append({ + # "role": msg["role"], + # "content": msg["content"] + # }) + + # 当前用户指令 + messages.append({"role": "user", "content": user_input}) + + # print(f"🧠 Prepared messages: {messages}") + payload = { "model": "deepseek-chat", - "messages": [ - {"role": "system", "content": system_prompt}, - {"role": "user", "content": user_input} - ], + "messages": messages, "temperature": 0.1, - "max_tokens": 500, + "max_tokens": 1024, "stream": False } @@ -192,42 +259,126 @@ def parse_command(self, user_input: str) -> Dict: if json_start >= 0 and json_end > json_start: json_str = content[json_start:json_end] print(f"🧠 Extracted JSON: {json_str}") - parsed_data = json.loads(json_str) - - # Convert string dates to datetime objects - if parsed_data.get('start_time'): - parsed_data['start_time'] = datetime.fromisoformat(parsed_data['start_time']) - if parsed_data.get('end_time'): - parsed_data['end_time'] = datetime.fromisoformat(parsed_data['end_time']) - - print(f"🧠 Final parsed data: {parsed_data}") - return parsed_data + raw = json.loads(json_str) + + # Support dual-field or legacy formats + if isinstance(raw, dict) and 'payload' in raw: + payload = raw.get('payload', {}) or {} + # Convert string dates to datetime objects + if payload.get('start_time'): + payload['start_time'] = datetime.fromisoformat(payload['start_time']) + if payload.get('end_time'): + payload['end_time'] = datetime.fromisoformat(payload['end_time']) + + result = { + 'assistant_message': raw.get('assistant_message') or None, + 'payload': payload, + 'search_info': raw.get('search_info') or None + } + print(f"🧠 Final parsed result (dual): {result}") + return result + else: + # Legacy: direct intent object + parsed_data = raw if isinstance(raw, dict) else {} + if parsed_data.get('start_time'): + parsed_data['start_time'] = datetime.fromisoformat(parsed_data['start_time']) + if parsed_data.get('end_time'): + parsed_data['end_time'] = datetime.fromisoformat(parsed_data['end_time']) + + result = { + 'assistant_message': None, + 'payload': parsed_data, + 'search_info': None + } + print(f"🧠 Final parsed result (legacy): {result}") + return result else: print(f"❌ No JSON found in response") # Fallback: return basic structure return { + 'assistant_message': None, + 'payload': { + 'intent': None, + 'title': None, + 'start_time': None, + 'end_time': None, + 'description': None, + 'location': None, + # 'target_event': None + }, + 'search_info': None + } + + except Exception as e: + print(f"DeepSeek API error: {e}") + # Return empty structure on error + return { + 'assistant_message': None, + 'payload': { 'intent': None, 'title': None, 'start_time': None, 'end_time': None, 'description': None, 'location': None, - 'target_event': None - } + # 'target_event': None + }, + 'search_info': None + } + + def generate_comment(self, target, intial_message) -> str: + """Generate comment based on parsed payload""" + sys_smg = """你是一个细心的日程管理助手。如果用户提供了日程，请你根据target的日程请求(用户的日程格式可能是iCalendar格式,如果是,请你先尝试理解日程再进行分析)，对用户的行程进行评价，比如有什么比较重要的行程（比如考试，会议，乘搭飞机，乘搭高铁，面试等等）需要提醒用户注意，有什么特别的时间点（作业提交时间，考试时间，提前时间提醒的设置）需要关注一下。 + 一般行程包括了以下信息：事件title，时间，地点，描述，优先级，提醒设置，如果你觉得，当前你看到的某些日程缺少了一些你觉得有必要记录的信息，你可以对此提出自己的建议，比如赶飞机的行程你觉得提前两个小时设置一个提醒比较好，你可以建议用户对这个行程添加一个提醒。 + 如果target中没有提供日程，而是提醒的话，比如 查询到多个事件，请提供更具体的事件信息这样的提醒，且这样的提醒无法与最开始的response 的 initial_message 实现一致性, 优先遵守target提供的信息，纠正initial_message中的错误。如果两者有一致性，那就整合两者的信息，与此同时如果target中有日程的话，尽量保留target中的内容。 + 【输出格式】 + - 你首先需要用自然语言输出日程的详细信息（比如事件title，时间，地点，描述，优先级，提醒设置，重复设置等,如果没有设置不用提及，若是无效提醒也不用提及），再对用户的行程进行评价。 + - 输出字符串，不能包含任何Markdown格式或者LaTeX格式的内容。 + - target中的信息有更高的优先度，另外如果target中列出了事件的话，尽量列出每个事件的关键信息，可以适当换行，容易阅读。 + - 尽量不要提供错误信息 + + 【行为建议(作为示例)】 + - 特别注意一下,像是23:59, 12:00AM, 12:00PM, 00:00 这样的时间点,有时候用户会理解错,比如把中午12点理解成12:00PM,你需要提醒用户注意一下。 + - 考试提醒：如果用户设置了一个考试提醒，你可以建议用户在考试前两个小时设置一个提醒，以确保不会忘记考试,如果考试就在本周,你可以建议用户尽快复习,另外如果对方没有设置考试地点,你也可以建议用户设置考试地点。 + - 会议提醒：如果用户设置了一个会议提醒，你可以建议用户在会议开始前一个小时设置一个提醒，以确保不会错过会议,你可以提醒用户记得提前准备相关资料,如果是比较正式的会议着装上要注意。 + - 作业提醒：如果用户设置了一个作业提交提醒，你可以建议用户在作业提交前一个小时设置一个提醒，以确保不会忘记提交作业,另外如果用户备注里面提到了比如说迟交惩罚,你也可以再强调以下。 + - 如果你看到某个时间段,比如某一周,某一天等等,用户有多个行程排在了一起,且有的行程很重要,比如考试等等,你需要做出冲突预警,提醒用户注意这些行程之间的时间冲突,或者时间间隔特别紧张,非常建议对方提早完成或者提早准备。 + - 如果用户创建了一个行程，你发现缺少一些比较重要的信息，你可以建议用户添加这些信息，比如考试地点，会议地点，作业提交时间等等。 + - 每个提醒包含 { action, trigger, description, repeat, duration, attach, uid, x_wr_alarmuid, related }；触发器支持相对（如 "-PT10M"、"-PT1H"、"-P1D"）或绝对时间（ISO）。当 assistant_message 承诺会提醒（例如“提前十五分钟提醒”），必须在 payload.component_attributes.alarms 中给出对应提醒，默认 action="DISPLAY"，并设置 related="START"；相对触发统一使用 RFC5545 时长字符串（如 "-PT15M"）。 + """ + # Build messages first to avoid referencing before assignment + user_content = f"target: {target}; initial_message: {intial_message}" + print(f"将根据这个iCalendar事件生成评价：{user_content}") + messages = [ + {'role': 'system', 'content': sys_smg}, + {'role': 'user', 'content': user_content} + ] + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {self.api_key}" + } + payload = { + "model": "deepseek-chat", + "messages": messages, + "temperature": 0.1, + "max_tokens": 1024, + "stream": False + } + + try: + response = requests.post(self.api_url, headers=headers, json=payload, timeout=30) + response.raise_for_status() + data = response.json() + content = data["choices"][0]["message"]["content"].strip() except Exception as e: - print(f"DeepSeek API error: {e}") - # Return empty structure on error - return { - 'intent': None, - 'title': None, - 'start_time': None, - 'end_time': None, - 'description': None, - 'location': None, - 'target_event': None - } + print(f"DeepSeek API error (generate_comment): {e}") + content = None + + print(f"🧠 API Response: {content if content else '无法生成评价'}") + return content + # Test function def test_deepseek_parsing(): """Test DeepSeek parsing with example commands""" diff --git a/templates/index.html b/templates/index.html index d64b58b..40d0da8 100644 --- a/templates/index.html +++ b/templates/index.html @@ -3,7 +3,7 @@ - AI日程管理助手 + 日历助手