File size: 4,253 Bytes
5c244a3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 | """
AI API routes.
Provides endpoints for AI-powered todo features.
"""
from typing import List
from fastapi import APIRouter, HTTPException, status, Depends
from pydantic import BaseModel
from src.api.deps import get_current_user_id, get_db
from src.services.ai_service import ai_service
from sqlmodel import Session, select
from src.models.todo import Todo
from uuid import UUID
class AIGenerateRequest(BaseModel):
"""Request schema for AI todo generation."""
goal: str
class AIGenerateResponse(BaseModel):
"""Response schema for AI todo generation."""
todos: List[dict]
message: str
class AISummarizeResponse(BaseModel):
"""Response schema for AI todo summarization."""
summary: str
breakdown: dict
urgent_todos: List[str]
class AIPrioritizeResponse(BaseModel):
"""Response schema for AI todo prioritization."""
prioritized_todos: List[dict]
message: str
router = APIRouter()
@router.post(
'/generate-todo',
response_model=AIGenerateResponse,
summary='Generate todos with AI',
description='Generate todo suggestions from a goal using AI',
)
async def generate_todos(
request: AIGenerateRequest,
current_user_id: str = Depends(get_current_user_id),
):
"""Generate todos from a goal using AI."""
try:
result = ai_service.generate_todos(request.goal)
return AIGenerateResponse(**result)
except ValueError as e:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=str(e),
)
except Exception as e:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"AI service error: {str(e)}",
)
@router.post(
'/summarize',
response_model=AISummarizeResponse,
summary='Summarize todos with AI',
description='Get an AI-powered summary of todos',
)
async def summarize_todos(
current_user_id: str = Depends(get_current_user_id),
db: Session = Depends(get_db),
):
"""Summarize todos using AI."""
try:
# Get user's todos
query = select(Todo).where(Todo.user_id == UUID(current_user_id))
todos = db.exec(query).all()
# Convert to dict format
todos_dict = [
{
"title": t.title,
"description": t.description,
"priority": t.priority.value,
"due_date": t.due_date.isoformat() if t.due_date else None,
}
for t in todos
]
result = ai_service.summarize_todos(todos_dict)
return AISummarizeResponse(**result)
except ValueError as e:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=str(e),
)
except Exception as e:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"AI service error: {str(e)}",
)
@router.post(
'/prioritize',
response_model=AIPrioritizeResponse,
summary='Prioritize todos with AI',
description='Get AI-powered todo prioritization',
)
async def prioritize_todos(
current_user_id: str = Depends(get_current_user_id),
db: Session = Depends(get_db),
):
"""Prioritize todos using AI."""
try:
# Get user's todos
query = select(Todo).where(Todo.user_id == UUID(current_user_id))
todos = db.exec(query).all()
# Convert to dict format with IDs
todos_dict = [
{
"id": str(t.id),
"title": t.title,
"description": t.description,
"priority": t.priority.value,
"due_date": t.due_date.isoformat() if t.due_date else None,
}
for t in todos
]
result = ai_service.prioritize_todos(todos_dict)
return AIPrioritizeResponse(**result)
except ValueError as e:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=str(e),
)
except Exception as e:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"AI service error: {str(e)}",
)
__all__ = ['router']
|