task_manager_agent.py

"""
Personal Task Manager Agent for Hugging Face AI Agent Course Final Project

This agent helps users manage tasks, set priorities, and organize information
through natural language understanding and simple task management functionality.
"""

import json
import re
import datetime
from typing import List, Dict, Any, Optional, Tuple

class Task:
    """Represents a single task with properties like description, priority, due date, etc."""
    
    def __init__(self, description: str, priority: str = "medium", 
                 due_date: Optional[str] = None, category: str = "general"):
        self.description = description
        self.priority = priority.lower()  # "high", "medium", or "low"
        self.due_date = due_date  # Format: YYYY-MM-DD
        self.category = category.lower()
        self.completed = False
        self.creation_date = datetime.datetime.now().strftime("%Y-%m-%d")
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert task to dictionary for JSON serialization"""
        return {
            "description": self.description,
            "priority": self.priority,
            "due_date": self.due_date,
            "category": self.category,
            "completed": self.completed,
            "creation_date": self.creation_date
        }
    
    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'Task':
        """Create a Task object from a dictionary"""
        task = cls(
            description=data["description"],
            priority=data["priority"],
            due_date=data["due_date"],
            category=data["category"]
        )
        task.completed = data["completed"]
        task.creation_date = data["creation_date"]
        return task
    
    def __str__(self) -> str:
        """String representation of the task"""
        status = "✓" if self.completed else "□"
        due_str = f" (Due: {self.due_date})" if self.due_date else ""
        return f"{status} [{self.priority.upper()}] {self.description}{due_str} - {self.category}"


class TaskManager:
    """Manages a collection of tasks with functionality to add, update, and query tasks"""
    
    def __init__(self):
        self.tasks: List[Task] = []
    
    def add_task(self, task: Task) -> int:
        """Add a new task and return its index"""
        self.tasks.append(task)
        return len(self.tasks) - 1
    
    def update_task(self, index: int, **kwargs) -> bool:
        """Update task properties at the given index"""
        if 0 <= index < len(self.tasks):
            task = self.tasks[index]
            for key, value in kwargs.items():
                if hasattr(task, key):
                    setattr(task, key, value)
            return True
        return False
    
    def complete_task(self, index: int) -> bool:
        """Mark a task as completed"""
        return self.update_task(index, completed=True)
    
    def delete_task(self, index: int) -> bool:
        """Delete a task at the given index"""
        if 0 <= index < len(self.tasks):
            self.tasks.pop(index)
            return True
        return False
    
    def get_tasks(self, 
                  category: Optional[str] = None, 
                  priority: Optional[str] = None,
                  completed: Optional[bool] = None) -> List[Tuple[int, Task]]:
        """Get tasks with optional filtering, returns list of (index, task) tuples"""
        result = []
        for i, task in enumerate(self.tasks):
            if (category is None or task.category == category.lower()) and \
               (priority is None or task.priority == priority.lower()) and \
               (completed is None or task.completed == completed):
                result.append((i, task))
        return result
    
    def get_all_tasks(self) -> List[Tuple[int, Task]]:
        """Get all tasks with their indices"""
        return [(i, task) for i, task in enumerate(self.tasks)]
    
    def get_categories(self) -> List[str]:
        """Get list of unique categories"""
        return list(set(task.category for task in self.tasks))
    
    def save_to_file(self, filename: str) -> bool:
        """Save tasks to a JSON file"""
        try:
            with open(filename, 'w') as f:
                json.dump([task.to_dict() for task in self.tasks], f, indent=2)
            return True
        except Exception as e:
            print(f"Error saving tasks: {e}")
            return False
    
    def load_from_file(self, filename: str) -> bool:
        """Load tasks from a JSON file"""
        try:
            with open(filename, 'r') as f:
                data = json.load(f)
                self.tasks = [Task.from_dict(item) for item in data]
            return True
        except Exception as e:
            print(f"Error loading tasks: {e}")
            return False


class TaskManagerAgent:
    """
    AI agent that understands natural language requests related to task management
    and performs appropriate actions using the TaskManager.
    """
    
    def __init__(self):
        self.task_manager = TaskManager()
        # Common patterns for understanding user intent
        self.patterns = {
            "add_task": [
                r"add (?:a )?(?:new )?task(?: to)?(?: do)?(?: called)?(?: to)?(?: about)? (.+)",
                r"create (?:a )?(?:new )?task(?: to)?(?: do)?(?: called)?(?: to)?(?: about)? (.+)",
                r"remind me to (.+)",
            ],
            "list_tasks": [
                r"(?:show|list|display|get)(?: all)?(?: my)? tasks",
                r"what (?:are my|do i have for) tasks",
                r"show me what i need to do",
            ],
            "complete_task": [
                r"(?:mark|set) task (?:number )?(\d+) (?:as )?(?:done|complete|finished)",
                r"complete task (?:number )?(\d+)",
                r"i (?:have )?(?:done|completed|finished) task (?:number )?(\d+)",
            ],
            "delete_task": [
                r"(?:delete|remove) task (?:number )?(\d+)",
                r"get rid of task (?:number )?(\d+)",
            ],
            "filter_tasks": [
                r"(?:show|list|display|get) (?:all )?(\w+) tasks",
                r"(?:show|list|display|get) tasks (?:with|that are) (\w+)",
            ],
            "help": [
                r"(?:help|assist|guide) me",
                r"what can you do",
                r"how do (?:you|i|this) work",
            ]
        }
    
    def _extract_task_details(self, description: str) -> Dict[str, Any]:
        """Extract task details like priority, due date, and category from description"""
        details = {"description": description}
        
        # Extract priority
        priority_match = re.search(r"(?:priority|important|urgency):?\s*(high|medium|low)", description, re.I)
        if priority_match:
            details["priority"] = priority_match.group(1).lower()
            # Remove the priority text from description
            details["description"] = re.sub(r"(?:priority|important|urgency):?\s*(high|medium|low)", "", details["description"], flags=re.I).strip()
        
        # Extract due date (simple format: YYYY-MM-DD or MM/DD/YYYY)
        date_match = re.search(r"(?:due|by|on):?\s*(\d{4}-\d{2}-\d{2}|\d{1,2}/\d{1,2}/\d{4})", description, re.I)
        if date_match:
            date_str = date_match.group(1)
            # Convert MM/DD/YYYY to YYYY-MM-DD if needed
            if "/" in date_str:
                month, day, year = date_str.split("/")
                date_str = f"{year}-{month.zfill(2)}-{day.zfill(2)}"
            details["due_date"] = date_str
            # Remove the date text from description
            details["description"] = re.sub(r"(?:due|by|on):?\s*(\d{4}-\d{2}-\d{2}|\d{1,2}/\d{1,2}/\d{4})", "", details["description"], flags=re.I).strip()
        
        # Extract category
        category_match = re.search(r"(?:category|tag|type):?\s*(\w+)", description, re.I)
        if category_match:
            details["category"] = category_match.group(1).lower()
            # Remove the category text from description
            details["description"] = re.sub(r"(?:category|tag|type):?\s*(\w+)", "", details["description"], flags=re.I).strip()
        
        return details
    
    def process_query(self, query: str) -> str:
        """
        Process a natural language query and perform the appropriate task management action.
        Returns a response string.
        """
        # Check for intent matches
        for intent, patterns in self.patterns.items():
            for pattern in patterns:
                match = re.search(pattern, query, re.I)
                if match:
                    # Call the appropriate method based on intent
                    if intent == "add_task" and match.group(1):
                        return self._handle_add_task(match.group(1))
                    elif intent == "list_tasks":
                        return self._handle_list_tasks()
                    elif intent == "complete_task" and match.group(1):
                        return self._handle_complete_task(int(match.group(1)))
                    elif intent == "delete_task" and match.group(1):
                        return self._handle_delete_task(int(match.group(1)))
                    elif intent == "filter_tasks" and match.group(1):
                        return self._handle_filter_tasks(match.group(1))
                    elif intent == "help":
                        return self._handle_help()
        
        # If no pattern matches, try to understand as a general query
        return self._handle_general_query(query)
    
    def _handle_add_task(self, description: str) -> str:
        """Handle adding a new task"""
        details = self._extract_task_details(description)
        task = Task(
            description=details["description"],
            priority=details.get("priority", "medium"),
            due_date=details.get("due_date"),
            category=details.get("category", "general")
        )
        index = self.task_manager.add_task(task)
        return f"Added task {index}: {task}"
    
    def _handle_list_tasks(self) -> str:
        """Handle listing all tasks"""
        tasks = self.task_manager.get_all_tasks()
        if not tasks:
            return "You don't have any tasks yet."
        
        result = "Here are your tasks:\n"
        for index, task in tasks:
            result += f"{index}: {task}\n"
        return result
    
    def _handle_complete_task(self, index: int) -> str:
        """Handle marking a task as complete"""
        if self.task_manager.complete_task(index):
            return f"Marked task {index} as completed."
        return f"Task {index} not found."
    
    def _handle_delete_task(self, index: int) -> str:
        """Handle deleting a task"""
        if self.task_manager.delete_task(index):
            return f"Deleted task {index}."
        return f"Task {index} not found."
    
    def _handle_filter_tasks(self, filter_term: str) -> str:
        """Handle filtering tasks by category or priority"""
        # Check if filter is a priority
        if filter_term.lower() in ["high", "medium", "low"]:
            tasks = self.task_manager.get_tasks(priority=filter_term.lower())
            filter_type = "priority"
        else:
            # Assume it's a category
            tasks = self.task_manager.get_tasks(category=filter_term.lower())
            filter_type = "category"
        
        if not tasks:
            return f"No tasks found with {filter_type} '{filter_term}'."
        
        result = f"Tasks with {filter_type} '{filter_term}':\n"
        for index, task in tasks:
            result += f"{index}: {task}\n"
        return result
    
    def _handle_help(self) -> str:
        """Handle help request"""
        return """
I can help you manage your tasks. Here's what you can ask me to do:

- Add a task: "Add a new task to buy groceries"
- Add with details: "Add task to call mom priority:high due:2023-05-20 category:personal"
- List tasks: "Show me my tasks" or "What do I need to do?"
- Complete a task: "Mark task 2 as done" or "I completed task 3"
- Delete a task: "Delete task 1" or "Remove task 4"
- Filter tasks: "Show high priority tasks" or "List personal tasks"
- Get help: "Help me" or "What can you do?"

Try one of these commands to get started!
"""
    
    def _handle_general_query(self, query: str) -> str:
        """Handle queries that don't match specific patterns"""
        # Check if it might be a task addition without explicit "add task" prefix
        if not any(re.search(pattern, query, re.I) for patterns in self.patterns.values() for pattern in patterns):
            # If query is short and looks like a task, add it
            if len(query.split()) <= 10 and not query.endswith("?"):
                return self._handle_add_task(query)
        
        return f"I'm not sure how to help with '{query}'. Type 'help' to see what I can do."
    
    def save_state(self, filename: str = "tasks.json") -> bool:
        """Save the current state of tasks to a file"""
        return self.task_manager.save_to_file(filename)
    
    def load_state(self, filename: str = "tasks.json") -> bool:
        """Load tasks from a file"""
        return self.task_manager.load_from_file(filename)


# Example usage for testing
if __name__ == "__main__":
    agent = TaskManagerAgent()
    
    # Test with some example queries
    test_queries = [
        "add task to buy groceries",
        "add task to call mom priority:high due:2023-05-20 category:personal",
        "show my tasks",
        "mark task 0 as done",
        "show high priority tasks",
        "help"
    ]
    
    for query in test_queries:
        print(f"\nQuery: {query}")
        response = agent.process_query(query)
        print(f"Response: {response}")