Текстовый файл из учебного комплекта, не свидетельство установленного runtime. Вернуться к учебнику · Исходный файл
1 from __future__ import annotations 2 3 import asyncio 4 import json 5 import logging 6 import math 7 import mimetypes 8 import os 9 import re 10 import shutil 11 import subprocess 12 import time 13 from urllib.parse import urlsplit, urlunsplit 14 import wave 15 from dataclasses import dataclass, field 16 from pathlib import Path 17 from typing import TYPE_CHECKING, Any, Awaitable, Callable 18 19 if TYPE_CHECKING: 20 import aiohttp 21 22 logger = logging.getLogger("whisper_webui.file_mode") 23 24 TARGET_SAMPLE_RATE = 16000 25 PCM_BYTES_PER_SECOND = TARGET_SAMPLE_RATE * 2 26 SUPPORTED_FILE_SUFFIXES = {".wav", ".mp3", ".m4a", ".mp4"} 27 DIARIZATION_WAV_MAX_UPLOAD_MB_DEFAULT = 64 28 DIARIZATION_AUDIO_MAX_UPLOAD_MB_DEFAULT = 64 29 PUNCT_NO_SPACE_BEFORE = tuple(".,!?;:%)]}") 30 PUNCT_NO_SPACE_AFTER = tuple("([{\"") 31 PUNCT_PREFERRED_BREAK_AFTER = (".", "!", "?", "…") 32 SendEvent = Callable[[dict[str, Any]], Awaitable[bool]] 33 34 35 def _env_bool(name: str, default: bool = False) -> bool: 36 raw = str(os.getenv(name, "1" if default else "0") or "").strip().lower() 37 return raw in {"1", "true", "yes", "on"} 38 39 40 def _env_int(name: str, default: int) -> int: 41 raw = str(os.getenv(name, str(default)) or "").strip() 42 try: 43 return int(raw) 44 except ValueError: 45 return int(default) 46 47 48 def _env_float(name: str, default: float) -> float: 49 raw = str(os.getenv(name, str(default)) or "").strip() 50 try: 51 return float(raw) 52 except ValueError: 53 return float(default) 54 55 56 def _env_json_string_map(name: str) -> dict[str, str]: 57 raw = str(os.getenv(name, "") or "").strip() 58 if not raw: 59 return {} 60 try: 61 payload = json.loads(raw) 62 except json.JSONDecodeError: 63 logger.warning("Ignoring invalid %s JSON", name) 64 return {} 65 if not isinstance(payload, dict): 66 logger.warning("Ignoring %s because it is not a JSON object", name) 67 return {} 68 fields: dict[str, str] = {} 69 for key, value in payload.items(): 70 text_key = str(key or "").strip() 71 text_value = str(value or "").strip() 72 if text_key and text_value: 73 fields[text_key] = text_value 74 return fields 75 76 77 def _default_diarization_health_url(diarization_http_url: str) -> str: 78 raw = str(diarization_http_url or "").strip() 79 if not raw: 80 return "" 81 try: 82 parsed = urlsplit(raw) 83 except ValueError: 84 return "" 85 if not parsed.scheme or not parsed.netloc: 86 return "" 87 return urlunsplit(parsed._replace(path="/healthz", query="", fragment="")) 88 89 90 def parse_optional_bool(value: object) -> bool | None: 91 if isinstance(value, bool): 92 return value 93 raw = "" if value is None else str(value).strip().lower() 94 if not raw: 95 return None 96 if raw in {"1", "true", "yes", "on"}: 97 return True 98 if raw in {"0", "false", "no", "off"}: 99 return False 100 raise ValueError("invalid boolean value") 101 102 103 def resolve_optional_bool(value: object, *, default: bool) -> bool: 104 parsed = parse_optional_bool(value) 105 return bool(default) if parsed is None else parsed 106 107 108 def _normalize_suffix(file_name: str) -> str: 109 return Path(file_name or "").suffix.lower().strip() 110 111 112 def normalize_for_compare(value: str) -> str: 113 return re.sub(r"\s+", " ", str(value or "").strip()).strip() 114 115 116 def _compact_error_text(value: str) -> str: 117 return normalize_for_compare(value)[:300] 118 119 120 def _normalize_line_endings(value: str) -> str: 121 return str(value or "").replace("\r\n", "\n").replace("\r", "\n").strip() 122 123 124 def guess_media_content_type(file_name: str) -> str: 125 guessed, _ = mimetypes.guess_type(file_name or "") 126 return guessed or "application/octet-stream" 127 128 129 def normalize_diarization_audio_format(value: object) -> str: 130 raw = str(value or "").strip().lower() 131 if raw in {"auto", "wav"}: 132 return raw 133 return "aac" 134 135 136 def _missing_audio_track_message(file_path: Path) -> str: 137 name = Path(file_path).name or "uploaded file" 138 return f'uploaded file "{name}" does not contain an audio track' 139 140 141 @dataclass(slots=True) 142 class ParsedTranscribeResponse: 143 payload: dict[str, Any] 144 text: str 145 raw_text: str 146 content_type: str 147 parsed_as_json: bool 148 json_unwrap_depth: int = 0 149 150 151 def _decode_transcribe_json_candidate(raw_text: str) -> tuple[dict[str, Any] | None, int]: 152 candidate: Any = raw_text 153 for depth in range(0, 3): 154 if not isinstance(candidate, str): 155 break 156 try: 157 decoded = json.loads(candidate) 158 except json.JSONDecodeError: 159 break 160 if isinstance(decoded, dict): 161 return dict(decoded), depth + 1 162 if isinstance(decoded, str): 163 candidate = decoded.lstrip("\ufeff").strip() 164 continue 165 break 166 return None, 0 167 168 169 def _parse_transcribe_response(*, body: bytes, content_type: str | None) -> ParsedTranscribeResponse: 170 raw_text = body.decode("utf-8", errors="replace").lstrip("\ufeff") 171 normalized_raw = _normalize_line_endings(raw_text) 172 normalized_content_type = str(content_type or "").lower() 173 payload_obj, unwrap_depth = _decode_transcribe_json_candidate(raw_text) 174 175 if isinstance(payload_obj, dict): 176 payload = dict(payload_obj) 177 text = _normalize_line_endings(str(payload.get("text") or "")) 178 payload["text"] = text 179 return ParsedTranscribeResponse( 180 payload=payload, 181 text=text, 182 raw_text=raw_text, 183 content_type=normalized_content_type, 184 parsed_as_json=True, 185 json_unwrap_depth=unwrap_depth, 186 ) 187 188 payload = {"text": normalized_raw, "words": []} 189 return ParsedTranscribeResponse( 190 payload=payload, 191 text=normalized_raw, 192 raw_text=raw_text, 193 content_type=normalized_content_type, 194 parsed_as_json=False, 195 json_unwrap_depth=0, 196 ) 197 198 199 def _transcribe_response_debug_info(parsed: ParsedTranscribeResponse) -> dict[str, Any]: 200 payload_keys = sorted(str(key) for key in parsed.payload.keys()) 201 words = parsed.payload.get("words") 202 words_count = len(words) if isinstance(words, list) else None 203 debug: dict[str, Any] = { 204 "linto_content_type": parsed.content_type or None, 205 "parsed_as_json": parsed.parsed_as_json, 206 "json_unwrap_depth": parsed.json_unwrap_depth, 207 "payload_keys": payload_keys, 208 "body_preview": _compact_error_text(parsed.raw_text), 209 } 210 if words_count is not None: 211 debug["words_count"] = words_count 212 if parsed.text: 213 debug["text_preview"] = _compact_error_text(parsed.text) 214 return debug 215 216 217 def extract_transcribe_text(*, body: bytes, content_type: str | None) -> str: 218 return _parse_transcribe_response(body=body, content_type=content_type).text 219 220 221 @dataclass(slots=True) 222 class FileModeSettings: 223 enabled: bool 224 max_upload_bytes: int 225 linto_http_url: str 226 linto_http_host: str 227 linto_http_language: str 228 diarization_enabled: bool 229 diarization_http_url: str 230 diarization_http_host: str 231 diarization_timeout_s: float 232 diarization_retry_limit: int 233 diarization_num_speakers: int | None 234 diarization_min_speakers: int | None 235 diarization_max_speakers: int | None 236 diarization_turn_break_s: float 237 diarization_syntax_postprocess_default: bool 238 diarization_syntax_sentence_gap_s: float 239 diarization_syntax_dominance_threshold: float 240 job_storage_dir: Path 241 job_event_prefix: str 242 job_session_event_prefix: str 243 diar_event_prefix: str 244 diar_text_event_prefix: str 245 diar_text_session_event_prefix: str 246 chunk_seconds: int 247 chunk_overlap_seconds: int 248 chunk_retry_limit: int 249 text_soft_break_s: float 250 text_hard_break_s: float 251 diarization_health_url: str = "" 252 diarization_health_host: str = "" 253 diarization_health_timeout_s: float = 5.0 254 diarization_audio_format: str = "aac" 255 diarization_wav_max_upload_mb: int = DIARIZATION_WAV_MAX_UPLOAD_MB_DEFAULT 256 diarization_audio_max_upload_mb: int = DIARIZATION_AUDIO_MAX_UPLOAD_MB_DEFAULT 257 diarization_extra_fields: dict[str, str] = field(default_factory=dict) 258 file_job_lock_url: str = "" 259 file_job_lock_timeout_s: float = 5.0 260 file_job_lock_heartbeat_s: float = 30.0 261 linto_http_proxy_url: str = "" 262 diarization_http_proxy_url: str = "" 263 264 def event_subject(self, job_id: str) -> str: 265 prefix = self.job_event_prefix 266 return f"{prefix}{job_id}" if prefix.endswith(".") else f"{prefix}.{job_id}" 267 268 def session_event_subject(self, session_id: str) -> str: 269 prefix = self.job_session_event_prefix 270 return f"{prefix}{session_id}" if prefix.endswith(".") else f"{prefix}.{session_id}" 271 272 def diar_event_subject(self, job_id: str) -> str: 273 prefix = self.diar_event_prefix 274 return f"{prefix}{job_id}" if prefix.endswith(".") else f"{prefix}.{job_id}" 275 276 def diar_text_event_subject(self, job_id: str) -> str: 277 prefix = self.diar_text_event_prefix 278 return f"{prefix}{job_id}" if prefix.endswith(".") else f"{prefix}.{job_id}" 279 280 def diar_text_session_subject(self, session_id: str) -> str: 281 prefix = self.diar_text_session_event_prefix 282 return f"{prefix}{session_id}" if prefix.endswith(".") else f"{prefix}.{session_id}" 283 284 285 def load_file_mode_settings() -> FileModeSettings: 286 upload_mb = max(16, _env_int("WEBUI_FILE_MAX_UPLOAD_MB", 1024)) 287 linto_http_url = str(os.getenv("LINTO_HTTP_URL", "") or "").strip() 288 linto_http_host = str(os.getenv("LINTO_HTTP_HOST", "") or "").strip() 289 linto_http_language = str(os.getenv("LINTO_HTTP_LANGUAGE", "ru") or "").strip() 290 diarization_http_url = str(os.getenv("DIARIZATION_HTTP_URL", "") or "").strip() 291 diarization_http_host = str(os.getenv("DIARIZATION_HTTP_HOST", "") or "").strip() 292 diarization_health_url = str( 293 os.getenv("DIARIZATION_HEALTH_URL", _default_diarization_health_url(diarization_http_url)) or "" 294 ).strip() 295 diarization_health_host = str(os.getenv("DIARIZATION_HEALTH_HOST", diarization_http_host) or "").strip() 296 diarization_audio_format = normalize_diarization_audio_format(os.getenv("DIARIZATION_AUDIO_FORMAT", "aac")) 297 diarization_num_speakers = _env_int("DIARIZATION_NUM_SPEAKERS", 0) 298 diarization_min_speakers = _env_int("DIARIZATION_MIN_SPEAKERS", 0) 299 diarization_max_speakers = _env_int("DIARIZATION_MAX_SPEAKERS", 0) 300 chunk_seconds = max(30, _env_int("FILE_CHUNK_SECONDS", 300)) 301 chunk_overlap_seconds = max(0, _env_int("FILE_CHUNK_OVERLAP_SECONDS", 15)) 302 if chunk_overlap_seconds >= chunk_seconds: 303 chunk_overlap_seconds = max(0, chunk_seconds - 1) 304 return FileModeSettings( 305 enabled=_env_bool("WEBUI_FILE_TRANSCRIBE_ENABLED", False), 306 max_upload_bytes=upload_mb * 1024 * 1024, 307 linto_http_url=linto_http_url, 308 linto_http_proxy_url=str(os.getenv("LINTO_HTTP_PROXY_URL", "") or "").strip(), 309 linto_http_host=linto_http_host, 310 linto_http_language=linto_http_language, 311 diarization_enabled=_env_bool("DIARIZATION_ENABLED", False), 312 diarization_http_url=diarization_http_url, 313 diarization_http_proxy_url=str(os.getenv("DIARIZATION_HTTP_PROXY_URL", "") or "").strip(), 314 diarization_http_host=diarization_http_host, 315 diarization_timeout_s=max(1.0, _env_float("DIARIZATION_TIMEOUT_S", 1800.0)), 316 diarization_retry_limit=max(0, _env_int("DIARIZATION_RETRY_LIMIT", 0)), 317 diarization_num_speakers=diarization_num_speakers if diarization_num_speakers > 0 else None, 318 diarization_min_speakers=diarization_min_speakers if diarization_min_speakers > 0 else None, 319 diarization_max_speakers=diarization_max_speakers if diarization_max_speakers > 0 else None, 320 diarization_turn_break_s=max(0.1, _env_float("DIARIZATION_TURN_BREAK_S", 1.0)), 321 diarization_syntax_postprocess_default=_env_bool("DIARIZATION_SYNTAX_POSTPROCESS_DEFAULT", False), 322 diarization_syntax_sentence_gap_s=max(0.1, _env_float("DIARIZATION_SYNTAX_SENTENCE_GAP_S", 1.2)), 323 diarization_syntax_dominance_threshold=min( 324 1.0, 325 max(0.0, _env_float("DIARIZATION_SYNTAX_DOMINANCE_THRESHOLD", 0.65)), 326 ), 327 job_storage_dir=Path( 328 str(os.getenv("FILE_JOB_STORAGE_DIR", "локальный источник;) or "локальный источник;) 329 ).resolve(), 330 job_event_prefix=str( 331 os.getenv("ASR_FILE_EVENT_PREFIX", "inference.whisper.file.event.") or "" 332 ).strip(), 333 job_session_event_prefix=str( 334 os.getenv("ASR_FILE_SESSION_EVENT_PREFIX", "inference.whisper.file.session.") or "" 335 ).strip(), 336 diar_event_prefix=str( 337 os.getenv("ASR_FILE_DIAR_EVENT_PREFIX", "inference.whisper.file.diar.event.") or "" 338 ).strip(), 339 diar_text_event_prefix=str( 340 os.getenv("ASR_FILE_DIAR_TEXT_EVENT_PREFIX", "inference.whisper.file.diar_text.event.") or "" 341 ).strip(), 342 diar_text_session_event_prefix=str( 343 os.getenv("ASR_FILE_DIAR_TEXT_SESSION_EVENT_PREFIX", "inference.whisper.file.diar_text.session.") or "" 344 ).strip(), 345 chunk_seconds=chunk_seconds, 346 chunk_overlap_seconds=chunk_overlap_seconds, 347 chunk_retry_limit=max(0, _env_int("FILE_CHUNK_RETRY_LIMIT", 2)), 348 text_soft_break_s=max(0.1, _env_float("FILE_TEXT_SOFT_BREAK_S", 0.8)), 349 text_hard_break_s=max(0.2, _env_float("FILE_TEXT_HARD_BREAK_S", 1.8)), 350 diarization_health_url=diarization_health_url, 351 diarization_health_host=diarization_health_host, 352 diarization_health_timeout_s=max(1.0, _env_float("DIARIZATION_HEALTH_TIMEOUT_S", 5.0)), 353 diarization_audio_format=diarization_audio_format, 354 diarization_wav_max_upload_mb=max( 355 1, 356 _env_int("DIARIZATION_WAV_MAX_UPLOAD_MB", DIARIZATION_WAV_MAX_UPLOAD_MB_DEFAULT), 357 ), 358 diarization_audio_max_upload_mb=max( 359 1, 360 _env_int("DIARIZATION_AUDIO_MAX_UPLOAD_MB", DIARIZATION_AUDIO_MAX_UPLOAD_MB_DEFAULT), 361 ), 362 diarization_extra_fields=_env_json_string_map("DIARIZATION_EXTRA_FIELDS_JSON"), 363 file_job_lock_url=str(os.getenv("FILE_JOB_LOCK_URL", "") or "").strip().rstrip("/"), 364 file_job_lock_timeout_s=max(1.0, _env_float("FILE_JOB_LOCK_TIMEOUT_S", 5.0)), 365 file_job_lock_heartbeat_s=max(5.0, _env_float("FILE_JOB_LOCK_HEARTBEAT_S", 30.0)), 366 ) 367 368 369 @dataclass(slots=True) 370 class TimedWord: 371 text: str 372 start_s: float 373 end_s: float 374 375 376 @dataclass(slots=True) 377 class RenderedTimedWord: 378 text: str 379 start_s: float 380 end_s: float 381 compare_key: str 382 piece_index: int 383 timeline_index: int | None = None 384 385 386 @dataclass(slots=True) 387 class MediaChunk: 388 index: int 389 total: int 390 start_s: float 391 end_s: float 392 keep_from_s: float 393 path: Path 394 395 396 class MissingWordTimestampsError(RuntimeError): 397 def __init__(self, message: str, *, debug: dict[str, Any]) -> None: 398 super().__init__(message) 399 self.debug = dict(debug) 400 401 402 def _missing_word_timestamps_debug(*, payload: dict[str, Any], chunk: MediaChunk) -> dict[str, Any]: 403 raw_debug = payload.get("_response_debug") 404 debug: dict[str, Any] = { 405 "chunk_index": chunk.index, 406 "chunk_total": chunk.total, 407 "chunk_start_s": round(chunk.start_s, 2), 408 "chunk_end_s": round(chunk.end_s, 2), 409 "keep_from_s": round(chunk.keep_from_s, 2), 410 } 411 if isinstance(raw_debug, dict): 412 debug.update(raw_debug) 413 elif "_response_debug" in payload: 414 debug["response_debug_type"] = type(raw_debug).__name__ 415 words = payload.get("words") 416 if isinstance(words, list): 417 debug["words_count"] = len(words) 418 text = _normalize_line_endings(str(payload.get("text") or "")) 419 if text: 420 debug["text_preview"] = _compact_error_text(text) 421 return debug 422 423 424 class TimedTranscriptBuilder: 425 _BOUNDARY_LOOKBACK_WORDS = 12 426 _BOUNDARY_MATCH_TOLERANCE_S = 0.05 427 428 def __init__( 429 self, 430 *, 431 soft_break_s: float, 432 hard_break_s: float, 433 require_word_timestamps: bool = False, 434 ) -> None: 435 self._soft_break_s = max(0.1, float(soft_break_s)) 436 self._hard_break_s = max(self._soft_break_s, float(hard_break_s)) 437 self._require_word_timestamps = bool(require_word_timestamps) 438 self._pieces: list[str] = [] 439 self._last_end_s: float | None = None 440 self._last_text = "" 441 self._current_line_len = 0 442 self._recent_words: list[RenderedTimedWord] = [] 443 self._timeline_words: list[TimedWord] = [] 444 self._force_continuation_separator = False 445 446 def append_chunk(self, *, payload: dict[str, Any], chunk: MediaChunk) -> str: 447 words = _extract_timed_words(payload) 448 if words: 449 appended = self._append_words( 450 words=[ 451 TimedWord( 452 text=word.text, 453 start_s=chunk.start_s + word.start_s, 454 end_s=chunk.start_s + word.end_s, 455 ) 456 for word in words 457 if ((chunk.start_s + word.start_s + chunk.start_s + word.end_s) / 2.0) 458 >= (chunk.keep_from_s - self._BOUNDARY_MATCH_TOLERANCE_S) 459 ], 460 overlap_start_s=chunk.start_s, 461 ) 462 return appended 463 464 text = _normalize_line_endings(str(payload.get("text") or "")) 465 if not text: 466 return "" 467 if self._require_word_timestamps or chunk.index > 1: 468 raise MissingWordTimestampsError( 469 "LinTO chunk response is missing word timestamps", 470 debug=_missing_word_timestamps_debug(payload=payload, chunk=chunk), 471 ) 472 return self._append_raw_text(text, start_s=chunk.start_s, end_s=chunk.end_s) 473 474 def _append_raw_text(self, text: str, *, start_s: float, end_s: float) -> str: 475 separator = self._separator(text=text, start_s=start_s) 476 piece = f"{separator}{text}" 477 self._append_piece(piece=piece, text=text, start_s=start_s, end_s=end_s, timeline_index=None) 478 return piece 479 480 def _append_words(self, *, words: list[TimedWord], overlap_start_s: float | None = None) -> str: 481 overlap_len = self._find_overlap_prefix_length(words=words, overlap_start_s=overlap_start_s) 482 if overlap_len > 0: 483 self._heal_previous_boundary_after_overlap() 484 words = words[overlap_len:] 485 if words: 486 self._force_continuation_separator = True 487 delta: list[str] = [] 488 for word in words: 489 text = str(word.text or "").strip() 490 if not text: 491 continue 492 separator = self._separator(text=text, start_s=word.start_s) 493 piece = f"{separator}{text}" 494 timeline_index = len(self._timeline_words) 495 self._timeline_words.append(TimedWord(text=text, start_s=word.start_s, end_s=word.end_s)) 496 self._append_piece( 497 piece=piece, 498 text=text, 499 start_s=word.start_s, 500 end_s=word.end_s, 501 timeline_index=timeline_index, 502 ) 503 delta.append(piece) 504 return "".join(delta) 505 506 def _append_piece( 507 self, 508 *, 509 piece: str, 510 text: str, 511 start_s: float, 512 end_s: float, 513 timeline_index: int | None, 514 ) -> None: 515 self._pieces.append(piece) 516 piece_index = len(self._pieces) - 1 517 self._last_end_s = end_s 518 self._last_text = text 519 if "\n" in piece: 520 self._current_line_len = len(piece.rsplit("\n", 1)[-1]) 521 else: 522 self._current_line_len += len(piece) 523 compare_key = self._normalize_boundary_token(text) 524 if compare_key: 525 self._recent_words.append( 526 RenderedTimedWord( 527 text=text, 528 start_s=start_s, 529 end_s=end_s, 530 compare_key=compare_key, 531 piece_index=piece_index, 532 timeline_index=timeline_index, 533 ) 534 ) 535 if len(self._recent_words) > (self._BOUNDARY_LOOKBACK_WORDS * 2): 536 self._recent_words = self._recent_words[-(self._BOUNDARY_LOOKBACK_WORDS * 2) :] 537 538 def _separator(self, *, text: str, start_s: float) -> str: 539 if not self._pieces: 540 return "" 541 if text.startswith(PUNCT_NO_SPACE_BEFORE): 542 self._force_continuation_separator = False 543 return "" 544 if self._last_text.endswith(PUNCT_NO_SPACE_AFTER): 545 self._force_continuation_separator = False 546 return "" 547 if self._force_continuation_separator: 548 self._force_continuation_separator = False 549 return " " 550 last_end_s = self._last_end_s if self._last_end_s is not None else start_s 551 gap_s = max(0.0, float(start_s) - float(last_end_s)) 552 if gap_s >= self._hard_break_s: 553 return "\n\n" 554 if gap_s >= self._soft_break_s: 555 return "\n" 556 if self._should_break_after_sentence(): 557 return "\n" 558 return " " 559 560 def _should_break_after_sentence(self) -> bool: 561 return self._current_line_len >= 140 and self._last_text.endswith(PUNCT_PREFERRED_BREAK_AFTER) 562 563 def _recent_boundary_words(self, *, overlap_start_s: float | None) -> list[RenderedTimedWord]: 564 if overlap_start_s is None: 565 return self._recent_words[-self._BOUNDARY_LOOKBACK_WORDS :] 566 return [ 567 word 568 for word in self._recent_words[-(self._BOUNDARY_LOOKBACK_WORDS * 2) :] 569 if ((word.start_s + word.end_s) / 2.0) >= (float(overlap_start_s) - self._BOUNDARY_MATCH_TOLERANCE_S) 570 ][-self._BOUNDARY_LOOKBACK_WORDS :] 571 572 def _find_overlap_prefix_length(self, *, words: list[TimedWord], overlap_start_s: float | None) -> int: 573 if not words or not self._recent_words: 574 return 0 575 recent_words = self._recent_boundary_words(overlap_start_s=overlap_start_s) 576 if not recent_words: 577 return 0 578 recent_keys = [word.compare_key for word in recent_words if word.compare_key] 579 incoming_keys = [ 580 compare_key 581 for compare_key in (self._normalize_boundary_token(word.text) for word in words[: self._BOUNDARY_LOOKBACK_WORDS]) 582 if compare_key 583 ] 584 if not recent_keys or not incoming_keys: 585 return 0 586 for overlap_len in range(min(len(recent_keys), len(incoming_keys)), 0, -1): 587 if recent_keys[-overlap_len:] == incoming_keys[:overlap_len]: 588 return overlap_len 589 return 0 590 591 def _heal_previous_boundary_after_overlap(self) -> None: 592 if not self._recent_words: 593 return 594 last_word = self._recent_words[-1] 595 stripped_text = re.sub(r"(?:\.{3}|…)+$", "", last_word.text).rstrip() 596 if not stripped_text or stripped_text == last_word.text: 597 return 598 self._replace_last_word_text(piece_index=last_word.piece_index, old_text=last_word.text, new_text=stripped_text) 599 last_word.text = stripped_text 600 last_word.compare_key = self._normalize_boundary_token(stripped_text) 601 if last_word.timeline_index is not None and last_word.timeline_index < len(self._timeline_words): 602 self._timeline_words[last_word.timeline_index].text = stripped_text 603 604 def _replace_last_word_text(self, *, piece_index: int, old_text: str, new_text: str) -> None: 605 if piece_index < 0 or piece_index >= len(self._pieces): 606 return 607 old_piece = self._pieces[piece_index] 608 if old_text and not old_piece.endswith(old_text): 609 return 610 new_piece = f"{old_piece[:-len(old_text)]}{new_text}" if old_text else old_piece 611 self._pieces[piece_index] = new_piece 612 self._last_text = new_text 613 self._current_line_len = max(0, self._current_line_len - (len(old_piece) - len(new_piece))) 614 615 @staticmethod 616 def _normalize_boundary_token(text: str) -> str: 617 normalized = normalize_for_compare(text).casefold() 618 normalized = re.sub(r"(^[\W_]+|[\W_]+$)", "", normalized, flags=re.UNICODE) 619 return normalized.strip() 620 621 def render(self) -> str: 622 return "".join(self._pieces).strip() 623 624 def timed_words(self) -> list[TimedWord]: 625 return [ 626 TimedWord(text=word.text, start_s=word.start_s, end_s=word.end_s) 627 for word in self._timeline_words 628 ] 629 630 631 class FileJobStore: 632 _DURABLE_STATE_KEYS = ( 633 "job_id", 634 "session_id", 635 "file_name", 636 "language", 637 "backend", 638 "page_mode", 639 "requested_diarization", 640 "run_diarization", 641 "syntax_postprocess", 642 "diarization_backend_status", 643 "diarization_fallback_reason", 644 "bench_config", 645 "diarization_audio_format", 646 ) 647 648 def __init__(self, root_dir: Path) -> None: 649 self._root_dir = Path(root_dir).resolve() 650 self._root_dir.mkdir(parents=True, exist_ok=True) 651 652 def job_dir(self, job_id: str) -> Path: 653 return self._root_dir / str(job_id).strip() 654 655 def ensure_job_dir(self, job_id: str) -> Path: 656 path = self.job_dir(job_id) 657 path.mkdir(parents=True, exist_ok=True) 658 return path 659 660 def source_path(self, job_id: str, file_name: str) -> Path: 661 job_dir = self.ensure_job_dir(job_id) 662 suffix = _normalize_suffix(file_name) or ".bin" 663 return job_dir / f"input{suffix}" 664 665 def normalized_audio_path(self, job_id: str) -> Path: 666 return self.ensure_job_dir(job_id) / "normalized.wav" 667 668 def diarization_audio_path(self, job_id: str) -> Path: 669 return self.ensure_job_dir(job_id) / "diarization_input.m4a" 670 671 def chunks_dir(self, job_id: str) -> Path: 672 path = self.ensure_job_dir(job_id) / "chunks" 673 path.mkdir(parents=True, exist_ok=True) 674 return path 675 676 def state_path(self, job_id: str) -> Path: 677 return self.ensure_job_dir(job_id) / "state.json" 678 679 def transcript_path(self, job_id: str) -> Path: 680 return self.ensure_job_dir(job_id) / "transcript.txt" 681 682 def transcription_payload_path(self, job_id: str) -> Path: 683 return self.ensure_job_dir(job_id) / "transcription.json" 684 685 def diarization_state_path(self, job_id: str) -> Path: 686 return self.ensure_job_dir(job_id) / "diarization_state.json" 687 688 def diarization_payload_path(self, job_id: str) -> Path: 689 return self.ensure_job_dir(job_id) / "diarization.json" 690 691 def diarization_text_path(self, job_id: str) -> Path: 692 return self.ensure_job_dir(job_id) / "diarization_text.json" 693 694 def combined_payload_path(self, job_id: str) -> Path: 695 return self.ensure_job_dir(job_id) / "combined.json" 696 697 def cancel_path(self, job_id: str) -> Path: 698 return self.ensure_job_dir(job_id) / "cancel.requested" 699 700 def merged_state_payload(self, job_id: str, payload: dict[str, Any]) -> dict[str, Any]: 701 stored_payload = dict(payload) 702 existing_payload = self.read_state(job_id) or {} 703 for key in self._DURABLE_STATE_KEYS: 704 if key not in stored_payload and key in existing_payload: 705 stored_payload[key] = existing_payload[key] 706 return stored_payload 707 708 def write_state(self, job_id: str, payload: dict[str, Any]) -> dict[str, Any]: 709 stored_payload = self.merged_state_payload(job_id, payload) 710 self.state_path(job_id).write_text( 711 json.dumps(stored_payload, ensure_ascii=False, indent=2), 712 encoding="utf-8", 713 ) 714 return stored_payload 715 716 def read_state(self, job_id: str) -> dict[str, Any] | None: 717 path = self.state_path(job_id) 718 if not path.exists(): 719 return None 720 try: 721 return json.loads(path.read_text(encoding="utf-8")) 722 except Exception: 723 return None 724 725 def write_transcript(self, job_id: str, text: str) -> None: 726 self.transcript_path(job_id).write_text(str(text or ""), encoding="utf-8") 727 728 def read_transcript(self, job_id: str) -> str: 729 path = self.transcript_path(job_id) 730 if not path.exists(): 731 return "" 732 try: 733 return path.read_text(encoding="utf-8") 734 except Exception: 735 return "" 736 737 def write_transcription_payload(self, job_id: str, payload: dict[str, Any]) -> None: 738 self.transcription_payload_path(job_id).write_text( 739 json.dumps(payload, ensure_ascii=False, indent=2), 740 encoding="utf-8", 741 ) 742 743 def read_transcription_payload(self, job_id: str) -> dict[str, Any] | None: 744 path = self.transcription_payload_path(job_id) 745 if not path.exists(): 746 return None 747 try: 748 return json.loads(path.read_text(encoding="utf-8")) 749 except Exception: 750 return None 751 752 def write_diarization_state(self, job_id: str, payload: dict[str, Any]) -> None: 753 self.diarization_state_path(job_id).write_text( 754 json.dumps(payload, ensure_ascii=False, indent=2), 755 encoding="utf-8", 756 ) 757 758 def read_diarization_state(self, job_id: str) -> dict[str, Any] | None: 759 path = self.diarization_state_path(job_id) 760 if not path.exists(): 761 return None 762 try: 763 return json.loads(path.read_text(encoding="utf-8")) 764 except Exception: 765 return None 766 767 def write_diarization_payload(self, job_id: str, payload: dict[str, Any]) -> None: 768 self.diarization_payload_path(job_id).write_text( 769 json.dumps(payload, ensure_ascii=False, indent=2), 770 encoding="utf-8", 771 ) 772 773 def request_cancel(self, job_id: str) -> None: 774 self.cancel_path(job_id).touch() 775 776 def clear_cancel_request(self, job_id: str) -> None: 777 self.cancel_path(job_id).unlink(missing_ok=True) 778 779 def is_cancel_requested(self, job_id: str) -> bool: 780 return self.cancel_path(job_id).exists() 781 782 def read_diarization_payload(self, job_id: str) -> dict[str, Any] | None: 783 path = self.diarization_payload_path(job_id) 784 if not path.exists(): 785 return None 786 try: 787 return json.loads(path.read_text(encoding="utf-8")) 788 except Exception: 789 return None 790 791 def write_diarization_text(self, job_id: str, payload: dict[str, Any]) -> None: 792 self.diarization_text_path(job_id).write_text( 793 json.dumps(payload, ensure_ascii=False, indent=2), 794 encoding="utf-8", 795 ) 796 797 def read_diarization_text(self, job_id: str) -> dict[str, Any] | None: 798 path = self.diarization_text_path(job_id) 799 if not path.exists(): 800 return None 801 try: 802 return json.loads(path.read_text(encoding="utf-8")) 803 except Exception: 804 return None 805 806 def write_combined_payload(self, job_id: str, payload: dict[str, Any]) -> None: 807 self.combined_payload_path(job_id).write_text( 808 json.dumps(payload, ensure_ascii=False, indent=2), 809 encoding="utf-8", 810 ) 811 812 def read_combined_payload(self, job_id: str) -> dict[str, Any] | None: 813 path = self.combined_payload_path(job_id) 814 if not path.exists(): 815 return None 816 try: 817 return json.loads(path.read_text(encoding="utf-8")) 818 except Exception: 819 return None 820 821 def list_job_ids(self) -> list[str]: 822 if not self._root_dir.exists(): 823 return [] 824 job_ids: list[str] = [] 825 for path in self._root_dir.iterdir(): 826 if not path.is_dir(): 827 continue 828 job_id = path.name.strip() 829 if job_id: 830 job_ids.append(job_id) 831 return sorted(job_ids) 832 833 def delete_job(self, job_id: str) -> None: 834 path = self.job_dir(job_id) 835 if not path.exists(): 836 return 837 shutil.rmtree(path, ignore_errors=True) 838 839 def probe_wav_duration_s(file_path: Path) -> float: 840 path = Path(file_path).resolve() 841 with wave.open(str(path), "rb") as wav_file: 842 frame_rate = wav_file.getframerate() or TARGET_SAMPLE_RATE 843 frame_count = wav_file.getnframes() 844 return max(0.0, float(frame_count) / float(frame_rate)) 845 846 847 def write_normalized_audio_wav( 848 file_path: Path, 849 *, 850 output_path: Path, 851 sample_rate: int, 852 chunk_ms: int = 1000, 853 ) -> Path: 854 target = Path(output_path).resolve() 855 target.parent.mkdir(parents=True, exist_ok=True) 856 with wave.open(str(target), "wb") as wav_file: 857 wav_file.setnchannels(1) 858 wav_file.setsampwidth(2) 859 wav_file.setframerate(sample_rate) 860 for pcm_bytes in iter_pcm_s16le_chunks(file_path, sample_rate=sample_rate, chunk_ms=chunk_ms): 861 wav_file.writeframes(pcm_bytes) 862 return target 863 864 865 @dataclass(slots=True) 866 class DiarizationAudioArtifact: 867 path: Path 868 requested_audio_format: str 869 effective_audio_format: str 870 artifact_strategy: str 871 artifact_size_bytes: int 872 wav_guard_fallback: bool = False 873 source_copy_guard_fallback: bool = False 874 875 @property 876 def artifact_size_mb(self) -> float: 877 return round(float(self.artifact_size_bytes) / 1024.0 / 1024.0, 3) 878 879 def as_payload(self) -> dict[str, Any]: 880 return { 881 "path_name": self.path.name, 882 "requested_audio_format": self.requested_audio_format, 883 "effective_audio_format": self.effective_audio_format, 884 "artifact_strategy": self.artifact_strategy, 885 "artifact_size_bytes": self.artifact_size_bytes, 886 "artifact_size_mb": self.artifact_size_mb, 887 "wav_guard_fallback": self.wav_guard_fallback, 888 "source_copy_guard_fallback": self.source_copy_guard_fallback, 889 } 890 891 892 def _ensure_nonempty_artifact(path: Path, *, error_prefix: str) -> None: 893 if not path.exists() or path.stat().st_size <= 0: 894 raise RuntimeError(f"{error_prefix}: output file is empty") 895 896 897 def _audio_codec_name(file_path: Path) -> str: 898 command = [ 899 "ffprobe", 900 "-v", 901 "error", 902 "-select_streams", 903 "a:0", 904 "-show_entries", 905 "stream=codec_name", 906 "-of", 907 "json", 908 str(Path(file_path).resolve()), 909 ] 910 try: 911 result = subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) 912 except (FileNotFoundError, subprocess.CalledProcessError): 913 return "" 914 try: 915 payload = json.loads(result.stdout.decode("utf-8", errors="replace") or "{}") 916 except json.JSONDecodeError: 917 return "" 918 streams = payload.get("streams") if isinstance(payload, dict) else None 919 if not isinstance(streams, list) or not streams: 920 return "" 921 first = streams[0] 922 if not isinstance(first, dict): 923 return "" 924 return str(first.get("codec_name") or "").strip().lower() 925 926 927 def source_supports_diarization_audio_copy(file_path: Path) -> bool: 928 suffix = Path(file_path).suffix.lower() 929 return suffix in {".m4a", ".mp4"} and _audio_codec_name(file_path) == "aac" 930 931 932 def write_diarization_audio_copy_artifact( 933 file_path: Path, 934 *, 935 output_path: Path, 936 ) -> Path: 937 source = Path(file_path).resolve() 938 target = Path(output_path).resolve() 939 target.parent.mkdir(parents=True, exist_ok=True) 940 command = [ 941 "ffmpeg", 942 "-nostdin", 943 "-hide_banner", 944 "-loglevel", 945 "error", 946 "-y", 947 "-i", 948 str(source), 949 "-vn", 950 "-sn", 951 "-dn", 952 "-map", 953 "0:a:0", 954 "-c:a", 955 "copy", 956 str(target), 957 ] 958 try: 959 subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) 960 except FileNotFoundError as exc: 961 raise RuntimeError("ffmpeg is required to copy the diarization audio track") from exc 962 except subprocess.CalledProcessError as exc: 963 details = exc.stderr.decode("utf-8", errors="replace").strip() or str(exc) 964 raise RuntimeError(f"failed to copy audio-only track for speaker separation: {details}") from exc 965 _ensure_nonempty_artifact( 966 target, 967 error_prefix="failed to copy audio-only track for speaker separation", 968 ) 969 return target 970 971 972 def write_diarization_audio_artifact( 973 file_path: Path, 974 *, 975 output_path: Path, 976 ) -> Path: 977 source = Path(file_path).resolve() 978 target = Path(output_path).resolve() 979 target.parent.mkdir(parents=True, exist_ok=True) 980 command = [ 981 "ffmpeg", 982 "-nostdin", 983 "-hide_banner", 984 "-loglevel", 985 "error", 986 "-y", 987 "-i", 988 str(source), 989 "-vn", 990 "-sn", 991 "-dn", 992 "-c:a", 993 "aac", 994 "-b:a", 995 "64k", 996 str(target), 997 ] 998 try: 999 subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) 1000 except FileNotFoundError as exc: 1001 raise RuntimeError("ffmpeg is required to prepare the diarization audio track") from exc 1002 except subprocess.CalledProcessError as exc: 1003 details = exc.stderr.decode("utf-8", errors="replace").strip() or str(exc) 1004 raise RuntimeError(f"failed to prepare audio-only track for speaker separation: {details}") from exc 1005 _ensure_nonempty_artifact( 1006 target, 1007 error_prefix="failed to prepare audio-only track for speaker separation", 1008 ) 1009 return target 1010 1011 1012 def prepare_diarization_audio_artifact( 1013 *, 1014 source_path: Path, 1015 normalized_path: Path, 1016 diarization_audio_path: Path, 1017 requested_audio_format: object, 1018 wav_max_upload_mb: int = DIARIZATION_WAV_MAX_UPLOAD_MB_DEFAULT, 1019 audio_max_upload_mb: int = DIARIZATION_AUDIO_MAX_UPLOAD_MB_DEFAULT, 1020 ) -> DiarizationAudioArtifact: 1021 requested = normalize_diarization_audio_format(requested_audio_format) 1022 normalized = Path(normalized_path).resolve() 1023 target = Path(diarization_audio_path).resolve() 1024 wav_limit_bytes = max(1, int(wav_max_upload_mb)) * 1024 * 1024 1025 audio_limit_bytes = max(1, int(audio_max_upload_mb)) * 1024 * 1024 1026 1027 if requested == "wav" and normalized.exists() and normalized.stat().st_size <= wav_limit_bytes: 1028 target.unlink(missing_ok=True) 1029 return DiarizationAudioArtifact( 1030 path=normalized, 1031 requested_audio_format=requested, 1032 effective_audio_format="wav", 1033 artifact_strategy="wav", 1034 artifact_size_bytes=int(normalized.stat().st_size), 1035 wav_guard_fallback=False, 1036 ) 1037 1038 wav_guard_fallback = requested == "wav" 1039 if requested in {"auto", "wav"} and source_supports_diarization_audio_copy(source_path): 1040 write_diarization_audio_copy_artifact(source_path, output_path=target) 1041 artifact_size = int(target.stat().st_size) 1042 if artifact_size <= audio_limit_bytes: 1043 return DiarizationAudioArtifact( 1044 path=target, 1045 requested_audio_format=requested, 1046 effective_audio_format="aac", 1047 artifact_strategy="source_aac_copy", 1048 artifact_size_bytes=artifact_size, 1049 wav_guard_fallback=wav_guard_fallback, 1050 ) 1051 logger.info( 1052 "diarization source AAC copy is %.3f MB, above %.3f MB upload guard; falling back to compact AAC transcode", 1053 artifact_size / 1024.0 / 1024.0, 1054 audio_limit_bytes / 1024.0 / 1024.0, 1055 ) 1056 write_diarization_audio_artifact(normalized, output_path=target) 1057 return DiarizationAudioArtifact( 1058 path=target, 1059 requested_audio_format=requested, 1060 effective_audio_format="aac", 1061 artifact_strategy="aac_transcode", 1062 artifact_size_bytes=int(target.stat().st_size), 1063 wav_guard_fallback=wav_guard_fallback, 1064 source_copy_guard_fallback=True, 1065 ) 1066 1067 write_diarization_audio_artifact(normalized, output_path=target) 1068 return DiarizationAudioArtifact( 1069 path=target, 1070 requested_audio_format=requested, 1071 effective_audio_format="aac", 1072 artifact_strategy="aac_transcode", 1073 artifact_size_bytes=int(target.stat().st_size), 1074 wav_guard_fallback=wav_guard_fallback, 1075 ) 1076 1077 1078 def estimate_chunk_total(*, duration_s: float, chunk_seconds: int, overlap_seconds: int) -> int: 1079 duration_s = max(0.0, float(duration_s)) 1080 chunk_seconds = max(1, int(chunk_seconds)) 1081 overlap_seconds = max(0, min(int(overlap_seconds), chunk_seconds - 1)) 1082 if duration_s <= chunk_seconds: 1083 return 1 1084 step_s = max(1, chunk_seconds - overlap_seconds) 1085 return int(math.ceil((duration_s - chunk_seconds) / float(step_s))) + 1 1086 1087 1088 def iter_normalized_wav_chunk_files( 1089 file_path: Path, 1090 *, 1091 output_dir: Path, 1092 sample_rate: int, 1093 chunk_seconds: int, 1094 overlap_seconds: int, 1095 total_chunks: int, 1096 ): 1097 path = Path(file_path).resolve() 1098 if not path.exists(): 1099 raise FileNotFoundError(path) 1100 1101 output_dir = Path(output_dir).resolve() 1102 output_dir.mkdir(parents=True, exist_ok=True) 1103 chunk_seconds = max(1, int(chunk_seconds)) 1104 overlap_seconds = max(0, min(int(overlap_seconds), chunk_seconds - 1)) 1105 sample_rate = max(1, int(sample_rate)) 1106 1107 window_frames = chunk_seconds * sample_rate 1108 overlap_frames = overlap_seconds * sample_rate 1109 step_frames = max(sample_rate, window_frames - overlap_frames) 1110 emitted = 0 1111 1112 with wave.open(str(path), "rb") as wav_file: 1113 channels = wav_file.getnchannels() 1114 sample_width = wav_file.getsampwidth() 1115 frame_rate = wav_file.getframerate() 1116 frame_count = wav_file.getnframes() 1117 if channels != 1 or sample_width != 2 or frame_rate != sample_rate: 1118 raise ValueError( 1119 f"normalized wav must be mono s16 {sample_rate}Hz; got channels={channels} sample_width={sample_width} frame_rate={frame_rate}" 1120 ) 1121 1122 while True: 1123 start_frame = emitted * step_frames 1124 if start_frame >= frame_count: 1125 break 1126 1127 frames_to_read = min(window_frames, frame_count - start_frame) 1128 if frames_to_read <= 0: 1129 break 1130 1131 wav_file.setpos(start_frame) 1132 chunk_bytes = wav_file.readframes(frames_to_read) 1133 if not chunk_bytes: 1134 break 1135 1136 chunk_path = _write_chunk_wav(output_dir, emitted, chunk_bytes, sample_rate) 1137 start_s = float(start_frame) / float(sample_rate) 1138 end_s = start_s + (frames_to_read / float(sample_rate)) 1139 keep_from_s = 0.0 if emitted == 0 else start_s + overlap_seconds 1140 emitted += 1 1141 yield MediaChunk( 1142 index=emitted, 1143 total=max(total_chunks, emitted), 1144 start_s=start_s, 1145 end_s=end_s, 1146 keep_from_s=keep_from_s, 1147 path=chunk_path, 1148 ) 1149 1150 async def transcribe_http_json( 1151 *, 1152 settings: FileModeSettings, 1153 file_path: Path, 1154 file_name: str, 1155 language: str | None, 1156 http: "aiohttp.ClientSession | None" = None, 1157 cancel_event: asyncio.Event | None = None, 1158 ) -> dict[str, Any]: 1159 status, body, content_type = await _post_to_linto( 1160 settings=settings, 1161 file_path=file_path, 1162 file_name=file_name, 1163 language=language, 1164 accept="application/json", 1165 http=http, 1166 cancel_event=cancel_event, 1167 ) 1168 parsed = _parse_transcribe_response(body=body, content_type=content_type) 1169 payload = dict(parsed.payload) 1170 payload["_response_debug"] = _transcribe_response_debug_info(parsed) 1171 if status >= 400: 1172 details = payload.get("text") or parsed.raw_text 1173 raise RuntimeError(f"LinTO HTTP error {status}: {_compact_error_text(str(details))}") 1174 return payload 1175 1176 1177 async def transcribe_http_text( 1178 *, 1179 settings: FileModeSettings, 1180 file_path: Path, 1181 file_name: str, 1182 language: str | None, 1183 http: "aiohttp.ClientSession | None" = None, 1184 cancel_event: asyncio.Event | None = None, 1185 ) -> str: 1186 status, body, content_type = await _post_to_linto( 1187 settings=settings, 1188 file_path=file_path, 1189 file_name=file_name, 1190 language=language, 1191 accept="text/plain", 1192 http=http, 1193 cancel_event=cancel_event, 1194 ) 1195 text = extract_transcribe_text(body=body, content_type=content_type) 1196 if status >= 400: 1197 details = text or body.decode("utf-8", errors="replace") 1198 raise RuntimeError(f"LinTO HTTP error {status}: {_compact_error_text(details)}") 1199 return text 1200 1201 1202 async def _post_to_linto( 1203 *, 1204 settings: FileModeSettings, 1205 file_path: Path, 1206 file_name: str, 1207 language: str | None, 1208 accept: str, 1209 http: "aiohttp.ClientSession | None", 1210 cancel_event: asyncio.Event | None, 1211 ) -> tuple[int, bytes, str | None]: 1212 import aiohttp 1213 1214 from src.model_transport import model_http_session 1215 1216 target_url = str(settings.linto_http_url or "").strip() 1217 target_host = str(settings.linto_http_host or "").strip() 1218 target_language = str(settings.linto_http_language or "").strip() 1219 1220 if not target_url: 1221 raise RuntimeError("LINTO_HTTP_URL is required for HTTP file transcription mode") 1222 1223 headers: dict[str, str] = {"Accept": accept} 1224 if target_host: 1225 headers["Host"] = target_host 1226 1227 params: dict[str, str] = {} 1228 resolved_language = _normalize_line_endings(language or target_language) 1229 if resolved_language: 1230 params["language"] = resolved_language 1231 1232 timeout = aiohttp.ClientTimeout(total=None, connect=10.0) 1233 proxy_url = settings.linto_http_proxy_url.strip() 1234 async with model_http_session(proxy_url=proxy_url, http=http, timeout=timeout) as session: 1235 with Path(file_path).open("rb") as handle: 1236 form = aiohttp.FormData() 1237 form.add_field( 1238 "file", 1239 handle, 1240 filename=file_name, 1241 content_type=guess_media_content_type(file_name), 1242 ) 1243 async with session.post( 1244 target_url, 1245 params=params or None, 1246 data=form, 1247 headers=headers, 1248 allow_redirects=False, 1249 ) as response: 1250 if cancel_event is not None and cancel_event.is_set(): 1251 raise asyncio.CancelledError() 1252 body = await response.read() 1253 return response.status, body, response.headers.get("Content-Type") 1254 1255 1256 def _extract_timed_words(payload: dict[str, Any]) -> list[TimedWord]: 1257 raw_words = payload.get("words") 1258 if not isinstance(raw_words, list): 1259 return [] 1260 1261 words: list[TimedWord] = [] 1262 for item in raw_words: 1263 if not isinstance(item, dict): 1264 continue 1265 text = str(item.get("word") or item.get("text") or "").strip() 1266 if not text: 1267 continue 1268 try: 1269 start_s = float(item.get("start")) 1270 end_s = float(item.get("end")) 1271 except (TypeError, ValueError): 1272 continue 1273 if end_s < start_s: 1274 end_s = start_s 1275 words.append(TimedWord(text=text, start_s=start_s, end_s=end_s)) 1276 return words 1277 1278 1279 def _write_chunk_wav(output_dir: Path, zero_index: int, pcm_bytes: bytes, sample_rate: int) -> Path: 1280 chunk_path = output_dir / f"chunk_{zero_index + 1:04d}.wav" 1281 with wave.open(str(chunk_path), "wb") as wav_file: 1282 wav_file.setnchannels(1) 1283 wav_file.setsampwidth(2) 1284 wav_file.setframerate(sample_rate) 1285 wav_file.writeframes(pcm_bytes) 1286 return chunk_path 1287 1288 1289 1290 1291 def iter_pcm_s16le_chunks( 1292 file_path: Path, 1293 *, 1294 sample_rate: int, 1295 chunk_ms: int, 1296 ): 1297 path = Path(file_path).resolve() 1298 if not path.exists(): 1299 raise FileNotFoundError(path) 1300 1301 suffix = _normalize_suffix(path.name) 1302 if suffix and suffix not in SUPPORTED_FILE_SUFFIXES: 1303 raise ValueError(f"unsupported file extension: {suffix}") 1304 1305 try: 1306 import av 1307 except ImportError as exc: 1308 raise RuntimeError("PyAV is required for HTTP file transcription mode") from exc 1309 1310 samples_per_chunk = max(1, int(sample_rate * max(1, chunk_ms) / 1000)) 1311 bytes_per_chunk = samples_per_chunk * 2 1312 pending = bytearray() 1313 resampler = av.audio.resampler.AudioResampler(format="s16", layout="mono", rate=sample_rate) 1314 1315 with av.open(str(path), mode="r", metadata_errors="ignore") as container: 1316 if not any(stream.type == "audio" for stream in container.streams): 1317 raise ValueError(_missing_audio_track_message(path)) 1318 1319 for frame in _iter_audio_frames(container, av): 1320 for output_frame in resampler.resample(frame): 1321 pending.extend(output_frame.to_ndarray().tobytes()) 1322 while len(pending) >= bytes_per_chunk: 1323 yield bytes(pending[:bytes_per_chunk]) 1324 del pending[:bytes_per_chunk] 1325 1326 for output_frame in resampler.resample(None): 1327 pending.extend(output_frame.to_ndarray().tobytes()) 1328 while len(pending) >= bytes_per_chunk: 1329 yield bytes(pending[:bytes_per_chunk]) 1330 del pending[:bytes_per_chunk] 1331 1332 if pending: 1333 yield bytes(pending) 1334 1335 1336 def _iter_audio_frames(container, av_module): 1337 try: 1338 frames = container.decode(audio=0) 1339 except Exception as exc: 1340 raise ValueError(f"failed to decode audio: {exc}") from exc 1341 1342 iterator = iter(frames) 1343 while True: 1344 try: 1345 yield next(iterator) 1346 except StopIteration: 1347 return 1348 except Exception as exc: 1349 invalid_data_error = getattr(getattr(av_module, "error", None), "InvalidDataError", None) 1350 if invalid_data_error and isinstance(exc, invalid_data_error): 1351 continue 1352 raise ValueError(f"invalid audio frame: {exc}") from exc