Config interface ok
This commit is contained in:
@@ -0,0 +1,300 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
# Copyright (c) 2025-2026 Renato Xavier da Silveira Rosa
|
||||
# Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
|
||||
# See [LICENSE](./LICENSE) or [BSD-3-Clause-Clear](https://spdx.org/licenses/BSD-3-Clause-Clear.html)
|
||||
"""config.py
|
||||
|
||||
The industry-standard hierarchy for application configuration is:
|
||||
1. Command-Line Arguments (argparse) (Highest priority – explicit runtime overrides).
|
||||
2. Environment Variables (os.environ) (Medium priority – deployment or session-specific configuration).
|
||||
3. Configuration Files (e.g., .env, .ini, .yaml files)
|
||||
4. Hardcoded Defaults (Lowest priority – fallback safety net)
|
||||
"""
|
||||
# stdlib
|
||||
from argparse import ArgumentParser
|
||||
from collections import UserDict
|
||||
from pathlib import Path
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import sys
|
||||
|
||||
|
||||
# Pip packages
|
||||
from dotenv import load_dotenv
|
||||
|
||||
# Local imports
|
||||
from . import tts_aedocw, tts_generic, tts_kokoro
|
||||
from . import logger, PathLike
|
||||
from . import DataDict
|
||||
|
||||
class App(DataDict):
|
||||
# Globals/Defaults
|
||||
DEFAULT_ENV_FILE=".env"
|
||||
ENV_FILE_OVERRIDE=False
|
||||
WHICH = ["ffmpeg"] # Needed in $PATH
|
||||
DEBUG = False
|
||||
LOGLEVEL = logging.INFO
|
||||
BACKENDS = {
|
||||
# keys are the backend names, values are the corresponding TTS classes
|
||||
"default": tts_generic.GenericTTSBackend,
|
||||
"edge": tts_aedocw.TTSEdge,
|
||||
# aedocw-backed implementations — these expect the corresponding
|
||||
# package "console scripts"/entrypoints to be available in the
|
||||
# environment (or an explicit backend_cmd to be provided).
|
||||
"epub2tts": tts_aedocw.AedocwEpub2TTS,
|
||||
"epub2tts-edge": tts_aedocw.AedocwEpub2TTSEdge,
|
||||
"epub2tts-chatterbox": tts_aedocw.AedocwChatterbox,
|
||||
"epub2tts-kokoro": tts_aedocw.AedocwKokoro,
|
||||
"generic-epub2tts": tts_aedocw.Epub2TTS,
|
||||
"kokoro": tts_kokoro.KokoroBackend,
|
||||
}
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args,**kwargs)
|
||||
self._argv = sys.argv
|
||||
|
||||
self._args_list = args if args is not None else self._argv
|
||||
|
||||
|
||||
# Setup logger
|
||||
self.logger = logger
|
||||
self.logger.setLevel(App.LOGLEVEL)
|
||||
if App.DEBUG or "--debug" in self._args_list:
|
||||
self.debug = True
|
||||
self.loglevel = logging.DEBUG
|
||||
logger.info("Loglevel = DEBUG")
|
||||
if "--loglevel" in self._args_list:
|
||||
logger.warning("Ignoring '--loglevel' flag because DEBUG/'--debug' was also set.")
|
||||
if "--verbose" in self._args_list:
|
||||
logger.warning("Ignoring '--verbose' flag because '--debug' was also set.")
|
||||
|
||||
for i, a in enumerate(self._args_list):
|
||||
if a.startswith("--"):
|
||||
match a:
|
||||
case "--env_file":
|
||||
if self.get("env_file", None):
|
||||
self.env_file = self._args_list[i+1]
|
||||
case "--loglevel":
|
||||
self.loglevel = self._args_list[i+1]
|
||||
case "--verbose":
|
||||
self.loglevel = logging.INFO
|
||||
case "--logfile":
|
||||
self.logfile = self._args_list[i+1]
|
||||
|
||||
for var_name in ["env_file","loglevel","logfile"]:
|
||||
flag = f"--{var_name}="
|
||||
if a.startswith(flag):
|
||||
self[var_name] = a.replace(flag, "")
|
||||
|
||||
if "env_file" not in self._data:
|
||||
self._data["env_file"] = App.DEFAULT_ENV_FILE
|
||||
|
||||
# for path_flag in ["env_file"]:
|
||||
# if not Path(self[path_flag]).exists():
|
||||
# self.logger.warning(f"Could not read %s. Ignoring '%s' and using '%s'", path_flag, self[path_flag], App.DEFAULT_ENV_FILE)
|
||||
# self[path_flag] = App["DEFAULT_" + path_flag.upper()]
|
||||
|
||||
# Set loglevel from laste --loglevel flag found in args/sys.argv
|
||||
if "loglevel" not in self._data:
|
||||
self.loglevel = self.LOGLEVEL
|
||||
self.logger.setLevel(self.loglevel)
|
||||
|
||||
# load .env/env_file
|
||||
load_dotenv(self.env_file, override=os.environ.get("ENV_FILE_OVERRIDE",App.ENV_FILE_OVERRIDE))
|
||||
|
||||
|
||||
# Now we can call argparser with correct defaults and logger/loglevel
|
||||
# Get ArgumentParser and pased *args or read from sys.argv
|
||||
self._parsed_args = self.get_parser().parse_args(args or sys.argv[1:])
|
||||
for k,v in self._parsed_args.__dict__.items():
|
||||
self[k] = v
|
||||
|
||||
if self._parsed_args.loglevel:
|
||||
self.loglevel = self._parsed_args.loglevel
|
||||
self.logger.setLevel(self.loglevel)
|
||||
elif self._parsed_args.verbose:
|
||||
self.loglevel = logging.INFO
|
||||
self.logger.setLevel(self.loglevel)
|
||||
|
||||
if getattr(self, "logfile", None) or getattr(self._parsed_args, "logfile", None):
|
||||
logfile = getattr(self, "logfile", None) or self._parsed_args.logfile
|
||||
fh = logging.FileHandler(logfile)
|
||||
fh.setLevel(self.loglevel)
|
||||
fh.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
|
||||
self.logger.addHandler(fh)
|
||||
|
||||
# Validate and sanitize output
|
||||
self.output_path = App.path(self.output)
|
||||
if len(self.input) > 1 and not self.output_path.is_dir():
|
||||
self.logger.error("Output must be a directory when multiple input files are provided. Invalid: '%s'", self.output_path)
|
||||
sys.exit(1)
|
||||
del self._data["output"]
|
||||
|
||||
# Validate and sanitize input
|
||||
self.input_files = {}
|
||||
self.input_dirs = []
|
||||
self.info("Paths provided as input: %d", len(self.input))
|
||||
for i, input_str in enumerate(self.input):
|
||||
self.info("[Input %d] %s", i+1, input_str)
|
||||
input_path = App.path(input_str)
|
||||
if not input_path.exists():
|
||||
self.warning("Skipping path (do not exist): '%s'.", input_path)
|
||||
continue
|
||||
if input_path.is_dir():
|
||||
self.input_dir.append(input_path)
|
||||
self.info("Added directory to queue: '%s'", input_path)
|
||||
else:
|
||||
self.input_files[input_str] = {"path":input_path}
|
||||
self.input_files[input_str]["filename"] = self.input_files[input_str]["path"].name
|
||||
self.input_files[input_str]["stem"] = self.input_files[input_str]["path"].stem
|
||||
self.input_files[input_str]["suffix"] = self.input_files[input_str]["path"].suffix
|
||||
#self.input_files[input_str]["filetype"] = ebook_meta self.input_files[input_str]["path"]
|
||||
self.info("Added file to queue: %s", self.input_files[input_str])
|
||||
del self._data["input"]
|
||||
|
||||
|
||||
# Sanitize replace_map
|
||||
self.debug(self.replace)
|
||||
if self.replace and len(self.replace) % 2 == 0:
|
||||
self.replace_map = { old: new for old, new in self.replace}
|
||||
del self.replace
|
||||
|
||||
|
||||
# print args
|
||||
self.logger.debug(self)
|
||||
|
||||
# Execute actions
|
||||
if self._parsed_args.check_env:
|
||||
self.check_env()
|
||||
|
||||
|
||||
self.engine = App.BACKENDS[args.backend](**vars(args))
|
||||
if len(args.input) > 1 and not output_dest.is_dir():
|
||||
self.logger.error(f"Output must be a directory when multiple input files are provided")
|
||||
sys.exit(1)
|
||||
for file in args.input:
|
||||
self.logger.info(file)
|
||||
if not Path(file).exists():
|
||||
self.logger.error(f"Input file %s does not exist", file)
|
||||
sys.exit(1)
|
||||
output_dest = Path(args.output)
|
||||
|
||||
# Log functions
|
||||
def log(self, *args,**kwargs):
|
||||
self.logger.log(self.loglevel, *args, **kwargs)
|
||||
|
||||
# 10=DEBUG
|
||||
def debug(self, *args,**kwargs):
|
||||
self.logger.debug(*args, **kwargs)
|
||||
|
||||
# 20=INFO
|
||||
def info(self, *args,**kwargs):
|
||||
self.logger.info(*args, **kwargs)
|
||||
|
||||
# 30=WARN/WARNING
|
||||
def warning(self, *args,**kwargs):
|
||||
self.logger.warning(*args, **kwargs)
|
||||
|
||||
# 40=ERROR
|
||||
def error(self, *args,**kwargs):
|
||||
self.logger.error(*args, **kwargs)
|
||||
|
||||
# 50=CRITICAL/FATAL
|
||||
def critical(self, *args,**kwargs):
|
||||
self.logger.critical(*args, **kwargs)
|
||||
|
||||
# Utility methods
|
||||
def check_env(self, which=None):
|
||||
"""Check anvironment for necessary binaries"""
|
||||
if not which:
|
||||
which = self.WHICH
|
||||
if isinstance(which, str):
|
||||
if " " in which:
|
||||
which = which.split()
|
||||
which = [which]
|
||||
for bin in which:
|
||||
if sys.platform == "win32":
|
||||
p = shutil.which(bin + ".exe")
|
||||
else:
|
||||
p = shutil.which(bin)
|
||||
if not p:
|
||||
self.logger.error(f"%s is either NOT installed or "
|
||||
"NOT in your system PATH.", bin)
|
||||
return False
|
||||
else:
|
||||
self.logger.info(f"%s found at: %s", bin, p)
|
||||
return True
|
||||
|
||||
def get_parser(self):
|
||||
p = ArgumentParser(description="Convert EPUB to audio using TTS")
|
||||
p.add_argument(
|
||||
"--version",
|
||||
help="Show version and exit",
|
||||
action="version",
|
||||
version=f"%(prog)s {__import__('epub_tts').__version__}")
|
||||
|
||||
# General options
|
||||
p.add_argument(
|
||||
"--debug", action="store_true",
|
||||
help="Enable debug logging and verbose diagnostics")
|
||||
|
||||
# Input options and processing
|
||||
p.add_argument(
|
||||
"-i", "--input",
|
||||
nargs="+", required=True,
|
||||
help="Input EPUB file")
|
||||
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(
|
||||
"--check-env",action="store_true",
|
||||
help="Check the runtime environment and exit")
|
||||
p.add_argument(
|
||||
"-b", "--backend", default="default",
|
||||
choices=App.BACKENDS.keys(),
|
||||
help="Backend to use for TTS")
|
||||
|
||||
# Output options
|
||||
p.add_argument(
|
||||
"-o", "--output",
|
||||
help="Output audio file", required=True)
|
||||
p.add_argument(
|
||||
"-c", "--cover",
|
||||
help="Path to cover image to embed or use "
|
||||
"for output metadata")
|
||||
|
||||
# Speech options
|
||||
p.add_argument(
|
||||
"-l", "--language",
|
||||
help="Language to use for TTS (see your "
|
||||
"backend's documentation for available voices)")
|
||||
p.add_argument(
|
||||
"-v", "--voice",
|
||||
help="Voice to use for TTS (see your backend's "
|
||||
"documentation for available voices)")
|
||||
p.add_argument(
|
||||
"--speed", type=float, default=1.0,
|
||||
help="Playback speed multiplier (default: 1.0) "
|
||||
"(not all backends support this)")
|
||||
p.add_argument(
|
||||
"--short-pause", type=int, default=None,
|
||||
help="Short pause duration in milliseconds between "
|
||||
"phrases or sentences (not all backends support this)")
|
||||
p.add_argument(
|
||||
"--long-pause", type=int, default=None,
|
||||
help="Long pause duration in milliseconds between "
|
||||
"sections or paragraphs (not all backends support this)")
|
||||
p.add_argument(
|
||||
"--notitles", action="store_true",
|
||||
help="Do not read chapter titles")
|
||||
|
||||
return p
|
||||
|
||||
@staticmethod
|
||||
def path(fpath: PathLike = None):
|
||||
return Path(fpath).resolve()
|
||||
Reference in New Issue
Block a user