Files
mediaplayer/sentence_api/models.py
2026-08-16 15:39:52 +08:00

128 lines
4.0 KiB
Python

from datetime import datetime
from typing import Dict, List, Literal, Optional
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
class SentenceBoundary(BaseModel):
model_config = ConfigDict(extra="forbid")
index: int = Field(ge=0)
start_ms: int = Field(ge=0)
end_ms: int = Field(gt=0)
text: Optional[str] = None
language: Optional[str] = None
reference_speech_duration_ms: Optional[int] = Field(default=None, gt=0)
@model_validator(mode="after")
def validate_range(self):
if self.end_ms <= self.start_ms:
raise ValueError("end_ms must be greater than start_ms")
return self
class SentenceBoundaryDocument(BaseModel):
model_config = ConfigDict(extra="forbid")
video_hash: str
duration_ms: int = Field(gt=0)
algorithm_version: str = Field(min_length=1)
sentences: List[SentenceBoundary]
@field_validator("video_hash")
@classmethod
def validate_video_hash(cls, value: str) -> str:
normalized = value.strip().lower()
if len(normalized) != 64 or any(char not in "0123456789abcdef" for char in normalized):
raise ValueError("video_hash must be a SHA-256 hex digest")
return normalized
@model_validator(mode="after")
def validate_sentences(self):
previous_end = 0
for expected_index, sentence in enumerate(self.sentences):
if sentence.index != expected_index:
raise ValueError("sentence indexes must be contiguous and zero-based")
if sentence.start_ms < previous_end:
raise ValueError("sentences must be sorted and non-overlapping")
if sentence.end_ms > self.duration_ms:
raise ValueError("sentence end_ms cannot exceed duration_ms")
previous_end = sentence.end_ms
return self
VideoStatus = Literal["uploaded", "processing", "ready", "failed"]
class VideoSummary(BaseModel):
model_config = ConfigDict(extra="forbid")
video_hash: str
title: str
filename: str
content_type: str
size_bytes: int = Field(ge=0)
duration_ms: Optional[int] = Field(default=None, gt=0)
language: Optional[str] = None
status: VideoStatus
error_message: Optional[str] = None
sentence_count: int = Field(ge=0)
stream_url: str
created_at: datetime
updated_at: datetime
class VideoListResponse(BaseModel):
videos: List[VideoSummary]
class VideoUploadResponse(BaseModel):
video_hash: str
status: VideoStatus
detail_url: str
class VideoDetailResponse(BaseModel):
video: VideoSummary
boundaries: Optional[SentenceBoundaryDocument] = None
class SentenceTextUpdate(BaseModel):
text: str = Field(min_length=1, max_length=4000)
language: Optional[str] = Field(default=None, max_length=32)
class TextSubstitution(BaseModel):
expected: str
actual: str
class AssessmentResult(BaseModel):
model_config = ConfigDict(extra="forbid")
attempt_id: str
scoring_version: str
overall_score: float = Field(ge=0, le=100)
passed: bool
pass_score: float = Field(ge=0, le=100)
content_score: float = Field(ge=0, le=100)
completeness_score: float = Field(ge=0, le=100)
fluency_score: float = Field(ge=0, le=100)
pronunciation_score: Optional[float] = Field(default=None, ge=0, le=100)
prosody_score: Optional[float] = Field(default=None, ge=0, le=100)
duration_score: float = Field(ge=0, le=100)
pause_score: float = Field(ge=0, le=100)
speech_rate_score: float = Field(ge=0, le=100)
reference_text: str
recognized_text: str
reference_duration_ms: int = Field(gt=0)
reference_speech_duration_ms: int = Field(gt=0)
student_recording_duration_ms: int = Field(gt=0)
student_speech_duration_ms: int = Field(gt=0)
duration_ratio: float = Field(gt=0)
missing_tokens: List[str]
extra_tokens: List[str]
substitutions: List[TextSubstitution]
feedback: str
details: Dict[str, str] = Field(default_factory=dict)