# -*- coding: utf-8 -*-
|
import logging
|
import os
|
import subprocess
|
import tempfile
|
from typing import List, Optional, Tuple
|
|
from app.schemas import AsrHit, KeywordConfig
|
from app.video_io import get_ffmpeg_cmd
|
|
logger = logging.getLogger(__name__)
|
|
_whisper_model = None
|
|
|
def _get_whisper():
|
global _whisper_model
|
if _whisper_model is None:
|
from faster_whisper import WhisperModel
|
model_size = os.environ.get("WHISPER_MODEL", "tiny")
|
_whisper_model = WhisperModel(model_size, device="cpu", compute_type="int8")
|
logger.info("加载 Whisper 模型: %s", model_size)
|
return _whisper_model
|
|
|
def extract_audio_wav(video_path: str) -> str:
|
out = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
|
out.close()
|
cmd = [
|
get_ffmpeg_cmd("ffmpeg"), "-y", "-i", video_path,
|
"-vn", "-acodec", "pcm_s16le", "-ar", "16000", "-ac", "1",
|
out.name,
|
]
|
subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
return out.name
|
|
|
def transcribe(video_path: str) -> List[Tuple[str, float, float]]:
|
wav = extract_audio_wav(video_path)
|
try:
|
model = _get_whisper()
|
segments, _ = model.transcribe(wav, language="zh", vad_filter=True)
|
return [(seg.text.strip(), seg.start, seg.end) for seg in segments if seg.text.strip()]
|
except Exception as e:
|
logger.warning("ASR 失败: %s", e)
|
return []
|
finally:
|
if os.path.isfile(wav):
|
os.remove(wav)
|
|
|
def match_keywords(
|
segments: List[Tuple[str, float, float]],
|
keywords: KeywordConfig,
|
) -> List[AsrHit]:
|
hits: List[AsrHit] = []
|
for text, start, _end in segments:
|
for kw in keywords.storefront:
|
if kw in text:
|
hits.append(AsrHit(keyword=kw, time_sec=round(start, 2)))
|
break
|
for kw in keywords.handover:
|
if kw in text:
|
hits.append(AsrHit(keyword=kw, time_sec=round(start, 2)))
|
break
|
return hits
|
|
|
def best_asr_time(hits: List[AsrHit], keywords: List[str]) -> Optional[float]:
|
for hit in hits:
|
if hit.keyword in keywords:
|
return hit.time_sec
|
return None
|
|
|
def asr_available() -> bool:
|
try:
|
import faster_whisper # noqa: F401
|
return True
|
except ImportError:
|
return False
|
