diff --git a/pyproject.toml b/pyproject.toml index 34bedf6..afb1f3e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,6 +29,8 @@ classifiers = [ dependencies = [ "audioop-lts; python_version >= '3.13'", "beautifulsoup4", + "colorlog", + "dotenv", "ebooklib", "kokoro>=0.9.4", "load-dotenv>=0.1.0", @@ -37,6 +39,7 @@ dependencies = [ "nltk", "numpy", # --index-url https://download.pytorch.org/whl/cu132 "pillow", # --index-url https://download.pytorch.org/whl/cu132 + "pyyaml", "pydub", "soundfile", "torch", # --index-url https://download.pytorch.org/whl/cu132 diff --git a/src/epub_tts/logger.py b/src/epub_tts/logger.py index 19ae090..dde3452 100644 --- a/src/epub_tts/logger.py +++ b/src/epub_tts/logger.py @@ -12,6 +12,18 @@ import os import re import sys 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 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) # %(message)s The result of record.getMessage(), computed just as # the record is emitted -formatter = logging.Formatter(format_str) -logging.basicConfig( - # 50=CRITICAL/FATAL, 40=ERROR, 30=WARN/WARNING, 20=INFO, 10=DEBUG, 0=NOTSET - # loglevel_map = logging.getLevelNamesMapping() - level=logging.INFO, - format=format_str, -) -logging.captureWarnings(True) +# formatter = logging.Formatter(format_str) +# logging.basicConfig( +# # 50=CRITICAL/FATAL, 40=ERROR, 30=WARN/WARNING, 20=INFO, 10=DEBUG, 0=NOTSET +# # loglevel_map = logging.getLevelNamesMapping() +# level=logging.INFO, +# format=format_str, +# ) -logger = logging.getLogger(__name__) - -loglevel_map = logging.getLevelNamesMapping() +# True regex pattern: +#locale.nl_langinfo(locale.YESEXPR) +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): pp = pprint.PrettyPrinter( indent=4, @@ -75,7 +144,7 @@ def get_level( verbose: bool = False, debug: bool = False, quiet: bool = False, - default: int = logging.INFO, + loglevel: int = logging.INFO, ) -> int: # Find lowest priority loglevel if "--debug" in cmdline: @@ -86,65 +155,107 @@ def get_level( quiet = True 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 - for i, a in enumerate(cmdline): - if a == "--loglevel": - loglevel = loglevel_map.get(cmdline[i+1], None) - elif a.startswith("--loglevel="): - loglevel = loglevel_map.get(a.replace("--loglevel=", ""), None) + if verbose: + if quiet: + return logging.INFO + #msg.append(", ignoring '--quiet' loglevel flag which was also set") + logger.info("Returning loglevel 'INFO' ('--verbose' flag was set).") + return logging.INFO - # lower = more priority - if loglevel_map[loglevel] > loglevel_map["INFO"]: - config.logger.debug("self.loglevel='%s'(%d) > INFO", config.loglevel, loglevel_map[config.loglevel]) - config.loglevel = "INFO" + if quiet: + return loglevel -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. + # msg = [f"Returning default loglevel '{loglevel_map[loglevel]}'."] + # logger.info("".join(msg)) + # return loglevel + # logger.info("Returning loglevel 'DEBUG' and ignoring other loglevel flags (" + # f"{"--verbose" if "--verbose" in cmdline} is also set" + # f"{"--quiet" if "--quiet" in cmdline}" + # ")") + # 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). - --debug (or debug_level=True) takes priority, then --verbose (or verbose=True) - is compared to --loglevel (or loglevel=) and the lower one is proritized. - """ + # for i, a in enumerate(cmdline): + # if a == "--loglevel": + # 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.setLevel(loglevel_map[config.loglevel]) + # config.logger.info("Setting loglevel to %s",config.loglevel) + # config.logger.setLevel(loglevel_map[config.loglevel]) - # Get logfile - for i, a in enumerate(config._args_list): - if a == "--logfile": - config.logfile = config._args_list[i+1] - elif a.startswith("--logfile="): - config.logfile = a.replace("--logfile=", "") + # # Get logfile + # for i, a in enumerate(config._args_list): + # if a == "--logfile": + # config.logfile = config._args_list[i+1] + # elif a.startswith("--logfile="): + # config.logfile = a.replace("--logfile=", "") - if "logfile" in config and config.logfile: - if logfile: - config.logger.warning("Overriding --logfile='%s' from setup_logger(logfile='%s')", config.logfile, logfile) - config.logfile = logfile - fh = logging.FileHandler(logfile) - fh.setLevel(config.loglevel) - fh.setFormatter(logger.formatter) - config.logger.addHandler(fh) + # if "logfile" in config and config.logfile: + # if logfile: + # config.logger.warning("Overriding --logfile='%s' from setup_logger(logfile='%s')", config.logfile, logfile) + # config.logfile = logfile + # fh = logging.FileHandler(logfile) + # fh.setLevel(config.loglevel) + # fh.setFormatter(logger.formatter) + # config.logger.addHandler(fh) - # if self.debug_level: - # logger.info("Loglevel: DEBUG") - # if self.loglevel: - # 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.") - # self.verbose = False - # if self.loglevel and self.verbose: - # self.warning("Conflict: trying to --loglevel=%s (%d) and --verbose.", self.loglevel, loglevel_map[self.loglevel]) - # 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]) + # # if self.debug_level: + # # logger.info("Loglevel: DEBUG") + # # if self.loglevel: + # # 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.") + # # self.verbose = False + # # if self.loglevel and self.verbose: + # # self.warning("Conflict: trying to --loglevel=%s (%d) and --verbose.", self.loglevel, loglevel_map[self.loglevel]) + # # 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]) - - return config.logger \ No newline at end of file + + # return config.logger \ No newline at end of file diff --git a/src/epub_tts/spantrack.py b/src/epub_tts/spantrack.py index 3f29acc..57a0de2 100644 --- a/src/epub_tts/spantrack.py +++ b/src/epub_tts/spantrack.py @@ -11,22 +11,26 @@ import logging from datetime import datetime as dt 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 load_dotenv() os.environ["NLTK_DATA"] = 'C:\\Users\\renat\\AppData\\Roaming\\nltk_data' 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(): + logger.info("Start main function.") root = Path('D:\\DATA\\EBOOKS\\_SEM_DRM\\') globs = [ - "Night School*.epub", - "The Midnight*.epub", + #"Night School*.epub", + "Midnight Line*.epub", "Past Tense*.epub", "Blue Moon*.epub", "The Sentinel*.epub", @@ -44,22 +48,45 @@ def main(): #"--lang", # Language, default=Language.EN_US, "--voice","am_liam", ]) - result_output = [] - for glob in globs: - file_list = [f for f in root.glob(glob)] - if len(file_list) > 1: - logger.error("Glob pattern found more than 1 file: %s", f) - continue - file = str(file_list[0]) - start = dt.now() - logger.info(f"[{start.strftime(timefmt)}] Narrating '{file}'") - cli.main([file] + cmd) - finish = dt.now() - logger.info(f"[{finish .strftime(timefmt)}] Success! '{file}'") + result_output = {"pending":[],"success":[],"error":[]} + logger.info("Start main loop: ") + logger.info("try (for glob in globs: %s", ", ".join([f"'{str(g)}'" for g in globs])) + logger.info("except: %s", ", ".join([f"{e.__name__}" for e in caught_exceptions]) ) + logger.info("finally: sys.exit( len(errors) + len(pending) )") + try: + for glob_i,glob in enumerate(globs): + file_list = [f for f in root.glob(glob)] + if len(file_list) != 1: + logger.error(f"Glob pattern '{glob}' found {"more" if len(file_list)>1 else "less"} than 1 file: %s", file_list) + result_output["error"].append(glob) + 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_hours= elapsed_time.total_seconds // 3600 - elapsed_minutes = (elapsed_time.total_seconds % 3600) // 60 - elapsed_seconds = elapsed_time.total_seconds % 60 - f"Total elapsed time: {elapsed_hours}h {elapsed_minutes}min {elapsed_seconds}s" - result_output.append() \ No newline at end of file + elapsed_time = finish_dt-start_dt + elapsed_total_seconds = elapsed_time.total_seconds() + elapsed_hours= elapsed_total_seconds // 3600 + elapsed_minutes = (elapsed_total_seconds % 3600) // 60 + elapsed_seconds = elapsed_total_seconds % 60 + 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"]) + ) \ No newline at end of file