import shutil import subprocess import uuid from pathlib import Path from typing import List, Optional from .audio_metrics import ( AudioAnalysisError, analyze_samples, decode_audio_mono, refine_sentence_end_ms, refine_sentence_start_ms, ) from .config import Settings from .generate_boundaries import ALGORITHM_VERSION, make_entry from .models import SentenceBoundary, SentenceBoundaryDocument from .oss import VolcanoOSSUploader from .repository import VideoRepository from .transcription import Transcript, Transcriber, split_sentences_at_punctuation MOSS_ALGORITHM_VERSION = "moss-period-v2" class VideoProcessor: def __init__( self, settings: Settings, repository: VideoRepository, transcriber: Transcriber, oss_uploader: Optional[VolcanoOSSUploader] = None, ): self.settings = settings self.repository = repository self.transcriber = transcriber self.oss_uploader = oss_uploader def process(self, video_hash: str) -> None: video = self.repository.get_video(video_hash) if video is None: raise ValueError(f"Unknown video: {video_hash}") media_path = self.settings.videos_dir / video["stored_filename"] if not media_path.is_file(): raise FileNotFoundError(f"Stored video is missing: {media_path.name}") self.repository.mark_processing(video_hash) work_path: Optional[Path] = None try: if self.transcriber.available: work_path = self.settings.work_dir / f"{video_hash}-{uuid.uuid4().hex}.wav" extract_audio(media_path, work_path) transcript = self.transcriber.transcribe(work_path, video.get("language")) document = document_from_transcript( video_hash=video_hash, duration_ms=_media_duration_ms(media_path), transcript=transcript, language=video.get("language"), audio_path=work_path, end_padding_ms=self.settings.moss_end_padding_ms, ) if not document.sentences: raise RuntimeError("MOSS returned no timestamped speech segments.") self._finish_processing(document, transcript.text) else: entry, _ = make_entry(media_path, video_hash=video_hash) document = SentenceBoundaryDocument( video_hash=video_hash, duration_ms=entry["duration_ms"], algorithm_version=ALGORITHM_VERSION, sentences=entry["sentences"], ) self._finish_processing(document, None) except Exception as exc: self.repository.mark_failed(video_hash, str(exc)) raise finally: if work_path is not None: work_path.unlink(missing_ok=True) def _finish_processing( self, document: SentenceBoundaryDocument, transcription: Optional[str], ) -> None: if self.oss_uploader is None or not self.oss_uploader.enabled: self.repository.save_processing_result(document, transcription, ready=True) return # Keep the course invisible while the mandatory OSS upload is running. self.repository.save_processing_result(document, transcription, ready=False) video = self.repository.get_video(document.video_hash) if video is None: raise ValueError(f"Unknown video: {document.video_hash}") remote_url = self.oss_uploader.upload( video_hash=document.video_hash, stored_filename=video["stored_filename"], local_path=self.settings.videos_dir / video["stored_filename"], content_type=video.get("content_type"), ) self.repository.mark_oss_uploaded(document.video_hash, remote_url) def extract_audio(media_path: Path, output_path: Path) -> None: ffmpeg = shutil.which("ffmpeg") if ffmpeg is None: raise RuntimeError("ffmpeg is required for MOSS transcription but was not found.") output_path.parent.mkdir(parents=True, exist_ok=True) command = [ ffmpeg, "-hide_banner", "-loglevel", "error", "-y", "-i", str(media_path), "-vn", "-ac", "1", "-ar", "16000", "-c:a", "pcm_s16le", str(output_path), ] completed = subprocess.run(command, capture_output=True, text=True, timeout=7200) if completed.returncode != 0: message = completed.stderr.strip() or "unknown ffmpeg error" raise RuntimeError(f"Could not extract video audio: {message[-2000:]}") def document_from_transcript( *, video_hash: str, duration_ms: int, transcript: Transcript, language: Optional[str], audio_path: Path, end_padding_ms: int = 300, ) -> SentenceBoundaryDocument: samples, sample_rate = decode_audio_mono(audio_path) sentences: List[SentenceBoundary] = [] previous_end = 0 sentence_segments = split_sentences_at_punctuation(transcript.segments) for index, sentence_segment in enumerate(sentence_segments): raw_start_ms = int(round(sentence_segment.start_seconds * 1000)) start_ms = max(previous_end, raw_start_ms) raw_end_ms = int(round(sentence_segment.end_seconds * 1000)) end_ms = min( duration_ms, raw_end_ms + end_padding_ms, ) if index + 1 < len(sentence_segments): next_start_ms = int( round(sentence_segments[index + 1].start_seconds * 1000) ) end_ms = min(end_ms, next_start_ms) end_ms = refine_sentence_end_ms( samples, sample_rate, raw_end_ms=raw_end_ms, padded_end_ms=end_ms ) refined_start_ms = refine_sentence_start_ms( samples, sample_rate, raw_start_ms=raw_start_ms, previous_end_ms=previous_end, ) if previous_end <= refined_start_ms and end_ms - refined_start_ms >= 100: start_ms = refined_start_ms if not sentence_segment.text.strip() or end_ms <= start_ms: continue start_sample = max(0, int(start_ms / 1000 * sample_rate)) end_sample = min(samples.size, int(end_ms / 1000 * sample_rate)) try: metrics = analyze_samples(samples[start_sample:end_sample], sample_rate) speech_duration_ms = metrics.speech_duration_ms except AudioAnalysisError: speech_duration_ms = end_ms - start_ms sentences.append( SentenceBoundary( index=len(sentences), start_ms=start_ms, end_ms=end_ms, text=sentence_segment.text.strip(), language=language, reference_speech_duration_ms=max(1, speech_duration_ms), ) ) previous_end = end_ms return SentenceBoundaryDocument( video_hash=video_hash, duration_ms=duration_ms, algorithm_version=MOSS_ALGORITHM_VERSION, sentences=sentences, ) def _media_duration_ms(path: Path) -> int: from .generate_boundaries import media_duration_ms return media_duration_ms(path)