from __future__ import annotations import asyncio import json import logging import math import mimetypes import os import re import shutil import subprocess import time from urllib.parse import urlsplit, urlunsplit import wave from dataclasses import dataclass, field from pathlib import Path from typing import TYPE_CHECKING, Any, Awaitable, Callable if TYPE_CHECKING: import aiohttp logger = logging.getLogger("whisper_webui.file_mode") TARGET_SAMPLE_RATE = 16000 PCM_BYTES_PER_SECOND = TARGET_SAMPLE_RATE * 2 SUPPORTED_FILE_SUFFIXES = {".wav", ".mp3", ".m4a", ".mp4"} DIARIZATION_WAV_MAX_UPLOAD_MB_DEFAULT = 64 DIARIZATION_AUDIO_MAX_UPLOAD_MB_DEFAULT = 64 PUNCT_NO_SPACE_BEFORE = tuple(".,!?;:%)]}") PUNCT_NO_SPACE_AFTER = tuple("([{\"") PUNCT_PREFERRED_BREAK_AFTER = (".", "!", "?", "…") SendEvent = Callable[[dict[str, Any]], Awaitable[bool]] def _env_bool(name: str, default: bool = False) -> bool: raw = str(os.getenv(name, "1" if default else "0") or "").strip().lower() return raw in {"1", "true", "yes", "on"} def _env_int(name: str, default: int) -> int: raw = str(os.getenv(name, str(default)) or "").strip() try: return int(raw) except ValueError: return int(default) def _env_float(name: str, default: float) -> float: raw = str(os.getenv(name, str(default)) or "").strip() try: return float(raw) except ValueError: return float(default) def _env_json_string_map(name: str) -> dict[str, str]: raw = str(os.getenv(name, "") or "").strip() if not raw: return {} try: payload = json.loads(raw) except json.JSONDecodeError: logger.warning("Ignoring invalid %s JSON", name) return {} if not isinstance(payload, dict): logger.warning("Ignoring %s because it is not a JSON object", name) return {} fields: dict[str, str] = {} for key, value in payload.items(): text_key = str(key or "").strip() text_value = str(value or "").strip() if text_key and text_value: fields[text_key] = text_value return fields def _default_diarization_health_url(diarization_http_url: str) -> str: raw = str(diarization_http_url or "").strip() if not raw: return "" try: parsed = urlsplit(raw) except ValueError: return "" if not parsed.scheme or not parsed.netloc: return "" return urlunsplit(parsed._replace(path="/healthz", query="", fragment="")) def parse_optional_bool(value: object) -> bool | None: if isinstance(value, bool): return value raw = "" if value is None else str(value).strip().lower() if not raw: return None if raw in {"1", "true", "yes", "on"}: return True if raw in {"0", "false", "no", "off"}: return False raise ValueError("invalid boolean value") def resolve_optional_bool(value: object, *, default: bool) -> bool: parsed = parse_optional_bool(value) return bool(default) if parsed is None else parsed def _normalize_suffix(file_name: str) -> str: return Path(file_name or "").suffix.lower().strip() def normalize_for_compare(value: str) -> str: return re.sub(r"\s+", " ", str(value or "").strip()).strip() def _compact_error_text(value: str) -> str: return normalize_for_compare(value)[:300] def _normalize_line_endings(value: str) -> str: return str(value or "").replace("\r\n", "\n").replace("\r", "\n").strip() def guess_media_content_type(file_name: str) -> str: guessed, _ = mimetypes.guess_type(file_name or "") return guessed or "application/octet-stream" def normalize_diarization_audio_format(value: object) -> str: raw = str(value or "").strip().lower() if raw in {"auto", "wav"}: return raw return "aac" def _missing_audio_track_message(file_path: Path) -> str: name = Path(file_path).name or "uploaded file" return f'uploaded file "{name}" does not contain an audio track' @dataclass(slots=True) class ParsedTranscribeResponse: payload: dict[str, Any] text: str raw_text: str content_type: str parsed_as_json: bool json_unwrap_depth: int = 0 def _decode_transcribe_json_candidate(raw_text: str) -> tuple[dict[str, Any] | None, int]: candidate: Any = raw_text for depth in range(0, 3): if not isinstance(candidate, str): break try: decoded = json.loads(candidate) except json.JSONDecodeError: break if isinstance(decoded, dict): return dict(decoded), depth + 1 if isinstance(decoded, str): candidate = decoded.lstrip("\ufeff").strip() continue break return None, 0 def _parse_transcribe_response(*, body: bytes, content_type: str | None) -> ParsedTranscribeResponse: raw_text = body.decode("utf-8", errors="replace").lstrip("\ufeff") normalized_raw = _normalize_line_endings(raw_text) normalized_content_type = str(content_type or "").lower() payload_obj, unwrap_depth = _decode_transcribe_json_candidate(raw_text) if isinstance(payload_obj, dict): payload = dict(payload_obj) text = _normalize_line_endings(str(payload.get("text") or "")) payload["text"] = text return ParsedTranscribeResponse( payload=payload, text=text, raw_text=raw_text, content_type=normalized_content_type, parsed_as_json=True, json_unwrap_depth=unwrap_depth, ) payload = {"text": normalized_raw, "words": []} return ParsedTranscribeResponse( payload=payload, text=normalized_raw, raw_text=raw_text, content_type=normalized_content_type, parsed_as_json=False, json_unwrap_depth=0, ) def _transcribe_response_debug_info(parsed: ParsedTranscribeResponse) -> dict[str, Any]: payload_keys = sorted(str(key) for key in parsed.payload.keys()) words = parsed.payload.get("words") words_count = len(words) if isinstance(words, list) else None debug: dict[str, Any] = { "linto_content_type": parsed.content_type or None, "parsed_as_json": parsed.parsed_as_json, "json_unwrap_depth": parsed.json_unwrap_depth, "payload_keys": payload_keys, "body_preview": _compact_error_text(parsed.raw_text), } if words_count is not None: debug["words_count"] = words_count if parsed.text: debug["text_preview"] = _compact_error_text(parsed.text) return debug def extract_transcribe_text(*, body: bytes, content_type: str | None) -> str: return _parse_transcribe_response(body=body, content_type=content_type).text @dataclass(slots=True) class FileModeSettings: enabled: bool max_upload_bytes: int linto_http_url: str linto_http_host: str linto_http_language: str diarization_enabled: bool diarization_http_url: str diarization_http_host: str diarization_timeout_s: float diarization_retry_limit: int diarization_num_speakers: int | None diarization_min_speakers: int | None diarization_max_speakers: int | None diarization_turn_break_s: float diarization_syntax_postprocess_default: bool diarization_syntax_sentence_gap_s: float diarization_syntax_dominance_threshold: float job_storage_dir: Path job_event_prefix: str job_session_event_prefix: str diar_event_prefix: str diar_text_event_prefix: str diar_text_session_event_prefix: str chunk_seconds: int chunk_overlap_seconds: int chunk_retry_limit: int text_soft_break_s: float text_hard_break_s: float diarization_health_url: str = "" diarization_health_host: str = "" diarization_health_timeout_s: float = 5.0 diarization_audio_format: str = "aac" diarization_wav_max_upload_mb: int = DIARIZATION_WAV_MAX_UPLOAD_MB_DEFAULT diarization_audio_max_upload_mb: int = DIARIZATION_AUDIO_MAX_UPLOAD_MB_DEFAULT diarization_extra_fields: dict[str, str] = field(default_factory=dict) file_job_lock_url: str = "" file_job_lock_timeout_s: float = 5.0 file_job_lock_heartbeat_s: float = 30.0 linto_http_proxy_url: str = "" diarization_http_proxy_url: str = "" def event_subject(self, job_id: str) -> str: prefix = self.job_event_prefix return f"{prefix}{job_id}" if prefix.endswith(".") else f"{prefix}.{job_id}" def session_event_subject(self, session_id: str) -> str: prefix = self.job_session_event_prefix return f"{prefix}{session_id}" if prefix.endswith(".") else f"{prefix}.{session_id}" def diar_event_subject(self, job_id: str) -> str: prefix = self.diar_event_prefix return f"{prefix}{job_id}" if prefix.endswith(".") else f"{prefix}.{job_id}" def diar_text_event_subject(self, job_id: str) -> str: prefix = self.diar_text_event_prefix return f"{prefix}{job_id}" if prefix.endswith(".") else f"{prefix}.{job_id}" def diar_text_session_subject(self, session_id: str) -> str: prefix = self.diar_text_session_event_prefix return f"{prefix}{session_id}" if prefix.endswith(".") else f"{prefix}.{session_id}" def load_file_mode_settings() -> FileModeSettings: upload_mb = max(16, _env_int("WEBUI_FILE_MAX_UPLOAD_MB", 1024)) linto_http_url = str(os.getenv("LINTO_HTTP_URL", "") or "").strip() linto_http_host = str(os.getenv("LINTO_HTTP_HOST", "") or "").strip() linto_http_language = str(os.getenv("LINTO_HTTP_LANGUAGE", "ru") or "").strip() diarization_http_url = str(os.getenv("DIARIZATION_HTTP_URL", "") or "").strip() diarization_http_host = str(os.getenv("DIARIZATION_HTTP_HOST", "") or "").strip() diarization_health_url = str( os.getenv("DIARIZATION_HEALTH_URL", _default_diarization_health_url(diarization_http_url)) or "" ).strip() diarization_health_host = str(os.getenv("DIARIZATION_HEALTH_HOST", diarization_http_host) or "").strip() diarization_audio_format = normalize_diarization_audio_format(os.getenv("DIARIZATION_AUDIO_FORMAT", "aac")) diarization_num_speakers = _env_int("DIARIZATION_NUM_SPEAKERS", 0) diarization_min_speakers = _env_int("DIARIZATION_MIN_SPEAKERS", 0) diarization_max_speakers = _env_int("DIARIZATION_MAX_SPEAKERS", 0) chunk_seconds = max(30, _env_int("FILE_CHUNK_SECONDS", 300)) chunk_overlap_seconds = max(0, _env_int("FILE_CHUNK_OVERLAP_SECONDS", 15)) if chunk_overlap_seconds >= chunk_seconds: chunk_overlap_seconds = max(0, chunk_seconds - 1) return FileModeSettings( enabled=_env_bool("WEBUI_FILE_TRANSCRIBE_ENABLED", False), max_upload_bytes=upload_mb * 1024 * 1024, linto_http_url=linto_http_url, linto_http_proxy_url=str(os.getenv("LINTO_HTTP_PROXY_URL", "") or "").strip(), linto_http_host=linto_http_host, linto_http_language=linto_http_language, diarization_enabled=_env_bool("DIARIZATION_ENABLED", False), diarization_http_url=diarization_http_url, diarization_http_proxy_url=str(os.getenv("DIARIZATION_HTTP_PROXY_URL", "") or "").strip(), diarization_http_host=diarization_http_host, diarization_timeout_s=max(1.0, _env_float("DIARIZATION_TIMEOUT_S", 1800.0)), diarization_retry_limit=max(0, _env_int("DIARIZATION_RETRY_LIMIT", 0)), diarization_num_speakers=diarization_num_speakers if diarization_num_speakers > 0 else None, diarization_min_speakers=diarization_min_speakers if diarization_min_speakers > 0 else None, diarization_max_speakers=diarization_max_speakers if diarization_max_speakers > 0 else None, diarization_turn_break_s=max(0.1, _env_float("DIARIZATION_TURN_BREAK_S", 1.0)), diarization_syntax_postprocess_default=_env_bool("DIARIZATION_SYNTAX_POSTPROCESS_DEFAULT", False), diarization_syntax_sentence_gap_s=max(0.1, _env_float("DIARIZATION_SYNTAX_SENTENCE_GAP_S", 1.2)), diarization_syntax_dominance_threshold=min( 1.0, max(0.0, _env_float("DIARIZATION_SYNTAX_DOMINANCE_THRESHOLD", 0.65)), ), job_storage_dir=Path( str(os.getenv("FILE_JOB_STORAGE_DIR", "локальный источник") or "локальный источник") ).resolve(), job_event_prefix=str( os.getenv("ASR_FILE_EVENT_PREFIX", "inference.whisper.file.event.") or "" ).strip(), job_session_event_prefix=str( os.getenv("ASR_FILE_SESSION_EVENT_PREFIX", "inference.whisper.file.session.") or "" ).strip(), diar_event_prefix=str( os.getenv("ASR_FILE_DIAR_EVENT_PREFIX", "inference.whisper.file.diar.event.") or "" ).strip(), diar_text_event_prefix=str( os.getenv("ASR_FILE_DIAR_TEXT_EVENT_PREFIX", "inference.whisper.file.diar_text.event.") or "" ).strip(), diar_text_session_event_prefix=str( os.getenv("ASR_FILE_DIAR_TEXT_SESSION_EVENT_PREFIX", "inference.whisper.file.diar_text.session.") or "" ).strip(), chunk_seconds=chunk_seconds, chunk_overlap_seconds=chunk_overlap_seconds, chunk_retry_limit=max(0, _env_int("FILE_CHUNK_RETRY_LIMIT", 2)), text_soft_break_s=max(0.1, _env_float("FILE_TEXT_SOFT_BREAK_S", 0.8)), text_hard_break_s=max(0.2, _env_float("FILE_TEXT_HARD_BREAK_S", 1.8)), diarization_health_url=diarization_health_url, diarization_health_host=diarization_health_host, diarization_health_timeout_s=max(1.0, _env_float("DIARIZATION_HEALTH_TIMEOUT_S", 5.0)), diarization_audio_format=diarization_audio_format, diarization_wav_max_upload_mb=max( 1, _env_int("DIARIZATION_WAV_MAX_UPLOAD_MB", DIARIZATION_WAV_MAX_UPLOAD_MB_DEFAULT), ), diarization_audio_max_upload_mb=max( 1, _env_int("DIARIZATION_AUDIO_MAX_UPLOAD_MB", DIARIZATION_AUDIO_MAX_UPLOAD_MB_DEFAULT), ), diarization_extra_fields=_env_json_string_map("DIARIZATION_EXTRA_FIELDS_JSON"), file_job_lock_url=str(os.getenv("FILE_JOB_LOCK_URL", "") or "").strip().rstrip("/"), file_job_lock_timeout_s=max(1.0, _env_float("FILE_JOB_LOCK_TIMEOUT_S", 5.0)), file_job_lock_heartbeat_s=max(5.0, _env_float("FILE_JOB_LOCK_HEARTBEAT_S", 30.0)), ) @dataclass(slots=True) class TimedWord: text: str start_s: float end_s: float @dataclass(slots=True) class RenderedTimedWord: text: str start_s: float end_s: float compare_key: str piece_index: int timeline_index: int | None = None @dataclass(slots=True) class MediaChunk: index: int total: int start_s: float end_s: float keep_from_s: float path: Path class MissingWordTimestampsError(RuntimeError): def __init__(self, message: str, *, debug: dict[str, Any]) -> None: super().__init__(message) self.debug = dict(debug) def _missing_word_timestamps_debug(*, payload: dict[str, Any], chunk: MediaChunk) -> dict[str, Any]: raw_debug = payload.get("_response_debug") debug: dict[str, Any] = { "chunk_index": chunk.index, "chunk_total": chunk.total, "chunk_start_s": round(chunk.start_s, 2), "chunk_end_s": round(chunk.end_s, 2), "keep_from_s": round(chunk.keep_from_s, 2), } if isinstance(raw_debug, dict): debug.update(raw_debug) elif "_response_debug" in payload: debug["response_debug_type"] = type(raw_debug).__name__ words = payload.get("words") if isinstance(words, list): debug["words_count"] = len(words) text = _normalize_line_endings(str(payload.get("text") or "")) if text: debug["text_preview"] = _compact_error_text(text) return debug class TimedTranscriptBuilder: _BOUNDARY_LOOKBACK_WORDS = 12 _BOUNDARY_MATCH_TOLERANCE_S = 0.05 def __init__( self, *, soft_break_s: float, hard_break_s: float, require_word_timestamps: bool = False, ) -> None: self._soft_break_s = max(0.1, float(soft_break_s)) self._hard_break_s = max(self._soft_break_s, float(hard_break_s)) self._require_word_timestamps = bool(require_word_timestamps) self._pieces: list[str] = [] self._last_end_s: float | None = None self._last_text = "" self._current_line_len = 0 self._recent_words: list[RenderedTimedWord] = [] self._timeline_words: list[TimedWord] = [] self._force_continuation_separator = False def append_chunk(self, *, payload: dict[str, Any], chunk: MediaChunk) -> str: words = _extract_timed_words(payload) if words: appended = self._append_words( words=[ TimedWord( text=word.text, start_s=chunk.start_s + word.start_s, end_s=chunk.start_s + word.end_s, ) for word in words if ((chunk.start_s + word.start_s + chunk.start_s + word.end_s) / 2.0) >= (chunk.keep_from_s - self._BOUNDARY_MATCH_TOLERANCE_S) ], overlap_start_s=chunk.start_s, ) return appended text = _normalize_line_endings(str(payload.get("text") or "")) if not text: return "" if self._require_word_timestamps or chunk.index > 1: raise MissingWordTimestampsError( "LinTO chunk response is missing word timestamps", debug=_missing_word_timestamps_debug(payload=payload, chunk=chunk), ) return self._append_raw_text(text, start_s=chunk.start_s, end_s=chunk.end_s) def _append_raw_text(self, text: str, *, start_s: float, end_s: float) -> str: separator = self._separator(text=text, start_s=start_s) piece = f"{separator}{text}" self._append_piece(piece=piece, text=text, start_s=start_s, end_s=end_s, timeline_index=None) return piece def _append_words(self, *, words: list[TimedWord], overlap_start_s: float | None = None) -> str: overlap_len = self._find_overlap_prefix_length(words=words, overlap_start_s=overlap_start_s) if overlap_len > 0: self._heal_previous_boundary_after_overlap() words = words[overlap_len:] if words: self._force_continuation_separator = True delta: list[str] = [] for word in words: text = str(word.text or "").strip() if not text: continue separator = self._separator(text=text, start_s=word.start_s) piece = f"{separator}{text}" timeline_index = len(self._timeline_words) self._timeline_words.append(TimedWord(text=text, start_s=word.start_s, end_s=word.end_s)) self._append_piece( piece=piece, text=text, start_s=word.start_s, end_s=word.end_s, timeline_index=timeline_index, ) delta.append(piece) return "".join(delta) def _append_piece( self, *, piece: str, text: str, start_s: float, end_s: float, timeline_index: int | None, ) -> None: self._pieces.append(piece) piece_index = len(self._pieces) - 1 self._last_end_s = end_s self._last_text = text if "\n" in piece: self._current_line_len = len(piece.rsplit("\n", 1)[-1]) else: self._current_line_len += len(piece) compare_key = self._normalize_boundary_token(text) if compare_key: self._recent_words.append( RenderedTimedWord( text=text, start_s=start_s, end_s=end_s, compare_key=compare_key, piece_index=piece_index, timeline_index=timeline_index, ) ) if len(self._recent_words) > (self._BOUNDARY_LOOKBACK_WORDS * 2): self._recent_words = self._recent_words[-(self._BOUNDARY_LOOKBACK_WORDS * 2) :] def _separator(self, *, text: str, start_s: float) -> str: if not self._pieces: return "" if text.startswith(PUNCT_NO_SPACE_BEFORE): self._force_continuation_separator = False return "" if self._last_text.endswith(PUNCT_NO_SPACE_AFTER): self._force_continuation_separator = False return "" if self._force_continuation_separator: self._force_continuation_separator = False return " " last_end_s = self._last_end_s if self._last_end_s is not None else start_s gap_s = max(0.0, float(start_s) - float(last_end_s)) if gap_s >= self._hard_break_s: return "\n\n" if gap_s >= self._soft_break_s: return "\n" if self._should_break_after_sentence(): return "\n" return " " def _should_break_after_sentence(self) -> bool: return self._current_line_len >= 140 and self._last_text.endswith(PUNCT_PREFERRED_BREAK_AFTER) def _recent_boundary_words(self, *, overlap_start_s: float | None) -> list[RenderedTimedWord]: if overlap_start_s is None: return self._recent_words[-self._BOUNDARY_LOOKBACK_WORDS :] return [ word for word in self._recent_words[-(self._BOUNDARY_LOOKBACK_WORDS * 2) :] if ((word.start_s + word.end_s) / 2.0) >= (float(overlap_start_s) - self._BOUNDARY_MATCH_TOLERANCE_S) ][-self._BOUNDARY_LOOKBACK_WORDS :] def _find_overlap_prefix_length(self, *, words: list[TimedWord], overlap_start_s: float | None) -> int: if not words or not self._recent_words: return 0 recent_words = self._recent_boundary_words(overlap_start_s=overlap_start_s) if not recent_words: return 0 recent_keys = [word.compare_key for word in recent_words if word.compare_key] incoming_keys = [ compare_key for compare_key in (self._normalize_boundary_token(word.text) for word in words[: self._BOUNDARY_LOOKBACK_WORDS]) if compare_key ] if not recent_keys or not incoming_keys: return 0 for overlap_len in range(min(len(recent_keys), len(incoming_keys)), 0, -1): if recent_keys[-overlap_len:] == incoming_keys[:overlap_len]: return overlap_len return 0 def _heal_previous_boundary_after_overlap(self) -> None: if not self._recent_words: return last_word = self._recent_words[-1] stripped_text = re.sub(r"(?:\.{3}|…)+$", "", last_word.text).rstrip() if not stripped_text or stripped_text == last_word.text: return self._replace_last_word_text(piece_index=last_word.piece_index, old_text=last_word.text, new_text=stripped_text) last_word.text = stripped_text last_word.compare_key = self._normalize_boundary_token(stripped_text) if last_word.timeline_index is not None and last_word.timeline_index < len(self._timeline_words): self._timeline_words[last_word.timeline_index].text = stripped_text def _replace_last_word_text(self, *, piece_index: int, old_text: str, new_text: str) -> None: if piece_index < 0 or piece_index >= len(self._pieces): return old_piece = self._pieces[piece_index] if old_text and not old_piece.endswith(old_text): return new_piece = f"{old_piece[:-len(old_text)]}{new_text}" if old_text else old_piece self._pieces[piece_index] = new_piece self._last_text = new_text self._current_line_len = max(0, self._current_line_len - (len(old_piece) - len(new_piece))) @staticmethod def _normalize_boundary_token(text: str) -> str: normalized = normalize_for_compare(text).casefold() normalized = re.sub(r"(^[\W_]+|[\W_]+$)", "", normalized, flags=re.UNICODE) return normalized.strip() def render(self) -> str: return "".join(self._pieces).strip() def timed_words(self) -> list[TimedWord]: return [ TimedWord(text=word.text, start_s=word.start_s, end_s=word.end_s) for word in self._timeline_words ] class FileJobStore: _DURABLE_STATE_KEYS = ( "job_id", "session_id", "file_name", "language", "backend", "page_mode", "requested_diarization", "run_diarization", "syntax_postprocess", "diarization_backend_status", "diarization_fallback_reason", "bench_config", "diarization_audio_format", ) def __init__(self, root_dir: Path) -> None: self._root_dir = Path(root_dir).resolve() self._root_dir.mkdir(parents=True, exist_ok=True) def job_dir(self, job_id: str) -> Path: return self._root_dir / str(job_id).strip() def ensure_job_dir(self, job_id: str) -> Path: path = self.job_dir(job_id) path.mkdir(parents=True, exist_ok=True) return path def source_path(self, job_id: str, file_name: str) -> Path: job_dir = self.ensure_job_dir(job_id) suffix = _normalize_suffix(file_name) or ".bin" return job_dir / f"input{suffix}" def normalized_audio_path(self, job_id: str) -> Path: return self.ensure_job_dir(job_id) / "normalized.wav" def diarization_audio_path(self, job_id: str) -> Path: return self.ensure_job_dir(job_id) / "diarization_input.m4a" def chunks_dir(self, job_id: str) -> Path: path = self.ensure_job_dir(job_id) / "chunks" path.mkdir(parents=True, exist_ok=True) return path def state_path(self, job_id: str) -> Path: return self.ensure_job_dir(job_id) / "state.json" def transcript_path(self, job_id: str) -> Path: return self.ensure_job_dir(job_id) / "transcript.txt" def transcription_payload_path(self, job_id: str) -> Path: return self.ensure_job_dir(job_id) / "transcription.json" def diarization_state_path(self, job_id: str) -> Path: return self.ensure_job_dir(job_id) / "diarization_state.json" def diarization_payload_path(self, job_id: str) -> Path: return self.ensure_job_dir(job_id) / "diarization.json" def diarization_text_path(self, job_id: str) -> Path: return self.ensure_job_dir(job_id) / "diarization_text.json" def combined_payload_path(self, job_id: str) -> Path: return self.ensure_job_dir(job_id) / "combined.json" def cancel_path(self, job_id: str) -> Path: return self.ensure_job_dir(job_id) / "cancel.requested" def merged_state_payload(self, job_id: str, payload: dict[str, Any]) -> dict[str, Any]: stored_payload = dict(payload) existing_payload = self.read_state(job_id) or {} for key in self._DURABLE_STATE_KEYS: if key not in stored_payload and key in existing_payload: stored_payload[key] = existing_payload[key] return stored_payload def write_state(self, job_id: str, payload: dict[str, Any]) -> dict[str, Any]: stored_payload = self.merged_state_payload(job_id, payload) self.state_path(job_id).write_text( json.dumps(stored_payload, ensure_ascii=False, indent=2), encoding="utf-8", ) return stored_payload def read_state(self, job_id: str) -> dict[str, Any] | None: path = self.state_path(job_id) if not path.exists(): return None try: return json.loads(path.read_text(encoding="utf-8")) except Exception: return None def write_transcript(self, job_id: str, text: str) -> None: self.transcript_path(job_id).write_text(str(text or ""), encoding="utf-8") def read_transcript(self, job_id: str) -> str: path = self.transcript_path(job_id) if not path.exists(): return "" try: return path.read_text(encoding="utf-8") except Exception: return "" def write_transcription_payload(self, job_id: str, payload: dict[str, Any]) -> None: self.transcription_payload_path(job_id).write_text( json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8", ) def read_transcription_payload(self, job_id: str) -> dict[str, Any] | None: path = self.transcription_payload_path(job_id) if not path.exists(): return None try: return json.loads(path.read_text(encoding="utf-8")) except Exception: return None def write_diarization_state(self, job_id: str, payload: dict[str, Any]) -> None: self.diarization_state_path(job_id).write_text( json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8", ) def read_diarization_state(self, job_id: str) -> dict[str, Any] | None: path = self.diarization_state_path(job_id) if not path.exists(): return None try: return json.loads(path.read_text(encoding="utf-8")) except Exception: return None def write_diarization_payload(self, job_id: str, payload: dict[str, Any]) -> None: self.diarization_payload_path(job_id).write_text( json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8", ) def request_cancel(self, job_id: str) -> None: self.cancel_path(job_id).touch() def clear_cancel_request(self, job_id: str) -> None: self.cancel_path(job_id).unlink(missing_ok=True) def is_cancel_requested(self, job_id: str) -> bool: return self.cancel_path(job_id).exists() def read_diarization_payload(self, job_id: str) -> dict[str, Any] | None: path = self.diarization_payload_path(job_id) if not path.exists(): return None try: return json.loads(path.read_text(encoding="utf-8")) except Exception: return None def write_diarization_text(self, job_id: str, payload: dict[str, Any]) -> None: self.diarization_text_path(job_id).write_text( json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8", ) def read_diarization_text(self, job_id: str) -> dict[str, Any] | None: path = self.diarization_text_path(job_id) if not path.exists(): return None try: return json.loads(path.read_text(encoding="utf-8")) except Exception: return None def write_combined_payload(self, job_id: str, payload: dict[str, Any]) -> None: self.combined_payload_path(job_id).write_text( json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8", ) def read_combined_payload(self, job_id: str) -> dict[str, Any] | None: path = self.combined_payload_path(job_id) if not path.exists(): return None try: return json.loads(path.read_text(encoding="utf-8")) except Exception: return None def list_job_ids(self) -> list[str]: if not self._root_dir.exists(): return [] job_ids: list[str] = [] for path in self._root_dir.iterdir(): if not path.is_dir(): continue job_id = path.name.strip() if job_id: job_ids.append(job_id) return sorted(job_ids) def delete_job(self, job_id: str) -> None: path = self.job_dir(job_id) if not path.exists(): return shutil.rmtree(path, ignore_errors=True) def probe_wav_duration_s(file_path: Path) -> float: path = Path(file_path).resolve() with wave.open(str(path), "rb") as wav_file: frame_rate = wav_file.getframerate() or TARGET_SAMPLE_RATE frame_count = wav_file.getnframes() return max(0.0, float(frame_count) / float(frame_rate)) def write_normalized_audio_wav( file_path: Path, *, output_path: Path, sample_rate: int, chunk_ms: int = 1000, ) -> Path: target = Path(output_path).resolve() target.parent.mkdir(parents=True, exist_ok=True) with wave.open(str(target), "wb") as wav_file: wav_file.setnchannels(1) wav_file.setsampwidth(2) wav_file.setframerate(sample_rate) for pcm_bytes in iter_pcm_s16le_chunks(file_path, sample_rate=sample_rate, chunk_ms=chunk_ms): wav_file.writeframes(pcm_bytes) return target @dataclass(slots=True) class DiarizationAudioArtifact: path: Path requested_audio_format: str effective_audio_format: str artifact_strategy: str artifact_size_bytes: int wav_guard_fallback: bool = False source_copy_guard_fallback: bool = False @property def artifact_size_mb(self) -> float: return round(float(self.artifact_size_bytes) / 1024.0 / 1024.0, 3) def as_payload(self) -> dict[str, Any]: return { "path_name": self.path.name, "requested_audio_format": self.requested_audio_format, "effective_audio_format": self.effective_audio_format, "artifact_strategy": self.artifact_strategy, "artifact_size_bytes": self.artifact_size_bytes, "artifact_size_mb": self.artifact_size_mb, "wav_guard_fallback": self.wav_guard_fallback, "source_copy_guard_fallback": self.source_copy_guard_fallback, } def _ensure_nonempty_artifact(path: Path, *, error_prefix: str) -> None: if not path.exists() or path.stat().st_size <= 0: raise RuntimeError(f"{error_prefix}: output file is empty") def _audio_codec_name(file_path: Path) -> str: command = [ "ffprobe", "-v", "error", "-select_streams", "a:0", "-show_entries", "stream=codec_name", "-of", "json", str(Path(file_path).resolve()), ] try: result = subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) except (FileNotFoundError, subprocess.CalledProcessError): return "" try: payload = json.loads(result.stdout.decode("utf-8", errors="replace") or "{}") except json.JSONDecodeError: return "" streams = payload.get("streams") if isinstance(payload, dict) else None if not isinstance(streams, list) or not streams: return "" first = streams[0] if not isinstance(first, dict): return "" return str(first.get("codec_name") or "").strip().lower() def source_supports_diarization_audio_copy(file_path: Path) -> bool: suffix = Path(file_path).suffix.lower() return suffix in {".m4a", ".mp4"} and _audio_codec_name(file_path) == "aac" def write_diarization_audio_copy_artifact( file_path: Path, *, output_path: Path, ) -> Path: source = Path(file_path).resolve() target = Path(output_path).resolve() target.parent.mkdir(parents=True, exist_ok=True) command = [ "ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "error", "-y", "-i", str(source), "-vn", "-sn", "-dn", "-map", "0:a:0", "-c:a", "copy", str(target), ] try: subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) except FileNotFoundError as exc: raise RuntimeError("ffmpeg is required to copy the diarization audio track") from exc except subprocess.CalledProcessError as exc: details = exc.stderr.decode("utf-8", errors="replace").strip() or str(exc) raise RuntimeError(f"failed to copy audio-only track for speaker separation: {details}") from exc _ensure_nonempty_artifact( target, error_prefix="failed to copy audio-only track for speaker separation", ) return target def write_diarization_audio_artifact( file_path: Path, *, output_path: Path, ) -> Path: source = Path(file_path).resolve() target = Path(output_path).resolve() target.parent.mkdir(parents=True, exist_ok=True) command = [ "ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "error", "-y", "-i", str(source), "-vn", "-sn", "-dn", "-c:a", "aac", "-b:a", "64k", str(target), ] try: subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) except FileNotFoundError as exc: raise RuntimeError("ffmpeg is required to prepare the diarization audio track") from exc except subprocess.CalledProcessError as exc: details = exc.stderr.decode("utf-8", errors="replace").strip() or str(exc) raise RuntimeError(f"failed to prepare audio-only track for speaker separation: {details}") from exc _ensure_nonempty_artifact( target, error_prefix="failed to prepare audio-only track for speaker separation", ) return target def prepare_diarization_audio_artifact( *, source_path: Path, normalized_path: Path, diarization_audio_path: Path, requested_audio_format: object, wav_max_upload_mb: int = DIARIZATION_WAV_MAX_UPLOAD_MB_DEFAULT, audio_max_upload_mb: int = DIARIZATION_AUDIO_MAX_UPLOAD_MB_DEFAULT, ) -> DiarizationAudioArtifact: requested = normalize_diarization_audio_format(requested_audio_format) normalized = Path(normalized_path).resolve() target = Path(diarization_audio_path).resolve() wav_limit_bytes = max(1, int(wav_max_upload_mb)) * 1024 * 1024 audio_limit_bytes = max(1, int(audio_max_upload_mb)) * 1024 * 1024 if requested == "wav" and normalized.exists() and normalized.stat().st_size <= wav_limit_bytes: target.unlink(missing_ok=True) return DiarizationAudioArtifact( path=normalized, requested_audio_format=requested, effective_audio_format="wav", artifact_strategy="wav", artifact_size_bytes=int(normalized.stat().st_size), wav_guard_fallback=False, ) wav_guard_fallback = requested == "wav" if requested in {"auto", "wav"} and source_supports_diarization_audio_copy(source_path): write_diarization_audio_copy_artifact(source_path, output_path=target) artifact_size = int(target.stat().st_size) if artifact_size <= audio_limit_bytes: return DiarizationAudioArtifact( path=target, requested_audio_format=requested, effective_audio_format="aac", artifact_strategy="source_aac_copy", artifact_size_bytes=artifact_size, wav_guard_fallback=wav_guard_fallback, ) logger.info( "diarization source AAC copy is %.3f MB, above %.3f MB upload guard; falling back to compact AAC transcode", artifact_size / 1024.0 / 1024.0, audio_limit_bytes / 1024.0 / 1024.0, ) write_diarization_audio_artifact(normalized, output_path=target) return DiarizationAudioArtifact( path=target, requested_audio_format=requested, effective_audio_format="aac", artifact_strategy="aac_transcode", artifact_size_bytes=int(target.stat().st_size), wav_guard_fallback=wav_guard_fallback, source_copy_guard_fallback=True, ) write_diarization_audio_artifact(normalized, output_path=target) return DiarizationAudioArtifact( path=target, requested_audio_format=requested, effective_audio_format="aac", artifact_strategy="aac_transcode", artifact_size_bytes=int(target.stat().st_size), wav_guard_fallback=wav_guard_fallback, ) def estimate_chunk_total(*, duration_s: float, chunk_seconds: int, overlap_seconds: int) -> int: duration_s = max(0.0, float(duration_s)) chunk_seconds = max(1, int(chunk_seconds)) overlap_seconds = max(0, min(int(overlap_seconds), chunk_seconds - 1)) if duration_s <= chunk_seconds: return 1 step_s = max(1, chunk_seconds - overlap_seconds) return int(math.ceil((duration_s - chunk_seconds) / float(step_s))) + 1 def iter_normalized_wav_chunk_files( file_path: Path, *, output_dir: Path, sample_rate: int, chunk_seconds: int, overlap_seconds: int, total_chunks: int, ): path = Path(file_path).resolve() if not path.exists(): raise FileNotFoundError(path) output_dir = Path(output_dir).resolve() output_dir.mkdir(parents=True, exist_ok=True) chunk_seconds = max(1, int(chunk_seconds)) overlap_seconds = max(0, min(int(overlap_seconds), chunk_seconds - 1)) sample_rate = max(1, int(sample_rate)) window_frames = chunk_seconds * sample_rate overlap_frames = overlap_seconds * sample_rate step_frames = max(sample_rate, window_frames - overlap_frames) emitted = 0 with wave.open(str(path), "rb") as wav_file: channels = wav_file.getnchannels() sample_width = wav_file.getsampwidth() frame_rate = wav_file.getframerate() frame_count = wav_file.getnframes() if channels != 1 or sample_width != 2 or frame_rate != sample_rate: raise ValueError( f"normalized wav must be mono s16 {sample_rate}Hz; got channels={channels} sample_width={sample_width} frame_rate={frame_rate}" ) while True: start_frame = emitted * step_frames if start_frame >= frame_count: break frames_to_read = min(window_frames, frame_count - start_frame) if frames_to_read <= 0: break wav_file.setpos(start_frame) chunk_bytes = wav_file.readframes(frames_to_read) if not chunk_bytes: break chunk_path = _write_chunk_wav(output_dir, emitted, chunk_bytes, sample_rate) start_s = float(start_frame) / float(sample_rate) end_s = start_s + (frames_to_read / float(sample_rate)) keep_from_s = 0.0 if emitted == 0 else start_s + overlap_seconds emitted += 1 yield MediaChunk( index=emitted, total=max(total_chunks, emitted), start_s=start_s, end_s=end_s, keep_from_s=keep_from_s, path=chunk_path, ) async def transcribe_http_json( *, settings: FileModeSettings, file_path: Path, file_name: str, language: str | None, http: "aiohttp.ClientSession | None" = None, cancel_event: asyncio.Event | None = None, ) -> dict[str, Any]: status, body, content_type = await _post_to_linto( settings=settings, file_path=file_path, file_name=file_name, language=language, accept="application/json", http=http, cancel_event=cancel_event, ) parsed = _parse_transcribe_response(body=body, content_type=content_type) payload = dict(parsed.payload) payload["_response_debug"] = _transcribe_response_debug_info(parsed) if status >= 400: details = payload.get("text") or parsed.raw_text raise RuntimeError(f"LinTO HTTP error {status}: {_compact_error_text(str(details))}") return payload async def transcribe_http_text( *, settings: FileModeSettings, file_path: Path, file_name: str, language: str | None, http: "aiohttp.ClientSession | None" = None, cancel_event: asyncio.Event | None = None, ) -> str: status, body, content_type = await _post_to_linto( settings=settings, file_path=file_path, file_name=file_name, language=language, accept="text/plain", http=http, cancel_event=cancel_event, ) text = extract_transcribe_text(body=body, content_type=content_type) if status >= 400: details = text or body.decode("utf-8", errors="replace") raise RuntimeError(f"LinTO HTTP error {status}: {_compact_error_text(details)}") return text async def _post_to_linto( *, settings: FileModeSettings, file_path: Path, file_name: str, language: str | None, accept: str, http: "aiohttp.ClientSession | None", cancel_event: asyncio.Event | None, ) -> tuple[int, bytes, str | None]: import aiohttp from src.model_transport import model_http_session target_url = str(settings.linto_http_url or "").strip() target_host = str(settings.linto_http_host or "").strip() target_language = str(settings.linto_http_language or "").strip() if not target_url: raise RuntimeError("LINTO_HTTP_URL is required for HTTP file transcription mode") headers: dict[str, str] = {"Accept": accept} if target_host: headers["Host"] = target_host params: dict[str, str] = {} resolved_language = _normalize_line_endings(language or target_language) if resolved_language: params["language"] = resolved_language timeout = aiohttp.ClientTimeout(total=None, connect=10.0) proxy_url = settings.linto_http_proxy_url.strip() async with model_http_session(proxy_url=proxy_url, http=http, timeout=timeout) as session: with Path(file_path).open("rb") as handle: form = aiohttp.FormData() form.add_field( "file", handle, filename=file_name, content_type=guess_media_content_type(file_name), ) async with session.post( target_url, params=params or None, data=form, headers=headers, allow_redirects=False, ) as response: if cancel_event is not None and cancel_event.is_set(): raise asyncio.CancelledError() body = await response.read() return response.status, body, response.headers.get("Content-Type") def _extract_timed_words(payload: dict[str, Any]) -> list[TimedWord]: raw_words = payload.get("words") if not isinstance(raw_words, list): return [] words: list[TimedWord] = [] for item in raw_words: if not isinstance(item, dict): continue text = str(item.get("word") or item.get("text") or "").strip() if not text: continue try: start_s = float(item.get("start")) end_s = float(item.get("end")) except (TypeError, ValueError): continue if end_s < start_s: end_s = start_s words.append(TimedWord(text=text, start_s=start_s, end_s=end_s)) return words def _write_chunk_wav(output_dir: Path, zero_index: int, pcm_bytes: bytes, sample_rate: int) -> Path: chunk_path = output_dir / f"chunk_{zero_index + 1:04d}.wav" with wave.open(str(chunk_path), "wb") as wav_file: wav_file.setnchannels(1) wav_file.setsampwidth(2) wav_file.setframerate(sample_rate) wav_file.writeframes(pcm_bytes) return chunk_path def iter_pcm_s16le_chunks( file_path: Path, *, sample_rate: int, chunk_ms: int, ): path = Path(file_path).resolve() if not path.exists(): raise FileNotFoundError(path) suffix = _normalize_suffix(path.name) if suffix and suffix not in SUPPORTED_FILE_SUFFIXES: raise ValueError(f"unsupported file extension: {suffix}") try: import av except ImportError as exc: raise RuntimeError("PyAV is required for HTTP file transcription mode") from exc samples_per_chunk = max(1, int(sample_rate * max(1, chunk_ms) / 1000)) bytes_per_chunk = samples_per_chunk * 2 pending = bytearray() resampler = av.audio.resampler.AudioResampler(format="s16", layout="mono", rate=sample_rate) with av.open(str(path), mode="r", metadata_errors="ignore") as container: if not any(stream.type == "audio" for stream in container.streams): raise ValueError(_missing_audio_track_message(path)) for frame in _iter_audio_frames(container, av): for output_frame in resampler.resample(frame): pending.extend(output_frame.to_ndarray().tobytes()) while len(pending) >= bytes_per_chunk: yield bytes(pending[:bytes_per_chunk]) del pending[:bytes_per_chunk] for output_frame in resampler.resample(None): pending.extend(output_frame.to_ndarray().tobytes()) while len(pending) >= bytes_per_chunk: yield bytes(pending[:bytes_per_chunk]) del pending[:bytes_per_chunk] if pending: yield bytes(pending) def _iter_audio_frames(container, av_module): try: frames = container.decode(audio=0) except Exception as exc: raise ValueError(f"failed to decode audio: {exc}") from exc iterator = iter(frames) while True: try: yield next(iterator) except StopIteration: return except Exception as exc: invalid_data_error = getattr(getattr(av_module, "error", None), "InvalidDataError", None) if invalid_data_error and isinstance(exc, invalid_data_error): continue raise ValueError(f"invalid audio frame: {exc}") from exc