""" 转写服务 处理音频文件的语音转文字，支持： - 双引擎（Whisper / 讯飞）切换 - 讯飞多模型（13 个）切换 - 实时进度上报 - 自动转换上传音频为讯飞要求的 PCM 16kHz 16bit mono 格式 """ import json import os import subprocess import wave from typing import Any, Optional from sqlalchemy.orm import Session from app.config import settings from app.services.asr import get_asr_engine, ASREngine, ASRResult from app.services.mock_data import generate_mock_transcript from app.services.progress_tracker import get_progress_tracker from app.utils.logger import logger from app.utils.exceptions import NotFoundException class TranscriptService: """转写服务""" def __init__(self, provider: str | None = None, model_id: str | None = None): """ 初始化转写服务 Args: provider: ASR 引擎类型，不传则使用默认配置 model_id: 讯飞模型 ID """ self._provider = provider or settings.ASR_PROVIDER self._model_id = model_id or getattr(settings, "XUNFEI_MODEL", "realtime_llm") self._engine: ASREngine | None = None self._mock_enabled = settings.MOCK_ENABLED or not settings.TRANSCRIPTION_ENABLED @property def provider(self) -> str: return self._provider @property def model_id(self) -> str: return self._model_id def get_engine(self) -> ASREngine: """获取当前 ASR 引擎（懒加载）""" if self._engine is None: self._engine = get_asr_engine(self._provider, self._model_id) logger.info( f"初始化 ASR 引擎: provider={self._provider}, model={self._model_id}" ) return self._engine def switch_provider(self, provider: str, model_id: Optional[str] = None) -> None: """ 切换 ASR 引擎 Args: provider: "whisper" | "xunfei" model_id: 讯飞模型 ID（仅 provider="xunfei" 时使用） """ if provider not in ("whisper", "xunfei"): raise ValueError(f"不支持的 ASR 引擎: {provider}") self._provider = provider if model_id: self._model_id = model_id self._engine = None # 强制重新初始化 logger.info(f"切换 ASR 引擎: provider={provider}, model={self._model_id}") @staticmethod def _resolve_user_engine(user_settings: Any | None, fallback_provider: str, fallback_model: str) -> tuple[str, str]: """根据 user_settings 解析当前用户实际使用的引擎与模型 优先级: 用户设置 > 全局环境变量默认值 返回 (provider, model_id) """ if user_settings is None: logger.info(f"ASR引擎选择: 用户无设置记录，使用全局默认 provider={fallback_provider}, model={fallback_model}") return fallback_provider, fallback_model # 显式读取用户偏好，空字符串也视为未设置 user_provider = (user_settings.asr_provider or "").strip() user_model = (getattr(user_settings, "asr_model", "") or "").strip() provider = user_provider if user_provider else fallback_provider model_id = user_model if user_model else fallback_model logger.info( f"ASR引擎选择: user_provider='{user_provider}', user_model='{user_model}' " f"→ 实际使用 provider={provider}, model={model_id}" ) return provider, model_id def transcribe( self, audio_path: str, *, consultation_id: Optional[str] = None, provider: Optional[str] = None, model_id: Optional[str] = None, **kwargs: Any, ) -> dict[str, Any]: """ 转写音频文件（自动检测格式并转换为讯飞要求的 PCM 16kHz 16bit mono） Args: audio_path: 音频文件路径 consultation_id: 用于上报进度 provider: 临时覆盖 provider model_id: 临时覆盖 model_id **kwargs: 透传给引擎的额外参数 """ # 如果指定了 provider/model_id，临时切换 original_provider, original_model = self._provider, self._model_id if provider: self._provider = provider if model_id: self._model_id = model_id if self._provider != original_provider or self._model_id != original_model: self._engine = None # 重置以加载新引擎 tracker = get_progress_tracker() if consultation_id else None if tracker is not None: tracker.init(consultation_id, engine=self._engine_name_for_log()) tracker.update(consultation_id, stage="preparing", percent=0, message="正在准备转写…") # 阶段1：音频格式转换 (0-20%) if tracker is not None: tracker.update(consultation_id, stage="converting", percent=5, message="正在检测音频格式…") converted_path = self._ensure_audio_format(audio_path) if tracker is not None: tracker.update(consultation_id, stage="uploading", percent=20, message="音频格式转换完成，准备连接云端引擎…") if self._mock_enabled: logger.info(f"使用 Mock 转写: {audio_path}") if tracker is not None: tracker.update(consultation_id, stage="uploading", percent=30, message="Mock 引擎初始化中…") tracker.update(consultation_id, stage="transcribing", percent=60, message="Mock 数据生成中…") tracker.update(consultation_id, stage="done", percent=100, message="Mock 转写完成") return generate_mock_transcript() # 阶段2：上传/连接讯飞 (20-40%) - 由引擎内部驱动 if tracker is not None: tracker.update(consultation_id, stage="uploading", percent=20, message="正在连接云端语音引擎…") logger.info( f"开始转写: {converted_path} provider={self._provider} " f"model={self._model_id}" ) try: engine = self.get_engine() # 注入进度回调：写到 ProgressTracker def _on_progress(percent: int, stage: str) -> None: if tracker is not None and consultation_id: tracker.update( consultation_id, stage=stage, percent=percent, message=stage_msg(stage), ) if hasattr(engine, "set_progress_callback"): engine.set_progress_callback(_on_progress) result: ASRResult = engine.transcribe(converted_path, **kwargs) response = { "segments": [ { "speaker": seg.speaker, "text": seg.text, "start_time": seg.start_time, "end_time": seg.end_time, "confidence": seg.confidence, } for seg in result.segments ], "full_text": result.full_text, "confidence": result.confidence, "audio_duration": result.audio_duration, "engine": result.engine, "word_count": result.word_count, "speaker_turns": result.speaker_turns, } if tracker is not None and consultation_id: tracker.update(consultation_id, stage="done", percent=100, message="转写完成") logger.info( f"转写完成: {result.engine}, " f"{len(result.segments)}段, {result.word_count}字, " f"conf={result.confidence:.2f}" ) return response except Exception as e: logger.error(f"转写失败: {e}") if tracker is not None and consultation_id: tracker.update(consultation_id, stage="error", percent=100, message=f"转写失败: {e}", error=str(e)) raise finally: # 恢复原 provider/model self._provider, self._model_id = original_provider, original_model self._engine = None def _engine_name_for_log(self) -> str: return f"{self._provider}:{self._model_id}" def _ensure_audio_format(self, audio_path: str) -> str: """ 确保音频为讯飞要求的格式：PCM 16kHz 16bit mono WAV """ if not os.path.exists(audio_path): raise FileNotFoundError(f"音频文件不存在: {audio_path}") try: with wave.open(audio_path, "rb") as w: channels = w.getnchannels() sample_width = w.getsampwidth() frame_rate = w.getframerate() needs_convert = ( channels != 1 or sample_width != 2 or frame_rate != 16000 ) if not needs_convert: logger.info( f"音频格式正确: {frame_rate}Hz/{sample_width*8}bit/{channels}ch, " f"跳过转换" ) return audio_path logger.info( f"音频需转换: {frame_rate}Hz/{sample_width*8}bit/{channels}ch → 16000Hz/16bit/1ch, " f"path={audio_path}" ) except Exception as exc: logger.info(f"非 WAV 格式, 需转换: {audio_path} ({exc})") needs_convert = True base, _ = os.path.splitext(audio_path) converted_path = f"{base}_16k.wav" cmd = [ "ffmpeg", "-y", "-i", audio_path, "-ac", "1", "-ar", "16000", "-sample_fmt", "s16", converted_path, ] try: logger.info(f"FFmpeg 转换开始: {os.path.basename(audio_path)} -> {os.path.basename(converted_path)}") result = subprocess.run(cmd, capture_output=True, text=True, timeout=120) if result.returncode != 0: logger.error(f"FFmpeg 转换失败: {result.stderr[:500]}") raise RuntimeError("音频格式转换失败, 请确认已安装 FFmpeg") converted_size = os.path.getsize(converted_path) logger.info( f"音频转换完成: {os.path.basename(converted_path)}, " f"size={converted_size / 1024:.1f}KB" ) return converted_path except FileNotFoundError: logger.error("FFmpeg 未安装") raise RuntimeError( "需要 FFmpeg 进行音频格式转换。" "请安装 FFmpeg: https://ffmpeg.org/download.html" ) except subprocess.TimeoutExpired: logger.error(f"FFmpeg 转换超时: {audio_path}") raise RuntimeError("音频转换超时（超120秒）") # ============================================================ # DB 方法 # ============================================================ @staticmethod def get_transcript( db: Session, consultation_id: str, provider: Optional[str] = None, model_id: Optional[str] = None, ) -> dict: """获取转写结果（旧路由兼容）""" from app.models.consultation import Consultation, ConsultationStatus from app.models.transcript import Transcript from app.models.user_settings import UserSettings consultation = TranscriptService._get_or_404(db, consultation_id) # 解析用户偏好：优先使用用户设置，未设置时回退到全局默认 user_settings = db.query(UserSettings).filter( UserSettings.user_id == consultation.doctor_id ).first() provider, model_id = TranscriptService._resolve_user_engine( user_settings, fallback_provider=provider or settings.ASR_PROVIDER, fallback_model=model_id or getattr(settings, "XUNFEI_MODEL", "realtime_llm"), ) # 已转写 → 直接返回 if consultation.transcript and consultation.transcript.content: content = json.loads(consultation.transcript.content) TranscriptService._save_audio_attachment(db, consultation) return {**content, "consultation_id": consultation_id} # 未转写 → 如果有录音文件，触发转写 if consultation.audio_path and os.path.exists(consultation.audio_path): if settings.MOCK_ENABLED or not settings.TRANSCRIPTION_ENABLED: mock_data = generate_mock_transcript() transcript = Transcript( consultation_id=consultation_id, content=json.dumps(mock_data, ensure_ascii=False), confidence=mock_data.get("confidence", 0.0), ) db.add(transcript) consultation.status = ConsultationStatus.TRANSCRIBED db.commit() TranscriptService._save_audio_attachment(db, consultation) return {**mock_data, "consultation_id": consultation_id} try: svc = TranscriptService(provider=provider, model_id=model_id) result = svc.transcribe( consultation.audio_path, consultation_id=consultation_id, ) transcript = Transcript( consultation_id=consultation_id, content=json.dumps(result, ensure_ascii=False), confidence=result.get("confidence", 0.0), ) db.add(transcript) consultation.status = ConsultationStatus.TRANSCRIBED db.commit() TranscriptService._save_audio_attachment(db, consultation) return {**result, "consultation_id": consultation_id} except Exception as e: logger.error(f"转写失败: {e}") raise raise NotFoundException("未找到录音文件, 请先上传音频") @staticmethod def extract_and_generate(db: Session, consultation_id: str) -> dict: """ LCPS 2.0: 病史提取 + SOAP 生成合并为一步。 直接调用 SoapService.generate() 完成从对话到完整 SOAP 的全流程。 """ from app.services.soap_service import SoapService return SoapService.generate(db, consultation_id) @staticmethod def get_extraction(db: Session, consultation_id: str) -> dict: """获取 AI 分析结果 — 直接读取已生成的 SOAP 病历""" from app.services.soap_service import SoapService return SoapService.get(db, consultation_id) @staticmethod def _get_or_404(db: Session, consultation_id: str): from app.models.consultation import Consultation consultation = db.query(Consultation).filter( Consultation.id == consultation_id ).first() if not consultation: raise NotFoundException(f"问诊记录不存在: {consultation_id}") return consultation @staticmethod def _save_audio_attachment(db: Session, consultation): """转录完成后将音频文件保存为附件（如尚未保存）""" from app.models.attachment import Attachment audio_path = consultation.audio_path if not audio_path or not os.path.exists(audio_path): return # 检查是否已存在该音频文件的附件记录 existing = db.query(Attachment).filter( Attachment.consultation_id == consultation.id, Attachment.file_path == audio_path, ).first() if existing: return # 根据扩展名确定 MIME 类型 ext = os.path.splitext(audio_path)[1].lower() mime_map = { ".wav": "audio/wav", ".webm": "audio/webm", ".mp3": "audio/mpeg", ".m4a": "audio/mp4", ".ogg": "audio/ogg", ".mp4": "audio/mp4", } mime_type = mime_map.get(ext, "audio/wav") try: file_size = os.path.getsize(audio_path) attachment = Attachment( consultation_id=consultation.id, filename=os.path.basename(audio_path), original_name=f"录音文件{ext}", file_path=audio_path, file_size=file_size, mime_type=mime_type, category="audio", ) db.add(attachment) db.commit() logger.info(f"音频附件已保存: consultation={consultation.id}, path={audio_path}") except Exception as e: logger.warning(f"保存音频附件失败: {e}") db.rollback() def stage_msg(stage: str) -> str: """进度阶段 → 人类可读描述""" return { "preparing": "准备中…", "converting": "音频格式转换中…", "uploading": "音频上传/连接中…", "transcribing": "语音转文字中…", "extracting": "病史提取中…", "generating": "AI 生成 SOAP 中…", "done": "完成", "error": "处理失败", }.get(stage, "处理中…")