Текстовый файл из учебного комплекта, не свидетельство установленного runtime. Вернуться к учебнику · Исходный файл
1 from __future__ import annotations 2 3 import asyncio 4 import json 5 import logging 6 import os 7 import signal 8 import time 9 import uuid 10 import wave 11 from dataclasses import dataclass, field 12 from pathlib import Path 13 from tempfile import NamedTemporaryFile 14 from typing import Any 15 16 import aiohttp 17 import nats 18 import numpy as np 19 from nats.aio.msg import Msg 20 21 from src.diarization import ( 22 attribute_words, 23 attributed_words_payload, 24 build_speaker_turns, 25 diarization_backend_ready, 26 diarization_segments_payload, 27 diarize_http_json, 28 parse_diarization_result, 29 render_speaker_text, 30 speaker_turns_payload, 31 ) 32 from src.diarization_syntax import apply_syntax_postprocess 33 from src.webui.file_mode import ( 34 FileJobStore, 35 FileModeSettings, 36 MissingWordTimestampsError, 37 TimedTranscriptBuilder, 38 estimate_chunk_total, 39 iter_normalized_wav_chunk_files, 40 load_file_mode_settings, 41 prepare_diarization_audio_artifact, 42 probe_wav_duration_s, 43 resolve_optional_bool, 44 transcribe_http_json, 45 transcribe_http_text, 46 write_normalized_audio_wav, 47 ) 48 from src.webui.job_lock import FileJobLockClient, job_lock_enabled, page_label 49 from src.utils.audio_utils import resample_audio 50 from src.utils.silero_onnx_vad import SileroOnnxVAD, download_model_file, get_speech_timestamps 51 from src.utils.wire_packet import parse_wire_packet 52 53 54 logger = logging.getLogger("stt_whisper_to_nats") 55 FILE_CHUNK_PROGRESS_INTERVAL_S = 5.0 56 FILE_JOB_TERMINAL_CLEANUP_DELAY_S = 30.0 57 DEFAULT_VAD_MODEL_URL = ( 58 "https://github.com/snakers4/silero-vad/raw/refs/heads/master/src/silero_vad/data/silero_vad.onnx" 59 ) 60 61 62 def _env(name: str, default: str = "") -> str: 63 return str(os.getenv(name, default) or "").strip() 64 65 66 def _env_bool(name: str, default: bool = False) -> bool: 67 raw = _env(name, "1" if default else "0").lower() 68 return raw in {"1", "true", "yes", "on"} 69 70 71 @dataclass(slots=True) 72 class PhraseStreamState: 73 out_subject: str 74 session_id: str | None 75 sample_rate: int 76 buffer: list[np.ndarray] 77 buffer_samples: int 78 check_in_progress: bool = False 79 flush_requested: bool = False 80 lock: asyncio.Lock = field(default_factory=asyncio.Lock, repr=False) 81 82 83 class SttWhisperToNatsService: 84 def __init__(self, settings: FileModeSettings) -> None: 85 self.service_name = _env("STACK_SERVICE_NAME", "stt_whisper_to_nats") 86 self.nats_url = _env("NATS_URL") 87 self.in_prefix = _env("ASR_IN_PREFIX") 88 self.out_prefix = _env("ASR_OUT_PREFIX") 89 self._settings = settings 90 self.min_phrase_ms = int(_env("MIN_PHRASE_MS", "100")) 91 self.vad_models_dir = Path(_env("VAD_MODELS", "/app/models/vad")) 92 self.vad_model_url = _env("VAD_MODEL_URL", DEFAULT_VAD_MODEL_URL) 93 self.vad_sample_rate = int(_env("VAD_SAMPLE_RATE", "16000")) 94 self.min_speech_duration_ms = int(_env("MIN_SPEECH_DURATION_MS", "250")) 95 self.min_silence_duration_ms = int(_env("MIN_SILENCE_DURATION_MS", "500")) 96 self.max_speech_duration_s = float(_env("MAX_SPEECH_DURATION_S", "30.0")) 97 self.speech_pad_ms = int(_env("SPEECH_PAD_MS", "30")) 98 self.vad_threshold = float(_env("VAD_THRESHOLD", "0.8")) 99 self.buffer_check_interval_s = float(_env("BUFFER_CHECK_INTERVAL_S", "1.0")) 100 self.max_concurrency = max(1, int(_env("WHISPER_MAX_CONCURRENCY", "1"))) 101 102 if not self.nats_url: 103 raise ValueError("NATS_URL is required") 104 if not self.in_prefix: 105 raise ValueError("ASR_IN_PREFIX is required") 106 if not self.out_prefix: 107 raise ValueError("ASR_OUT_PREFIX is required") 108 if not self.in_prefix.endswith("."): 109 raise ValueError("ASR_IN_PREFIX must end with '.'") 110 if not self.out_prefix.endswith("."): 111 self.out_prefix = f"{self.out_prefix}." 112 if not self._settings.linto_http_url: 113 raise ValueError("LINTO_HTTP_URL is required for live phrase transcription") 114 115 self.in_subscribe = f"{self.in_prefix}>" 116 self.nc: nats.NATS | None = None 117 self.http: aiohttp.ClientSession | None = None 118 self.stop_event = asyncio.Event() 119 self.tasks: set[asyncio.Task] = set() 120 self.streams: dict[str, PhraseStreamState] = {} 121 self.streams_lock = asyncio.Lock() 122 self._vad_model: SileroOnnxVAD | None = None 123 self._vad_model_lock = asyncio.Lock() 124 self._vad_run_lock = asyncio.Lock() 125 self._semaphore = asyncio.Semaphore(self.max_concurrency) 126 127 async def run(self) -> None: 128 timeout = aiohttp.ClientTimeout(total=None, connect=10.0) 129 self.http = aiohttp.ClientSession(timeout=timeout) 130 self.nc = await nats.connect( 131 servers=[self.nats_url], 132 name=self.service_name, 133 max_reconnect_attempts=-1, 134 reconnect_time_wait=2, 135 ping_interval=10, 136 ) 137 await self.nc.subscribe(self.in_subscribe, cb=self._handle_message) 138 logger.info("Connected to NATS %s", self.nats_url) 139 logger.info("Subscribed to %s", self.in_subscribe) 140 logger.info("LinTO HTTP target %s", self._settings.linto_http_url) 141 if self._settings.linto_http_host: 142 logger.info("LinTO HTTP host override %s", self._settings.linto_http_host) 143 144 await self.stop_event.wait() 145 await self._shutdown() 146 147 def stop(self) -> None: 148 self.stop_event.set() 149 150 async def _shutdown(self) -> None: 151 for task in list(self.tasks): 152 task.cancel() 153 154 async with self.streams_lock: 155 self.streams.clear() 156 157 if self.nc: 158 try: 159 await self.nc.drain() 160 except Exception: 161 await self.nc.close() 162 163 if self.http: 164 await self.http.close() 165 166 async def _handle_message(self, msg: Msg) -> None: 167 task = asyncio.create_task(self._process_message(msg)) 168 self.tasks.add(task) 169 task.add_done_callback(self.tasks.discard) 170 171 def _derive_out_subject(self, in_subject: str) -> str | None: 172 if not in_subject.startswith(self.in_prefix): 173 logger.error( 174 "Dropping message: subject does not match ASR_IN_PREFIX (subject=%s prefix=%s)", 175 in_subject, 176 self.in_prefix, 177 ) 178 return None 179 suffix = in_subject[len(self.in_prefix):] 180 if not suffix: 181 logger.error("Dropping message: empty token suffix in subject=%s", in_subject) 182 return None 183 return f"{self.out_prefix}{suffix}" 184 185 async def _process_message(self, msg: Msg) -> None: 186 out_subject = self._derive_out_subject(str(msg.subject or "")) 187 if not out_subject: 188 return 189 190 try: 191 meta, audio = parse_wire_packet(msg.data) 192 except Exception as exc: 193 logger.warning("Invalid packet on %s: %s", msg.subject, exc) 194 await self._publish_json(out_subject, {"error": "invalid_packet", "details": str(exc)}) 195 return 196 197 msg_type = str(meta.get("type") or "").strip() 198 if msg_type not in {"frame", "end"}: 199 await self._publish_json( 200 out_subject, 201 {"error": "unsupported_packet_type", "details": f"type={msg_type or '-'}"}, 202 ) 203 return 204 205 if msg_type == "frame": 206 await self._handle_frame(out_subject, meta, audio) 207 return 208 209 await self._handle_end(out_subject, meta) 210 211 async def _handle_frame(self, out_subject: str, meta: dict[str, Any], audio: np.ndarray) -> None: 212 session_id = _clean_optional_text(meta.get("session_id")) 213 sample_rate = int(meta.get("sample_rate") or self.vad_sample_rate) 214 state = self.streams.get(out_subject) 215 if not state: 216 async with self.streams_lock: 217 state = self.streams.get(out_subject) 218 if not state: 219 state = PhraseStreamState( 220 out_subject=out_subject, 221 session_id=session_id, 222 sample_rate=sample_rate, 223 buffer=[], 224 buffer_samples=0, 225 ) 226 self.streams[out_subject] = state 227 logger.info("Phrase stream opened: %s sample_rate=%s", out_subject, sample_rate) 228 229 schedule_check = False 230 async with state.lock: 231 state.session_id = state.session_id or session_id 232 state.sample_rate = sample_rate or state.sample_rate or self.vad_sample_rate 233 if audio.size: 234 state.buffer.append(audio) 235 state.buffer_samples += int(audio.size) 236 duration_s = ( 237 state.buffer_samples / max(1, state.sample_rate) 238 if state.buffer_samples 239 else 0.0 240 ) 241 schedule_check = duration_s >= self.buffer_check_interval_s and not state.check_in_progress 242 243 if schedule_check: 244 task = asyncio.create_task(self._check_stream_buffer(out_subject, flush=False)) 245 self.tasks.add(task) 246 task.add_done_callback(self.tasks.discard) 247 248 async def _handle_end(self, out_subject: str, meta: dict[str, Any]) -> None: 249 state = self.streams.get(out_subject) 250 if not state: 251 return 252 async with state.lock: 253 if not state.session_id: 254 state.session_id = _clean_optional_text(meta.get("session_id")) 255 state.flush_requested = True 256 schedule_check = not state.check_in_progress 257 if schedule_check: 258 task = asyncio.create_task(self._check_stream_buffer(out_subject, flush=True)) 259 self.tasks.add(task) 260 task.add_done_callback(self.tasks.discard) 261 262 async def _check_stream_buffer(self, out_subject: str, *, flush: bool) -> None: 263 state = self.streams.get(out_subject) 264 if not state: 265 return 266 267 async with state.lock: 268 if state.check_in_progress: 269 return 270 state.check_in_progress = True 271 effective_flush = bool(flush or state.flush_requested) 272 snapshot_chunks = list(state.buffer) 273 snapshot_samples = int(state.buffer_samples) 274 snapshot_sr = int(state.sample_rate or self.vad_sample_rate) 275 session_id = state.session_id 276 277 try: 278 if not snapshot_chunks: 279 return 280 281 audio = np.concatenate(snapshot_chunks) 282 if snapshot_sr != self.vad_sample_rate: 283 audio = resample_audio(audio, snapshot_sr, self.vad_sample_rate) 284 if not effective_flush and (len(audio) / self.vad_sample_rate) < 1.0: 285 return 286 287 timestamps = await self._get_timestamps(audio) 288 if not timestamps: 289 if (len(audio) / self.vad_sample_rate) > 5.0: 290 async with state.lock: 291 del state.buffer[: len(snapshot_chunks)] 292 state.buffer_samples = max(0, int(state.buffer_samples) - snapshot_samples) 293 return 294 295 should_flush = effective_flush 296 if not should_flush: 297 last_end_sample = int(timestamps[-1]["end"]) 298 silence_after = (len(audio) - last_end_sample) / self.vad_sample_rate 299 should_flush = silence_after >= (self.min_silence_duration_ms / 1000.0) 300 if not should_flush: 301 return 302 303 for ts in timestamps: 304 start_sample = int(ts["start"]) 305 end_sample = int(ts["end"]) 306 segment_audio = audio[start_sample:end_sample] 307 if not segment_audio.size: 308 continue 309 phrase_id = str(uuid.uuid4()) 310 task = asyncio.create_task( 311 self._transcribe_and_publish( 312 reply_subject=out_subject, 313 session_id=session_id, 314 phrase_id=phrase_id, 315 audio=segment_audio, 316 sample_rate=self.vad_sample_rate, 317 ) 318 ) 319 self.tasks.add(task) 320 task.add_done_callback(self.tasks.discard) 321 322 async with state.lock: 323 del state.buffer[: len(snapshot_chunks)] 324 state.buffer_samples = max(0, int(state.buffer_samples) - snapshot_samples) 325 finally: 326 needs_final_flush = False 327 async with state.lock: 328 state.check_in_progress = False 329 needs_final_flush = bool(state.flush_requested and not flush) 330 331 if effective_flush: 332 async with self.streams_lock: 333 self.streams.pop(out_subject, None) 334 logger.info("Phrase stream closed: %s", out_subject) 335 elif needs_final_flush: 336 task = asyncio.create_task(self._check_stream_buffer(out_subject, flush=True)) 337 self.tasks.add(task) 338 task.add_done_callback(self.tasks.discard) 339 340 async def _transcribe_and_publish( 341 self, 342 *, 343 reply_subject: str, 344 session_id: str | None, 345 phrase_id: str, 346 audio: np.ndarray, 347 sample_rate: int, 348 ) -> None: 349 if not self.http: 350 raise RuntimeError("HTTP session is not initialized") 351 352 duration_s = float(len(audio) / max(1, sample_rate)) 353 min_samples = int(sample_rate * self.min_phrase_ms / 1000) 354 if audio.size < min_samples: 355 audio = np.pad(audio, (0, min_samples - audio.size), mode="constant") 356 357 pending_payload: dict[str, Any] = {"pending": True, "phrase_id": phrase_id} 358 pending_payload.update(self._session_field(session_id)) 359 await self._publish_json(reply_subject, pending_payload) 360 361 wav_path: Path | None = None 362 try: 363 wav_path = await asyncio.to_thread(self._write_phrase_wav, phrase_id, audio, sample_rate) 364 async with self._semaphore: 365 text = await transcribe_http_text( 366 settings=self._settings, 367 file_path=wav_path, 368 file_name=f"{phrase_id}.wav", 369 language=self._settings.linto_http_language or None, 370 http=self.http, 371 ) 372 except Exception as exc: 373 logger.exception("Phrase transcription failed for %s: %s", reply_subject, exc) 374 await self._publish_json( 375 reply_subject, 376 { 377 "phrase_id": phrase_id, 378 "error": "transcription_failed", 379 "details": str(exc), 380 **self._session_field(session_id), 381 }, 382 ) 383 return 384 finally: 385 if wav_path is not None: 386 try: 387 wav_path.unlink(missing_ok=True) 388 except Exception: 389 pass 390 391 cleaned_text = str(text or "").strip() 392 if not cleaned_text: 393 logger.info("Ignoring empty phrase result for %s phrase_id=%s", reply_subject, phrase_id) 394 return 395 await self._publish_json( 396 reply_subject, 397 { 398 "phrase_id": phrase_id, 399 "text": cleaned_text, 400 "duration": duration_s, 401 **self._session_field(session_id), 402 }, 403 ) 404 405 async def _get_timestamps(self, audio_array: np.ndarray) -> list[dict[str, Any]]: 406 await self._ensure_vad_loaded() 407 if not self._vad_model: 408 return [] 409 410 async with self._vad_run_lock: 411 return await asyncio.to_thread( 412 get_speech_timestamps, 413 audio_array, 414 self._vad_model, 415 threshold=self.vad_threshold, 416 sampling_rate=self.vad_sample_rate, 417 min_speech_duration_ms=self.min_speech_duration_ms, 418 min_silence_duration_ms=self.min_silence_duration_ms, 419 max_speech_duration_s=self.max_speech_duration_s, 420 speech_pad_ms=self.speech_pad_ms, 421 return_seconds=False, 422 ) 423 424 async def _ensure_vad_loaded(self) -> None: 425 if self._vad_model: 426 return 427 async with self._vad_model_lock: 428 if self._vad_model: 429 return 430 431 filename = os.path.basename(self.vad_model_url) 432 base_path = self.vad_models_dir.expanduser() 433 model_path = base_path if base_path.suffix.lower() == ".onnx" else (base_path / filename) 434 model_path.parent.mkdir(parents=True, exist_ok=True) 435 if not model_path.exists(): 436 await asyncio.to_thread(download_model_file, str(model_path), self.vad_model_url) 437 self._vad_model = await asyncio.to_thread(SileroOnnxVAD, str(model_path)) 438 logger.info("Silero VAD model loaded from %s", model_path) 439 440 @staticmethod 441 def _write_phrase_wav(phrase_id: str, audio: np.ndarray, sample_rate: int) -> Path: 442 pcm = np.clip(audio, -1.0, 1.0) 443 pcm_i16 = (pcm * 32767.0).astype("<i2") 444 with NamedTemporaryFile(prefix=f"{phrase_id}-", suffix=".wav", delete=False) as handle: 445 wav_path = Path(handle.name) 446 with wave.open(str(wav_path), "wb") as wav_file: 447 wav_file.setnchannels(1) 448 wav_file.setsampwidth(2) 449 wav_file.setframerate(sample_rate) 450 wav_file.writeframes(pcm_i16.tobytes()) 451 return wav_path 452 453 @staticmethod 454 def _session_field(session_id: str | None) -> dict[str, str]: 455 if not session_id: 456 return {} 457 return {"session_id": session_id} 458 459 async def _publish_json(self, subject: str, payload: dict[str, Any]) -> None: 460 if not self.nc: 461 return 462 await self.nc.publish(subject, json.dumps(payload, ensure_ascii=False).encode("utf-8")) 463 464 465 def _clean_optional_text(value: object) -> str | None: 466 if not isinstance(value, str): 467 return None 468 cleaned = value.strip() 469 return cleaned or None 470 471 472 def _normalize_diarization_runtime_label(value: object) -> str: 473 raw = str(value or "").strip() 474 if not raw: 475 return "" 476 lowered = raw.lower() 477 if lowered in {"gpu", "cuda"}: 478 return "GPU" 479 if lowered == "cpu": 480 return "CPU" 481 return raw.upper() 482 483 484 def _diarization_wait_details(runtime_label: object) -> str: 485 label = _normalize_diarization_runtime_label(runtime_label) 486 if label: 487 return ( 488 f"Plain text ready. Speaker separation is still running on {label} " 489 "and can take several minutes for longer files." 490 ) 491 return "Plain text ready. Speaker separation is still running and can take several minutes for longer files." 492 493 494 def _diar_text_done_payload( 495 *, 496 job_id: str, 497 file_name: str, 498 duration_s: float, 499 speaker_count: int, 500 speaker_text: str, 501 speaker_turns: list[dict[str, Any]], 502 speaker_text_raw: str, 503 speaker_turns_raw: list[dict[str, Any]], 504 speaker_text_syntax: str, 505 speaker_turns_syntax: list[dict[str, Any]], 506 syntax_postprocess_enabled: bool, 507 syntax_postprocess_applied: bool, 508 syntax_sentence_count: int, 509 syntax_low_confidence_count: int, 510 session_id: str | None, 511 ) -> dict[str, Any]: 512 payload: dict[str, Any] = { 513 "type": "diar_text_done", 514 "job_id": job_id, 515 "file_name": file_name, 516 "phase": "done", 517 "duration_s": round(duration_s, 3), 518 "speaker_count": speaker_count, 519 "text": speaker_text, 520 "turns": speaker_turns, 521 "text_raw": speaker_text_raw, 522 "turns_raw": speaker_turns_raw, 523 "text_syntax": speaker_text_syntax, 524 "turns_syntax": speaker_turns_syntax, 525 "syntax_postprocess_enabled": bool(syntax_postprocess_enabled), 526 "syntax_postprocess_applied": bool(syntax_postprocess_applied), 527 "syntax_sentence_count": int(syntax_sentence_count), 528 "syntax_low_confidence_count": int(syntax_low_confidence_count), 529 } 530 if session_id: 531 payload["session_id"] = session_id 532 return payload 533 534 535 def _diar_done_signal_payload( 536 *, 537 job_id: str, 538 file_name: str, 539 duration_s: float, 540 model: str, 541 speaker_count: int, 542 speakers: list[str], 543 syntax_postprocess_enabled: bool, 544 syntax_postprocess_applied: bool, 545 syntax_sentence_count: int, 546 syntax_low_confidence_count: int, 547 session_id: str | None, 548 ) -> dict[str, Any]: 549 payload: dict[str, Any] = { 550 "type": "diar_done", 551 "job_id": job_id, 552 "file_name": file_name, 553 "phase": "done", 554 "duration_s": round(duration_s, 3), 555 "model": model, 556 "speaker_count": int(speaker_count), 557 "speakers": list(speakers), 558 "syntax_postprocess_enabled": bool(syntax_postprocess_enabled), 559 "syntax_postprocess_applied": bool(syntax_postprocess_applied), 560 "syntax_sentence_count": int(syntax_sentence_count), 561 "syntax_low_confidence_count": int(syntax_low_confidence_count), 562 } 563 if session_id: 564 payload["session_id"] = session_id 565 return payload 566 567 568 def _diar_text_done_signal_payload(payload: dict[str, Any]) -> dict[str, Any]: 569 signal = { 570 "type": "diar_text_done", 571 "job_id": payload.get("job_id"), 572 "file_name": payload.get("file_name"), 573 "phase": payload.get("phase") or "done", 574 "duration_s": payload.get("duration_s"), 575 "speaker_count": payload.get("speaker_count"), 576 "syntax_postprocess_enabled": payload.get("syntax_postprocess_enabled"), 577 "syntax_postprocess_applied": payload.get("syntax_postprocess_applied"), 578 "syntax_sentence_count": payload.get("syntax_sentence_count"), 579 "syntax_low_confidence_count": payload.get("syntax_low_confidence_count"), 580 } 581 if payload.get("session_id"): 582 signal["session_id"] = payload.get("session_id") 583 return {key: value for key, value in signal.items() if value is not None} 584 585 586 class FileJobWorkerService: 587 def __init__(self, settings: FileModeSettings) -> None: 588 self.service_name = _env("STACK_SERVICE_NAME", "stt_whisper_to_nats") 589 self.nats_url = _env("NATS_URL") 590 self._settings = settings 591 self._diarization_runtime_label = _normalize_diarization_runtime_label(_env("DIARIZATION_RUNTIME_LABEL")) 592 self._job_store = FileJobStore(settings.job_storage_dir) 593 self._http: aiohttp.ClientSession | None = None 594 self._nc: nats.NATS | None = None 595 self._active_task: asyncio.Task | None = None 596 self._running_job: str | None = None 597 self._running_session_id: str | None = None 598 self._lifecycle_lock = asyncio.Lock() 599 self._cleanup_tasks: dict[str, asyncio.Task] = {} 600 self._cancel_reasons: dict[str, str] = {} 601 602 @property 603 def active_job_id(self) -> str | None: 604 return self._running_job 605 606 @property 607 def active_session_id(self) -> str | None: 608 return self._running_session_id 609 610 async def start(self) -> None: 611 if self._http is None: 612 timeout = aiohttp.ClientTimeout(total=None, connect=10.0) 613 self._http = aiohttp.ClientSession(timeout=timeout) 614 if self._nc is None and self.nats_url: 615 self._nc = await nats.connect( 616 servers=[self.nats_url], 617 name=f"{self.service_name}-file-runner", 618 max_reconnect_attempts=-1, 619 reconnect_time_wait=2, 620 ping_interval=10, 621 ) 622 logger.info("File runner connected to NATS %s", self.nats_url) 623 elif self._nc is None: 624 logger.warning("File runner started without NATS_URL; live file-mode signals are disabled") 625 626 async def close(self) -> None: 627 await self.cancel_active(reason="service shutdown") 628 cleanup_tasks = list(self._cleanup_tasks.values()) 629 self._cleanup_tasks.clear() 630 for task in cleanup_tasks: 631 task.cancel() 632 if cleanup_tasks: 633 await asyncio.gather(*cleanup_tasks, return_exceptions=True) 634 if self._nc: 635 try: 636 await self._nc.drain() 637 except Exception: 638 await self._nc.close() 639 self._nc = None 640 if self._http: 641 await self._http.close() 642 self._http = None 643 644 async def diarization_backend_status(self) -> tuple[bool, str]: 645 await self.start() 646 assert self._http is not None 647 return await diarization_backend_ready(settings=self._settings, http=self._http) 648 649 def _job_state_fields(self, payload: dict[str, Any]) -> dict[str, Any]: 650 run_diarization = resolve_optional_bool( 651 payload.get("run_diarization"), 652 default=self._settings.diarization_enabled, 653 ) 654 return { 655 "requested_diarization": resolve_optional_bool( 656 payload.get("requested_diarization"), 657 default=run_diarization, 658 ), 659 "run_diarization": run_diarization, 660 "syntax_postprocess": bool(run_diarization) and resolve_optional_bool( 661 payload.get("syntax_postprocess"), 662 default=self._settings.diarization_syntax_postprocess_default, 663 ), 664 "diarization_backend_status": str(payload.get("diarization_backend_status") or "").strip(), 665 "diarization_fallback_reason": str(payload.get("diarization_fallback_reason") or "").strip(), 666 } 667 668 async def start_job(self, payload: dict[str, Any]) -> None: 669 job_id = str(payload.get("job_id") or "").strip() 670 session_id = str(payload.get("session_id") or "").strip() or None 671 file_name = str(payload.get("file_name") or "audio.bin").strip() or "audio.bin" 672 if not job_id: 673 raise ValueError("job_id is required") 674 job_state_fields = self._job_state_fields(payload) 675 await self.start() 676 async with self._lifecycle_lock: 677 await self._cancel_active_locked(reason="replaced by a new file job") 678 lock_lease = await self._acquire_job_lock(payload=payload, job_state_fields=job_state_fields) 679 try: 680 await self._publish_job_event( 681 job_id, 682 { 683 "type": "starting", 684 "job_id": job_id, 685 "file_name": file_name, 686 "phase": "starting", 687 **job_state_fields, 688 }, 689 session_id=session_id, 690 ) 691 task = asyncio.create_task( 692 self._process_job_with_lock(payload, lock_lease=lock_lease), 693 name=f"file-job-{job_id}", 694 ) 695 except Exception: 696 if lock_lease is not None: 697 await lock_lease.release() 698 raise 699 self._active_task = task 700 self._running_job = job_id 701 self._running_session_id = session_id 702 task.add_done_callback(self._clear_active_task) 703 704 async def _acquire_job_lock(self, *, payload: dict[str, Any], job_state_fields: dict[str, Any]) -> Any | None: 705 if not job_lock_enabled(self._settings): 706 return None 707 assert self._http is not None 708 job_id = str(payload.get("job_id") or "").strip() 709 run_diarization = resolve_optional_bool(job_state_fields.get("run_diarization"), default=False) 710 page_mode = str(payload.get("page_mode") or ("diarize" if run_diarization else "transcribe_only")).strip() 711 step = "transcribe+diarize" if run_diarization else "transcribe" 712 client = FileJobLockClient(settings=self._settings, http=self._http) 713 lease = await client.acquire( 714 owner=f"{self.service_name}:file:{job_id}", 715 page_mode=page_mode, 716 page_label_value=page_label(page_mode), 717 job_id=job_id, 718 session_id=str(payload.get("session_id") or "").strip(), 719 file_name=str(payload.get("file_name") or "audio.bin").strip() or "audio.bin", 720 step=step, 721 ) 722 if lease is not None: 723 await lease.start_heartbeat() 724 return lease 725 726 async def cancel_active(self, *, session_id: str | None = None, reason: str = "cancel requested") -> str | None: 727 async with self._lifecycle_lock: 728 if session_id and self._running_session_id != session_id: 729 return None 730 return await self._cancel_active_locked(reason=reason) 731 732 async def _cancel_active_locked(self, *, reason: str) -> str | None: 733 job_id = str(self._running_job or "").strip() 734 task = self._active_task 735 if not job_id or task is None: 736 return None 737 self._job_store.request_cancel(job_id) 738 self._cancel_reasons[job_id] = str(reason or "cancel requested").strip() or "cancel requested" 739 logger.info("Cancelling active file job %s: %s", job_id, self._cancel_reasons[job_id]) 740 if not task.done(): 741 task.cancel() 742 await asyncio.gather(task, return_exceptions=True) 743 self._clear_active_task(task) 744 self._job_store.delete_job(job_id) 745 return job_id 746 747 def _clear_active_task(self, task: asyncio.Task) -> None: 748 if self._active_task is not task: 749 return 750 self._active_task = None 751 self._running_job = None 752 self._running_session_id = None 753 754 async def _process_job_with_lock(self, payload: dict[str, Any], *, lock_lease: Any | None) -> None: 755 try: 756 await self._process_job(payload) 757 finally: 758 if lock_lease is not None: 759 await lock_lease.release() 760 761 async def _process_job(self, payload: dict[str, Any]) -> None: 762 assert self._http is not None 763 764 job_id = str(payload.get("job_id") or "").strip() 765 file_name = str(payload.get("file_name") or "audio.bin").strip() or "audio.bin" 766 file_path = str(payload.get("file_path") or "").strip() 767 language = str(payload.get("language") or "").strip() or None 768 session_id = str(payload.get("session_id") or "").strip() or None 769 job_state_fields = self._job_state_fields(payload) 770 run_diarization = bool(job_state_fields["run_diarization"]) 771 syntax_postprocess = bool(job_state_fields["syntax_postprocess"]) 772 if not job_id or not file_path: 773 logger.warning("Dropping file job with missing job_id/file_path: %s", payload) 774 return 775 776 source_path = Path(file_path).resolve() 777 if not source_path.exists(): 778 await self._publish_job_event( 779 job_id, 780 { 781 "type": "job_error", 782 "job_id": job_id, 783 "file_name": file_name, 784 "phase": "prepare_media", 785 "details": f"source file not found: {source_path}", 786 }, 787 session_id=session_id, 788 ) 789 if run_diarization: 790 await self._publish_diar_error( 791 job_id, 792 file_name=file_name, 793 details=f"source file not found: {source_path}", 794 phase="prepare_media", 795 session_id=session_id, 796 ) 797 self._schedule_job_cleanup(job_id) 798 return 799 800 diarization_task: asyncio.Task | None = None 801 diarization_cancel_event: asyncio.Event | None = None 802 diarization_requires_error = bool(run_diarization) 803 diarization_config_error = "" 804 job_done_published = False 805 cleanup_job = False 806 duration_s = 0.0 807 try: 808 if self._job_store.is_cancel_requested(job_id): 809 await self._publish_job_event( 810 job_id, 811 { 812 "type": "job_error", 813 "job_id": job_id, 814 "file_name": file_name, 815 "phase": "cancelled", 816 "details": "cancel requested before processing started", 817 }, 818 session_id=session_id, 819 ) 820 if diarization_requires_error: 821 await self._publish_diar_error( 822 job_id, 823 file_name=file_name, 824 details="cancel requested before processing started", 825 phase="cancelled", 826 session_id=session_id, 827 ) 828 cleanup_job = True 829 return 830 831 if not self._settings.linto_http_url: 832 await self._publish_job_event( 833 job_id, 834 { 835 "type": "job_error", 836 "job_id": job_id, 837 "file_name": file_name, 838 "phase": "prepare_media", 839 "details": "LINTO_HTTP_URL is required for sequential file transcription", 840 }, 841 session_id=session_id, 842 ) 843 if diarization_requires_error: 844 await self._publish_diar_error( 845 job_id, 846 file_name=file_name, 847 details="LINTO_HTTP_URL is required for sequential file transcription", 848 phase="prepare_media", 849 session_id=session_id, 850 ) 851 cleanup_job = True 852 return 853 854 normalized_path = self._job_store.normalized_audio_path(job_id) 855 await self._publish_job_event( 856 job_id, 857 { 858 "type": "prepare_media", 859 "job_id": job_id, 860 "file_name": file_name, 861 "phase": "prepare_audio", 862 }, 863 session_id=session_id, 864 ) 865 await asyncio.to_thread( 866 write_normalized_audio_wav, 867 source_path, 868 output_path=normalized_path, 869 sample_rate=16000, 870 ) 871 872 if self._job_store.is_cancel_requested(job_id): 873 await self._publish_job_event( 874 job_id, 875 { 876 "type": "job_error", 877 "job_id": job_id, 878 "file_name": file_name, 879 "phase": "cancelled", 880 "details": "cancel requested after audio preparation", 881 }, 882 session_id=session_id, 883 ) 884 if diarization_requires_error: 885 await self._publish_diar_error( 886 job_id, 887 file_name=file_name, 888 details="cancel requested after audio preparation", 889 phase="cancelled", 890 session_id=session_id, 891 ) 892 cleanup_job = True 893 return 894 895 duration_s = probe_wav_duration_s(normalized_path) 896 chunk_total = estimate_chunk_total( 897 duration_s=duration_s, 898 chunk_seconds=self._settings.chunk_seconds, 899 overlap_seconds=self._settings.chunk_overlap_seconds, 900 ) 901 await self._publish_job_event( 902 job_id, 903 { 904 "type": "prepare_media", 905 "job_id": job_id, 906 "file_name": file_name, 907 "phase": "prepare_media", 908 "duration_s": round(duration_s, 2), 909 "chunk_total": chunk_total, 910 }, 911 session_id=session_id, 912 ) 913 914 if run_diarization: 915 if not self._settings.diarization_enabled: 916 diarization_config_error = "file diarization is disabled" 917 elif self._settings.diarization_http_url: 918 diarization_artifact = await asyncio.to_thread( 919 prepare_diarization_audio_artifact, 920 source_path=source_path, 921 normalized_path=normalized_path, 922 diarization_audio_path=self._job_store.diarization_audio_path(job_id), 923 requested_audio_format=self._settings.diarization_audio_format, 924 wav_max_upload_mb=self._settings.diarization_wav_max_upload_mb, 925 audio_max_upload_mb=self._settings.diarization_audio_max_upload_mb, 926 ) 927 diarization_input_path = diarization_artifact.path 928 diarization_cancel_event = asyncio.Event() 929 diarization_task = asyncio.create_task( 930 self._run_diarization( 931 job_id=job_id, 932 upload_path=diarization_input_path, 933 file_name=diarization_input_path.name, 934 job_file_name=file_name, 935 session_id=session_id, 936 cancel_event=diarization_cancel_event, 937 ), 938 name=f"diarization-{job_id}", 939 ) 940 await self._publish_diar_progress( 941 job_id, 942 file_name=file_name, 943 details="Speaker separation started. It runs in parallel with transcription.", 944 phase="diarize_start", 945 session_id=session_id, 946 duration_s=duration_s, 947 ) 948 else: 949 diarization_config_error = "DIARIZATION_HTTP_URL is required for file diarization" 950 951 transcript = TimedTranscriptBuilder( 952 soft_break_s=self._settings.text_soft_break_s, 953 hard_break_s=self._settings.text_hard_break_s, 954 require_word_timestamps=True, 955 ) 956 completed_chunks = 0 957 for chunk in iter_normalized_wav_chunk_files( 958 normalized_path, 959 output_dir=self._job_store.chunks_dir(job_id), 960 sample_rate=16000, 961 chunk_seconds=self._settings.chunk_seconds, 962 overlap_seconds=self._settings.chunk_overlap_seconds, 963 total_chunks=chunk_total, 964 ): 965 if self._job_store.is_cancel_requested(job_id): 966 await self._publish_job_event( 967 job_id, 968 { 969 "type": "job_error", 970 "job_id": job_id, 971 "file_name": file_name, 972 "phase": "cancelled", 973 "chunk_index": chunk.index, 974 "chunk_total": chunk.total, 975 "details": "cancel requested", 976 }, 977 session_id=session_id, 978 ) 979 if diarization_requires_error: 980 await self._cancel_background_task(diarization_task, diarization_cancel_event) 981 await self._publish_diar_error( 982 job_id, 983 file_name=file_name, 984 details="cancel requested", 985 phase="cancelled", 986 session_id=session_id, 987 duration_s=duration_s, 988 ) 989 cleanup_job = True 990 return 991 992 payload_json: dict[str, Any] | None = None 993 last_error = "" 994 for attempt in range(1, self._settings.chunk_retry_limit + 2): 995 transcribe_task: asyncio.Task | None = None 996 await self._publish_job_event( 997 job_id, 998 { 999 "type": "chunk_started", 1000 "job_id": job_id, 1001 "file_name": file_name, 1002 "phase": "stream", 1003 "chunk_index": chunk.index, 1004 "chunk_total": chunk.total, 1005 "attempt": attempt, 1006 "chunk_start_s": round(chunk.start_s, 2), 1007 "chunk_end_s": round(chunk.end_s, 2), 1008 }, 1009 session_id=session_id, 1010 ) 1011 try: 1012 started_at = time.monotonic() 1013 transcribe_task = asyncio.create_task( 1014 transcribe_http_json( 1015 settings=self._settings, 1016 file_path=chunk.path, 1017 file_name=chunk.path.name, 1018 language=language, 1019 http=self._http, 1020 ) 1021 ) 1022 while True: 1023 try: 1024 payload_json = await asyncio.wait_for( 1025 asyncio.shield(transcribe_task), 1026 timeout=FILE_CHUNK_PROGRESS_INTERVAL_S, 1027 ) 1028 break 1029 except asyncio.TimeoutError: 1030 await self._publish_job_event( 1031 job_id, 1032 { 1033 "type": "chunk_progress", 1034 "job_id": job_id, 1035 "file_name": file_name, 1036 "phase": "transcribe_wait", 1037 "chunk_index": chunk.index, 1038 "chunk_total": chunk.total, 1039 "attempt": attempt, 1040 "chunk_start_s": round(chunk.start_s, 2), 1041 "chunk_end_s": round(chunk.end_s, 2), 1042 "elapsed_s": round(time.monotonic() - started_at, 1), 1043 }, 1044 session_id=session_id, 1045 ) 1046 break 1047 except Exception as exc: 1048 if transcribe_task is not None and not transcribe_task.done(): 1049 transcribe_task.cancel() 1050 try: 1051 await transcribe_task 1052 except BaseException: 1053 pass 1054 last_error = str(exc) 1055 if attempt > self._settings.chunk_retry_limit: 1056 break 1057 await self._publish_job_event( 1058 job_id, 1059 { 1060 "type": "chunk_retry", 1061 "job_id": job_id, 1062 "file_name": file_name, 1063 "phase": "retry", 1064 "chunk_index": chunk.index, 1065 "chunk_total": chunk.total, 1066 "attempt": attempt + 1, 1067 "details": str(exc), 1068 }, 1069 session_id=session_id, 1070 ) 1071 1072 if payload_json is None: 1073 await self._publish_job_event( 1074 job_id, 1075 { 1076 "type": "job_error", 1077 "job_id": job_id, 1078 "file_name": file_name, 1079 "phase": "transcribe", 1080 "chunk_index": chunk.index, 1081 "chunk_total": chunk.total, 1082 "details": last_error or "chunk transcription failed", 1083 }, 1084 session_id=session_id, 1085 ) 1086 if diarization_requires_error: 1087 await self._cancel_background_task(diarization_task, diarization_cancel_event) 1088 await self._publish_diar_error( 1089 job_id, 1090 file_name=file_name, 1091 details=last_error or "chunk transcription failed", 1092 phase="transcribe", 1093 session_id=session_id, 1094 duration_s=duration_s, 1095 ) 1096 cleanup_job = True 1097 return 1098 1099 try: 1100 transcript.append_chunk(payload=payload_json, chunk=chunk) 1101 except MissingWordTimestampsError as exc: 1102 await self._publish_job_event( 1103 job_id, 1104 { 1105 "type": "job_error", 1106 "job_id": job_id, 1107 "file_name": file_name, 1108 "phase": "merge", 1109 "chunk_index": chunk.index, 1110 "chunk_total": chunk.total, 1111 "chunk_start_s": round(chunk.start_s, 2), 1112 "chunk_end_s": round(chunk.end_s, 2), 1113 "details": str(exc), 1114 "debug": exc.debug, 1115 }, 1116 session_id=session_id, 1117 ) 1118 if diarization_requires_error: 1119 await self._cancel_background_task(diarization_task, diarization_cancel_event) 1120 await self._publish_diar_error( 1121 job_id, 1122 file_name=file_name, 1123 details=str(exc), 1124 phase="merge", 1125 session_id=session_id, 1126 duration_s=duration_s, 1127 ) 1128 cleanup_job = True 1129 return 1130 except Exception as exc: 1131 await self._publish_job_event( 1132 job_id, 1133 { 1134 "type": "job_error", 1135 "job_id": job_id, 1136 "file_name": file_name, 1137 "phase": "merge", 1138 "chunk_index": chunk.index, 1139 "chunk_total": chunk.total, 1140 "chunk_start_s": round(chunk.start_s, 2), 1141 "chunk_end_s": round(chunk.end_s, 2), 1142 "details": str(exc), 1143 }, 1144 session_id=session_id, 1145 ) 1146 if diarization_requires_error: 1147 await self._cancel_background_task(diarization_task, diarization_cancel_event) 1148 await self._publish_diar_error( 1149 job_id, 1150 file_name=file_name, 1151 details=str(exc), 1152 phase="merge", 1153 session_id=session_id, 1154 duration_s=duration_s, 1155 ) 1156 cleanup_job = True 1157 return 1158 1159 completed_chunks += 1 1160 current_text = transcript.render() 1161 self._job_store.write_transcript(job_id, current_text) 1162 await self._publish_job_event( 1163 job_id, 1164 { 1165 "type": "chunk_done", 1166 "job_id": job_id, 1167 "file_name": file_name, 1168 "phase": "stream", 1169 "chunk_index": chunk.index, 1170 "chunk_total": chunk.total, 1171 "chunks_completed": completed_chunks, 1172 "chunk_start_s": round(chunk.start_s, 2), 1173 "chunk_end_s": round(chunk.end_s, 2), 1174 "chars_total": len(current_text), 1175 }, 1176 session_id=session_id, 1177 ) 1178 try: 1179 chunk.path.unlink(missing_ok=True) 1180 except Exception: 1181 pass 1182 1183 final_text = transcript.render() 1184 self._job_store.write_transcript(job_id, final_text) 1185 await self._publish_job_event( 1186 job_id, 1187 { 1188 "type": "job_done", 1189 "job_id": job_id, 1190 "file_name": file_name, 1191 "phase": "done", 1192 "chunk_total": max(chunk_total, completed_chunks or 1), 1193 "chunks_completed": completed_chunks, 1194 "chars_total": len(final_text), 1195 }, 1196 session_id=session_id, 1197 ) 1198 job_done_published = True 1199 1200 if not run_diarization: 1201 cleanup_job = True 1202 return 1203 1204 if diarization_config_error: 1205 await self._publish_diar_error( 1206 job_id, 1207 file_name=file_name, 1208 details=diarization_config_error, 1209 phase="diarize", 1210 session_id=session_id, 1211 duration_s=duration_s, 1212 ) 1213 cleanup_job = True 1214 return 1215 1216 if diarization_task is None: 1217 await self._publish_diar_error( 1218 job_id, 1219 file_name=file_name, 1220 details="diarization task was not started", 1221 phase="diarize", 1222 session_id=session_id, 1223 duration_s=duration_s, 1224 ) 1225 cleanup_job = True 1226 return 1227 1228 if not diarization_task.done(): 1229 await self._publish_diar_progress( 1230 job_id, 1231 file_name=file_name, 1232 details=_diarization_wait_details(self._diarization_runtime_label), 1233 phase="diarize_wait", 1234 session_id=session_id, 1235 duration_s=duration_s, 1236 ) 1237 1238 diarization_payload = parse_diarization_result(await diarization_task) 1239 attributed_words = attribute_words( 1240 words=transcript.timed_words(), 1241 exclusive_segments=diarization_payload.exclusive_segments, 1242 ) 1243 raw_speaker_turns = build_speaker_turns( 1244 words=attributed_words, 1245 break_s=self._settings.diarization_turn_break_s, 1246 ) 1247 raw_speaker_text = render_speaker_text(turns=raw_speaker_turns) 1248 final_words = attributed_words 1249 final_speaker_turns = raw_speaker_turns 1250 final_speaker_text = raw_speaker_text 1251 syntax_speaker_turns: list[Any] = [] 1252 syntax_speaker_text = "" 1253 syntax_sentence_count = 0 1254 syntax_low_confidence_count = 0 1255 syntax_postprocess_applied = False 1256 if syntax_postprocess: 1257 syntax_result = apply_syntax_postprocess( 1258 words=attributed_words, 1259 sentence_gap_s=self._settings.diarization_syntax_sentence_gap_s, 1260 dominance_threshold=self._settings.diarization_syntax_dominance_threshold, 1261 turn_break_s=self._settings.diarization_turn_break_s, 1262 ) 1263 final_words = syntax_result.words 1264 final_speaker_turns = syntax_result.turns 1265 final_speaker_text = syntax_result.text 1266 syntax_speaker_turns = syntax_result.turns 1267 syntax_speaker_text = syntax_result.text 1268 syntax_sentence_count = syntax_result.sentence_count 1269 syntax_low_confidence_count = syntax_result.low_confidence_count 1270 syntax_postprocess_applied = True 1271 speaker_count = max(diarization_payload.speaker_count, len(diarization_payload.speakers)) 1272 diarization_result_payload: dict[str, Any] = { 1273 "type": "diar_done", 1274 "job_id": job_id, 1275 "file_name": file_name, 1276 "phase": "done", 1277 "duration_s": round(duration_s, 3), 1278 "model": diarization_payload.model, 1279 "speaker_count": speaker_count, 1280 "speakers": diarization_payload.speakers, 1281 "segments": diarization_segments_payload(diarization_payload.segments), 1282 "exclusive_segments": diarization_segments_payload(diarization_payload.exclusive_segments), 1283 "words": attributed_words_payload(final_words), 1284 "turns": speaker_turns_payload(final_speaker_turns), 1285 "words_raw": attributed_words_payload(attributed_words), 1286 "turns_raw": speaker_turns_payload(raw_speaker_turns), 1287 "text": final_speaker_text, 1288 "text_raw": raw_speaker_text, 1289 "text_syntax": syntax_speaker_text, 1290 "turns_syntax": speaker_turns_payload(syntax_speaker_turns), 1291 "syntax_postprocess_enabled": syntax_postprocess, 1292 "syntax_postprocess_applied": syntax_postprocess_applied, 1293 "syntax_sentence_count": syntax_sentence_count, 1294 "syntax_low_confidence_count": syntax_low_confidence_count, 1295 } 1296 if session_id: 1297 diarization_result_payload["session_id"] = session_id 1298 diar_done_signal = _diar_done_signal_payload( 1299 job_id=job_id, 1300 file_name=file_name, 1301 duration_s=duration_s, 1302 model=diarization_payload.model, 1303 speaker_count=speaker_count, 1304 speakers=diarization_payload.speakers, 1305 syntax_postprocess_enabled=syntax_postprocess, 1306 syntax_postprocess_applied=syntax_postprocess_applied, 1307 syntax_sentence_count=syntax_sentence_count, 1308 syntax_low_confidence_count=syntax_low_confidence_count, 1309 session_id=session_id, 1310 ) 1311 diar_text_payload = _diar_text_done_payload( 1312 job_id=job_id, 1313 file_name=file_name, 1314 duration_s=duration_s, 1315 speaker_count=speaker_count, 1316 speaker_text=final_speaker_text, 1317 speaker_turns=speaker_turns_payload(final_speaker_turns), 1318 speaker_text_raw=raw_speaker_text, 1319 speaker_turns_raw=speaker_turns_payload(raw_speaker_turns), 1320 speaker_text_syntax=syntax_speaker_text, 1321 speaker_turns_syntax=speaker_turns_payload(syntax_speaker_turns), 1322 syntax_postprocess_enabled=syntax_postprocess, 1323 syntax_postprocess_applied=syntax_postprocess_applied, 1324 syntax_sentence_count=syntax_sentence_count, 1325 syntax_low_confidence_count=syntax_low_confidence_count, 1326 session_id=session_id, 1327 ) 1328 await self._publish_diar_payload( 1329 job_id, 1330 diar_done_signal, 1331 store_payload=diarization_result_payload, 1332 ) 1333 await self._publish_diar_text_payload( 1334 job_id, 1335 _diar_text_done_signal_payload(diar_text_payload), 1336 store_payload=diar_text_payload, 1337 session_id=session_id, 1338 ) 1339 cleanup_job = True 1340 except asyncio.CancelledError: 1341 cancel_details = self._cancel_reasons.pop(job_id, None) or "cancel requested" 1342 logger.info( 1343 "File job cancelled (job_id=%s session_id=%s file_name=%s): %s", 1344 job_id, 1345 session_id or "-", 1346 file_name, 1347 cancel_details, 1348 ) 1349 await self._cancel_background_task(diarization_task, diarization_cancel_event) 1350 if not job_done_published: 1351 await self._publish_job_event( 1352 job_id, 1353 { 1354 "type": "job_error", 1355 "job_id": job_id, 1356 "file_name": file_name, 1357 "phase": "cancelled", 1358 "details": cancel_details, 1359 }, 1360 session_id=session_id, 1361 ) 1362 if diarization_requires_error: 1363 await self._publish_diar_error( 1364 job_id, 1365 file_name=file_name, 1366 details=cancel_details, 1367 phase="cancelled", 1368 session_id=session_id, 1369 duration_s=duration_s or None, 1370 ) 1371 cleanup_job = True 1372 return 1373 except Exception as exc: 1374 logger.exception("File job failed: %s", exc) 1375 if not job_done_published: 1376 await self._publish_job_event( 1377 job_id, 1378 { 1379 "type": "job_error", 1380 "job_id": job_id, 1381 "file_name": file_name, 1382 "phase": "worker_error", 1383 "details": str(exc), 1384 }, 1385 session_id=session_id, 1386 ) 1387 if diarization_requires_error: 1388 await self._publish_diar_error( 1389 job_id, 1390 file_name=file_name, 1391 details=str(exc), 1392 phase="worker_error", 1393 session_id=session_id, 1394 duration_s=duration_s or None, 1395 ) 1396 cleanup_job = True 1397 finally: 1398 await self._cancel_background_task(diarization_task, diarization_cancel_event) 1399 if cleanup_job: 1400 self._schedule_job_cleanup(job_id) 1401 1402 async def _publish_job_event( 1403 self, 1404 job_id: str, 1405 payload: dict[str, Any], 1406 *, 1407 session_id: str | None = None, 1408 ) -> None: 1409 effective_session_id = str(session_id or payload.get("session_id") or "").strip() or None 1410 event_payload = dict(payload) 1411 if effective_session_id and not event_payload.get("session_id"): 1412 event_payload["session_id"] = effective_session_id 1413 event_payload = self._job_store.write_state(job_id, event_payload) 1414 if not self._nc: 1415 return 1416 encoded = json.dumps(event_payload, ensure_ascii=False).encode("utf-8") 1417 await self._nc.publish(self._settings.event_subject(job_id), encoded) 1418 if effective_session_id: 1419 await self._nc.publish(self._settings.session_event_subject(effective_session_id), encoded) 1420 1421 async def _run_diarization( 1422 self, 1423 *, 1424 job_id: str, 1425 upload_path: Path, 1426 file_name: str, 1427 job_file_name: str, 1428 session_id: str | None, 1429 cancel_event: asyncio.Event, 1430 ) -> dict[str, Any]: 1431 assert self._http is not None 1432 1433 last_error = "" 1434 for attempt in range(1, self._settings.diarization_retry_limit + 2): 1435 if cancel_event.is_set(): 1436 raise asyncio.CancelledError() 1437 try: 1438 logger.info( 1439 "Diarization attempt %s started (job_id=%s session_id=%s file_name=%s path=%s)", 1440 attempt, 1441 job_id, 1442 session_id or "-", 1443 job_file_name, 1444 upload_path, 1445 ) 1446 return await diarize_http_json( 1447 settings=self._settings, 1448 file_path=upload_path, 1449 file_name=file_name, 1450 job_id=job_id, 1451 session_id=session_id, 1452 job_file_name=job_file_name, 1453 http=self._http, 1454 cancel_event=cancel_event, 1455 ) 1456 except asyncio.CancelledError: 1457 logger.info( 1458 "Diarization task cancelled (job_id=%s session_id=%s file_name=%s)", 1459 job_id, 1460 session_id or "-", 1461 job_file_name, 1462 ) 1463 raise 1464 except Exception as exc: 1465 last_error = str(exc) 1466 logger.warning( 1467 "Diarization attempt %s failed (job_id=%s session_id=%s file_name=%s): %s", 1468 attempt, 1469 job_id, 1470 session_id or "-", 1471 job_file_name, 1472 exc, 1473 ) 1474 if attempt > self._settings.diarization_retry_limit: 1475 break 1476 raise RuntimeError(last_error or "diarization failed") 1477 1478 @staticmethod 1479 async def _cancel_background_task(task: asyncio.Task | None, cancel_event: asyncio.Event | None = None) -> None: 1480 if cancel_event is not None: 1481 cancel_event.set() 1482 if task is None or task.done(): 1483 return 1484 logger.info("Cancelling background task %s", task.get_name()) 1485 task.cancel() 1486 try: 1487 await task 1488 except BaseException: 1489 pass 1490 1491 def _schedule_job_cleanup(self, job_id: str, *, delay_s: float = FILE_JOB_TERMINAL_CLEANUP_DELAY_S) -> None: 1492 token = str(job_id or "").strip() 1493 if not token: 1494 return 1495 existing = self._cleanup_tasks.pop(token, None) 1496 if existing is not None: 1497 existing.cancel() 1498 task = asyncio.create_task(self._delete_job_after_delay(token, delay_s), name=f"cleanup-{token}") 1499 self._cleanup_tasks[token] = task 1500 1501 def _clear(_task: asyncio.Task) -> None: 1502 if self._cleanup_tasks.get(token) is _task: 1503 self._cleanup_tasks.pop(token, None) 1504 1505 task.add_done_callback(_clear) 1506 1507 async def _delete_job_after_delay(self, job_id: str, delay_s: float) -> None: 1508 if delay_s > 0: 1509 await asyncio.sleep(delay_s) 1510 self._job_store.delete_job(job_id) 1511 1512 async def _publish_diar_payload( 1513 self, 1514 job_id: str, 1515 payload: dict[str, Any], 1516 *, 1517 store_payload: dict[str, Any] | None = None, 1518 ) -> None: 1519 self._job_store.write_diarization_state(job_id, payload) 1520 if store_payload is not None: 1521 self._job_store.write_diarization_payload(job_id, store_payload) 1522 if not self._nc: 1523 return 1524 encoded = json.dumps(payload, ensure_ascii=False).encode("utf-8") 1525 await self._nc.publish(self._settings.diar_event_subject(job_id), encoded) 1526 1527 async def _publish_diar_text_payload( 1528 self, 1529 job_id: str, 1530 payload: dict[str, Any], 1531 *, 1532 session_id: str | None = None, 1533 store_payload: dict[str, Any] | None = None, 1534 ) -> None: 1535 self._job_store.write_diarization_text(job_id, store_payload or payload) 1536 if not self._nc: 1537 return 1538 encoded = json.dumps(payload, ensure_ascii=False).encode("utf-8") 1539 await self._nc.publish(self._settings.diar_text_event_subject(job_id), encoded) 1540 effective_session_id = str(session_id or payload.get("session_id") or "").strip() or None 1541 if effective_session_id: 1542 await self._nc.publish(self._settings.diar_text_session_subject(effective_session_id), encoded) 1543 1544 async def _publish_diar_progress( 1545 self, 1546 job_id: str, 1547 *, 1548 file_name: str, 1549 details: str, 1550 phase: str, 1551 session_id: str | None = None, 1552 duration_s: float | None = None, 1553 ) -> None: 1554 base_payload: dict[str, Any] = { 1555 "type": "diar_progress", 1556 "job_id": job_id, 1557 "file_name": file_name, 1558 "phase": phase, 1559 "details": str(details or "").strip() or "speaker separation is in progress", 1560 } 1561 if duration_s is not None: 1562 base_payload["duration_s"] = round(float(duration_s), 3) 1563 if session_id: 1564 base_payload["session_id"] = session_id 1565 await self._publish_diar_payload(job_id, base_payload) 1566 await self._publish_diar_text_payload( 1567 job_id, 1568 { 1569 **base_payload, 1570 "type": "diar_text_progress", 1571 }, 1572 session_id=session_id, 1573 ) 1574 1575 async def _publish_diar_error( 1576 self, 1577 job_id: str, 1578 *, 1579 file_name: str, 1580 details: str, 1581 phase: str, 1582 session_id: str | None = None, 1583 duration_s: float | None = None, 1584 ) -> None: 1585 base_payload: dict[str, Any] = { 1586 "type": "diar_error", 1587 "job_id": job_id, 1588 "file_name": file_name, 1589 "phase": phase, 1590 "details": str(details or "").strip() or "unknown diarization error", 1591 } 1592 if duration_s is not None: 1593 base_payload["duration_s"] = round(float(duration_s), 3) 1594 if session_id: 1595 base_payload["session_id"] = session_id 1596 await self._publish_diar_payload(job_id, base_payload) 1597 await self._publish_diar_text_payload( 1598 job_id, 1599 { 1600 **base_payload, 1601 "type": "diar_text_error", 1602 }, 1603 session_id=session_id, 1604 ) 1605 1606 1607 def configure_logging() -> None: 1608 level = _env("LOG_LEVEL", "INFO").upper() 1609 logging.basicConfig( 1610 level=getattr(logging, level, logging.INFO), 1611 format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", 1612 ) 1613 1614 1615 async def _main_async() -> None: 1616 stream_bridge_enabled = _env_bool("STREAM_BRIDGE_ENABLED", True) 1617 webui_enabled = _env_bool("WEBUI_ENABLED", True) 1618 file_mode_settings = load_file_mode_settings() 1619 1620 if not any((stream_bridge_enabled, webui_enabled)): 1621 raise ValueError("At least one runtime must be enabled: WEBUI_ENABLED or STREAM_BRIDGE_ENABLED") 1622 1623 bridge_service = SttWhisperToNatsService(file_mode_settings) if stream_bridge_enabled else None 1624 tasks: list[asyncio.Task] = [] 1625 stop_event = asyncio.Event() 1626 1627 def request_stop() -> None: 1628 if stop_event.is_set(): 1629 return 1630 stop_event.set() 1631 if bridge_service is not None: 1632 bridge_service.stop() 1633 1634 loop = asyncio.get_running_loop() 1635 for sig in (signal.SIGINT, signal.SIGTERM): 1636 try: 1637 loop.add_signal_handler(sig, request_stop) 1638 except NotImplementedError: 1639 pass 1640 1641 if webui_enabled: 1642 from src.webui.server import run_webui 1643 1644 host = _env("WEBUI_HOST", "0.0.0.0") 1645 port = int(_env("WEBUI_PORT", "8090") or "8090") 1646 base_path = _env("WEBUI_BASE_PATH", "") 1647 tasks.append(asyncio.create_task(run_webui(host=host, port=port, base_path=base_path), name="webui")) 1648 1649 if bridge_service is not None: 1650 tasks.append(asyncio.create_task(bridge_service.run(), name="stream-bridge")) 1651 1652 stopper = asyncio.create_task(stop_event.wait(), name="stop-wait") 1653 try: 1654 done, pending = await asyncio.wait(tasks + [stopper], return_when=asyncio.FIRST_COMPLETED) 1655 if stopper not in done: 1656 for task in done: 1657 if task.cancelled(): 1658 continue 1659 exc = task.exception() 1660 if exc is not None: 1661 logger.exception("Runtime task failed: %s", exc) 1662 else: 1663 logger.warning("Runtime task exited unexpectedly: %s", task.get_name()) 1664 request_stop() 1665 await asyncio.wait(tasks, timeout=5.0) 1666 finally: 1667 request_stop() 1668 stopper.cancel() 1669 for task in tasks: 1670 if task.get_name() == "webui": 1671 task.cancel() 1672 await asyncio.gather(*tasks, return_exceptions=True) 1673 await asyncio.gather(stopper, return_exceptions=True) 1674 1675 1676 def main() -> None: 1677 configure_logging() 1678 try: 1679 asyncio.run(_main_async()) 1680 except KeyboardInterrupt: 1681 pass 1682 except Exception as exc: 1683 logger.exception("Service terminated with error: %s", exc) 1684 raise SystemExit(1) from exc 1685 1686 1687 if __name__ == "__main__": 1688 main()