From 28ffc1820b244a2fef64e66993e8331432f1f06b Mon Sep 17 00:00:00 2001 From: RenatoXSR Date: Fri, 7 Aug 2026 10:16:22 -0300 Subject: [PATCH] working backend --- src/epub_tts/__init__.py | 5 +++ src/epub_tts/__main__.py | 21 ++++------ src/epub_tts/tts_aedocw.py | 15 +++++++- src/epub_tts/tts_generic.py | 75 +++++++++++++++++++++++------------- src/epub_tts/vendor_utils.py | 21 ++++++---- 5 files changed, 89 insertions(+), 48 deletions(-) diff --git a/src/epub_tts/__init__.py b/src/epub_tts/__init__.py index 5318be8..d0810e0 100644 --- a/src/epub_tts/__init__.py +++ b/src/epub_tts/__init__.py @@ -12,3 +12,8 @@ try: except PackageNotFoundError: # package is not installed pass + +from typing import Union +from pathlib import Path + +PathLike = Union[str, Path] \ No newline at end of file diff --git a/src/epub_tts/__main__.py b/src/epub_tts/__main__.py index f9ac1d1..fda2ab3 100644 --- a/src/epub_tts/__main__.py +++ b/src/epub_tts/__main__.py @@ -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), ) diff --git a/src/epub_tts/tts_aedocw.py b/src/epub_tts/tts_aedocw.py index a051d49..3886f44 100644 --- a/src/epub_tts/tts_aedocw.py +++ b/src/epub_tts/tts_aedocw.py @@ -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): diff --git a/src/epub_tts/tts_generic.py b/src/epub_tts/tts_generic.py index 248aa5f..4a60a65 100644 --- a/src/epub_tts/tts_generic.py +++ b/src/epub_tts/tts_generic.py @@ -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 - \ No newline at end of file + + + 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)) \ No newline at end of file diff --git a/src/epub_tts/vendor_utils.py b/src/epub_tts/vendor_utils.py index 9ea88ad..f5966ee 100644 --- a/src/epub_tts/vendor_utils.py +++ b/src/epub_tts/vendor_utils.py @@ -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] +