working backend

This commit is contained in:
2026-08-07 10:16:22 -03:00
parent 3553426a72
commit 28ffc1820b
5 changed files with 89 additions and 48 deletions
+5
View File
@@ -12,3 +12,8 @@ try:
except PackageNotFoundError:
# package is not installed
pass
from typing import Union
from pathlib import Path
PathLike = Union[str, Path]
+8 -13
View File
@@ -57,6 +57,7 @@ def main(args=None):
p.add_argument("-o", "--output", help="Output audio file", required=True)
p.add_argument("-l", "--language", help="Language to use for TTS")
p.add_argument("-v", "--voice", help="Voice to use for TTS")
p.add_argument("-r", "--replace", action="append", nargs=2, help="Replace text in the intermediate output. Specify pairs of old_text new_text. Can be used multiple times.")
p.add_argument(
"-b",
"--backend",
@@ -65,6 +66,8 @@ def main(args=None):
choices=backend.keys(),
)
args = p.parse_args(args or sys.argv[1:])
setattr(args, 'replace_map',
{old: new for old, new in args.replace} if args.replace else None)
print(
f"Converting {args.input} to {args.output} "
f"using {args.backend} backend "
@@ -72,20 +75,12 @@ def main(args=None):
f"{k}={v}" for k, v in vars(args).items()
if k not in ("input", "output", "backend") and v is not None)
)
#print(args.replace_map)
input_path = Path(args.input).resolve()
if not input_path.exists():
raise FileNotFoundError(f"Input file {input_path} does not exist.")
output_path = Path(args.output).resolve()
if not output_path.parent.exists():
os.makedirs(output_path.parent, exist_ok=True)
backend_cls = backend[args.backend]
backend_instance = backend_cls(
backend_cmd=None,
)
backend_instance.run(
input_path, output_path,
backend[args.backend]().run(
args.input, args.output,
**vars(args),
)
+14 -1
View File
@@ -14,7 +14,7 @@ corresponding console script is not available in the current environment.
"""
from .tts_generic import GenericTTSBackend
from . import PathLike
class AedocwBackend(GenericTTSBackend):
"""Generic aedocw backend wrapper.
@@ -35,6 +35,9 @@ class AedocwBackend(GenericTTSBackend):
"language": "--language",
}
INTERMEDIATE_TXT = True
INTERMEDIATE_CALL = GenericTTSBackend._replace_map
class AedocwEpub2TTS(AedocwBackend):
@@ -45,6 +48,16 @@ class AedocwEpub2TTS(AedocwBackend):
class AedocwEpub2TTSEdge(AedocwBackend):
"""Backend configured for github.com/aedocw/epub2tts-edge."""
DEFAULT_SPEAKER = "en-US-AndrewNeural"
CMD = "epub2tts-edge"
REPO = "epub2tts-edge"
def gen_input_flag(self, input_path) -> list[str]:
"""Return backend flag for input file or directory."""
p = self._normalize_path(input_path)
return [str(p)] # epub2tts-edge expects a single path, no flag
gen_output_flag = None
class TTSEdge(AedocwEpub2TTSEdge):
+48 -27
View File
@@ -28,15 +28,13 @@ Classes:
run(input_source: PathLike, output_dest: PathLike) -> subprocess.CompletedProcess: Execute backend command.
"""
import re
import subprocess
from pathlib import Path
from typing import Optional, Union
PathLike = Union[str, Path]
from typing import Optional, Callable
from .vendor_utils import build_run_command
from . import PathLike
class GenericTTSBackend:
"""Generic TTS backend for handling text-to-speech conversion.
@@ -46,11 +44,13 @@ class GenericTTSBackend:
language (Optional[str]): Language code for TTS output.
voice (Optional[str]): Voice name for TTS output.
"""
FLAGS = {}
INTERMEDIATE_TXT = False
DEFAULT_SPEAKER = None
DEFAULT_SAMPLE = None
REPO: Optional[str] = None
CMD: Optional[str] = None
FLAGS: Optional[dict] = {}
INTERMEDIATE_TXT: Optional[bool] = False
INTERMEDIATE_CALL: Optional[Callable] = None
DEFAULT_SPEAKER: Optional[str] = None
DEFAULT_SAMPLE: Optional[str] = None
SUPPORTED_AUDIO_FORMATS = {
# subclasses may override this
".m4b": "m4b",
@@ -60,11 +60,8 @@ class GenericTTSBackend:
".ogg": "ogg",
}
_INPUT_FLAG_MAP = lambda self, is_dir, path: [
"--input-dir", path] if is_dir else ["--input-file", path]
_OUTPUT_FLAG_MAP = lambda self, is_dir, path: [
"--output-dir", path] if is_dir else ["--output-file", path]
INPUT_FLAG = {True: "--input-dir", False: "--input-file"}
OUTPUT_FLAG = {True: "--output-dir", False: "--output-file"}
def __init__(self,**kwargs):
for k,v in kwargs.items():
@@ -78,20 +75,23 @@ class GenericTTSBackend:
return Path(value).resolve()
def gen_input_flag(self, input_path: Path) -> str:
return self._INPUT_FLAG_MAP(
self._normalize_path(input_path).is_dir(), input_path)
def gen_input_flag(self, input_path: Path) -> list[str]:
p = self._normalize_path(input_path)
return [self.INPUT_FLAG[p.is_dir()], str(p)]
def gen_output_flag(self, output_path: Path) -> str:
return self._OUTPUT_FLAG_MAP(
self._normalize_path(output_path).is_dir(), output_path)
def gen_output_flag(self, output_path: Path) -> list[str]:
p = self._normalize_path(output_path)
return [self.OUTPUT_FLAG[p.is_dir()], str(p)]
def _build_command(self, input_source: PathLike, output_dest: PathLike) -> list[str]:
command = [self.backend_cmd]
if self.REPO is None or self.CMD is None:
command = [self.CMD]
else:
command = build_run_command(self.REPO, CMD=self.CMD)
command.extend(self.gen_input_flag(input_source))
command.extend(self.gen_output_flag(output_dest))
if self.gen_output_flag:
command.extend(self.gen_output_flag(output_dest))
for flag_name, var_name in self.FLAGS.items():
if getattr(self, flag_name, None) is not None:
command.extend([
@@ -104,9 +104,10 @@ class GenericTTSBackend:
output_dest: PathLike,
**kwargs
) -> subprocess.CompletedProcess:
command = self._build_command(input_source, output_dest)
print(command)
print(f"Running command: {' '.join(command)}")
completed = subprocess.run(
self._build_command(input_source, output_dest),
cwd=str(self._normalize_path(input_source).parent),
@@ -115,6 +116,12 @@ class GenericTTSBackend:
if self.INTERMEDIATE_TXT:
txt_file = self._normalize_path(input_source).with_suffix(".txt")
if txt_file.exists():
if (self.INTERMEDIATE_CALL is not None and
kwargs.get('replace_map', None) is not None):
self.INTERMEDIATE_CALL(
txt_file,
txt_file.with_stem(txt_file.stem + "_replaced"), kwargs.get('replace_map', {}))
txt_file = txt_file.with_stem(txt_file.stem + "_replaced")
completed = subprocess.run(
self._build_command(txt_file, output_dest),
cwd=str(self._normalize_path(input_source).parent),
@@ -124,4 +131,18 @@ class GenericTTSBackend:
return completed
def _replace_map(self,
input_path: PathLike,
output_path: PathLike,
replace_map: dict = None,
):
"""Run a simple regex substitution for the intermediate text extraction step."""
if replace_map is not None:
pattern = re.compile("|".join(re.escape(key) for key in replace_map.keys()))
with open(input_path, 'r') as src, open(output_path, 'w') as dest:
for line in src:
dest.write(
pattern.sub(lambda match: replace_map[match.group(0)], line))
+14 -7
View File
@@ -58,16 +58,23 @@ def ensure_venv(repo_name: str) -> Path:
def build_run_command(
repo_name: str,
module_name: str,
backend_cmd: Optional[str] = None,
*,
MOD: Optional[str] = None,
CMD: Optional[str] = None,
) -> list[str]:
if backend_cmd is None or backend_cmd == repo_name:
if not MOD and not CMD:
raise ValueError("Either MOD or CMD must be provided.")
if MOD:
python_executable = ensure_venv(repo_name)
return [str(python_executable), "-m", module_name]
else:
return [str(python_executable), "-m", MOD]
if CMD:
ensure_venv(repo_name)
venv_path = get_venv_path(repo_name)
return [str( venv_path / "Scripts" / f"{backend_cmd}.exe")
return [str( venv_path / "Scripts" / f"{CMD}.exe")
if sys.platform == "win32"
else str(venv_path / "bin" / backend_cmd)]
else str(venv_path / "bin" / CMD)]
# else:
python_executable = ensure_venv(repo_name)
return [str(python_executable), "-m", MOD]