Spaces:
Running
Running
File size: 3,228 Bytes
cccf200 | 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 | """
Analytics API endpoints for task analytics and visualizations.
"""
from fastapi import APIRouter, Depends, HTTPException
from sqlmodel import Session
from pydantic import BaseModel
from typing import List, Dict, Any
from src.db.session import get_session
from src.mcp.tools import (
get_task_statistics,
get_tasks_over_time,
get_completion_analytics,
get_productivity_hours
)
router = APIRouter()
class TaskStatistics(BaseModel):
"""Task statistics response model."""
total_tasks: int
completed_tasks: int
pending_tasks: int
completion_rate: float
class TimelineData(BaseModel):
"""Timeline data point."""
date: str
created: int
completed: int
class TasksOverTimeResponse(BaseModel):
"""Tasks over time response model."""
timeline: List[TimelineData]
days: int
class CompletionAnalytics(BaseModel):
"""Completion analytics response model."""
total: int
completed: int
pending: int
completion_rate: float
avg_completion_time_hours: float
class ProductivityHour(BaseModel):
"""Productivity by hour data point."""
hour: int
tasks_completed: int
class ProductivityResponse(BaseModel):
"""Productivity response model."""
productivity_by_hour: List[ProductivityHour]
@router.get("/{user_id}/analytics/overview", response_model=TaskStatistics)
async def get_analytics_overview(
user_id: int,
session: Session = Depends(get_session)
):
"""
Get overall task statistics.
Args:
user_id: User ID
session: Database session
Returns:
Task statistics
"""
result = get_task_statistics(user_id=user_id, session=session)
return TaskStatistics(**result)
@router.get("/{user_id}/analytics/timeline", response_model=TasksOverTimeResponse)
async def get_analytics_timeline(
user_id: int,
days: int = 30,
session: Session = Depends(get_session)
):
"""
Get task creation and completion trends over time.
Args:
user_id: User ID
days: Number of days to look back (default 30)
session: Database session
Returns:
Timeline data
"""
result = get_tasks_over_time(user_id=user_id, days=days, session=session)
return TasksOverTimeResponse(**result)
@router.get("/{user_id}/analytics/completion", response_model=CompletionAnalytics)
async def get_analytics_completion(
user_id: int,
session: Session = Depends(get_session)
):
"""
Get detailed completion analytics.
Args:
user_id: User ID
session: Database session
Returns:
Completion analytics
"""
result = get_completion_analytics(user_id=user_id, session=session)
return CompletionAnalytics(**result)
@router.get("/{user_id}/analytics/productivity", response_model=ProductivityResponse)
async def get_analytics_productivity(
user_id: int,
session: Session = Depends(get_session)
):
"""
Get productivity statistics by hour of day.
Args:
user_id: User ID
session: Database session
Returns:
Productivity by hour
"""
result = get_productivity_hours(user_id=user_id, session=session)
return ProductivityResponse(**result)
|