110 lines
4.2 KiB
Python
110 lines
4.2 KiB
Python
import uuid
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from .audio_metrics import analyze_audio
|
|
from .models import AssessmentResult, SentenceBoundaryDocument, TextSubstitution
|
|
from .repository import VideoRepository
|
|
from .scoring import SCORING_VERSION, score_attempt
|
|
from .transcription import Transcriber
|
|
|
|
|
|
class AssessmentService:
|
|
def __init__(
|
|
self,
|
|
repository: VideoRepository,
|
|
transcriber: Transcriber,
|
|
pass_score: float,
|
|
):
|
|
self.repository = repository
|
|
self.transcriber = transcriber
|
|
self.pass_score = pass_score
|
|
|
|
def assess(
|
|
self,
|
|
*,
|
|
document: SentenceBoundaryDocument,
|
|
sentence_index: int,
|
|
audio_path: Path,
|
|
language: Optional[str] = None,
|
|
retained_audio_filename: Optional[str] = None,
|
|
user_id: str = "",
|
|
) -> AssessmentResult:
|
|
if not self.transcriber.available:
|
|
raise RuntimeError("MOSS transcription is not configured on this server.")
|
|
sentence = next(
|
|
(item for item in document.sentences if item.index == sentence_index),
|
|
None,
|
|
)
|
|
if sentence is None:
|
|
raise LookupError("Sentence index was not found.")
|
|
if not sentence.text or not sentence.text.strip():
|
|
raise ValueError("The sentence has no reviewed reference text yet.")
|
|
|
|
student_metrics = analyze_audio(audio_path)
|
|
transcript = self.transcriber.transcribe(
|
|
audio_path,
|
|
language or sentence.language,
|
|
)
|
|
recognized_text = transcript.text.strip()
|
|
if not recognized_text:
|
|
raise ValueError("MOSS did not recognize any speech in the recording.")
|
|
|
|
reference_duration_ms = sentence.end_ms - sentence.start_ms
|
|
reference_speech_duration_ms = (
|
|
sentence.reference_speech_duration_ms or reference_duration_ms
|
|
)
|
|
score = score_attempt(
|
|
reference_text=sentence.text,
|
|
recognized_text=recognized_text,
|
|
reference_speech_duration_ms=reference_speech_duration_ms,
|
|
student_metrics=student_metrics,
|
|
)
|
|
attempt_id = uuid.uuid4().hex
|
|
result = AssessmentResult(
|
|
attempt_id=attempt_id,
|
|
scoring_version=SCORING_VERSION,
|
|
overall_score=score.overall_score,
|
|
passed=score.overall_score >= self.pass_score,
|
|
pass_score=self.pass_score,
|
|
content_score=score.content_score,
|
|
completeness_score=score.completeness_score,
|
|
fluency_score=score.fluency_score,
|
|
pronunciation_score=None,
|
|
prosody_score=None,
|
|
duration_score=score.duration_score,
|
|
pause_score=score.pause_score,
|
|
speech_rate_score=score.speech_rate_score,
|
|
reference_text=sentence.text,
|
|
recognized_text=recognized_text,
|
|
reference_duration_ms=reference_duration_ms,
|
|
reference_speech_duration_ms=reference_speech_duration_ms,
|
|
student_recording_duration_ms=student_metrics.recording_duration_ms,
|
|
student_speech_duration_ms=student_metrics.speech_duration_ms,
|
|
duration_ratio=score.duration_ratio,
|
|
missing_tokens=score.missing_tokens,
|
|
extra_tokens=score.extra_tokens,
|
|
substitutions=[
|
|
TextSubstitution(expected=expected, actual=actual)
|
|
for expected, actual in score.substitutions
|
|
],
|
|
feedback=score.feedback,
|
|
details={
|
|
"score_kind": "朗读内容与流畅度匹配分",
|
|
"content_weight": "0.80",
|
|
"fluency_weight": "0.20",
|
|
"duration_weight_within_fluency": "0.35",
|
|
"phoneme_scoring": "not_enabled",
|
|
},
|
|
)
|
|
if self.repository.get_video(document.video_hash) is not None:
|
|
self.repository.record_attempt(
|
|
attempt_id=attempt_id,
|
|
video_hash=document.video_hash,
|
|
sentence_index=sentence_index,
|
|
result=result.model_dump(mode="json"),
|
|
audio_filename=retained_audio_filename,
|
|
user_id=user_id,
|
|
)
|
|
return result
|