import math import unicodedata from dataclasses import dataclass from typing import List, Sequence, Tuple from .audio_metrics import AudioMetrics SCORING_VERSION = "asr-fluency-v1" @dataclass(frozen=True) class TextAlignment: reference_tokens: List[str] recognized_tokens: List[str] matches: int missing_tokens: List[str] extra_tokens: List[str] substitutions: List[Tuple[str, str]] content_score: float completeness_score: float @dataclass(frozen=True) class ScoreBreakdown: overall_score: float content_score: float completeness_score: float fluency_score: float duration_score: float pause_score: float speech_rate_score: float duration_ratio: float missing_tokens: List[str] extra_tokens: List[str] substitutions: List[Tuple[str, str]] feedback: str def tokenize(text: str) -> List[str]: normalized = unicodedata.normalize("NFKC", text).lower().strip() tokens: List[str] = [] word: List[str] = [] def flush_word() -> None: if word: token = "".join(word).strip("'") if token: tokens.append(token) word.clear() for char in normalized: if _is_cjk(char): flush_word() tokens.append(char) elif char.isalnum() or (char == "'" and word): word.append(char) else: flush_word() flush_word() return tokens def align_text(reference_text: str, recognized_text: str) -> TextAlignment: reference = tokenize(reference_text) recognized = tokenize(recognized_text) if not reference: raise ValueError("Reference text does not contain any scoreable tokens.") costs = [[0] * (len(recognized) + 1) for _ in range(len(reference) + 1)] for row in range(1, len(reference) + 1): costs[row][0] = row for column in range(1, len(recognized) + 1): costs[0][column] = column for row in range(1, len(reference) + 1): for column in range(1, len(recognized) + 1): substitution_cost = 0 if reference[row - 1] == recognized[column - 1] else 1 costs[row][column] = min( costs[row - 1][column] + 1, costs[row][column - 1] + 1, costs[row - 1][column - 1] + substitution_cost, ) row = len(reference) column = len(recognized) matches = 0 missing: List[str] = [] extra: List[str] = [] substitutions: List[Tuple[str, str]] = [] while row > 0 or column > 0: if row > 0 and column > 0: same = reference[row - 1] == recognized[column - 1] diagonal_cost = costs[row - 1][column - 1] + (0 if same else 1) if costs[row][column] == diagonal_cost: if same: matches += 1 else: substitutions.append((reference[row - 1], recognized[column - 1])) row -= 1 column -= 1 continue if row > 0 and costs[row][column] == costs[row - 1][column] + 1: missing.append(reference[row - 1]) row -= 1 else: extra.append(recognized[column - 1]) column -= 1 missing.reverse() extra.reverse() substitutions.reverse() distance = len(missing) + len(extra) + len(substitutions) accuracy = max(0.0, 1.0 - distance / len(reference)) * 100 attempted_reference_tokens = matches + len(substitutions) completeness = attempted_reference_tokens / len(reference) * 100 content = accuracy * 0.7 + completeness * 0.3 return TextAlignment( reference_tokens=reference, recognized_tokens=recognized, matches=matches, missing_tokens=missing, extra_tokens=extra, substitutions=substitutions, content_score=_round_score(content), completeness_score=_round_score(completeness), ) def score_attempt( *, reference_text: str, recognized_text: str, reference_speech_duration_ms: int, student_metrics: AudioMetrics, ) -> ScoreBreakdown: if reference_speech_duration_ms <= 0: raise ValueError("reference_speech_duration_ms must be positive") alignment = align_text(reference_text, recognized_text) ratio = student_metrics.speech_duration_ms / reference_speech_duration_ms duration = duration_similarity_score(ratio) pause = pause_score(student_metrics.internal_pause_ratio) reference_rate = len(alignment.reference_tokens) / (reference_speech_duration_ms / 1000) recognized_units = max(1, len(alignment.recognized_tokens)) student_rate = recognized_units / (student_metrics.speech_duration_ms / 1000) rate_ratio = student_rate / reference_rate if reference_rate > 0 else 1.0 rate = symmetric_rate_score(rate_ratio) fluency = duration * 0.35 + pause * 0.40 + rate * 0.25 overall = alignment.content_score * 0.80 + fluency * 0.20 feedback = build_feedback(alignment, ratio, student_metrics.internal_pause_ratio) return ScoreBreakdown( overall_score=_round_score(overall), content_score=alignment.content_score, completeness_score=alignment.completeness_score, fluency_score=_round_score(fluency), duration_score=_round_score(duration), pause_score=_round_score(pause), speech_rate_score=_round_score(rate), duration_ratio=round(ratio, 4), missing_tokens=alignment.missing_tokens, extra_tokens=alignment.extra_tokens, substitutions=alignment.substitutions, feedback=feedback, ) def duration_similarity_score(ratio: float) -> float: if ratio <= 0: return 0.0 if 0.80 <= ratio <= 1.30: return 100.0 if 0.65 <= ratio < 0.80: return _interpolate(ratio, 0.65, 0.80, 60, 100) if 1.30 < ratio <= 1.50: return _interpolate(ratio, 1.30, 1.50, 100, 60) if 0.50 <= ratio < 0.65: return _interpolate(ratio, 0.50, 0.65, 20, 60) if 1.50 < ratio <= 1.80: return _interpolate(ratio, 1.50, 1.80, 60, 20) if ratio < 0.50: return max(0.0, ratio / 0.50 * 20) return max(0.0, 20 - (ratio - 1.80) / 0.40 * 20) def pause_score(internal_pause_ratio: float) -> float: if internal_pause_ratio <= 0.15: return 100.0 if internal_pause_ratio <= 0.30: return _interpolate(internal_pause_ratio, 0.15, 0.30, 100, 70) if internal_pause_ratio <= 0.50: return _interpolate(internal_pause_ratio, 0.30, 0.50, 70, 20) return max(0.0, 20 - (internal_pause_ratio - 0.50) / 0.30 * 20) def symmetric_rate_score(ratio: float) -> float: if ratio <= 0: return 0.0 deviation = abs(math.log(ratio)) free_tolerance = math.log(1.20) if deviation <= free_tolerance: return 100.0 return max(0.0, 100 - (deviation - free_tolerance) / math.log(2.5) * 100) def build_feedback( alignment: TextAlignment, duration_ratio: float, internal_pause_ratio: float, ) -> str: messages = [] if alignment.missing_tokens: messages.append("存在漏读") if alignment.substitutions: messages.append("存在错读") if alignment.extra_tokens: messages.append("存在多读") if duration_ratio < 0.65: messages.append("朗读明显偏快或内容不完整") elif duration_ratio > 1.50: messages.append("朗读速度偏慢") if internal_pause_ratio > 0.30: messages.append("句内停顿偏多") return "；".join(messages) if messages else "内容和朗读节奏匹配良好" def _is_cjk(char: str) -> bool: codepoint = ord(char) return ( 0x3400 <= codepoint <= 0x4DBF or 0x4E00 <= codepoint <= 0x9FFF or 0xF900 <= codepoint <= 0xFAFF ) def _interpolate(value: float, start: float, end: float, start_score: float, end_score: float) -> float: return start_score + (value - start) / (end - start) * (end_score - start_score) def _round_score(value: float) -> float: return round(min(100.0, max(0.0, value)), 1)