feat: added colorful logging using rich, improved startup times by lazy-loading spacy, some other fixes

.gitignore

@@ -40,3 +40,6 @@ test_assets/
 dev_notes/
 .claude/
 .coverage
+
+# CodeGraph index (local, machine-specific)
+.codegraph/

abogen/entity_analysis.py

@@ -9,15 +9,26 @@ from collections import Counter
 from dataclasses import dataclass, field
 from typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple
 
-try:  # pragma: no cover - fallback when spaCy not available during tests
-    import spacy  # type: ignore[import-not-found]
-except Exception:  # pragma: no cover - spaCy optional during runtime bootstrap
-    spacy = None
-
 _Language = Any  # type: ignore[misc,assignment]
 Doc = Any  # type: ignore[misc,assignment]
 Span = Any  # type: ignore[misc,assignment]
 
+_SPACY: Any = None
+_SPACY_LOADED = False
+
+
+def _get_spacy() -> Any:
+    """Import spaCy lazily (it pulls in torch/thinc, ~2s at startup)."""
+    global _SPACY, _SPACY_LOADED
+    if not _SPACY_LOADED:
+        _SPACY_LOADED = True
+        try:  # pragma: no cover - fallback when spaCy not available during tests
+            import spacy  # type: ignore[import-not-found]
+        except Exception:  # pragma: no cover - spaCy optional during runtime bootstrap
+            spacy = None
+        _SPACY = spacy
+    return _SPACY
+
 
 _TITLE_PREFIXES = (
     "mr",
@@ -167,6 +178,7 @@ def _resolve_model_name(language: str) -> str:
 
 
 def _load_model(language: str) -> Any:
+    spacy = _get_spacy()
     if spacy is None:
         raise EntityModelError(
             "spaCy is not available. Install spaCy to enable entity extraction."

abogen/heteronym_overrides.py

@@ -5,10 +5,21 @@ import re
 from dataclasses import dataclass
 from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple
 
-try:  # pragma: no cover - optional dependency
+_SPACY: Any = None
+_SPACY_LOADED = False
+
+
+def _get_spacy() -> Any:
+    """Import spaCy lazily (it pulls in torch/thinc, ~2s at startup)."""
+    global _SPACY, _SPACY_LOADED
+    if not _SPACY_LOADED:
+        _SPACY_LOADED = True
+        try:  # pragma: no cover - optional dependency
             import spacy  # type: ignore
-except Exception:  # pragma: no cover - spaCy may be unavailable in minimal environments
+        except Exception:  # pragma: no cover - spaCy may be unavailable in minimal environments
             spacy = None
+        _SPACY = spacy
+    return _SPACY
 
 
 @dataclass(frozen=True)
@@ -184,6 +195,7 @@ def _build_replacement_sentence(
 
 
 def _load_spacy(language: str) -> Any:
+    spacy = _get_spacy()
     if spacy is None:
         return None
 
@@ -221,7 +233,7 @@ def extract_heteronym_overrides(
     if not lang.startswith("en"):
         return []
 
-    if spacy is None:
+    if _get_spacy() is None:
         return []
 
     nlp = _load_spacy(lang)

abogen/pyqt/book_handler.py

@@ -45,9 +45,9 @@ import urllib.parse
 import textwrap
 
 # Setup logging
-logging.basicConfig(
-    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
-)
+from abogen.utils import setup_console_logging
+
+setup_console_logging()
 
 _HTML_TAG_PATTERN = re.compile(r"<[^>]+>")
 _LEADING_DASH_PATTERN = re.compile(r"^\s*[-–—]\s*")

abogen/pyqt/gui.py

@@ -988,6 +988,11 @@ class abogen(QWidget):
         self.queued_items = []
         self.current_queue_index = 0
 
+        from abogen.utils import timed_log
+        import logging
+        _startup_log = logging.getLogger("abogen.startup")
+
+        with timed_log("GUI initUI (widget building)", logger=_startup_log):
             self.initUI()
         self.speed_slider.setValue(int(self.config.get("speed", _d["speed"]) * 100))
         self.update_speed_label()
@@ -1003,6 +1008,7 @@ class abogen(QWidget):
             if self.selected_profile_name:
                 from abogen.voice_profiles import load_profiles
 
+                with timed_log("voice profile load", logger=_startup_log):
                     entry = load_profiles().get(self.selected_profile_name, {})
                 if isinstance(entry, dict):
                     self.mixed_voice_state = entry.get("voices", [])

abogen/pyqt/main.py

@@ -1,3 +1,4 @@
+import logging
 import os
 import sys
 import platform
@@ -6,8 +7,14 @@ import platform
 from abogen import shutdown  # noqa: F401
 shutdown.register_shutdown()
 
+from abogen.utils import get_resource_path, setup_console_logging, timed_log  # noqa: E402
+
+_log = logging.getLogger("abogen.startup")
+setup_console_logging()
+
 # Fix PyTorch DLL loading issue ([WinError 1114]) on Windows before importing PyQt6
 if platform.system() == "Windows":
+    with timed_log("PyTorch DLLs (Windows)", logger=_log):
         import ctypes
         from importlib.util import find_spec
 
@@ -25,7 +32,8 @@ if platform.system() == "Windows":
 
 
 # Qt platform plugin detection (fixes #59)
-try:
+with timed_log("Qt platform plugin detection", logger=_log):
+    try:
         from PyQt6.QtCore import QLibraryInfo
 
         # Get the path to the plugins directory
@@ -39,17 +47,16 @@ try:
 
         if os.path.isdir(platform_dir):
             os.environ["QT_QPA_PLATFORM_PLUGIN_PATH"] = platform_dir
-        print("QT_QPA_PLATFORM_PLUGIN_PATH set to:", platform_dir)
+            _log.info("QT_QPA_PLATFORM_PLUGIN_PATH set to: %s", platform_dir)
         else:
-        print("PyQt6 platform plugins not found at", platform_dir)
-except ImportError:
-    print("PyQt6 not installed.")
+            _log.warning("PyQt6 platform plugins not found at %s", platform_dir)
+    except ImportError:
+        _log.warning("PyQt6 not installed.")
 
 
-from abogen.utils import get_resource_path
-
 # Pre-load "libxcb-cursor" on Linux (fixes #101)
 if platform.system() == "Linux":
+    with timed_log("libxcb-cursor preload (Linux)", logger=_log):
         arch = platform.machine().lower()
         lib_filename = {"x86_64": "libxcb-cursor-amd64.so.0", "amd64": "libxcb-cursor-amd64.so.0", "aarch64": "libxcb-cursor-arm64.so.0", "arm64": "libxcb-cursor-arm64.so.0"}.get(arch)
         if lib_filename:
@@ -71,6 +78,7 @@ if platform.system() == "Linux":
 
 # Set application ID for Windows taskbar icon
 if platform.system() == "Windows":
+    with timed_log("Windows AppUserModelID", logger=_log):
         try:
             from abogen.constants import PROGRAM_NAME, VERSION
             import ctypes
@@ -78,31 +86,34 @@ if platform.system() == "Windows":
             app_id = f"{PROGRAM_NAME}.{VERSION}"
             ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID(app_id)
         except Exception as e:
-        print("Warning: failed to set AppUserModelID:", e)
+            _log.warning("Failed to set AppUserModelID: %s", e)
 
-from PyQt6.QtWidgets import QApplication
-from PyQt6.QtGui import QIcon
-from PyQt6.QtCore import (
+with timed_log("PyQt6 imports", logger=_log):
+    from PyQt6.QtWidgets import QApplication
+    from PyQt6.QtGui import QIcon
+    from PyQt6.QtCore import (
         QLibraryInfo,
         qInstallMessageHandler,
         QtMsgType,
-)
+    )
 
 # Add the directory to Python path
 sys.path.insert(0, os.path.join(os.path.dirname(__file__)))
 
 # Set Hugging Face Hub environment variables
-os.environ["HF_HUB_DISABLE_TELEMETRY"] = "1"  # Disable Hugging Face telemetry
-os.environ["HF_HUB_ETAG_TIMEOUT"] = "10"  # Metadata request timeout (seconds)
-os.environ["HF_HUB_DOWNLOAD_TIMEOUT"] = "10"  # File download timeout (seconds)
-os.environ["HF_HUB_DISABLE_SYMLINKS_WARNING"] = "1"  # Disable symlinks warning
-from abogen.utils import load_config
-if load_config().get("disable_kokoro_internet", False):
-    print("INFO: Kokoro's internet access is disabled.")
+with timed_log("config load + HF env setup", logger=_log):
+    os.environ["HF_HUB_DISABLE_TELEMETRY"] = "1"  # Disable Hugging Face telemetry
+    os.environ["HF_HUB_ETAG_TIMEOUT"] = "10"  # Metadata request timeout (seconds)
+    os.environ["HF_HUB_DOWNLOAD_TIMEOUT"] = "10"  # File download timeout (seconds)
+    os.environ["HF_HUB_DISABLE_SYMLINKS_WARNING"] = "1"  # Disable symlinks warning
+    from abogen.utils import load_config
+    if load_config().get("disable_kokoro_internet", False):
+        _log.info("Kokoro's internet access is disabled.")
         os.environ["HF_HUB_OFFLINE"] = "1"  # Disable Hugging Face Hub internet access
 
-from abogen.pyqt.gui import abogen
-from abogen.constants import PROGRAM_NAME, VERSION
+with timed_log("GUI module import (abogen.pyqt.gui)", logger=_log):
+    from abogen.pyqt.gui import abogen
+    from abogen.constants import PROGRAM_NAME, VERSION
 
 # Set environment variables for AMD ROCm
 os.environ["MIOPEN_FIND_MODE"] = "FAST"
@@ -150,6 +161,7 @@ if platform.system() == "Linux":
 
 def main():
     """Main entry point for console usage."""
+    with timed_log("QApplication creation", logger=_log):
         app = QApplication(sys.argv)
 
     # Set application icon using get_resource_path from utils
@@ -164,8 +176,11 @@ def main():
         except AttributeError:
             pass
 
+    with timed_log("main window construction", logger=_log):
         ex = abogen()
+    with timed_log("window show", logger=_log):
         ex.show()
+    _log.info("App startup complete. Showing window.")
     sys.exit(app.exec())
 
 

abogen/spacy_contraction_resolver.py

@@ -6,10 +6,9 @@ from dataclasses import dataclass
 from functools import lru_cache
 from typing import Any, Dict, Optional, Tuple
 
-try:  # pragma: no cover - optional dependency
-    import spacy
-except Exception:  # pragma: no cover - spaCy unavailable at runtime
-    spacy = None
+# spaCy is intentionally NOT imported at module level: importing it pulls in
+# thinc -> torch, which costs seconds of startup time. It is imported lazily
+# inside _load_spacy_model below.
 
 # Lazy spaCy type hints to avoid a hard dependency at import time.
 Language = Any  # type: ignore[assignment]
@@ -37,7 +36,9 @@ _DEFAULT_MODEL = os.environ.get("ABOGEN_SPACY_MODEL", "en_core_web_sm")
 
 @lru_cache(maxsize=1)
 def _load_spacy_model(model: str = _DEFAULT_MODEL) -> Optional[Language]:
-    if spacy is None:
+    try:  # pragma: no cover - optional dependency
+        import spacy
+    except Exception:  # pragma: no cover - spaCy unavailable at runtime
         logger.debug("spaCy is not installed; skipping contraction disambiguation")
         return None
 

abogen/utils.py

@@ -6,7 +6,9 @@ import re
 import shutil
 import subprocess
 import sys
+import time
 import warnings
+from contextlib import contextmanager
 from threading import Thread
 from typing import Dict, Optional
 
@@ -29,6 +31,125 @@ _load_environment()
 
 warnings.filterwarnings("ignore")
 
+# --- Console log colorization via rich (mirrors AutoSubSync's approach) ---
+
+try:  # rich is a declared dependency, but degrade gracefully if unavailable
+    from rich.console import Console
+    from rich.highlighter import NullHighlighter
+    from rich.logging import RichHandler
+
+    _RICH_AVAILABLE = True
+except Exception:  # pragma: no cover - fallback to plain logging
+    Console = None
+    NullHighlighter = None
+    RichHandler = None
+    _RICH_AVAILABLE = False
+
+
+def _console_supports_color() -> bool:
+    if os.environ.get("NO_COLOR"):
+        return False
+    try:
+        return bool(sys.stderr.isatty())
+    except Exception:
+        return False
+
+
+_RICH_CONSOLE = None
+if Console is not None:
+    try:
+        _RICH_CONSOLE = Console(stderr=True, no_color=not _console_supports_color())
+    except Exception:  # pragma: no cover - defensive
+        _RICH_CONSOLE = None
+
+
+_ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;]*[A-Za-z]")
+
+if RichHandler is not None:
+
+    class RichConsoleHandler(RichHandler):
+        """RichHandler with default settings, except raw ANSI escapes are
+        stripped from messages first (werkzeug colorizes its own log lines
+        when attached to a TTY; without this they render as literal "[36m"
+        fragments)."""
+
+        def emit(self, record):
+            # Werkzeug logs its dev-server banner at INFO but hardcodes a
+            # "WARNING: " prefix into the message text. Promote the record so
+            # the level tag matches the content.
+            try:
+                message = _ANSI_ESCAPE_RE.sub("", record.getMessage())
+            except Exception:  # pragma: no cover - defensive
+                message = ""
+            if record.levelno < logging.WARNING and message.startswith("WARNING: "):
+                record.levelno = logging.WARNING
+                record.levelname = "WARNING"
+            super().emit(record)
+
+        def render_message(self, record, message):
+            message = _ANSI_ESCAPE_RE.sub("", message)
+            if message.startswith("WARNING: "):
+                message = message[len("WARNING: ") :]
+            return super().render_message(record, message)
+
+else:  # pragma: no cover - rich unavailable fallback
+    RichConsoleHandler = None  # type: ignore[assignment, misc]
+
+
+def console_handler(show_level=True):
+    """Build a colored console handler. Rich's RichHandler when available
+    (no timestamps, colored level tags), plain StreamHandler otherwise."""
+    if _RICH_CONSOLE is not None and RichConsoleHandler is not None:
+        return RichConsoleHandler(
+            console=_RICH_CONSOLE,
+            show_path=False,
+            show_time=False,
+            rich_tracebacks=True,
+        )
+    handler = logging.StreamHandler(sys.stderr)
+    prefix = "%(levelname)s - " if show_level else ""
+    handler.setFormatter(logging.Formatter(f"{prefix}%(message)s"))
+    return handler
+
+
+def setup_console_logging(level=logging.INFO):
+    """Configure the root logger once with a colored console handler."""
+    root = logging.getLogger()
+    if not root.handlers:
+        root.addHandler(console_handler())
+    root.setLevel(level)
+
+
+@contextmanager
+def timed_log(label, logger=None, level=logging.INFO):
+    """Context manager that logs the wall-clock time a block of code takes.
+
+    Used to surface which load/startup steps are slow. The elapsed time is
+    colorized: green < 1s, yellow 1-5s, red > 5s.
+    """
+    log = logger or logging.getLogger(__name__)
+    start = time.perf_counter()
+    try:
+        yield
+    finally:
+        elapsed = time.perf_counter() - start
+        if _RICH_AVAILABLE and _RICH_CONSOLE is not None and not _RICH_CONSOLE.no_color:
+            if elapsed >= 5.0:
+                color = "red"
+            elif elapsed >= 1.0:
+                color = "yellow"
+            else:
+                color = "green"
+            log.log(
+                level,
+                "Loaded %s in %s",
+                f"[cyan]{label}[/cyan]",
+                f"[{color}]{elapsed:.2f}s[/{color}]",
+                extra={"markup": True, "highlighter": NullHighlighter()},
+            )
+        else:
+            log.log(level, "Loaded %s in %.2fs", label, elapsed)
+
 
 def detect_encoding(file_path):
     try:
@@ -527,6 +648,10 @@ class LoadPipelineThread(Thread):
         try:
             from abogen.domain.pipeline_factory import create_pipeline_for_job
 
+            with timed_log(
+                f"TTS pipeline (lang={self.lang_code}, gpu={self.use_gpu})",
+                logger=logging.getLogger("abogen.startup"),
+            ):
                 backend = create_pipeline_for_job(
                     "kokoro", language=self.lang_code, use_gpu=self.use_gpu
                 )

abogen/webui/app.py

@@ -9,11 +9,19 @@ from flask import Flask
 
 from abogen import shutdown  # noqa: F401
 shutdown.register_shutdown()
-from abogen.utils import get_user_cache_path, get_user_output_path, get_user_settings_dir
+from abogen.utils import (
+    get_user_cache_path,
+    get_user_output_path,
+    get_user_settings_dir,
+    setup_console_logging,
+    timed_log,
+)
 
 from .conversion_runner import run_conversion_job
 from .service import build_service
 
+_logger = logging.getLogger("abogen.startup")
+
 
 class _SuppressSuccessfulAccessFilter(logging.Filter):
     """Filter out successful (HTTP 200) werkzeug access logs."""
@@ -79,8 +87,10 @@ def _get_secret_key() -> str:
 
 
 def create_app(config: Optional[dict[str, Any]] = None) -> Flask:
+    with timed_log("default directories", logger=_logger):
         uploads_dir, outputs_dir = _default_dirs()
 
+    with timed_log("Flask app creation + config", logger=_logger):
         app = Flask(
             __name__,
             static_folder="static",
@@ -102,6 +112,7 @@ def create_app(config: Optional[dict[str, Any]] = None) -> Flask:
             base_config.update(config)
         app.config.update(base_config)
 
+    with timed_log("conversion service (incl. queue state load)", logger=_logger):
         service = build_service(
             runner=run_conversion_job,
             output_root=Path(app.config["OUTPUT_FOLDER"]),
@@ -109,6 +120,7 @@ def create_app(config: Optional[dict[str, Any]] = None) -> Flask:
         )
     app.extensions["conversion_service"] = service
 
+    with timed_log("blueprint registration", logger=_logger):
         from abogen.webui.routes import (
             main_bp,
             jobs_bp,
@@ -137,6 +149,16 @@ def create_app(config: Optional[dict[str, Any]] = None) -> Flask:
 
 
 def main() -> None:
+    setup_console_logging()
+    # Route Flask's dev-server banner through our logger instead of click.echo.
+    import flask.cli as flask_cli
+
+    def _show_server_banner(debug, app_import_path):
+        _logger.info(" * Serving Flask app %r", app_import_path)
+        _logger.info(" * Debug mode: %s", "on" if debug else "off")
+
+    flask_cli.show_server_banner = _show_server_banner
+
     app = create_app()
     host = os.environ.get("ABOGEN_HOST", "0.0.0.0")
     port = int(os.environ.get("ABOGEN_PORT", "8808"))

abogen/webui/debug_tts_runner.py

@@ -47,6 +47,13 @@ def _resolve_voice_setting(value: str) -> tuple[str, Optional[str], Optional[str
 
 
 def _load_pipeline(language: Language, use_gpu: bool) -> Any:
+    import logging
+    from abogen.utils import timed_log
+
+    with timed_log(
+        f"TTS pipeline (lang={language}, gpu={use_gpu})",
+        logger=logging.getLogger("abogen.startup"),
+    ):
         device = "cpu"
         if use_gpu:
             device = _select_device()

abogen/webui/service.py

@@ -17,7 +17,7 @@ from typing import Any, Callable, Dict, Iterable, List, Optional, Mapping
 from abogen.domain.metadata_helpers import normalize_metadata_map
 
 from abogen.domain.enums import Language
-from abogen.utils import get_internal_cache_path, get_user_settings_dir
+from abogen.utils import console_handler, get_internal_cache_path, get_user_settings_dir
 
 
 
@@ -32,9 +32,7 @@ STATE_VERSION = 8
 
 _JOB_LOGGER = logging.getLogger("abogen.jobs")
 if not _JOB_LOGGER.handlers:
-    handler = logging.StreamHandler(sys.stdout)
-    handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s", "%Y-%m-%d %H:%M:%S"))
-    _JOB_LOGGER.addHandler(handler)
+    _JOB_LOGGER.addHandler(console_handler())
     _JOB_LOGGER.propagate = False
 _JOB_LOGGER.setLevel(logging.DEBUG)
 

pyproject.toml

@@ -50,6 +50,7 @@ dependencies = [
   "num2words>=0.5.13",
   "httpx>=0.27.0",
   "PyQt6>=6.5.0",
+  "rich>=13.0.0",
 ]
 
 classifiers = [

tests/test_pending_job_metadata.py

@@ -4,6 +4,10 @@ from pathlib import Path
 from types import SimpleNamespace
 
 
+_real_routes = sys.modules.get("abogen.webui.routes")
+# Import routes.utils.form without executing routes/__init__.py (which imports
+# every blueprint). Use a temporary namespace package, then restore the real
+# module so later tests can still `from abogen.webui.routes import ...`.
 routes_package = types.ModuleType("abogen.webui.routes")
 routes_package.__path__ = [
     str(Path(__file__).parents[1] / "abogen" / "webui" / "routes")
@@ -15,6 +19,11 @@ from abogen.webui.routes.utils.form import (  # noqa: E402
     load_settings,
 )
 
+if _real_routes is not None:
+    sys.modules["abogen.webui.routes"] = _real_routes
+else:
+    del sys.modules["abogen.webui.routes"]
+
 
 def test_user_metadata_overrides_extraction_fallback(tmp_path: Path) -> None:
     extraction = SimpleNamespace(