147 lines
4.2 KiB
Python
147 lines
4.2 KiB
Python
import asyncio
|
|
import json
|
|
import logging
|
|
from typing import Tuple, List, Dict, Any, Optional, Union
|
|
|
|
async def create_task(task_description: str, tags: List[str] = None, project: str = None,
|
|
due: str = None, priority: str = None) -> Tuple[bool, Optional[str]]:
|
|
"""
|
|
Create a new task using the Taskwarrior CLI.
|
|
|
|
Args:
|
|
task_description: Description of the task
|
|
tags: List of tags to apply to the task
|
|
project: Project to which the task belongs
|
|
due: Due date in the format that Taskwarrior accepts
|
|
priority: Priority of the task (H, M, L)
|
|
|
|
Returns:
|
|
Tuple containing:
|
|
- Success status (True if operation was successful)
|
|
- Task ID or error message
|
|
"""
|
|
try:
|
|
cmd = ["task", "add"]
|
|
|
|
# Add project if specified
|
|
if project:
|
|
cmd.append(f"project:{project}")
|
|
|
|
# Add tags if specified
|
|
if tags:
|
|
for tag in tags:
|
|
cmd.append(f"+{tag}")
|
|
|
|
# Add due date if specified
|
|
if due:
|
|
cmd.append(f"due:{due}")
|
|
|
|
# Add priority if specified
|
|
if priority and priority in ["H", "M", "L"]:
|
|
cmd.append(f"priority:{priority}")
|
|
|
|
# Add task description
|
|
cmd.append(task_description)
|
|
|
|
# Convert command list to string
|
|
cmd_str = " ".join(cmd)
|
|
|
|
process = await asyncio.create_subprocess_shell(
|
|
cmd_str,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
)
|
|
stdout, stderr = await process.communicate()
|
|
|
|
if process.returncode == 0:
|
|
return True, stdout.decode().strip()
|
|
else:
|
|
error_msg = stderr.decode().strip()
|
|
logging.error(f"Error creating task: {error_msg}")
|
|
return False, error_msg
|
|
except Exception as e:
|
|
logging.error(f"Exception during task creation: {e}")
|
|
return False, str(e)
|
|
|
|
async def list_tasks(filter_str: str = "") -> Tuple[List[Dict[str, Any]], bool]:
|
|
"""
|
|
List tasks from Taskwarrior.
|
|
|
|
Args:
|
|
filter_str: Optional filter string to pass to Taskwarrior
|
|
|
|
Returns:
|
|
Tuple containing:
|
|
- List of task dictionaries
|
|
- Success status (True if operation was successful)
|
|
"""
|
|
try:
|
|
cmd = f"task {filter_str} export"
|
|
|
|
process = await asyncio.create_subprocess_shell(
|
|
cmd,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
)
|
|
stdout, stderr = await process.communicate()
|
|
|
|
if process.returncode == 0:
|
|
tasks = json.loads(stdout.decode())
|
|
return tasks, True
|
|
else:
|
|
logging.error(f"Error listing tasks: {stderr.decode()}")
|
|
return [], False
|
|
except Exception as e:
|
|
logging.error(f"Exception during task listing: {e}")
|
|
return [], False
|
|
|
|
async def complete_task(task_id: str) -> bool:
|
|
"""
|
|
Mark a task as completed.
|
|
|
|
Args:
|
|
task_id: ID of the task to complete
|
|
|
|
Returns:
|
|
True if task was completed successfully, False otherwise
|
|
"""
|
|
try:
|
|
cmd = f"echo 'yes' | task {task_id} done"
|
|
|
|
process = await asyncio.create_subprocess_shell(
|
|
cmd,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
)
|
|
stdout, stderr = await process.communicate()
|
|
|
|
return process.returncode == 0
|
|
except Exception as e:
|
|
logging.error(f"Exception during task completion: {e}")
|
|
return False
|
|
|
|
async def delete_task(task_id: str) -> bool:
|
|
"""
|
|
Delete a task.
|
|
|
|
Args:
|
|
task_id: ID of the task to delete
|
|
|
|
Returns:
|
|
True if task was deleted successfully, False otherwise
|
|
"""
|
|
try:
|
|
cmd = f"echo 'yes' | task {task_id} delete"
|
|
|
|
process = await asyncio.create_subprocess_shell(
|
|
cmd,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
)
|
|
stdout, stderr = await process.communicate()
|
|
|
|
return process.returncode == 0
|
|
except Exception as e:
|
|
logging.error(f"Exception during task deletion: {e}")
|
|
return False
|