1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
|
"""The managed Python environment for the audiobook generator and its backends.
audiobook.py is meant to be launched from any Python (a bare system interpreter
is fine): on startup it bootstraps a single tool-managed venv at
``app/envs/tts`` and re-execs itself inside it. That venv holds both the
audiobook app's own ``requirements.txt`` dependencies and the backend TTS
packages (``qwen-tts``, ``faster-qwen3-tts[demo]``) the setup wizards pip
install, so nothing is ever installed into the launching interpreter's
environment.
A parent process never needs to "activate" an environment — activation is
just a shell convenience that puts an env's ``bin`` on PATH. Instead every
helper here resolves the env's binaries by absolute path
(``app/envs/tts/bin/python``, ``app/envs/tts/bin/qwen-tts-demo``), so the hub can
spawn servers in this env from any parent environment.
This module is imported before audiobook.py's third-party dependencies, so
it must stay stdlib-only (it may import ``backends.common``, which is also
stdlib-only, but never ``converter`` or the backend modules).
"""
import hashlib
import os
import sys
from pathlib import Path
from typing import List
from backends import common
# The tts-audiobook-generator checkout root (where audiobook.py lives).
TTS_ROOT = Path(__file__).resolve().parent.parent.parent
# One shared venv for the app requirements and every pip-installed backend.
ENV_DIR = TTS_ROOT / "app" / "envs" / "tts"
REQUIREMENTS_PATH = TTS_ROOT / "requirements.txt"
# Marker file recording the requirements.txt hash last installed into the env,
# so ensure_app_env() re-installs when requirements.txt changes.
MARKER_PATH = ENV_DIR / ".audiobook_env_ready"
def _is_windows() -> bool:
return sys.platform == "win32"
def env_python() -> Path:
"""Absolute path to the venv's python interpreter."""
return ENV_DIR / ("Scripts/python.exe" if _is_windows() else "bin/python")
def env_script(name: str) -> Path:
"""Absolute path to a console script installed in the venv (e.g. qwen-tts-demo)."""
subdir = "Scripts" if _is_windows() else "bin"
suffix = ".exe" if _is_windows() else ""
return ENV_DIR / subdir / f"{name}{suffix}"
def env_exists() -> bool:
"""True when the venv's python interpreter is present on disk."""
return env_python().is_file()
def is_managed_env() -> bool:
"""True when the current process is already running inside the managed venv."""
try:
return Path(sys.executable).resolve() == env_python().resolve()
except OSError:
return False
def create_env() -> int:
"""Create the venv with the launching interpreter (inherits its version).
pip is bootstrapped inside the venv by ensurepip. Returns the ``python -m
venv`` exit code; a non-zero result is reported with platform remediation.
"""
print(f"[INFO] creating managed environment at {ENV_DIR}...")
rc = common.run_console_subprocess(
[sys.executable, "-m", "venv", str(ENV_DIR)])
if rc != 0:
print(f"[ERROR] python -m venv failed (exit {rc}).")
if _is_windows():
print(" On Windows make sure the launcher has the venv module.")
else:
print(" On Debian/Ubuntu install the venv package, e.g.:")
print(" sudo apt install python3-venv")
return rc
def install_requirements() -> int:
"""pip install -r requirements.txt into the venv. Returns pip's exit code."""
print(f"[INFO] pip install -r {REQUIREMENTS_PATH} into {ENV_DIR}...")
return common.run_console_subprocess(
[str(env_python()), "-m", "pip", "install", "-r", str(REQUIREMENTS_PATH)])
def pip_install(packages: List[str], *, emit=None, cancel=None) -> int:
"""pip install PACKAGES into the venv, creating it first if needed.
Used by the qwen/faster setup wizards to install backend TTS packages
alongside the app requirements. Returns pip's exit code. With EMIT given
(the in-TUI task view) pip runs with ``--progress-bar off`` so its output
is clean status lines rather than carriage-return progress spam.
"""
if not env_exists() and create_env() != 0:
return 1
print(f"[INFO] pip install {' '.join(packages)} into {ENV_DIR}...")
argv = [str(env_python()), "-m", "pip", "install"]
if emit is not None:
argv.append("--progress-bar")
argv.append("off")
argv.extend(packages)
return common.run_console_subprocess(argv, emit=emit, cancel=cancel)
def pip_uninstall(packages: List[str], *, emit=None) -> int:
"""pip uninstall PACKAGES from the venv. Returns pip's exit code.
Used by the backends' ``uninstall`` action to remove pip-installed TTS
packages from the managed environment. A missing env is a no-op (there
is nothing to uninstall from), reported as success. With EMIT given
(the in-TUI task view) pip runs with its output piped and streamed to
EMIT, so nothing writes to the terminal behind curses.
"""
if not env_exists():
return 0
print(f"[INFO] pip uninstall {' '.join(packages)} from {ENV_DIR}...")
return common.run_console_subprocess(
[str(env_python()), "-m", "pip", "uninstall", "-y", *packages],
emit=emit)
def module_available(module: str) -> bool:
"""True when MODULE imports inside the venv (e.g. qwen_tts, faster_qwen3_tts).
A short subprocess probe against the venv's interpreter — the equivalent of
importlib.util.find_spec, but for the managed env rather than the current
one. Used by each backend's ``_is_installed``.
"""
if not env_exists():
return False
import subprocess
try:
result = subprocess.run(
[str(env_python()), "-c", f"import {module}"],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
timeout=30, check=False)
except (OSError, subprocess.TimeoutExpired):
return False
return result.returncode == 0
def _requirements_sha() -> str:
try:
data = REQUIREMENTS_PATH.read_bytes()
except OSError:
return ""
return hashlib.sha256(data).hexdigest()
def _marker_valid() -> bool:
try:
return MARKER_PATH.read_text(encoding="utf-8").strip() == _requirements_sha()
except OSError:
return False
def _write_marker() -> None:
try:
MARKER_PATH.write_text(_requirements_sha() + "\n", encoding="utf-8")
except OSError:
pass
def ensure_app_env() -> None:
"""Make sure the venv exists and has the current requirements.txt installed.
Creates the venv when missing, and (re)installs requirements.txt when it is
missing or has changed since the last install (tracked by a hash marker).
Raises RuntimeError on any failure so the caller can abort before re-exec.
"""
if not env_exists() and create_env() != 0:
raise RuntimeError("could not create the managed environment")
if not _marker_valid():
if install_requirements() != 0:
raise RuntimeError("pip install -r requirements.txt failed")
_write_marker()
def bootstrap(script_path: str) -> None:
"""Run audiobook.py inside the managed venv, creating it first if needed.
A no-op when the current process is already the venv's interpreter. Otherwise
ensures the env (and requirements) are ready, then replaces the process with
the venv's python running the same script and CLI args. Called at the top of
audiobook.py before any third-party import.
"""
if is_managed_env():
return
try:
ensure_app_env()
except RuntimeError as exc:
print(f"[FATAL] {exc}", file=sys.stderr)
sys.exit(1)
py = str(env_python())
target = str(Path(script_path).resolve())
print(f"[INFO] re-launching inside managed environment: {py}")
os.execv(py, [py, target, *sys.argv[1:]])
|
