renatoxsr/epub-tts
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

logger and spantrack.main() working great

Browse Source
This commit is contained in:
renatoxsr
2026-08-12 12:32:16 -03:00
parent ea16f87fc9
commit b45321a19f
3 changed files with 226 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+3
View File
@@ -29,6 +29,8 @@ classifiers = [
dependencies = [ dependencies = [
"audioop-lts; python_version >= '3.13'", "audioop-lts; python_version >= '3.13'",
"beautifulsoup4", "beautifulsoup4",
"colorlog",
"dotenv",
"ebooklib", "ebooklib",
"kokoro>=0.9.4", "kokoro>=0.9.4",
"load-dotenv>=0.1.0", "load-dotenv>=0.1.0",
@@ -37,6 +39,7 @@ dependencies = [
"nltk", "nltk",
"numpy", # --index-url https://download.pytorch.org/whl/cu132 "numpy", # --index-url https://download.pytorch.org/whl/cu132
"pillow", # --index-url https://download.pytorch.org/whl/cu132 "pillow", # --index-url https://download.pytorch.org/whl/cu132
"pyyaml",
"pydub", "pydub",
"soundfile", "soundfile",
"torch", # --index-url https://download.pytorch.org/whl/cu132 "torch", # --index-url https://download.pytorch.org/whl/cu132
src/epub_tts/logger.py
+174 -63
View File
@@ -12,6 +12,18 @@ import os
import re import re
import sys import sys
import pprint import pprint
import locale
import yaml
from typing import Any, Optional, Callable, Required
# pip packages
from dotenv import load_dotenv
import colorlog
logger = logging.getLogger(__name__)
loglevel_map = logging.getLevelNamesMapping()
# Globals # Globals
format_str = "[%(name)s][%(filename)s:%(lineno)04d][%(relativeCreated)s]::%(levelname).4s: (%(funcName)s) %(message)s" format_str = "[%(name)s][%(filename)s:%(lineno)04d][%(relativeCreated)s]::%(levelname).4s: (%(funcName)s) %(message)s"
@@ -41,21 +53,78 @@ format_str = "[%(name)s][%(filename)s:%(lineno)04d][%(relativeCreated)s]::%(leve
# %(processName)s Process name (if available) # %(processName)s Process name (if available)
# %(message)s The result of record.getMessage(), computed just as # %(message)s The result of record.getMessage(), computed just as
# the record is emitted # the record is emitted
formatter = logging.Formatter(format_str) # formatter = logging.Formatter(format_str)
logging.basicConfig( # logging.basicConfig(
# 50=CRITICAL/FATAL, 40=ERROR, 30=WARN/WARNING, 20=INFO, 10=DEBUG, 0=NOTSET # # 50=CRITICAL/FATAL, 40=ERROR, 30=WARN/WARNING, 20=INFO, 10=DEBUG, 0=NOTSET
# loglevel_map = logging.getLevelNamesMapping() # # loglevel_map = logging.getLevelNamesMapping()
level=logging.INFO, # level=logging.INFO,
format=format_str, # format=format_str,
) # )
logging.captureWarnings(True)
logger = logging.getLogger(__name__) # True regex pattern:
#locale.nl_langinfo(locale.YESEXPR)
loglevel_map = logging.getLevelNamesMapping() RE_TRUE = r"^[yY]|^[sS]|^[tT]|^[oO][nN]|1"
# False regex pattern: (use with caution!)
# locale.nl_langinfo(locale.NOEXPR)
RE_FALSE = r"^[nN]|^[fF]|^[oO][fF]+|0"
# None:
# matches any combination of word boundaries and whitespaces.
# \b cannot be used in a character range. See
RE_NONE = r"^\b*\s*\b*\s*$"
# Logging utils def user_bool(user_str: str, empty_is_none: bool = False, from_yaml=False) -> None|bool:
"""Use caution in this call, only if you are absolutely sure that user_str should be bool or none/empty"""
if from_yaml:
return yaml.YAMLObject().from_yaml(user_str)
if re.match(RE_TRUE, user_str):
return True
if re.match(RE_FALSE, user_str):
return False
if re.match(RE_NONE, user_str):
if empty_is_none:
return None
else:
return False
raise ValueError(user_str)
def get_env(env_name: str,
cast_type: Optional[Callable] = str,
default: Optional[str] = None,
) -> Any | str:
if not isinstance(cast_type, Callable):
raise ValueError(cast_type)
return cast_type(os.environ.get(env_name, default))
def get_logger(name):
return logging.getLogger(name)
def create_logger(name = __name__, loglevel: int = logging.INFO):
load_dotenv()
logging.captureWarnings(get_env("CAPTURE_WARNINGS", user_bool, "True"))
handler = colorlog.StreamHandler()
handler.setFormatter(colorlog.ColoredFormatter(
#"%(log_color)s%(levelname)-8s%(reset)s %(blue)s%(message)s",
f"%(log_color)s{format_str}%(reset)s",
log_colors={
'DEBUG': 'cyan',
'INFO': 'green',
'WARNING': 'orange',
'ERROR': 'red',
'CRITICAL': 'yellow',
}))
logger = colorlog.getLogger(name)
logger.addHandler(handler)
logger.setLevel(loglevel)
return logger
# Simple logger by printing
def print(message): def print(message):
pp = pprint.PrettyPrinter( pp = pprint.PrettyPrinter(
indent=4, indent=4,
@@ -75,7 +144,7 @@ def get_level(
verbose: bool = False, verbose: bool = False,
debug: bool = False, debug: bool = False,
quiet: bool = False, quiet: bool = False,
default: int = logging.INFO, loglevel: int = logging.INFO,
) -> int: ) -> int:
# Find lowest priority loglevel # Find lowest priority loglevel
if "--debug" in cmdline: if "--debug" in cmdline:
@@ -86,65 +155,107 @@ def get_level(
quiet = True quiet = True
if debug: if debug:
msg = ["Returning loglevel 'DEBUG'"]
if verbose or quiet:
msg.append(", ignoring other loglevel flags ")
if verbose and quiet:
msg.append(", ignoring other loglevel flags "
"(--verbose and --quiet) which are also set")
elif verbose:
msg.append("(--verbose), which is also set")
else:
msg.append("(--quiet), which is also set")
msg.append(".")
if not quiet:
logger.info("".join(msg))
return logging.DEBUG return logging.DEBUG
for i, a in enumerate(cmdline): if verbose:
if a == "--loglevel": if quiet:
loglevel = loglevel_map.get(cmdline[i+1], None) return logging.INFO
elif a.startswith("--loglevel="): #msg.append(", ignoring '--quiet' loglevel flag which was also set")
loglevel = loglevel_map.get(a.replace("--loglevel=", ""), None) logger.info("Returning loglevel 'INFO' ('--verbose' flag was set).")
return logging.INFO
# lower = more priority if quiet:
if loglevel_map[loglevel] > loglevel_map["INFO"]: return loglevel
config.logger.debug("self.loglevel='%s'(%d) > INFO", config.loglevel, loglevel_map[config.loglevel])
config.loglevel = "INFO"
def get_logger( # msg = [f"Returning default loglevel '{loglevel_map[loglevel]}'."]
verbose:bool = False, # logger.info("".join(msg))
debug:bool = False, # return loglevel
quiet: bool = False, # logger.info("Returning loglevel 'DEBUG' and ignoring other loglevel flags ("
loglevel: str|int = logging.INFO, # f"{"--verbose" if "--verbose" in cmdline} is also set"
logfile: str|Path = None, # f"{"--quiet" if "--quiet" in cmdline}"
): # ")")
"""Initialize the package logger from CLI flags and defaults. # return logging.DEBUG
# if "--verbose" in cmdline:
# verbose = True
# logger.info("Returning loglevel 'INFO'")
# return logging.INFO
# if "--quiet" in cmdline:
# logger.info("Returning loglevel 'ERROR'")
# return logging.ERROR
# return logging.DEBUG
Resolve the effective logging level from the class args (or sys.argv). # for i, a in enumerate(cmdline):
--debug (or debug_level=True) takes priority, then --verbose (or verbose=True) # if a == "--loglevel":
is compared to --loglevel (or loglevel=) and the lower one is proritized. # loglevel = loglevel_map.get(cmdline[i+1], None)
""" # elif a.startswith("--loglevel="):
# loglevel = loglevel_map.get(a.replace("--loglevel=", ""), None)
# # lower = more priority
# if loglevel_map[loglevel] > loglevel_map["INFO"]:
# logger.debug("self.loglevel='%s'(%d) > INFO", loglevel, loglevel_map[loglevel])
# loglevel = "INFO"
# def get_logger(
# verbose:bool = False,
# debug:bool = False,
# quiet: bool = False,
# loglevel: str|int = logging.INFO,
# logfile: str|Path = None,
# ):
# """Initialize the package logger from CLI flags and defaults.
# Resolve the effective logging level from the class args (or sys.argv).
# --debug (or debug_level=True) takes priority, then --verbose (or verbose=True)
# is compared to --loglevel (or loglevel=) and the lower one is proritized.
# """
config.logger.info("Setting loglevel to %s",config.loglevel) # config.logger.info("Setting loglevel to %s",config.loglevel)
config.logger.setLevel(loglevel_map[config.loglevel]) # config.logger.setLevel(loglevel_map[config.loglevel])
# Get logfile # # Get logfile
for i, a in enumerate(config._args_list): # for i, a in enumerate(config._args_list):
if a == "--logfile": # if a == "--logfile":
config.logfile = config._args_list[i+1] # config.logfile = config._args_list[i+1]
elif a.startswith("--logfile="): # elif a.startswith("--logfile="):
config.logfile = a.replace("--logfile=", "") # config.logfile = a.replace("--logfile=", "")
if "logfile" in config and config.logfile: # if "logfile" in config and config.logfile:
if logfile: # if logfile:
config.logger.warning("Overriding --logfile='%s' from setup_logger(logfile='%s')", config.logfile, logfile) # config.logger.warning("Overriding --logfile='%s' from setup_logger(logfile='%s')", config.logfile, logfile)
config.logfile = logfile # config.logfile = logfile
fh = logging.FileHandler(logfile) # fh = logging.FileHandler(logfile)
fh.setLevel(config.loglevel) # fh.setLevel(config.loglevel)
fh.setFormatter(logger.formatter) # fh.setFormatter(logger.formatter)
config.logger.addHandler(fh) # config.logger.addHandler(fh)
# if self.debug_level: # # if self.debug_level:
# logger.info("Loglevel: DEBUG") # # logger.info("Loglevel: DEBUG")
# if self.loglevel: # # if self.loglevel:
# logger.warning("Ignoring '--loglevel' flag because DEBUG/'--debug' was also set.") # # logger.warning("Ignoring '--loglevel' flag because DEBUG/'--debug' was also set.")
# if "--verbose" in self._args_list: # # if "--verbose" in self._args_list:
# logger.warning("Ignoring '--verbose' flag because '--debug' was also set.") # # logger.warning("Ignoring '--verbose' flag because '--debug' was also set.")
# self.verbose = False # # self.verbose = False
# if self.loglevel and self.verbose: # # if self.loglevel and self.verbose:
# self.warning("Conflict: trying to --loglevel=%s (%d) and --verbose.", self.loglevel, loglevel_map[self.loglevel]) # # self.warning("Conflict: trying to --loglevel=%s (%d) and --verbose.", self.loglevel, loglevel_map[self.loglevel])
# if loglevel_map[self.loglevel] < logging.INFO: # # if loglevel_map[self.loglevel] < logging.INFO:
# self.logger.info("Setting loglevel to %s (%d), which is more verbose than INFO.", self.loglevel, loglevel_map[self.loglevel]) # # self.logger.info("Setting loglevel to %s (%d), which is more verbose than INFO.", self.loglevel, loglevel_map[self.loglevel])
return config.logger # return config.logger
src/epub_tts/spantrack.py
+49 -22
View File
@@ -11,22 +11,26 @@ import logging
from datetime import datetime as dt from datetime import datetime as dt
from dotenv import load_dotenv from dotenv import load_dotenv
from .logger import logger from .logger import logger, create_logger
logger = create_logger(__name__)
from spantrack import cli, run_config from spantrack import cli, run_config
load_dotenv() load_dotenv()
os.environ["NLTK_DATA"] = 'C:\\Users\\renat\\AppData\\Roaming\\nltk_data' os.environ["NLTK_DATA"] = 'C:\\Users\\renat\\AppData\\Roaming\\nltk_data'
os.environ["HF_TOKEN"] = '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' os.environ["HF_TOKEN"] = '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'
timefmt = os.environ.get("TIMEFMT", r"%Y-%m-%dT%H:%M:%S.%f")
# Globals
time_format: str = os.environ.get("TIMEFMT", r"%Y-%m-%dT%H:%M:%S.%f")
caught_exceptions: tuple[Exception] = (KeyboardInterrupt, BaseException)
def main(): def main():
logger.info("Start main function.")
root = Path('D:\\DATA\\EBOOKS\\_SEM_DRM\\') root = Path('D:\\DATA\\EBOOKS\\_SEM_DRM\\')
globs = [ globs = [
"Night School*.epub", #"Night School*.epub",
"The Midnight*.epub", "Midnight Line*.epub",
"Past Tense*.epub", "Past Tense*.epub",
"Blue Moon*.epub", "Blue Moon*.epub",
"The Sentinel*.epub", "The Sentinel*.epub",
@@ -44,22 +48,45 @@ def main():
#"--lang", # Language, default=Language.EN_US, #"--lang", # Language, default=Language.EN_US,
"--voice","am_liam", "--voice","am_liam",
]) ])
result_output = [] result_output = {"pending":[],"success":[],"error":[]}
for glob in globs: logger.info("Start main loop: ")
file_list = [f for f in root.glob(glob)] logger.info("try (for glob in globs: %s", ", ".join([f"'{str(g)}'" for g in globs]))
if len(file_list) > 1: logger.info("except: %s", ", ".join([f"{e.__name__}" for e in caught_exceptions]) )
logger.error("Glob pattern found more than 1 file: %s", f) logger.info("finally: sys.exit( len(errors) + len(pending) )")
continue try:
file = str(file_list[0]) for glob_i,glob in enumerate(globs):
start = dt.now() file_list = [f for f in root.glob(glob)]
logger.info(f"[{start.strftime(timefmt)}] Narrating '{file}'") if len(file_list) != 1:
cli.main([file] + cmd) logger.error(f"Glob pattern '{glob}' found {"more" if len(file_list)>1 else "less"} than 1 file: %s", file_list)
finish = dt.now() result_output["error"].append(glob)
logger.info(f"[{finish .strftime(timefmt)}] Success! '{file}'") continue
result_output["pending"].append(str(file_list[0]))
logger.info(
"Files to parse:\n%s",
",\n".join([f"[{i+1}] '{str(f)}'" for i,f in enumerate(result_output["pending"])]))
for file in result_output["pending"]:
start_dt = dt.now()
logger.info(f"[{start_dt.strftime(time_format)}] Narrating '{file}'")
cli.main([file] + cmd)
finish_dt = dt.now()
logger.info(f"[{finish_dt .strftime(time_format)}] Success! '{file}'")
elapsed_time = finish-start elapsed_time = finish_dt-start_dt
elapsed_hours= elapsed_time.total_seconds // 3600 elapsed_total_seconds = elapsed_time.total_seconds()
elapsed_minutes = (elapsed_time.total_seconds % 3600) // 60 elapsed_hours= elapsed_total_seconds // 3600
elapsed_seconds = elapsed_time.total_seconds % 60 elapsed_minutes = (elapsed_total_seconds % 3600) // 60
f"Total elapsed time: {elapsed_hours}h {elapsed_minutes}min {elapsed_seconds}s" elapsed_seconds = elapsed_total_seconds % 60
result_output.append() f"Total elapsed time: {elapsed_hours}h {elapsed_minutes}min {elapsed_seconds}s"
result_output["success"].append(file)
except caught_exceptions as e:
logger.critical("Caught exception: %s", str(e))
finally:
logger.info("Exiting with results:\nDONE:%s\nLIST:%s\nERR:%s",
",\n".join([f"\t[{i+1}] '{str(f)}'" for i,f in enumerate(result_output["success"])]),
",\n".join([f"\t[{i+1}] '{str(f)}'" for i,f in enumerate(result_output["pending"])]),
",\n".join([f"\t[{i+1}] '{str(f)}'" for i,f in enumerate(result_output["error"])]),
)
sys.exit(
len(result_output["pending"]) +
len(result_output["error"])
)