import logging import shutil import subprocess import tempfile 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, ) logger = logging.getLogger(__name__) MOSS_ALGORITHM_VERSION = "moss-punctuation-v3" 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 extract_audio_segment( media_path: Path, output_path: Path, *, start_ms: int, duration_ms: int, ) -> None: """Extract one mono WAV segment for a focused Whisper request.""" 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", "-ss", f"{max(0, start_ms) / 1000:.3f}", "-t", f"{max(1, duration_ms) / 1000:.3f}", "-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=1800) if completed.returncode != 0: message = completed.stderr.strip() or "unknown ffmpeg error" raise RuntimeError(f"Could not extract sentence audio: {message[-2000:]}") def split_sentence_text( *, transcriber: Transcriber, media_path: Path, sentence: SentenceBoundary, text: str, language: Optional[str], ) -> List[SentenceBoundary]: """Split one stored sentence at explicit line breaks. When MOSS/Whisper is available, the sentence's own audio is transcribed and its word timestamps select the new boundary. If transcription is unavailable or fails, editing remains possible and boundaries fall back to text-length proportions so the admin UI does not lose the user's manual edit. """ lines = [line.strip() for line in text.splitlines() if line.strip()] if len(lines) < 2: return [ sentence.model_copy( update={ "text": text.strip(), "language": language if language is not None else sentence.language, } ) ] transcript: Optional[Transcript] = None if transcriber.available and media_path.is_file(): try: with tempfile.TemporaryDirectory(prefix="sentence-split-") as temp_dir: audio_path = Path(temp_dir) / "sentence.wav" extract_audio_segment( media_path, audio_path, start_ms=sentence.start_ms, duration_ms=sentence.end_ms - sentence.start_ms, ) transcript = transcriber.transcribe(audio_path, language) except Exception as exc: logger.warning("Whisper split-timing failed for sentence %s: %s", sentence.index, exc) boundaries = _estimate_split_boundaries_ms(sentence, lines, transcript) weights = [_text_weight(line) for line in lines] total_weight = max(1, sum(weights)) replacements: List[SentenceBoundary] = [] for boundary, line, weight in zip(boundaries, lines, weights): reference_duration = sentence.reference_speech_duration_ms or ( boundary.end_ms - boundary.start_ms ) replacements.append( boundary.model_copy( update={ "text": line, "language": language if language is not None else sentence.language, "reference_speech_duration_ms": max( 1, int(round(reference_duration * weight / total_weight)), ), } ) ) return replacements def _estimate_split_boundaries_ms( sentence: SentenceBoundary, lines: List[str], transcript: Optional[Transcript], ) -> List[SentenceBoundary]: estimates = _whisper_split_boundaries_ms(sentence, lines, transcript) if len(estimates) != len(lines) - 1: estimates = _proportional_split_boundaries_ms(sentence, lines) # Keep every split strictly inside the source sentence, even if a Whisper # timestamp is slightly outside or two estimates collapse to one point. previous = sentence.start_ms remaining = len(lines) for index, estimate in enumerate(list(estimates)): low = previous + 1 high = sentence.end_ms - (remaining - 1) if low >= high: estimates = _proportional_split_boundaries_ms(sentence, lines) break estimate = min(max(estimate, low), high) estimates[index] = estimate previous = estimate remaining -= 1 starts = [sentence.start_ms, *estimates] ends = [*estimates, sentence.end_ms] return [ SentenceBoundary( index=index, start_ms=start, end_ms=end, text=lines[index], language=sentence.language, reference_speech_duration_ms=max(1, end - start), ) for index, (start, end) in enumerate(zip(starts, ends)) ] def _whisper_split_boundaries_ms( sentence: SentenceBoundary, lines: List[str], transcript: Optional[Transcript], ) -> List[int]: if transcript is None or not transcript.segments: return [] source_ratios = _cumulative_ratios([_text_weight(line) for line in lines]) words = [word for segment in transcript.segments for word in (segment.words or [])] if len(words) >= 2: span_start = words[0].start_seconds span_end = max(word.end_seconds for word in words) if span_end <= span_start: return [] total_weight = sum(max(1, len(word.text.strip())) for word in words) cumulative = 0 samples: List[tuple[float, int]] = [] for word in words[:-1]: cumulative += max(1, len(word.text.strip())) timestamp_ms = sentence.start_ms + int( round( (word.end_seconds - span_start) / (span_end - span_start) * (sentence.end_ms - sentence.start_ms) ) ) samples.append((cumulative / total_weight, timestamp_ms)) return [ min(samples, key=lambda item: (abs(item[0] - ratio), item[1]))[1] for ratio in source_ratios ] # Some MOSS configurations omit word timestamps. Segment ends are still # much better than blind proportional allocation when they exist. if len(transcript.segments) < 2: return [] total_weight = sum(max(1, len(segment.text.strip())) for segment in transcript.segments) cumulative = 0 samples: List[tuple[float, int]] = [] for segment in transcript.segments[:-1]: cumulative += max(1, len(segment.text.strip())) timestamp_ms = sentence.start_ms + int(round(segment.end_seconds * 1000)) samples.append((cumulative / total_weight, timestamp_ms)) return [ min(samples, key=lambda item: (abs(item[0] - ratio), item[1]))[1] for ratio in source_ratios ] def _proportional_split_boundaries_ms( sentence: SentenceBoundary, lines: List[str], ) -> List[int]: duration = sentence.end_ms - sentence.start_ms return [ sentence.start_ms + int(round(ratio * duration)) for ratio in _cumulative_ratios([_text_weight(line) for line in lines]) ] def _cumulative_ratios(weights: List[int]) -> List[float]: total = max(1, sum(weights)) cumulative = 0 ratios = [] for weight in weights[:-1]: cumulative += weight ratios.append(cumulative / total) return ratios def _text_weight(text: str) -> int: return max(1, sum(1 for char in text if not char.isspace())) 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)