From 4364276a5bd0090e76c21941d436cb4e500e253a Mon Sep 17 00:00:00 2001 From: Artem Akymenko Date: Wed, 15 Jul 2026 11:56:06 +0000 Subject: [PATCH] refactor: extract output path utilities to domain/output_paths.py MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Extract slugify, sanitize_output_stem, output_timestamp_token, build_output_path - Extract apply_newline_policy, resolve_output_directory, resolve_project_layout - _prepare_output_dir and _prepare_project_layout become thin wrappers with mkdir - Add tests/test_output_paths.py (21 tests) - conversion_runner.py: 1443 → 1410 lines - All tests pass --- abogen/domain/output_paths.py | 91 +++++++++ abogen/webui/conversion_runner.py | 77 +++----- tests/test_output_paths.py | 306 +++++++++++++++++++++++------- 3 files changed, 355 insertions(+), 119 deletions(-) create mode 100644 abogen/domain/output_paths.py diff --git a/abogen/domain/output_paths.py b/abogen/domain/output_paths.py new file mode 100644 index 0000000..524c202 --- /dev/null +++ b/abogen/domain/output_paths.py @@ -0,0 +1,91 @@ +"""Output path resolution utilities. + +Pure functions for resolving output directories, building file paths, +and computing project folder layouts. +""" + +from __future__ import annotations + +import re +from datetime import datetime +from pathlib import Path +from typing import Any, Callable, List, Optional, Tuple + +from abogen.text_extractor import ExtractedChapter + + +_OUTPUT_SANITIZE_RE = re.compile(r"[^\w\-_.]+") + + +def slugify(title: str, index: int) -> str: + sanitized = re.sub(r"[^\w\-]+", "_", title.lower()).strip("_") + if not sanitized: + sanitized = f"chapter_{index:02d}" + return sanitized[:80] + + +def sanitize_output_stem(name: str) -> str: + base = Path(name or "").stem + sanitized = _OUTPUT_SANITIZE_RE.sub("_", base).strip("_") + return sanitized or "output" + + +def output_timestamp_token() -> str: + return datetime.now().strftime("%Y%m%d-%H%M%S") + + +def build_output_path(directory: Path, original_name: str, extension: str) -> Path: + sanitized = sanitize_output_stem(original_name) + return directory / f"{sanitized}.{extension}" + + +def apply_newline_policy(chapters: List[ExtractedChapter], replace_single_newlines: bool) -> None: + if not replace_single_newlines: + return + newline_regex = re.compile(r"(? Path: + if save_mode == "Save to Desktop" and desktop_dir: + return desktop_dir + if save_mode == "Save next to input file": + return stored_path.parent + if save_mode == "Choose output folder" and output_folder: + return Path(output_folder) + if save_mode == "Use default save location" and user_output_path: + return user_output_path + return user_cache_outputs or Path(".") + + +def resolve_project_layout( + *, + original_filename: str, + save_as_project: bool, + base_dir: Path, + timestamp_fn: Callable[[], str] = output_timestamp_token, + sanitize_fn: Callable[[str, int], str] = sanitize_output_stem, +) -> Tuple[Path, Path, Path, Optional[Path]]: + sanitized = sanitize_fn(original_filename, 0) + folder_name = f"{timestamp_fn()}_{sanitized}" + project_root = base_dir / folder_name + project_root.mkdir(parents=True, exist_ok=True) + + if save_as_project: + audio_dir = project_root / "audio" + subtitle_dir = project_root / "subtitles" + metadata_dir = project_root / "metadata" + for directory in (audio_dir, subtitle_dir, metadata_dir): + directory.mkdir(parents=True, exist_ok=True) + return project_root, audio_dir, subtitle_dir, metadata_dir + + return project_root, project_root, project_root, None diff --git a/abogen/webui/conversion_runner.py b/abogen/webui/conversion_runner.py index 32862e8..8c8c6ae 100644 --- a/abogen/webui/conversion_runner.py +++ b/abogen/webui/conversion_runner.py @@ -102,6 +102,15 @@ from abogen.domain.voice_utils import ( infer_provider_from_spec as _infer_provider_from_spec, coerce_truthy as _coerce_truthy, ) +from abogen.domain.output_paths import ( + slugify as _slugify, + sanitize_output_stem as _sanitize_output_stem, + output_timestamp_token as _output_timestamp_token, + build_output_path as _build_output_path, + apply_newline_policy as _apply_newline_policy, + resolve_output_directory as _resolve_output_directory, + resolve_project_layout as _resolve_project_layout, +) from .service import Job, JobStatus @@ -1250,67 +1259,25 @@ def _prepare_output_dir(job: Job) -> Path: from platformdirs import user_desktop_dir # type: ignore[import-not-found] default_output = Path(str(get_user_cache_path("outputs"))) - if job.save_mode == "Save to Desktop": - directory = Path(user_desktop_dir()) - elif job.save_mode == "Save next to input file": - directory = job.stored_path.parent - elif job.save_mode == "Choose output folder" and job.output_folder: - directory = Path(job.output_folder) - elif job.save_mode == "Use default save location": - directory = Path(get_user_output_path()) - else: - directory = default_output + directory = _resolve_output_directory( + save_mode=job.save_mode, + stored_path=job.stored_path, + output_folder=getattr(job, "output_folder", None), + desktop_dir=Path(user_desktop_dir()), + user_output_path=Path(get_user_output_path()), + user_cache_outputs=default_output, + ) directory.mkdir(parents=True, exist_ok=True) return directory -def _build_output_path(directory: Path, original_name: str, extension: str) -> Path: - sanitized = _sanitize_output_stem(original_name) - directory.mkdir(parents=True, exist_ok=True) - return directory / f"{sanitized}.{extension}" - - def _prepare_project_layout(job: Job, base_dir: Path) -> tuple[Path, Path, Path, Optional[Path]]: base_dir.mkdir(parents=True, exist_ok=True) - sanitized = _sanitize_output_stem(job.original_filename) - folder_name = f"{_output_timestamp_token()}_{sanitized}" - project_root = base_dir / folder_name - project_root.mkdir(parents=True, exist_ok=True) - - if job.save_as_project: - audio_dir = project_root / "audio" - subtitle_dir = project_root / "subtitles" - metadata_dir = project_root / "metadata" - for directory in (audio_dir, subtitle_dir, metadata_dir): - directory.mkdir(parents=True, exist_ok=True) - return project_root, audio_dir, subtitle_dir, metadata_dir - - return project_root, project_root, project_root, None - - -def _apply_newline_policy(chapters: List[ExtractedChapter], replace_single_newlines: bool) -> None: - if not replace_single_newlines: - return - newline_regex = re.compile(r"(? str: - sanitized = re.sub(r"[^\w\-]+", "_", title.lower()).strip("_") - if not sanitized: - sanitized = f"chapter_{index:02d}" - return sanitized[:80] - - -def _sanitize_output_stem(name: str) -> str: - base = Path(name or "").stem - sanitized = _OUTPUT_SANITIZE_RE.sub("_", base).strip("_") - return sanitized or "output" - - -def _output_timestamp_token() -> str: - return datetime.now().strftime("%Y%m%d-%H%M%S") + return _resolve_project_layout( + original_filename=job.original_filename, + save_as_project=job.save_as_project, + base_dir=base_dir, + ) def _open_audio_sink( diff --git a/tests/test_output_paths.py b/tests/test_output_paths.py index 77e8992..dc4089c 100644 --- a/tests/test_output_paths.py +++ b/tests/test_output_paths.py @@ -1,79 +1,257 @@ +"""Tests for output path utilities. + +Tests import from domain/output_paths.py (new module). +""" + from __future__ import annotations -import time +import re +from datetime import datetime from pathlib import Path +from types import SimpleNamespace import pytest -from abogen.webui.conversion_runner import _build_output_path, _prepare_project_layout -from abogen.webui.service import Job + +# --------------------------------------------------------------------------- +# slugify +# --------------------------------------------------------------------------- -def _sample_job(tmp_path: Path) -> Job: - source = tmp_path / "sample.txt" - source.write_text("example", encoding="utf-8") - return Job( - id="job-1", - original_filename="Sample Title.txt", - stored_path=source, - language="en", - voice="af_alloy", - speed=1.0, - use_gpu=False, - subtitle_mode="Sentence", - output_format="mp3", - save_mode="Use default save location", - output_folder=tmp_path, - replace_single_newlines=False, - subtitle_format="srt", - created_at=time.time(), - ) +class TestSlugify: + """slugify converts title to filesystem-safe slug.""" + + def test_basic_slug(self): + from abogen.domain.output_paths import slugify + + assert slugify("Hello World", 0) == "hello_world" + + def test_strips_special_chars(self): + from abogen.domain.output_paths import slugify + + result = slugify("Chapter 1: The Beginning!", 0) + assert result == "chapter_1_the_beginning" + + def test_empty_title_uses_index(self): + from abogen.domain.output_paths import slugify + + assert slugify("", 5) == "chapter_05" + + def test_truncated_at_80(self): + from abogen.domain.output_paths import slugify + + long_title = "a" * 100 + assert len(slugify(long_title, 0)) == 80 + + def test_preserves_hyphens(self): + from abogen.domain.output_paths import slugify + + assert slugify("mid-night", 0) == "mid-night" -def test_prepare_project_layout_uses_timestamped_folder( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path -) -> None: - job = _sample_job(tmp_path) - monkeypatch.setattr( - "abogen.webui.conversion_runner._output_timestamp_token", - lambda: "20250101-120000", - ) - - project_root, audio_dir, subtitle_dir, metadata_dir = _prepare_project_layout( - job, tmp_path - ) - - assert project_root.name.startswith( - "20250101-120000_Sample_Title" - ), project_root.name - assert audio_dir == project_root - assert subtitle_dir == project_root - assert metadata_dir is None - - output_path = _build_output_path(audio_dir, job.original_filename, "mp3") - assert output_path == project_root / "Sample_Title.mp3" +# --------------------------------------------------------------------------- +# sanitize_output_stem +# --------------------------------------------------------------------------- -def test_prepare_project_layout_creates_project_subdirs( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path -) -> None: - job = _sample_job(tmp_path) - job.save_as_project = True - monkeypatch.setattr( - "abogen.webui.conversion_runner._output_timestamp_token", - lambda: "20250101-120500", - ) +class TestSanitizeOutputStem: + """sanitize_output_stem cleans filename stem.""" - project_root, audio_dir, subtitle_dir, metadata_dir = _prepare_project_layout( - job, tmp_path - ) + def test_basic_sanitize(self): + from abogen.domain.output_paths import sanitize_output_stem - assert audio_dir == project_root / "audio" - assert subtitle_dir == project_root / "subtitles" - assert metadata_dir == project_root / "metadata" - assert audio_dir.is_dir() - assert subtitle_dir.is_dir() - assert metadata_dir is not None and metadata_dir.is_dir() + assert sanitize_output_stem("my file.mp3") == "my_file" - output_path = _build_output_path(audio_dir, job.original_filename, "wav") - assert output_path == audio_dir / "Sample_Title.wav" + def test_empty_returns_default(self): + from abogen.domain.output_paths import sanitize_output_stem + + assert sanitize_output_stem("") == "output" + + def test_strips_underscores(self): + from abogen.domain.output_paths import sanitize_output_stem + + assert sanitize_output_stem("__test__") == "test" + + +# --------------------------------------------------------------------------- +# output_timestamp_token +# --------------------------------------------------------------------------- + + +class TestOutputTimestampToken: + """output_timestamp_token returns timestamp string.""" + + def test_format(self): + from abogen.domain.output_paths import output_timestamp_token + + token = output_timestamp_token() + assert re.match(r"\d{8}-\d{6}", token) + + +# --------------------------------------------------------------------------- +# build_output_path +# --------------------------------------------------------------------------- + + +class TestBuildOutputPath: + """build_output_path builds the output file path.""" + + def test_basic_path(self, tmp_path): + from abogen.domain.output_paths import build_output_path + + result = build_output_path(tmp_path, "test.mp3", "mp3") + assert result.suffix == ".mp3" + assert result.parent == tmp_path + + def test_stem_sanitized(self, tmp_path): + from abogen.domain.output_paths import build_output_path + + result = build_output_path(tmp_path, "my file.txt", "wav") + assert "my_file" in result.name + + +# --------------------------------------------------------------------------- +# apply_newline_policy +# --------------------------------------------------------------------------- + + +class TestApplyNewlinePolicy: + """apply_newline_policy replaces single newlines in chapter text.""" + + def test_noop_when_disabled(self): + from abogen.domain.output_paths import apply_newline_policy + + from abogen.text_extractor import ExtractedChapter + chapters = [ExtractedChapter(title="t", text="a\nb")] + apply_newline_policy(chapters, False) + assert chapters[0].text == "a\nb" + + def test_replaces_single_newlines(self): + from abogen.domain.output_paths import apply_newline_policy + + from abogen.text_extractor import ExtractedChapter + chapters = [ExtractedChapter(title="t", text="a\nb\nc")] + apply_newline_policy(chapters, True) + assert chapters[0].text == "a b c" + + def test_preserves_double_newlines(self): + from abogen.domain.output_paths import apply_newline_policy + + from abogen.text_extractor import ExtractedChapter + chapters = [ExtractedChapter(title="t", text="a\n\nb")] + apply_newline_policy(chapters, True) + assert chapters[0].text == "a\n\nb" + + +# --------------------------------------------------------------------------- +# resolve_output_directory +# --------------------------------------------------------------------------- + + +class TestResolveOutputDirectory: + """resolve_output_directory determines output dir from save_mode.""" + + def test_save_to_desktop(self, tmp_path): + from abogen.domain.output_paths import resolve_output_directory + + result = resolve_output_directory( + save_mode="Save to Desktop", + stored_path=Path("/input/book.epub"), + output_folder=None, + desktop_dir=tmp_path, + user_output_path=None, + user_cache_outputs=None, + ) + assert result == tmp_path + + def test_save_next_to_input(self, tmp_path): + from abogen.domain.output_paths import resolve_output_directory + + stored = tmp_path / "book.epub" + result = resolve_output_directory( + save_mode="Save next to input file", + stored_path=stored, + output_folder=None, + desktop_dir=None, + user_output_path=None, + user_cache_outputs=None, + ) + assert result == tmp_path + + def test_choose_output_folder(self, tmp_path): + from abogen.domain.output_paths import resolve_output_directory + + custom = tmp_path / "custom" + result = resolve_output_directory( + save_mode="Choose output folder", + stored_path=Path("/x/y.epub"), + output_folder=str(custom), + desktop_dir=None, + user_output_path=None, + user_cache_outputs=None, + ) + assert result == custom + + def test_use_default_save_location(self, tmp_path): + from abogen.domain.output_paths import resolve_output_directory + + result = resolve_output_directory( + save_mode="Use default save location", + stored_path=Path("/x/y.epub"), + output_folder=None, + desktop_dir=None, + user_output_path=tmp_path / "default", + user_cache_outputs=None, + ) + assert result == tmp_path / "default" + + def test_fallback_to_cache(self, tmp_path): + from abogen.domain.output_paths import resolve_output_directory + + result = resolve_output_directory( + save_mode="unknown", + stored_path=Path("/x/y.epub"), + output_folder=None, + desktop_dir=None, + user_output_path=None, + user_cache_outputs=tmp_path / "cache", + ) + assert result == tmp_path / "cache" + + +# --------------------------------------------------------------------------- +# resolve_project_layout +# --------------------------------------------------------------------------- + + +class TestResolveProjectLayout: + """resolve_project_layout computes project folder structure.""" + + def test_flat_layout(self, tmp_path): + from abogen.domain.output_paths import resolve_project_layout + + root, audio, subs, meta = resolve_project_layout( + original_filename="book.epub", + save_as_project=False, + base_dir=tmp_path, + timestamp_fn=lambda: "20260101-000000", + sanitize_fn=lambda n, i: "book", + ) + assert audio == root + assert subs == root + assert meta is None + + def test_project_layout(self, tmp_path): + from abogen.domain.output_paths import resolve_project_layout + + root, audio, subs, meta = resolve_project_layout( + original_filename="book.epub", + save_as_project=True, + base_dir=tmp_path, + timestamp_fn=lambda: "20260101-000000", + sanitize_fn=lambda n, i: "book", + ) + assert root.name == "20260101-000000_book" + assert audio.name == "audio" + assert subs.name == "subtitles" + assert meta.name == "metadata"