From 53b850ef4132992e65368ceeab61bb9033105f23 Mon Sep 17 00:00:00 2001 From: Artem Akymenko Date: Tue, 21 Jul 2026 07:02:03 +0000 Subject: [PATCH] feat: unified conversion planner (build_conversion_plan) Pure function that takes ConversionRequest -> ConversionPlan. Handles chapter parsing, voice markers, chunks, intro/outro, output layout. Replaces duplicated planning logic in both PyQt and WebUI runners. --- abogen/application/conversion_planner.py | 400 +++++++++++++++++++++++ 1 file changed, 400 insertions(+) create mode 100644 abogen/application/conversion_planner.py diff --git a/abogen/application/conversion_planner.py b/abogen/application/conversion_planner.py new file mode 100644 index 0000000..4f60737 --- /dev/null +++ b/abogen/application/conversion_planner.py @@ -0,0 +1,400 @@ +"""Unified conversion planner. + +Pure functions that take a ConversionRequest and produce a ConversionPlan. +No side effects, no I/O — all complexity from both UIs in one place. + +This is Stage 2 of the conversion flow unification plan. +""" + +from __future__ import annotations + +import os +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +from abogen.application.conversion_models import ( + ChapterPlan, + ConversionPlan, + IntroOutroSpec, + OutputLayout, + SegmentPlan, +) +from abogen.application.conversion_request import ConversionRequest +from abogen.domain.chapter_overrides import apply_chapter_overrides +from abogen.domain.file_type import auto_select_relevant_chapters +from abogen.domain.intro_outro import resolve_intro, resolve_outro +from abogen.domain.metadata_extraction import extract_metadata_for_file +from abogen.domain.metadata_merge import merge_metadata +from abogen.domain.output_paths import ( + resolve_output_directory, + resolve_project_layout, + resolve_unique_path, + sanitize_output_stem, +) +from abogen.subtitle_utils import split_text_by_voice_markers + + +def build_conversion_plan(request: ConversionRequest) -> ConversionPlan: + """Build a complete conversion plan from a request. + + This is the single entry point that both UIs will call. + It handles all the planning logic that was previously duplicated + in both PyQt and WebUI conversion runners. + + Args: + request: Normalized conversion request + + Returns: + ConversionPlan with all chapters, segments, and output layout + + Raises: + ValueError: If request is invalid (no source, no chapters, etc.) + """ + # 1. Extract and validate source + source_text = _extract_source_text(request) + if not source_text or not source_text.strip(): + raise ValueError("No text content to convert") + + # 2. Extract metadata + metadata = _extract_metadata(request) + + # 3. Parse chapters + raw_chapters = _parse_chapters(source_text, request) + + # 4. Apply chapter selection/overrides + selected_chapters = _apply_selection(raw_chapters, request) + + # 5. Build segments for each chapter + chapters = _build_chapters(selected_chapters, request) + + # 6. Build intro/outro + intro, outro = _build_intro_outro(metadata, request) + + # 7. Resolve output layout + output_layout = _resolve_output_layout(request) + + return ConversionPlan( + request=request, + metadata=metadata, + chapters=chapters, + intro=intro, + outro=outro, + output_layout=output_layout, + ) + + +def _extract_source_text(request: ConversionRequest) -> Optional[str]: + """Extract text from request source.""" + if request.direct_text: + return request.direct_text + if request.source_path and request.source_path.exists(): + from abogen.subtitle_utils import clean_text + + encoding = "utf-8" + try: + with open(request.source_path, "r", encoding=encoding, errors="replace") as f: + text = f.read() + except Exception: + return None + return clean_text(text) + return None + + +def _extract_metadata(request: ConversionRequest) -> Dict[str, Any]: + """Extract metadata from source file.""" + if request.direct_text: + return dict(request.metadata_tags) + + if request.source_path and request.source_path.exists(): + try: + extraction = extract_metadata_for_file( + str(request.source_path), is_direct_text=False + ) + metadata = dict(extraction.metadata) if extraction.metadata else {} + except Exception: + metadata = {} + metadata = merge_metadata(metadata, request.metadata_tags) + return metadata + + return dict(request.metadata_tags) + + +def _parse_chapters( + source_text: str, request: ConversionRequest +) -> List[Tuple[str, str, str]]: + """Parse source text into raw chapters. + + Returns list of (title, body_text, default_voice) tuples. + """ + from abogen.domain.text_chapters import parse_chapters_from_text + + # For direct text, use the text as-is + if request.direct_text: + chapters = parse_chapters_from_text(source_text, default_title="text", clean=False) + else: + chapters = parse_chapters_from_text(source_text, default_title="text", clean=False) + + # Default voice from request + default_voice = request.voice or "M1" + + return [(title, text, default_voice) for title, text in chapters] + + +def _apply_selection( + raw_chapters: List[Tuple[str, str, str]], request: ConversionRequest +) -> List[Tuple[str, str, str]]: + """Apply chapter selection and overrides.""" + from abogen.text_extractor import ExtractedChapter + + # Convert to ExtractedChapter objects for auto_select_relevant_chapters + extracted = [ + ExtractedChapter(title=title, text=text) + for title, text, _ in raw_chapters + ] + + # If user specified chapters, apply overrides + if request.chapter_overrides: + selected, _, diagnostics = apply_chapter_overrides(extracted, request.chapter_overrides) + if selected: + # Map back to (title, text, voice) tuples + result = [] + for ch in selected: + # Find matching original chapter to get voice + voice = request.voice or "M1" + for orig_title, orig_text, orig_voice in raw_chapters: + if orig_title == ch.title: + voice = orig_voice + break + result.append((ch.title, ch.text or "", voice)) + return result + # If no chapters selected, fall through to auto-selection + + # Auto-select relevant chapters + from abogen.domain.file_type import infer_file_type + + file_type = infer_file_type(request.source_path) if request.source_path else "text" + result = auto_select_relevant_chapters(extracted, file_type) + filtered = result.kept + + if filtered: + # Map back to (title, text, voice) tuples + result = [] + for ch in filtered: + voice = request.voice or "M1" + for orig_title, orig_text, orig_voice in raw_chapters: + if orig_title == ch.title: + voice = orig_voice + break + result.append((ch.title, ch.text or "", voice)) + return result + + # Fall back to all chapters + return raw_chapters + + +def _build_chapters( + selected_chapters: List[Tuple[str, str, str]], request: ConversionRequest +) -> List[ChapterPlan]: + """Build ChapterPlan with SegmentPlan for each chapter.""" + chapters = [] + + for idx, (title, body_text, default_voice) in enumerate(selected_chapters, 1): + # Build segments for this chapter + segments = _build_segments(body_text, default_voice, request) + + chapter = ChapterPlan( + index=idx, + title=title, + original_title=title, + body_text=body_text, + segments=segments, + voice_spec=default_voice, + ) + chapters.append(chapter) + + return chapters + + +def _build_segments( + body_text: str, default_voice: str, request: ConversionRequest +) -> List[SegmentPlan]: + """Build SegmentPlan list for a chapter's body text. + + Handles voice markers (PyQt) and chunks (WebUI). + """ + segments = [] + + # Check for chunks (WebUI style) + if request.chunks: + # Group chunks by chapter (simplified — assume chunks are for current chapter) + for chunk_idx, chunk in enumerate(request.chunks): + chunk_text = chunk.get("normalized_text") or chunk.get("text", "") + if not chunk_text or not chunk_text.strip(): + continue + + chunk_voice = _resolve_chunk_voice(chunk, default_voice, request) + speaker_id = chunk.get("speaker_id", "narrator") + + segments.append( + SegmentPlan( + text=chunk_text.strip(), + voice_spec=chunk_voice, + kind="body", + speaker_id=speaker_id, + chunk_id=chunk.get("id"), + chunk_index=chunk.get("chunk_index", chunk_idx), + level=chunk.get("level", request.chunk_level), + source="chunk", + ) + ) + return segments + + # Check for voice markers (PyQt style) + # Detect markers even if validation fails (voice names may not be loaded yet) + from abogen.subtitle_utils import _VOICE_MARKER_SEARCH_PATTERN + + has_voice_markers = bool(_VOICE_MARKER_SEARCH_PATTERN.search(body_text)) + voice_segments, last_voice, valid_count, invalid_count = split_text_by_voice_markers( + body_text, default_voice + ) + + if has_voice_markers or (len(voice_segments) > 1): + # Voice markers were used + for voice_name, segment_text in voice_segments: + if not segment_text or not segment_text.strip(): + continue + segments.append( + SegmentPlan( + text=segment_text.strip(), + voice_spec=voice_name, + kind="body", + source="voice_marker", + ) + ) + return segments + + # No voice markers — single segment for entire body + if body_text and body_text.strip(): + segments.append( + SegmentPlan( + text=body_text.strip(), + voice_spec=default_voice, + kind="body", + source="chapter", + ) + ) + + return segments + + +def _resolve_chunk_voice( + chunk: Dict[str, Any], default_voice: str, request: ConversionRequest +) -> str: + """Resolve voice for a chunk.""" + # Check for speaker-based voice + speaker_id = chunk.get("speaker_id", "narrator") + if speaker_id and speaker_id != "narrator" and request.speakers: + speaker_config = request.speakers.get(speaker_id, {}) + if isinstance(speaker_config, dict): + voice = speaker_config.get("voice") + if voice: + return voice + + # Check for direct voice field + voice = chunk.get("voice") + if voice: + return voice + + return default_voice + + +def _build_intro_outro( + metadata: Dict[str, Any], request: ConversionRequest +) -> Tuple[Optional[IntroOutroSpec], Optional[IntroOutroSpec]]: + """Build intro and outro specs.""" + intro_spec = None + outro_spec = None + + # Intro + if request.read_title_intro: + resolved = resolve_intro( + metadata, + request.original_filename, + True, + request.voice or "M1", + request.voice or "M1", + [], + ) + if resolved.enabled: + intro_spec = IntroOutroSpec( + enabled=True, + text=resolved.text, + voice_spec=resolved.voice_spec, + kind="intro", + ) + + # Outro + if request.read_closing_outro: + resolved = resolve_outro( + metadata, + request.original_filename, + True, + request.voice or "M1", + request.voice or "M1", + [], + ) + if resolved.enabled: + outro_spec = IntroOutroSpec( + enabled=True, + text=resolved.text, + voice_spec=resolved.voice_spec, + kind="outro", + ) + + return intro_spec, outro_spec + + +def _resolve_output_layout(request: ConversionRequest) -> OutputLayout: + """Resolve output paths for the conversion.""" + # Determine base output directory + if request.save_mode == "custom_folder" and request.output_folder: + parent_dir = Path(request.output_folder) + elif request.source_path: + parent_dir = request.source_path.parent + else: + parent_dir = Path.cwd() + + # Determine base name + if request.original_filename: + base_name = sanitize_output_stem(request.original_filename) + elif request.source_path: + base_name = sanitize_output_stem(request.source_path.stem) + else: + base_name = "output" + + # Find unique output path + allowed_exts = {request.output_format, "srt", "ass", "vtt", "mp4", "m4b"} + unique_base = resolve_unique_path( + parent_dir, base_name, "", allowed_extensions=allowed_exts + ) + + # Resolve project layout + project_root = None + audio_dir = parent_dir + subtitle_dir = None + metadata_dir = None + + if request.save_as_project: + project_root, audio_dir, subtitle_dir, metadata_dir = resolve_project_layout( + original_filename=request.original_filename, + save_as_project=True, + base_dir=parent_dir, + ) + + return OutputLayout( + parent_dir=parent_dir, + project_root=project_root, + audio_dir=audio_dir, + subtitle_dir=subtitle_dir, + metadata_dir=metadata_dir, + )