import logging import shutil import subprocess import tempfile import uuid from pathlib import Path from typing import Any, Dict, 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, media_duration_ms from sentence_analysis import detect_sentence_boundaries 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-v4" SILENCE_ALGORITHM_VERSION = "silence-rms-v1" _CUES_ID = b"\x1e\x53\xb6\x80" _CLUSTER_ID = b"\x1f\x43\xb6\x75" _HEADER_SCAN_SIZE = 64 * 1024 _WEBM_SUFFIXES = {".webm", ".mkv"} def _webm_cues_at_front(filepath: Path) -> bool: """Return True if WebM/MKV CUES appear before the first Cluster.""" try: with open(filepath, "rb") as f: header = f.read(_HEADER_SCAN_SIZE) except OSError: return True # can't read → assume OK, don't touch cues_pos = header.find(_CUES_ID) cluster_pos = header.find(_CLUSTER_ID) if cues_pos != -1 and (cluster_pos == -1 or cues_pos < cluster_pos): return True return False def _ensure_webm_cues_front(filepath: Path) -> None: """Re-mux a WebM/MKV file (stream copy) if CUES are not at the front. YouTube-dl / yt-dlp writes CUES at the end by default, which slows HTTP Range seeking. This runs once during upload processing and takes seconds. """ if filepath.suffix.lower() not in _WEBM_SUFFIXES: return if _webm_cues_at_front(filepath): return ffmpeg = shutil.which("ffmpeg") if not ffmpeg: logger.warning("ffmpeg not found; skipping WebM CUES optimization.") return tmp = filepath.with_suffix(".cues_tmp" + filepath.suffix) try: result = subprocess.run( [ffmpeg, "-y", "-loglevel", "error", "-i", str(filepath), "-c", "copy", "-cues_to_front", "1", str(tmp)], capture_output=True, timeout=300, ) if result.returncode == 0 and tmp.is_file() and tmp.stat().st_size > 0: tmp.replace(filepath) logger.info("WebM CUES moved to front: %s", filepath.name) else: logger.warning( "WebM CUES re-mux failed for %s: rc=%d stderr=%s", filepath.name, result.returncode, result.stderr.decode()[-200:], ) except (subprocess.TimeoutExpired, OSError) as exc: logger.warning("WebM CUES re-mux error for %s: %s", filepath.name, exc) finally: tmp.unlink(missing_ok=True) 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: _ensure_webm_cues_front(media_path) 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, moss_segments=_serialize_moss_segments(transcript), ) 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], moss_segments: Optional[List[Dict[str, Any]]] = None, ) -> None: if self.oss_uploader is None or not self.oss_uploader.enabled: self.repository.save_processing_result( document, transcription, ready=True, moss_segments=moss_segments ) return # Keep the course invisible while the mandatory OSS upload is running. self.repository.save_processing_result( document, transcription, ready=False, moss_segments=moss_segments ) 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 generate_audio_boundary_document( *, video_hash: str, audio_path: Path, transcriber: Transcriber, language: Optional[str] = None, ) -> SentenceBoundaryDocument: """Generate sentence boundaries for a client-supplied audio file. The caller owns ``audio_path`` and is responsible for deleting it. Nothing in this function copies or registers the audio in application storage. """ duration_ms = media_duration_ms(audio_path) transcript = None if transcriber.available: try: transcript = transcriber.transcribe(audio_path, language) except Exception as exc: logger.warning("Audio boundary transcription failed; using silence fallback: %s", exc) if transcript is not None: document = document_from_transcript( video_hash=video_hash, duration_ms=duration_ms, transcript=transcript, language=language, audio_path=audio_path, ) if document.sentences: return document starts = detect_sentence_boundaries(audio_path) if not starts: raise ValueError("No sentence boundaries could be detected in the audio.") starts_ms = sorted({max(0, min(duration_ms - 1, int(round(start * 1000)))) for start in starts}) sentences: List[SentenceBoundary] = [] for index, start_ms in enumerate(starts_ms): end_ms = starts_ms[index + 1] if index + 1 < len(starts_ms) else duration_ms if end_ms <= start_ms: continue sentences.append( SentenceBoundary( index=len(sentences), start_ms=start_ms, end_ms=end_ms, text=None, language=language, ) ) if not sentences: raise ValueError("Detected audio boundaries do not form valid sentence ranges.") return SentenceBoundaryDocument( video_hash=video_hash, duration_ms=duration_ms, algorithm_version=SILENCE_ALGORITHM_VERSION, sentences=sentences, ) def _media_duration_ms(path: Path) -> int: from .generate_boundaries import media_duration_ms return media_duration_ms(path) def _serialize_moss_segments(transcript: Transcript) -> List[Dict[str, Any]]: rows: List[Dict[str, Any]] = [] for segment in transcript.segments: text = segment.text.strip() if not text: continue words = None if segment.words: words = [ { "start_seconds": word.start_seconds, "end_seconds": word.end_seconds, "text": word.text, } for word in segment.words ] rows.append( { "start_seconds": segment.start_seconds, "end_seconds": segment.end_seconds, "text": text, "speaker": segment.speaker, "words": words, } ) return rows