mirror of
https://github.com/denizsafak/abogen.git
synced 2026-07-18 13:40:27 +02:00
refactor: extract output path utilities to domain/output_paths.py
- 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
This commit is contained in:
Artem Akymenko
@@ -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"(?<!\n)\n(?!\n)")
|
||||
for chapter in chapters:
|
||||
chapter.text = newline_regex.sub(" ", chapter.text)
|
||||
|
||||
|
||||
def resolve_output_directory(
|
||||
*,
|
||||
save_mode: str,
|
||||
stored_path: Path,
|
||||
output_folder: Optional[str],
|
||||
desktop_dir: Optional[Path],
|
||||
user_output_path: Optional[Path],
|
||||
user_cache_outputs: Optional[Path],
|
||||
) -> 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
|
||||
@@ -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"(?<!\n)\n(?!\n)")
|
||||
for chapter in chapters:
|
||||
chapter.text = newline_regex.sub(" ", chapter.text)
|
||||
|
||||
|
||||
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")
|
||||
return _resolve_project_layout(
|
||||
original_filename=job.original_filename,
|
||||
save_as_project=job.save_as_project,
|
||||
base_dir=base_dir,
|
||||
)
|
||||
|
||||
|
||||
def _open_audio_sink(
|
||||
|
||||
+242
-64
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user