import json import uuid from typing import Dict, List, Any, Optional, Callable, Union from datetime import datetime, timedelta from utils.logging import setup_logger from utils.error_handling import handle_exceptions, AutomationError from utils.storage import load_data, save_data # Initialize logger logger = setup_logger(__name__) class Reminder: """Intelligent reminder for notifications""" def __init__(self, title: str, message: str, reminder_type: str, target_id: Optional[str] = None, context: Optional[Dict[str, Any]] = None): """Initialize a reminder Args: title: Reminder title message: Reminder message reminder_type: Type of reminder (task, goal, note, focus, etc.) target_id: ID of the target item (optional) context: Additional context data (optional) """ self.id = str(uuid.uuid4()) self.title = title self.message = message self.reminder_type = reminder_type self.target_id = target_id self.context = context or {} self.created_at = datetime.now().isoformat() self.updated_at = self.created_at self.due_at = None self.repeat_config = None self.notification_channels = ["app"] # Default to app notifications self.priority = "normal" # low, normal, high, urgent self.status = "pending" # pending, sent, dismissed, snoozed self.sent_at = None self.dismissed_at = None self.snoozed_until = None self.smart_features = { "auto_adjust": False, # Automatically adjust timing based on user behavior "context_aware": False, # Adjust based on context (location, time, etc.) "priority_boost": False, # Boost priority based on urgency "batch_similar": True, # Batch similar reminders together "quiet_hours": False # Respect quiet hours } self.quiet_hours = { "enabled": False, "start": "22:00", "end": "08:00" } @handle_exceptions def set_due_date(self, due_at: Union[str, datetime]) -> None: """Set reminder due date Args: due_at: Due date (datetime or ISO format string) """ if isinstance(due_at, datetime): self.due_at = due_at.isoformat() else: self.due_at = due_at self.updated_at = datetime.now().isoformat() @handle_exceptions def set_repeat(self, repeat_type: str, repeat_config: Dict[str, Any]) -> None: """Set reminder repeat configuration Args: repeat_type: Type of repeat (daily, weekly, monthly, custom) repeat_config: Repeat configuration """ self.repeat_config = { "type": repeat_type, "config": repeat_config } self.updated_at = datetime.now().isoformat() @handle_exceptions def set_notification_channels(self, channels: List[str]) -> None: """Set notification channels Args: channels: List of notification channels (app, email, telegram, etc.) """ self.notification_channels = channels self.updated_at = datetime.now().isoformat() @handle_exceptions def set_priority(self, priority: str) -> None: """Set reminder priority Args: priority: Priority level (low, normal, high, urgent) """ if priority in ["low", "normal", "high", "urgent"]: self.priority = priority self.updated_at = datetime.now().isoformat() @handle_exceptions def mark_as_sent(self) -> None: """Mark reminder as sent""" self.status = "sent" self.sent_at = datetime.now().isoformat() self.updated_at = datetime.now().isoformat() @handle_exceptions def dismiss(self) -> None: """Dismiss reminder""" self.status = "dismissed" self.dismissed_at = datetime.now().isoformat() self.updated_at = datetime.now().isoformat() @handle_exceptions def snooze(self, minutes: int = 15) -> None: """Snooze reminder Args: minutes: Snooze duration in minutes """ self.status = "snoozed" snooze_until = datetime.now() + timedelta(minutes=minutes) self.snoozed_until = snooze_until.isoformat() self.updated_at = datetime.now().isoformat() @handle_exceptions def enable_smart_feature(self, feature: str, enabled: bool = True) -> None: """Enable or disable a smart feature Args: feature: Feature name enabled: Whether to enable or disable """ if feature in self.smart_features: self.smart_features[feature] = enabled self.updated_at = datetime.now().isoformat() @handle_exceptions def set_quiet_hours(self, enabled: bool, start: str, end: str) -> None: """Set quiet hours Args: enabled: Whether quiet hours are enabled start: Start time (HH:MM) end: End time (HH:MM) """ self.quiet_hours = { "enabled": enabled, "start": start, "end": end } self.updated_at = datetime.now().isoformat() @handle_exceptions def is_due(self) -> bool: """Check if reminder is due Returns: True if reminder is due, False otherwise """ if not self.due_at or self.status in ["sent", "dismissed"]: return False if self.status == "snoozed" and self.snoozed_until: # Check if snooze period has ended now = datetime.now().isoformat() if now < self.snoozed_until: return False # Check if due date has passed now = datetime.now().isoformat() return now >= self.due_at @handle_exceptions def should_send_now(self) -> bool: """Check if reminder should be sent now Returns: True if reminder should be sent now, False otherwise """ if not self.is_due(): return False # Check quiet hours if self.smart_features.get("quiet_hours", False) and self.quiet_hours.get("enabled", False): now = datetime.now().time() start_time = datetime.strptime(self.quiet_hours["start"], "%H:%M").time() end_time = datetime.strptime(self.quiet_hours["end"], "%H:%M").time() # Check if current time is within quiet hours if start_time < end_time: # Normal case (e.g., 22:00 to 08:00) if start_time <= now <= end_time: return False else: # Overnight case (e.g., 22:00 to 08:00) if now >= start_time or now <= end_time: return False return True @handle_exceptions def to_dict(self) -> Dict[str, Any]: """Convert reminder to dictionary Returns: Reminder as dictionary """ return { "id": self.id, "title": self.title, "message": self.message, "reminder_type": self.reminder_type, "target_id": self.target_id, "context": self.context, "created_at": self.created_at, "updated_at": self.updated_at, "due_at": self.due_at, "repeat_config": self.repeat_config, "notification_channels": self.notification_channels, "priority": self.priority, "status": self.status, "sent_at": self.sent_at, "dismissed_at": self.dismissed_at, "snoozed_until": self.snoozed_until, "smart_features": self.smart_features, "quiet_hours": self.quiet_hours } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'Reminder': """Create reminder from dictionary Args: data: Reminder data Returns: Reminder instance """ reminder = cls(data["title"], data["message"], data["reminder_type"], data.get("target_id"), data.get("context", {})) reminder.id = data["id"] reminder.created_at = data["created_at"] reminder.updated_at = data["updated_at"] reminder.due_at = data.get("due_at") reminder.repeat_config = data.get("repeat_config") reminder.notification_channels = data.get("notification_channels", ["app"]) reminder.priority = data.get("priority", "normal") reminder.status = data.get("status", "pending") reminder.sent_at = data.get("sent_at") reminder.dismissed_at = data.get("dismissed_at") reminder.snoozed_until = data.get("snoozed_until") reminder.smart_features = data.get("smart_features", { "auto_adjust": False, "context_aware": False, "priority_boost": False, "batch_similar": True, "quiet_hours": False }) reminder.quiet_hours = data.get("quiet_hours", { "enabled": False, "start": "22:00", "end": "08:00" }) return reminder class ReminderManager: """Manager for intelligent reminders""" def __init__(self): """Initialize reminder manager""" self.reminders = {} self.notification_handlers = { "app": self._send_app_notification, "email": self._send_email_notification, "telegram": self._send_telegram_notification } self.load_reminders() @handle_exceptions def load_reminders(self) -> None: """Load reminders from storage""" try: reminders_data = load_data("reminders", default=[]) for reminder_data in reminders_data: reminder = Reminder.from_dict(reminder_data) self.reminders[reminder.id] = reminder logger.info(f"Loaded {len(self.reminders)} reminders") except Exception as e: logger.error(f"Failed to load reminders: {str(e)}") @handle_exceptions def save_reminders(self) -> None: """Save reminders to storage""" try: reminders_data = [reminder.to_dict() for reminder in self.reminders.values()] save_data("reminders", reminders_data) logger.info(f"Saved {len(self.reminders)} reminders") except Exception as e: logger.error(f"Failed to save reminders: {str(e)}") @handle_exceptions def create_reminder(self, title: str, message: str, reminder_type: str, target_id: Optional[str] = None, context: Optional[Dict[str, Any]] = None) -> Reminder: """Create a new reminder Args: title: Reminder title message: Reminder message reminder_type: Type of reminder target_id: ID of the target item (optional) context: Additional context data (optional) Returns: Created reminder """ reminder = Reminder(title, message, reminder_type, target_id, context) self.reminders[reminder.id] = reminder self.save_reminders() return reminder @handle_exceptions def get_reminder(self, reminder_id: str) -> Optional[Reminder]: """Get reminder by ID Args: reminder_id: Reminder ID Returns: Reminder if found, None otherwise """ return self.reminders.get(reminder_id) @handle_exceptions def update_reminder(self, reminder: Reminder) -> None: """Update reminder Args: reminder: Reminder to update """ if reminder.id in self.reminders: reminder.updated_at = datetime.now().isoformat() self.reminders[reminder.id] = reminder self.save_reminders() else: raise AutomationError(f"Reminder not found: {reminder.id}") @handle_exceptions def delete_reminder(self, reminder_id: str) -> None: """Delete reminder Args: reminder_id: Reminder ID """ if reminder_id in self.reminders: del self.reminders[reminder_id] self.save_reminders() else: raise AutomationError(f"Reminder not found: {reminder_id}") @handle_exceptions def get_all_reminders(self) -> List[Reminder]: """Get all reminders Returns: List of all reminders """ return list(self.reminders.values()) @handle_exceptions def get_pending_reminders(self) -> List[Reminder]: """Get pending reminders Returns: List of pending reminders """ return [ reminder for reminder in self.reminders.values() if reminder.status == "pending" or reminder.status == "snoozed" ] @handle_exceptions def get_due_reminders(self) -> List[Reminder]: """Get due reminders Returns: List of due reminders """ return [ reminder for reminder in self.reminders.values() if reminder.should_send_now() ] @handle_exceptions def process_due_reminders(self) -> int: """Process all due reminders Returns: Number of reminders processed """ due_reminders = self.get_due_reminders() # Group reminders by type if batch_similar is enabled batched_reminders = {} individual_reminders = [] for reminder in due_reminders: if reminder.smart_features.get("batch_similar", True): key = reminder.reminder_type if key not in batched_reminders: batched_reminders[key] = [] batched_reminders[key].append(reminder) else: individual_reminders.append(reminder) # Process batched reminders for reminder_type, reminders in batched_reminders.items(): if len(reminders) > 1: # Create a batched notification title = f"{len(reminders)} {reminder_type} reminders" message = "\n".join([f"• {r.title}" for r in reminders]) self._send_notification(title, message, reminders[0].priority, reminders[0].notification_channels, {"batch": True, "count": len(reminders)}) # Mark all as sent for reminder in reminders: self._process_after_send(reminder) else: # Only one reminder of this type, process individually individual_reminders.append(reminders[0]) # Process individual reminders for reminder in individual_reminders: self._send_notification(reminder.title, reminder.message, reminder.priority, reminder.notification_channels, {"reminder_id": reminder.id}) self._process_after_send(reminder) return len(due_reminders) def _process_after_send(self, reminder: Reminder) -> None: """Process reminder after sending Args: reminder: Reminder to process """ # Mark as sent reminder.mark_as_sent() # Handle repeat if configured if reminder.repeat_config: repeat_type = reminder.repeat_config.get("type") repeat_config = reminder.repeat_config.get("config", {}) # Create a new due date based on repeat configuration if reminder.due_at: due_date = datetime.fromisoformat(reminder.due_at) new_due_date = None if repeat_type == "daily": days = repeat_config.get("days", 1) new_due_date = due_date + timedelta(days=days) elif repeat_type == "weekly": weeks = repeat_config.get("weeks", 1) new_due_date = due_date + timedelta(weeks=weeks) elif repeat_type == "monthly": months = repeat_config.get("months", 1) # Simple month addition (not perfect for all edge cases) new_month = due_date.month + months new_year = due_date.year + (new_month - 1) // 12 new_month = ((new_month - 1) % 12) + 1 new_due_date = due_date.replace(year=new_year, month=new_month) elif repeat_type == "custom" and "interval_days" in repeat_config: interval_days = repeat_config.get("interval_days", 1) new_due_date = due_date + timedelta(days=interval_days) if new_due_date: # Create a new reminder with the new due date new_reminder = Reminder(reminder.title, reminder.message, reminder.reminder_type, reminder.target_id, reminder.context) new_reminder.set_due_date(new_due_date) new_reminder.set_repeat(repeat_type, repeat_config) new_reminder.set_notification_channels(reminder.notification_channels) new_reminder.set_priority(reminder.priority) # Copy smart features for feature, enabled in reminder.smart_features.items(): new_reminder.enable_smart_feature(feature, enabled) # Copy quiet hours new_reminder.set_quiet_hours( reminder.quiet_hours.get("enabled", False), reminder.quiet_hours.get("start", "22:00"), reminder.quiet_hours.get("end", "08:00") ) # Add the new reminder self.reminders[new_reminder.id] = new_reminder # Update the reminder self.update_reminder(reminder) def _send_notification(self, title: str, message: str, priority: str, channels: List[str], context: Dict[str, Any]) -> None: """Send notification through all specified channels Args: title: Notification title message: Notification message priority: Notification priority channels: Notification channels context: Additional context """ for channel in channels: handler = self.notification_handlers.get(channel) if handler: try: handler(title, message, priority, context) except Exception as e: logger.error(f"Failed to send {channel} notification: {str(e)}") def _send_app_notification(self, title: str, message: str, priority: str, context: Dict[str, Any]) -> None: """Send app notification Args: title: Notification title message: Notification message priority: Notification priority context: Additional context """ # Placeholder for actual app notification logger.info(f"App notification: {priority} - {title} - {message}") def _send_email_notification(self, title: str, message: str, priority: str, context: Dict[str, Any]) -> None: """Send email notification Args: title: Notification title message: Notification message priority: Notification priority context: Additional context """ # Placeholder for actual email notification logger.info(f"Email notification: {priority} - {title} - {message}") def _send_telegram_notification(self, title: str, message: str, priority: str, context: Dict[str, Any]) -> None: """Send telegram notification Args: title: Notification title message: Notification message priority: Notification priority context: Additional context """ # Placeholder for actual telegram notification logger.info(f"Telegram notification: {priority} - {title} - {message}") # Create a global instance of the reminder manager reminder_manager = ReminderManager()