1  """
  2  Internal task list management for Kamaji work mode.
  3  """
  4  
  5  import json
  6  from pathlib import Path
  7  from typing import List, Dict, Optional
  8  from datetime import datetime
  9  
 10  
 11  class TaskList:
 12      """Manages Kamaji's internal task list."""
 13  
 14      def __init__(self, task_file: Optional[Path] = None):
 15          """
 16          Initialize task list.
 17  
 18          Args:
 19              task_file: Path to task list JSON file. If None, uses default location.
 20          """
 21          if task_file is None:
 22              # Use .kamaji directory in home or project root
 23              kamaji_dir = Path.home() / ".kamaji"
 24              kamaji_dir.mkdir(exist_ok=True)
 25              task_file = kamaji_dir / "tasks.json"
 26  
 27          self.task_file = task_file
 28          self.tasks: List[Dict] = []
 29          self.load()
 30  
 31      def load(self):
 32          """Load tasks from file."""
 33          if self.task_file.exists():
 34              try:
 35                  with open(self.task_file, 'r') as f:
 36                      data = json.load(f)
 37                      self.tasks = data.get('tasks', [])
 38              except Exception as e:
 39                  print(f"Warning: Could not load tasks: {e}")
 40                  self.tasks = []
 41          else:
 42              self.tasks = []
 43  
 44      def save(self):
 45          """Save tasks to file."""
 46          try:
 47              with open(self.task_file, 'w') as f:
 48                  json.dump({
 49                      'tasks': self.tasks,
 50                      'last_updated': datetime.now().isoformat()
 51                  }, f, indent=2)
 52          except Exception as e:
 53              print(f"Warning: Could not save tasks: {e}")
 54  
 55      def add_task(self, description: str, priority: str = "medium", tags: Optional[List[str]] = None) -> Dict:
 56          """
 57          Add a new task.
 58  
 59          Args:
 60              description: Task description
 61              priority: Priority level (low, medium, high)
 62              tags: Optional list of tags
 63  
 64          Returns:
 65              The created task
 66          """
 67          task = {
 68              'id': len(self.tasks) + 1,
 69              'description': description,
 70              'priority': priority,
 71              'status': 'pending',
 72              'tags': tags or [],
 73              'created_at': datetime.now().isoformat(),
 74              'completed_at': None
 75          }
 76          self.tasks.append(task)
 77          self.save()
 78          return task
 79  
 80      def get_pending_tasks(self) -> List[Dict]:
 81          """Get all pending tasks."""
 82          return [t for t in self.tasks if t['status'] == 'pending']
 83  
 84      def get_task(self, task_id: int) -> Optional[Dict]:
 85          """Get task by ID."""
 86          for task in self.tasks:
 87              if task['id'] == task_id:
 88                  return task
 89          return None
 90  
 91      def complete_task(self, task_id: int):
 92          """Mark task as completed."""
 93          task = self.get_task(task_id)
 94          if task:
 95              task['status'] = 'completed'
 96              task['completed_at'] = datetime.now().isoformat()
 97              self.save()
 98  
 99      def remove_task(self, task_id: int):
100          """Remove a task."""
101          self.tasks = [t for t in self.tasks if t['id'] != task_id]
102          self.save()
103  
104      def clear_completed(self):
105          """Remove all completed tasks."""
106          self.tasks = [t for t in self.tasks if t['status'] != 'completed']
107          self.save()
108  
109      def get_next_task(self) -> Optional[Dict]:
110          """Get the next highest priority pending task."""
111          pending = self.get_pending_tasks()
112          if not pending:
113              return None
114  
115          # Sort by priority (high > medium > low)
116          priority_order = {'high': 0, 'medium': 1, 'low': 2}
117          pending.sort(key=lambda t: priority_order.get(t['priority'], 2))
118  
119          return pending[0]
120  
121      def format_task_list(self) -> str:
122          """Format task list for display."""
123          if not self.tasks:
124              return "No tasks in task list."
125  
126          lines = []
127          lines.append("\n📋 Task List:")
128          lines.append("=" * 60)
129  
130          # Pending tasks
131          pending = self.get_pending_tasks()
132          if pending:
133              lines.append("\n⏳ Pending Tasks:")
134              for task in pending:
135                  priority_emoji = {
136                      'high': '🔴',
137                      'medium': '🟡',
138                      'low': '🟢'
139                  }
140                  emoji = priority_emoji.get(task['priority'], '⚪')
141                  tags_str = f" [{', '.join(task['tags'])}]" if task['tags'] else ""
142                  lines.append(f"  {emoji} #{task['id']}: {task['description']}{tags_str}")
143  
144          # Completed tasks
145          completed = [t for t in self.tasks if t['status'] == 'completed']
146          if completed:
147              lines.append(f"\n✅ Completed Tasks: {len(completed)}")
148              for task in completed[-5:]:  # Show last 5
149                  lines.append(f"  ✓ #{task['id']}: {task['description']}")
150  
151          lines.append("=" * 60)
152          return '\n'.join(lines)
153  
154  
155  def get_task_list() -> TaskList:
156      """
157      Get the global task list instance.
158  
159      Returns:
160          TaskList instance
161      """
162      if not hasattr(get_task_list, '_instance'):
163          get_task_list._instance = TaskList()
164      return get_task_list._instance