| """Prepare the complete local speech runtime using backend/.venv only.""" |
|
|
| from __future__ import annotations |
|
|
| import hashlib |
| import os |
| import shutil |
| import sys |
| import tempfile |
| import urllib.request |
| import zipfile |
| from pathlib import Path, PurePosixPath |
|
|
|
|
| BACKEND_DIR = Path(__file__).resolve().parents[1] |
| EXPECTED_ENV = (BACKEND_DIR / ".venv").resolve() |
| DATA_DIR = BACKEND_DIR / "data" |
| MODEL_DIR = DATA_DIR / "models" |
| FFMPEG_DIR = DATA_DIR / "ffmpeg" |
| FFMPEG_ARCHIVE_URL = ( |
| "https://github.com/GyanD/codexffmpeg/releases/download/7.1.1/" |
| "ffmpeg-7.1.1-full_build-shared.zip" |
| ) |
| |
| FFMPEG_ARCHIVE_SHA256 = ( |
| "9f28727e8b472a04c1d2e520aaa425dca82721b995139b35710091130ea6e699" |
| ) |
| FFMPEG_DLL_PATTERNS = ( |
| "avcodec-*.dll", |
| "avdevice-*.dll", |
| "avfilter-*.dll", |
| "avformat-*.dll", |
| "avutil-*.dll", |
| "swresample-*.dll", |
| "swscale-*.dll", |
| ) |
| sys.path.insert(0, str(BACKEND_DIR)) |
|
|
|
|
| def require_project_environment() -> None: |
| active = Path(sys.prefix).resolve() |
| if active != EXPECTED_ENV: |
| raise SystemExit( |
| "Refusing to prepare models outside backend/.venv. " |
| f"Expected {EXPECTED_ENV}, got {active}." |
| ) |
|
|
|
|
| def configure_project_caches() -> None: |
| paths = { |
| "HF_HOME": MODEL_DIR / "huggingface", |
| "TORCH_HOME": MODEL_DIR / "torch", |
| "NLTK_DATA": MODEL_DIR / "nltk", |
| "WHISPERX_MODEL_DIR": MODEL_DIR / "whisperx", |
| } |
| for key, path in paths.items(): |
| path.mkdir(parents=True, exist_ok=True) |
| os.environ[key] = str(path) |
|
|
|
|
| def _ffmpeg_runtime_complete(directory: Path) -> bool: |
| return ( |
| (directory / "ffmpeg.exe").is_file() |
| and (directory / "ffprobe.exe").is_file() |
| and all(any(directory.glob(pattern)) for pattern in FFMPEG_DLL_PATTERNS) |
| ) |
|
|
|
|
| def _prepend_ffmpeg_to_path() -> None: |
| directory = str(FFMPEG_DIR.resolve()) |
| current = os.environ.get("PATH", "") |
| entries = [entry for entry in current.split(os.pathsep) if entry] |
| if not any(os.path.normcase(entry) == os.path.normcase(directory) for entry in entries): |
| os.environ["PATH"] = ( |
| f"{directory}{os.pathsep}{current}" if current else directory |
| ) |
|
|
|
|
| def _download_ffmpeg_archive(destination: Path) -> None: |
| request = urllib.request.Request( |
| FFMPEG_ARCHIVE_URL, |
| headers={"User-Agent": "MAESC speech runtime setup"}, |
| ) |
| digest = hashlib.sha256() |
| with urllib.request.urlopen(request, timeout=120) as response, destination.open( |
| "wb" |
| ) as output: |
| while chunk := response.read(1024 * 1024): |
| digest.update(chunk) |
| output.write(chunk) |
|
|
| if digest.hexdigest().lower() != FFMPEG_ARCHIVE_SHA256.lower(): |
| raise RuntimeError("本地语音运行时下载不完整，请检查网络后重新运行准备命令。") |
|
|
|
|
| def _extract_ffmpeg_bin(archive_path: Path, destination: Path) -> None: |
| destination.mkdir(parents=True, exist_ok=True) |
| with zipfile.ZipFile(archive_path) as archive: |
| for member in archive.infolist(): |
| archive_name = PurePosixPath(member.filename.replace("\\", "/")) |
| if ( |
| member.is_dir() |
| or len(archive_name.parts) < 2 |
| or archive_name.parts[-2].lower() != "bin" |
| ): |
| continue |
| filename = archive_name.name |
| if Path(filename).suffix.lower() not in {".exe", ".dll"}: |
| continue |
| with archive.open(member) as source, (destination / filename).open( |
| "wb" |
| ) as output: |
| shutil.copyfileobj(source, output) |
|
|
|
|
| def ensure_local_ffmpeg() -> Path: |
| """Download a pinned shared FFmpeg build into ignored project data once.""" |
| if _ffmpeg_runtime_complete(FFMPEG_DIR): |
| _prepend_ffmpeg_to_path() |
| return FFMPEG_DIR |
|
|
| DATA_DIR.mkdir(parents=True, exist_ok=True) |
| print("Preparing the local speech runtime (first run only)...", flush=True) |
| try: |
| with tempfile.TemporaryDirectory(prefix="maesc-ffmpeg-", dir=DATA_DIR) as temp: |
| temp_dir = Path(temp) |
| archive_path = temp_dir / "ffmpeg.zip" |
| extracted_dir = temp_dir / "ffmpeg" |
| _download_ffmpeg_archive(archive_path) |
| _extract_ffmpeg_bin(archive_path, extracted_dir) |
| if not _ffmpeg_runtime_complete(extracted_dir): |
| raise RuntimeError( |
| "本地语音运行时文件不完整，请检查网络后重新运行准备命令。" |
| ) |
|
|
| FFMPEG_DIR.mkdir(parents=True, exist_ok=True) |
| for source in extracted_dir.iterdir(): |
| os.replace(source, FFMPEG_DIR / source.name) |
| except Exception as error: |
| if isinstance(error, RuntimeError): |
| raise |
| raise RuntimeError( |
| "无法自动准备本地语音运行时，请检查网络后重新运行准备命令。" |
| ) from error |
|
|
| if not _ffmpeg_runtime_complete(FFMPEG_DIR): |
| raise RuntimeError("本地语音运行时准备失败，请重新运行准备命令。") |
| _prepend_ffmpeg_to_path() |
| return FFMPEG_DIR |
|
|
|
|
| def main() -> None: |
| require_project_environment() |
| configure_project_caches() |
| ensure_local_ffmpeg() |
| os.environ["LOCAL_SPEECH_ENABLED"] = "true" |
|
|
| import nltk |
|
|
| nltk_dir = os.environ["NLTK_DATA"] |
| for resource in ( |
| "averaged_perceptron_tagger_eng", |
| "cmudict", |
| "punkt", |
| "punkt_tab", |
| ): |
| if not nltk.download(resource, download_dir=nltk_dir, quiet=False): |
| raise RuntimeError(f"Failed to download NLTK resource: {resource}") |
|
|
| from app.services.speech_runtime import SpeechRuntime |
|
|
| runtime = SpeechRuntime() |
| runtime.preload() |
| if runtime.status != "ready": |
| raise RuntimeError(runtime.error or "WhisperX model preparation failed") |
| if runtime.pronunciation_status == "error": |
| raise RuntimeError( |
| runtime.pronunciation_error or "Pronunciation model preparation failed" |
| ) |
| runtime.close() |
| print(f"Speech runtime is ready under {DATA_DIR}") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
