File size: 4,109 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 | from __future__ import annotations
import asyncio
import re
from fastapi import APIRouter, Depends, HTTPException, UploadFile
from app.api.auth import get_current_user_id
from app.api.schemas import (
GrammarAnswerRequest,
GrammarAnswerResponse,
GrammarExerciseResponse,
MasterRequest,
PronunciationErrorResponse,
ScoreResponse,
)
from app.db.models import GrammarExercise
from app.db.session import get_session_factory
from app.pipelines.audio import ogg_opus_to_pcm16
from app.pipelines.pronunciation import score_word_pronunciation, word_ipa
from app.services.grammar_exercise import get_due_exercises, update_sm2
from app.services.pronunciation import get_top_errors, mark_mastered
from app.services.user import get_or_create_user
router = APIRouter(prefix="/practice", tags=["practice"])
def _normalize(text: str) -> str:
return re.sub(r"[^\w\s]", "", text.strip().lower())
@router.get("/pronunciation", response_model=list[PronunciationErrorResponse])
async def get_pronunciation_errors(
user_id: int = Depends(get_current_user_id),
) -> list[PronunciationErrorResponse]:
"""Get top pronunciation errors for practice."""
factory = get_session_factory()
async with factory() as s:
errors = await get_top_errors(s, user_id, limit=10)
ipas = await asyncio.gather(*[word_ipa(e.word_example) for e in errors])
return [
PronunciationErrorResponse(
word_example=e.word_example,
expected_phoneme=e.expected_phoneme,
actual_phoneme=e.actual_phoneme,
occurrence_count=e.occurrence_count,
ipa=ipa,
)
for e, ipa in zip(errors, ipas, strict=True)
]
@router.post("/pronunciation/score", response_model=ScoreResponse)
async def score_pronunciation(
audio: UploadFile,
word: str,
user_id: int = Depends(get_current_user_id),
) -> ScoreResponse:
"""Score pronunciation of a word from uploaded audio."""
ogg_bytes = await audio.read()
pcm = await ogg_opus_to_pcm16(ogg_bytes)
async with get_session_factory()() as s:
user = await get_or_create_user(s, user_id)
is_correct, score, tip, assessed = await score_word_pronunciation(
pcm, word, lang=user.interface_language
)
return ScoreResponse(is_correct=is_correct, score=score, tip=tip, assessed=assessed)
@router.post("/pronunciation/master")
async def master_word(
body: MasterRequest,
user_id: int = Depends(get_current_user_id),
) -> dict[str, str]:
"""Mark a word as mastered."""
factory = get_session_factory()
async with factory() as s:
await mark_mastered(s, user_id, body.word)
return {"status": "mastered"}
@router.get("/grammar", response_model=list[GrammarExerciseResponse])
async def get_grammar_exercises(
user_id: int = Depends(get_current_user_id),
) -> list[GrammarExerciseResponse]:
"""Get due grammar exercises."""
factory = get_session_factory()
async with factory() as s:
exercises = await get_due_exercises(s, user_id, limit=10)
return [
GrammarExerciseResponse(
id=e.id,
wrong_text=e.wrong_text,
correct_text=e.correct_text,
rule=e.rule,
)
for e in exercises
]
@router.post("/grammar/answer", response_model=GrammarAnswerResponse)
async def answer_grammar(
body: GrammarAnswerRequest,
user_id: int = Depends(get_current_user_id),
) -> GrammarAnswerResponse:
"""Check grammar exercise answer and update SM-2."""
factory = get_session_factory()
async with factory() as s:
ex = await s.get(GrammarExercise, body.exercise_id)
if ex is None or ex.user_id != user_id:
raise HTTPException(status_code=404, detail="exercise not found")
is_correct = _normalize(body.answer) == _normalize(ex.correct_text)
quality = 4 if is_correct else 1
await update_sm2(s, body.exercise_id, user_id, quality)
return GrammarAnswerResponse(
is_correct=is_correct,
correct_text=ex.correct_text,
quality=quality,
)
|