File size: 4,783 Bytes
3470cf9 | 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 | from __future__ import annotations
from fastapi import APIRouter, Depends, HTTPException
from app.api.auth import get_current_user_id
from app.api.schemas import (
AssessmentDetailResponse,
SpeechAggregatesResponse,
SpeechAssessmentResponse,
StatsResponse,
UserResponse,
)
from app.db.session import get_session_factory
from app.services.grammar_exercise import get_exercise_count
from app.services.pronunciation import get_error_count
from app.services.speech_assessment import (
get_assessment,
get_recent_assessments,
get_score_aggregates,
)
from app.services.stats import get_stats
from app.services.user import get_or_create_user
router = APIRouter(tags=["user"])
@router.get("/user", response_model=UserResponse)
async def get_user(user_id: int = Depends(get_current_user_id)) -> UserResponse:
"""Get current user profile."""
factory = get_session_factory()
async with factory() as s:
user = await get_or_create_user(s, user_id)
return UserResponse(
telegram_id=user.telegram_id,
level=user.level,
interface_language=user.interface_language,
created_at=user.created_at,
)
@router.get("/stats", response_model=StatsResponse)
async def get_user_stats(user_id: int = Depends(get_current_user_id)) -> StatsResponse:
"""Get learning statistics."""
factory = get_session_factory()
async with factory() as s:
user = await get_or_create_user(s, user_id)
st = await get_stats(s, user_id)
pron_count = await get_error_count(s, user_id)
gram_count = await get_exercise_count(s, user_id)
assessed = user.last_assessed_at.strftime("%d %b %Y") if user.last_assessed_at else "not yet"
return StatsResponse(
level=user.level,
last_assessed=assessed,
total_messages=st.total_messages,
total_corrections=st.total_corrections,
accuracy_pct=st.accuracy_pct,
streak_days=st.streak_days,
words_saved=st.words_saved,
pronunciation_errors=pron_count,
grammar_exercises=gram_count,
)
@router.get("/assessment/latest", response_model=SpeechAssessmentResponse | None)
async def get_latest_assessment(
user_id: int = Depends(get_current_user_id),
) -> SpeechAssessmentResponse | None:
"""Get the most recent speech assessment (6 metrics + transcript), or null."""
factory = get_session_factory()
async with factory() as s:
rows = await get_recent_assessments(s, user_id, limit=1)
if not rows:
return None
a = rows[0]
return SpeechAssessmentResponse(
accuracy=a.accuracy,
fluency=a.fluency,
prosody=a.prosody,
vocabulary=a.vocabulary,
grammar=a.grammar,
topic=a.topic,
stress_accuracy=a.stress_accuracy,
transcript=a.transcript,
created_at=a.created_at,
)
@router.get("/assessment/phrase/{assessment_id}", response_model=AssessmentDetailResponse)
async def get_phrase_analysis(
assessment_id: int,
user_id: int = Depends(get_current_user_id),
) -> AssessmentDetailResponse:
"""Full per-phrase analysis: six metrics, transcript, grammar fixes, pronunciation."""
factory = get_session_factory()
async with factory() as s:
user = await get_or_create_user(s, user_id)
a = await get_assessment(s, assessment_id, user_id)
if a is None:
raise HTTPException(status_code=404, detail="assessment not found")
return AssessmentDetailResponse(
id=a.id,
dialog_turn_id=a.dialog_turn_id,
accuracy=a.accuracy,
fluency=a.fluency,
prosody=a.prosody,
vocabulary=a.vocabulary,
grammar=a.grammar,
topic=a.topic,
stress_accuracy=a.stress_accuracy,
transcript=a.transcript,
corrections=a.corrections or [],
pronunciation_notes=a.pronunciation_notes or [],
stress_notes=a.stress_notes or [],
peaks=a.peaks,
duration_sec=a.duration_sec,
level=user.level,
created_at=a.created_at,
)
@router.get("/assessment/aggregates", response_model=SpeechAggregatesResponse)
async def get_assessment_aggregates(
user_id: int = Depends(get_current_user_id),
) -> SpeechAggregatesResponse:
"""Get averaged speech metrics over the user's recent turns."""
factory = get_session_factory()
async with factory() as s:
agg = await get_score_aggregates(s, user_id)
return SpeechAggregatesResponse(
accuracy=agg["accuracy"],
fluency=agg["fluency"],
prosody=agg["prosody"],
vocabulary=agg["vocabulary"],
grammar=agg["grammar"],
topic=agg["topic"],
stress_accuracy=agg["stress_accuracy"],
count=agg["count"] or 0,
)
|