abogen/abogen/web/routes.py

from __future__ import annotations

import io
import json
import mimetypes
import os
import re
import threading
import time
import uuid
from pathlib import Path
from typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple, cast

from flask import (
    Blueprint,
    abort,
    current_app,
    jsonify,
    redirect,
    render_template,
    request,
    send_file,
    url_for,
)
from flask.typing import ResponseReturnValue
from werkzeug.utils import secure_filename

import numpy as np
import soundfile as sf
from abogen.constants import (
    LANGUAGE_DESCRIPTIONS,
    SAMPLE_VOICE_TEXTS,
    SUBTITLE_FORMATS,
    SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION,
    SUPPORTED_SOUND_FORMATS,
    VOICES_INTERNAL,
)
from abogen.chunking import ChunkLevel, build_chunks_for_chapters
from abogen.utils import (
    calculate_text_length,
    clean_text,
    get_user_output_path,
    load_config,
    load_numpy_kpipeline,
    save_config,
)
from abogen.voice_profiles import (
    delete_profile,
    duplicate_profile,
    export_profiles_payload,
    import_profiles_data,
    load_profiles,
    normalize_voice_entries,
    remove_profile,
    save_profile,
    save_profiles,
    serialize_profiles,
)

from abogen.voice_formulas import get_new_voice
from abogen.speaker_analysis import analyze_speakers
from abogen.speaker_configs import (
    delete_config,
    get_config,
    list_configs,
    load_configs,
    random_voice,
    save_configs,
    upsert_config,
    slugify_label,
)
from abogen.text_extractor import extract_from_path
from .conversion_runner import SPLIT_PATTERN, SAMPLE_RATE, _select_device, _to_float32
from .service import ConversionService, Job, JobStatus, PendingJob

web_bp = Blueprint("web", __name__)
api_bp = Blueprint("api", __name__)


_preview_pipeline_lock = threading.RLock()
_preview_pipelines: Dict[Tuple[str, str], Any] = {}


_CHUNK_LEVEL_OPTIONS = [
    {"value": "paragraph", "label": "Paragraphs"},
    {"value": "sentence", "label": "Sentences"},
]

_SPEAKER_MODE_OPTIONS = [
    {"value": "single", "label": "Single Speaker"},
    {"value": "multi", "label": "Multi-Speaker"},
]

_CHUNK_LEVEL_VALUES = {option["value"] for option in _CHUNK_LEVEL_OPTIONS}
_SPEAKER_MODE_VALUES = {option["value"] for option in _SPEAKER_MODE_OPTIONS}


_DEFAULT_ANALYSIS_THRESHOLD = 3
def _build_narrator_roster(
    voice: str,
    voice_profile: Optional[str],
    existing: Optional[Mapping[str, Any]] = None,
) -> Dict[str, Any]:
    roster: Dict[str, Any] = {
        "narrator": {
            "id": "narrator",
            "label": "Narrator",
            "voice": voice,
        }
    }
    if voice_profile:
        roster["narrator"]["voice_profile"] = voice_profile
    existing_entry: Optional[Mapping[str, Any]] = None
    if existing is not None:
        existing_entry = existing.get("narrator") if isinstance(existing, Mapping) else None
    if isinstance(existing_entry, Mapping):
        roster_entry = roster["narrator"]
        for key in ("label", "voice", "voice_profile", "voice_formula", "pronunciation"):
            value = existing_entry.get(key)
            if value is not None and value != "":
                roster_entry[key] = value
    return roster


def _build_speaker_roster(
    analysis: Dict[str, Any],
    base_voice: str,
    voice_profile: Optional[str],
    existing: Optional[Mapping[str, Any]] = None,
    order: Optional[Iterable[str]] = None,
) -> Dict[str, Any]:
    roster = _build_narrator_roster(base_voice, voice_profile, existing)
    existing_map: Dict[str, Any] = dict(existing) if isinstance(existing, Mapping) else {}
    speakers = analysis.get("speakers", {}) if isinstance(analysis, dict) else {}
    ordered_ids: Iterable[str]
    if order is not None:
        ordered_ids = [sid for sid in order if sid in speakers]
    else:
        ordered_ids = speakers.keys()

    for speaker_id in ordered_ids:
        payload = speakers.get(speaker_id, {})
        if speaker_id == "narrator":
            continue
        if payload.get("suppressed"):
            continue
        previous = existing_map.get(speaker_id)
        roster[speaker_id] = {
            "id": speaker_id,
            "label": payload.get("label") or speaker_id.replace("_", " ").title(),
            "voice": base_voice,
            "analysis_confidence": payload.get("confidence"),
            "analysis_count": payload.get("count"),
            "gender": payload.get("gender", "unknown"),
        }
        if isinstance(previous, Mapping):
            for key in ("voice", "voice_profile", "voice_formula", "resolved_voice", "pronunciation"):
                value = previous.get(key)
                if value is not None and value != "":
                    roster[speaker_id][key] = value
    return roster


def _match_configured_speaker(
    config_speakers: Mapping[str, Any],
    roster_id: str,
    roster_label: str,
) -> Optional[Mapping[str, Any]]:
    if not config_speakers:
        return None
    entry = config_speakers.get(roster_id)
    if entry:
        return cast(Mapping[str, Any], entry)
    slug = slugify_label(roster_label)
    if slug != roster_id and slug in config_speakers:
        return cast(Mapping[str, Any], config_speakers[slug])
    lower_label = roster_label.strip().lower()
    for record in config_speakers.values():
        if not isinstance(record, Mapping):
            continue
        if str(record.get("label", "")).strip().lower() == lower_label:
            return record
    return None


def _apply_speaker_config_to_roster(
    roster: Mapping[str, Any],
    config: Optional[Mapping[str, Any]],
    *,
    allow_randomize: bool = False,
    persist_changes: bool = False,
    fallback_languages: Optional[Iterable[str]] = None,
) -> Tuple[Dict[str, Any], List[str], Optional[Dict[str, Any]]]:
    if not isinstance(roster, Mapping):
        effective_languages = [code for code in (fallback_languages or []) if isinstance(code, str) and code]
        return {}, effective_languages, None
    updated_roster: Dict[str, Any] = {key: dict(value) for key, value in roster.items() if isinstance(value, Mapping)}
    if not config:
        effective_languages = [code for code in (fallback_languages or []) if isinstance(code, str) and code]
        return updated_roster, effective_languages, None

    speakers_map = config.get("speakers")
    if not isinstance(speakers_map, Mapping):
        effective_languages = [code for code in (fallback_languages or []) if isinstance(code, str) and code]
        return updated_roster, effective_languages, None

    config_languages = config.get("languages")
    if isinstance(config_languages, list):
        allowed_languages = [code for code in config_languages if isinstance(code, str) and code]
    else:
        allowed_languages = []
    if not allowed_languages and fallback_languages:
        allowed_languages = [code for code in fallback_languages if isinstance(code, str) and code]

    default_voice = config.get("default_voice") if isinstance(config.get("default_voice"), str) else ""
    used_voices = {entry.get("resolved_voice") or entry.get("voice") for entry in updated_roster.values()} - {None}

    narrator_voice = ""
    narrator_entry = updated_roster.get("narrator") if isinstance(updated_roster, Mapping) else None
    if isinstance(narrator_entry, Mapping):
        narrator_voice = str(
            narrator_entry.get("resolved_voice")
            or narrator_entry.get("voice")
            or narrator_entry.get("default_voice")
            or ""
        ).strip()
        if narrator_voice:
            used_voices.add(narrator_voice)

    config_changed = False
    new_config_payload: Dict[str, Any] = {
        "language": config.get("language", "a"),
        "languages": allowed_languages,
        "default_voice": default_voice,
        "speakers": dict(speakers_map),
        "version": config.get("version", 1),
        "notes": config.get("notes", ""),
    }

    speakers_payload = new_config_payload["speakers"]

    for speaker_id, roster_entry in updated_roster.items():
        if speaker_id == "narrator":
            continue
        label = str(roster_entry.get("label") or speaker_id)
        config_entry = _match_configured_speaker(speakers_map, speaker_id, label)
        if config_entry is None:
            continue
        voice_id = str(config_entry.get("voice") or "").strip()
        voice_profile = str(config_entry.get("voice_profile") or "").strip()
        voice_formula = str(config_entry.get("voice_formula") or "").strip()
        resolved_voice = str(config_entry.get("resolved_voice") or "").strip()
        languages = config_entry.get("languages") if isinstance(config_entry.get("languages"), list) else []
        randomize_flag = bool(config_entry.get("randomize"))

        chosen_voice = resolved_voice or voice_formula or voice_id or roster_entry.get("voice")
        usable_languages = languages or allowed_languages
        should_randomize = randomize_flag and (allow_randomize or not chosen_voice)
        if should_randomize:
            exclusion = used_voices | {voice_id, resolved_voice, narrator_voice}
            randomized = random_voice(
                gender=config_entry.get("gender", "unknown"),
                allowed_languages=usable_languages,
                fallback_voice=default_voice or roster_entry.get("voice"),
                exclude=exclusion,
            )
            if randomized:
                chosen_voice = randomized
                voice_id = randomized
                resolved_voice = randomized
                config_changed = True
                used_voices.add(randomized)

        if chosen_voice:
            roster_entry["resolved_voice"] = chosen_voice
            roster_entry["voice"] = chosen_voice if not voice_profile and not voice_formula else roster_entry.get("voice", chosen_voice)
        if voice_profile:
            roster_entry["voice_profile"] = voice_profile
        if voice_formula:
            roster_entry["voice_formula"] = voice_formula
            roster_entry["resolved_voice"] = voice_formula
        if not voice_formula and not voice_profile and resolved_voice:
            roster_entry["resolved_voice"] = resolved_voice
        roster_entry["config_languages"] = usable_languages or []

        if chosen_voice:
            used_voices.add(chosen_voice)

        # persist updates back to config payload if required
        if persist_changes:
            slug = config_entry.get("id") or slugify_label(label)
            speakers_payload[slug] = {
                "id": slug,
                "label": label,
                "gender": config_entry.get("gender", "unknown"),
                "voice": voice_id,
                "voice_profile": voice_profile,
                "voice_formula": voice_formula,
                "resolved_voice": roster_entry.get("resolved_voice", resolved_voice or voice_id),
                "languages": usable_languages,
                "randomize": randomize_flag,
            }

    new_config = new_config_payload if (persist_changes and config_changed) else None
    return updated_roster, allowed_languages, new_config


def _filter_voice_catalog(
    catalog: Iterable[Mapping[str, Any]],
    *,
    gender: str,
    allowed_languages: Optional[Iterable[str]] = None,
) -> List[str]:
    allowed_set = {code.lower() for code in (allowed_languages or []) if isinstance(code, str) and code}
    gender_normalized = (gender or "unknown").lower()
    gender_code = ""
    if gender_normalized == "male":
        gender_code = "m"
    elif gender_normalized == "female":
        gender_code = "f"

    matches: List[str] = []
    seen: set[str] = set()

    def _consider(entry: Mapping[str, Any]) -> None:
        voice_id = entry.get("id")
        if not isinstance(voice_id, str) or not voice_id:
            return
        if voice_id in seen:
            return
        seen.add(voice_id)
        matches.append(voice_id)

    primary: List[Mapping[str, Any]] = []
    fallback: List[Mapping[str, Any]] = []
    for entry in catalog:
        if not isinstance(entry, Mapping):
            continue
        voice_lang = str(entry.get("language", "")).lower()
        voice_gender_code = str(entry.get("gender_code", "")).lower()
        if allowed_set and voice_lang not in allowed_set:
            continue
        if gender_code and voice_gender_code != gender_code:
            fallback.append(entry)
            continue
        primary.append(entry)

    for entry in primary:
        _consider(entry)

    if not matches:
        for entry in fallback:
            _consider(entry)

    if not matches:
        for entry in catalog:
            if isinstance(entry, Mapping):
                _consider(entry)

    return matches


def _inject_recommended_voices(
    roster: Mapping[str, Any],
    *,
    fallback_languages: Optional[Iterable[str]] = None,
) -> None:
    voice_catalog = _build_voice_catalog()
    fallback_list = [code for code in (fallback_languages or []) if isinstance(code, str) and code]
    for speaker_id, payload in roster.items():
        if not isinstance(payload, dict):
            continue
        languages = payload.get("config_languages")
        if isinstance(languages, list) and languages:
            language_list = languages
        else:
            language_list = fallback_list
        gender = str(payload.get("gender", "unknown"))
        payload["recommended_voices"] = _filter_voice_catalog(
            voice_catalog,
            gender=gender,
            allowed_languages=language_list,
        )


def _extract_speaker_config_form(form: Mapping[str, Any]) -> Tuple[str, Dict[str, Any], List[str]]:
    getter = getattr(form, "getlist", None)

    def _get_list(name: str) -> List[str]:
        if callable(getter):
            values = cast(Iterable[Any], getter(name))
            return [str(value).strip() for value in values if value]
        raw_value = form.get(name)
        if isinstance(raw_value, str):
            return [item.strip() for item in raw_value.split(",") if item.strip()]
        return []

    name = (form.get("config_name") or "").strip()
    language = str(form.get("config_language") or "a").strip() or "a"
    allowed_languages = []
    default_voice = (form.get("config_default_voice") or "").strip()
    notes = (form.get("config_notes") or "").strip()
    version = _coerce_int(form.get("config_version"), 1, minimum=1, maximum=9999)

    speaker_rows = _get_list("speaker_rows")
    speakers: Dict[str, Dict[str, Any]] = {}
    for row_key in speaker_rows:
        prefix = f"speaker-{row_key}-"
        label = (form.get(prefix + "label") or "").strip()
        if not label:
            continue
        raw_gender = (form.get(prefix + "gender") or "unknown").strip().lower()
        gender = raw_gender if raw_gender in {"male", "female", "unknown"} else "unknown"
        voice = (form.get(prefix + "voice") or "").strip()
        voice_profile = (form.get(prefix + "profile") or "").strip()
        voice_formula = (form.get(prefix + "formula") or "").strip()
        randomize_flag = form.get(prefix + "randomize") in {"on", "1", "true"}
        speaker_id = (form.get(prefix + "id") or "").strip() or slugify_label(label)
        speakers[speaker_id] = {
            "id": speaker_id,
            "label": label,
            "gender": gender,
            "voice": voice,
            "voice_profile": voice_profile,
            "voice_formula": voice_formula,
            "resolved_voice": voice_formula or voice,
            "languages": [],
            "randomize": randomize_flag,
        }

    payload = {
        "language": language,
        "languages": allowed_languages,
        "default_voice": default_voice,
        "speakers": speakers,
        "notes": notes,
        "version": version,
    }

    errors: List[str] = []
    if not name:
        errors.append("Configuration name is required.")
    if not speakers:
        errors.append("Add at least one speaker to the configuration.")

    return name, payload, errors


def _prepare_speaker_metadata(
    *,
    chapters: List[Dict[str, Any]],
    chunks: List[Dict[str, Any]],
    analysis_chunks: Optional[List[Dict[str, Any]]] = None,
    speaker_mode: str,
    voice: str,
    voice_profile: Optional[str],
    threshold: int,
    existing_roster: Optional[Mapping[str, Any]] = None,
    run_analysis: bool = True,
    speaker_config: Optional[Mapping[str, Any]] = None,
    apply_config: bool = False,
    allow_randomize: bool = False,
    persist_config: bool = False,
) -> tuple[List[Dict[str, Any]], Dict[str, Any], Dict[str, Any], List[str], Optional[Dict[str, Any]]]:
    chunk_list = [dict(chunk) for chunk in chunks]
    analysis_source = [dict(chunk) for chunk in (analysis_chunks or chunks)]
    threshold_value = max(1, int(threshold))
    analysis_enabled = speaker_mode == "multi" and run_analysis
    settings_state = _load_settings()
    global_random_languages = [
        code
        for code in settings_state.get("speaker_random_languages", [])
        if isinstance(code, str) and code
    ]

    if not analysis_enabled:
        for chunk in chunk_list:
            chunk["speaker_id"] = "narrator"
            chunk["speaker_label"] = "Narrator"
        analysis_payload = {
            "version": "1.0",
            "narrator": "narrator",
            "assignments": {str(chunk.get("id")): "narrator" for chunk in chunk_list},
            "speakers": {
                "narrator": {
                    "id": "narrator",
                    "label": "Narrator",
                    "count": len(chunk_list),
                    "confidence": "low",
                    "sample_quotes": [],
                    "suppressed": False,
                }
            },
            "suppressed": [],
            "stats": {
                "total_chunks": len(chunk_list),
                "explicit_chunks": 0,
                "active_speakers": 0,
                "unique_speakers": 1,
                "suppressed": 0,
            },
        }
        roster = _build_narrator_roster(voice, voice_profile, existing_roster)
        narrator_pron = roster["narrator"].get("pronunciation")
        if narrator_pron:
            analysis_payload["speakers"]["narrator"]["pronunciation"] = narrator_pron
        return chunk_list, roster, analysis_payload, [], None

    analysis_result = analyze_speakers(
        chapters,
        analysis_source,
        threshold=threshold_value,
        max_speakers=0,
    )
    analysis_payload = analysis_result.to_dict()
    speakers_payload = analysis_payload.get("speakers", {})
    ordered_ids = [
        sid
        for sid, meta in sorted(
            ((sid, meta) for sid, meta in speakers_payload.items() if sid != "narrator"),
            key=lambda item: item[1].get("count", 0),
            reverse=True,
        )
    ]
    analysis_payload["ordered_speakers"] = ordered_ids
    assignments = analysis_payload.get("assignments", {})
    suppressed_ids = analysis_payload.get("suppressed", [])
    suppressed_details: List[Dict[str, Any]] = []
    speakers_payload = analysis_payload.get("speakers", {})
    if isinstance(suppressed_ids, Iterable):
        for suppressed_id in suppressed_ids:
            speaker_meta = speakers_payload.get(suppressed_id) if isinstance(speakers_payload, dict) else None
            if isinstance(speaker_meta, dict):
                suppressed_details.append(
                    {
                        "id": suppressed_id,
                        "label": speaker_meta.get("label")
                        or str(suppressed_id).replace("_", " ").title(),
                        "pronunciation": speaker_meta.get("pronunciation"),
                    }
                )
            else:
                suppressed_details.append(
                    {
                        "id": suppressed_id,
                        "label": str(suppressed_id).replace("_", " ").title(),
                        "pronunciation": None,
                    }
                )
    analysis_payload["suppressed_details"] = suppressed_details
    roster = _build_speaker_roster(
        analysis_payload,
        voice,
        voice_profile,
        existing=existing_roster,
        order=analysis_payload.get("ordered_speakers"),
    )
    applied_languages: List[str] = []
    updated_config: Optional[Dict[str, Any]] = None
    if apply_config and speaker_config:
        roster, applied_languages, updated_config = _apply_speaker_config_to_roster(
            roster,
            speaker_config,
            allow_randomize=allow_randomize,
            persist_changes=persist_config,
            fallback_languages=global_random_languages,
        )
        speakers_payload = analysis_payload.get("speakers")
        if isinstance(speakers_payload, dict):
            for roster_id, roster_payload in roster.items():
                speaker_meta = speakers_payload.get(roster_id)
                if isinstance(speaker_meta, dict):
                    for key in ("voice", "voice_profile", "voice_formula", "resolved_voice"):
                        value = roster_payload.get(key)
                        if value:
                            speaker_meta[key] = value
    effective_languages: List[str] = []
    if applied_languages:
        effective_languages = applied_languages
    elif isinstance(analysis_payload.get("config_languages"), list):
        effective_languages = [
            code for code in analysis_payload.get("config_languages", []) if isinstance(code, str) and code
        ]
    elif global_random_languages:
        effective_languages = list(global_random_languages)

    if effective_languages:
        analysis_payload["config_languages"] = effective_languages
    speakers_payload = analysis_payload.get("speakers")
    if isinstance(speakers_payload, dict):
        for roster_id, roster_payload in roster.items():
            if roster_id in speakers_payload and isinstance(roster_payload, dict):
                pronunciation_value = roster_payload.get("pronunciation")
                if pronunciation_value:
                    speakers_payload[roster_id]["pronunciation"] = pronunciation_value

    fallback_languages = effective_languages or []
    _inject_recommended_voices(roster, fallback_languages=fallback_languages)

    for chunk in chunk_list:
        chunk_id = str(chunk.get("id"))
        speaker_id = assignments.get(chunk_id, "narrator")
        chunk["speaker_id"] = speaker_id
        speaker_meta = roster.get(speaker_id)
        chunk["speaker_label"] = speaker_meta.get("label") if isinstance(speaker_meta, dict) else speaker_id

    return chunk_list, roster, analysis_payload, applied_languages, updated_config


def _apply_prepare_form(
    pending: PendingJob, form: Mapping[str, Any]
) -> tuple[
    ChunkLevel,
    List[Dict[str, Any]],
    List[Dict[str, Any]],
    List[str],
    int,
    str,
    bool,
    bool,
    bool,
]:
    raw_chunk_level = (form.get("chunk_level") or pending.chunk_level or "paragraph").strip().lower()
    if raw_chunk_level not in _CHUNK_LEVEL_VALUES:
        raw_chunk_level = pending.chunk_level if pending.chunk_level in _CHUNK_LEVEL_VALUES else "paragraph"
    pending.chunk_level = raw_chunk_level
    chunk_level_literal = cast(ChunkLevel, pending.chunk_level)

    raw_speaker_mode = (form.get("speaker_mode") or pending.speaker_mode or "single").strip().lower()
    if raw_speaker_mode not in _SPEAKER_MODE_VALUES:
        raw_speaker_mode = "single"
    pending.speaker_mode = raw_speaker_mode

    pending.generate_epub3 = _coerce_bool(form.get("generate_epub3"), False)

    threshold_default = getattr(pending, "speaker_analysis_threshold", _DEFAULT_ANALYSIS_THRESHOLD)
    raw_threshold = form.get("speaker_analysis_threshold")
    if raw_threshold is not None:
        pending.speaker_analysis_threshold = _coerce_int(
            raw_threshold,
            threshold_default,
            minimum=1,
            maximum=25,
        )
    else:
        pending.speaker_analysis_threshold = threshold_default

    if not pending.speakers:
        narrator: Dict[str, Any] = {
            "id": "narrator",
            "label": "Narrator",
            "voice": pending.voice,
        }
        if pending.voice_profile:
            narrator["voice_profile"] = pending.voice_profile
        pending.speakers = {"narrator": narrator}
    else:
        existing_narrator = pending.speakers.get("narrator")
        if isinstance(existing_narrator, dict):
            existing_narrator.setdefault("id", "narrator")
            existing_narrator["label"] = existing_narrator.get("label", "Narrator")
            existing_narrator["voice"] = pending.voice
            if pending.voice_profile:
                existing_narrator["voice_profile"] = pending.voice_profile
            pending.speakers["narrator"] = existing_narrator

    selected_config = (form.get("applied_speaker_config") or "").strip()
    randomize_requested = str(form.get("randomize_speaker_config", "")).strip() in {"1", "true", "on"}
    apply_config_requested = (
        str(form.get("apply_speaker_config", "")).strip() in {"1", "true", "on"}
        or randomize_requested
    )
    persist_config_requested = str(form.get("save_speaker_config", "")).strip() in {"1", "true", "on"}

    pending.applied_speaker_config = selected_config or None

    if isinstance(pending.speakers, dict):
        for speaker_id, payload in list(pending.speakers.items()):
            if not isinstance(payload, dict):
                continue
            field_key = f"speaker-{speaker_id}-pronunciation"
            raw_value = form.get(field_key, "")
            pronunciation = raw_value.strip()
            if pronunciation:
                payload["pronunciation"] = pronunciation
            else:
                payload.pop("pronunciation", None)

            voice_value = (form.get(f"speaker-{speaker_id}-voice") or "").strip()
            if voice_value:
                payload["voice"] = voice_value
                payload["resolved_voice"] = voice_value
            else:
                payload.pop("voice", None)
                payload.pop("resolved_voice", None)

            randomize_flag = form.get(f"speaker-{speaker_id}-randomize") in {"on", "1", "true"}
            payload["randomize"] = randomize_flag

            lang_key = f"speaker-{speaker_id}-languages"
            languages: List[str] = []
            getter = getattr(form, "getlist", None)
            if callable(getter):
                values = cast(Iterable[str], getter(lang_key))
                languages = [code.strip() for code in values if code]
            else:
                raw_langs = form.get(lang_key)
                if isinstance(raw_langs, str):
                    languages = [item.strip() for item in raw_langs.split(",") if item.strip()]
            payload["config_languages"] = languages

    profiles = serialize_profiles()
    errors: List[str] = []
    raw_delay = form.get("chapter_intro_delay")
    if raw_delay is not None:
        raw_normalized = raw_delay.strip()
        if raw_normalized:
            try:
                pending.chapter_intro_delay = max(0.0, float(raw_normalized))
            except ValueError:
                errors.append("Enter a valid number for the chapter intro delay.")
        else:
            pending.chapter_intro_delay = 0.0

    overrides: List[Dict[str, Any]] = []
    selected_total = 0

    for index, chapter in enumerate(pending.chapters):
        enabled = form.get(f"chapter-{index}-enabled") == "on"
        title_input = (form.get(f"chapter-{index}-title") or "").strip()
        title = title_input or chapter.get("title") or f"Chapter {index + 1}"
        voice_selection = form.get(f"chapter-{index}-voice", "__default")
        formula_input = (form.get(f"chapter-{index}-formula") or "").strip()

        entry: Dict[str, Any] = {
            "id": chapter.get("id") or f"{index:04d}",
            "index": index,
            "order": index,
            "source_title": chapter.get("title") or title,
            "title": title,
            "text": chapter.get("text", ""),
            "enabled": enabled,
        }
        entry["characters"] = calculate_text_length(entry["text"])

        if enabled:
            if voice_selection.startswith("voice:"):
                entry["voice"] = voice_selection.split(":", 1)[1]
                entry["resolved_voice"] = entry["voice"]
            elif voice_selection.startswith("profile:"):
                profile_name = voice_selection.split(":", 1)[1]
                entry["voice_profile"] = profile_name
                profile_entry = profiles.get(profile_name) or {}
                formula_value = _formula_from_profile(profile_entry)
                if formula_value:
                    entry["voice_formula"] = formula_value
                    entry["resolved_voice"] = formula_value
                else:
                    errors.append(f"Profile '{profile_name}' has no configured voices.")
            elif voice_selection == "formula":
                if not formula_input:
                    errors.append(f"Provide a custom formula for chapter {index + 1}.")
                else:
                    try:
                        _parse_voice_formula(formula_input)
                    except ValueError as exc:
                        errors.append(str(exc))
                    else:
                        entry["voice_formula"] = formula_input
                        entry["resolved_voice"] = formula_input
            selected_total += entry["characters"]

        overrides.append(entry)
        pending.chapters[index] = dict(entry)

    enabled_overrides = [entry for entry in overrides if entry.get("enabled")]

    return (
        chunk_level_literal,
        overrides,
        enabled_overrides,
        errors,
        selected_total,
        selected_config,
        randomize_requested,
        apply_config_requested,
        persist_config_requested,
    )
_SUPPLEMENT_TITLE_PATTERNS: List[tuple[re.Pattern[str], float]] = [
    (re.compile(r"\btitle\s+page\b"), 3.0),
    (re.compile(r"\bcopyright\b"), 2.4),
    (re.compile(r"\btable\s+of\s+contents\b"), 2.8),
    (re.compile(r"\bcontents\b"), 2.0),
    (re.compile(r"\backnowledg(e)?ments?\b"), 2.0),
    (re.compile(r"\bdedication\b"), 2.0),
    (re.compile(r"\babout\s+the\s+author(s)?\b"), 2.4),
    (re.compile(r"\balso\s+by\b"), 2.0),
    (re.compile(r"\bpraise\s+for\b"), 2.0),
    (re.compile(r"\bcolophon\b"), 2.2),
    (re.compile(r"\bpublication\s+data\b"), 2.2),
    (re.compile(r"\btranscriber'?s?\s+note\b"), 2.2),
    (re.compile(r"\bglossary\b"), 2.0),
    (re.compile(r"\bindex\b"), 2.0),
    (re.compile(r"\bbibliograph(y|ies)\b"), 2.0),
    (re.compile(r"\breferences\b"), 1.8),
    (re.compile(r"\bappendix\b"), 1.9),
]

_CONTENT_TITLE_PATTERNS: List[re.Pattern[str]] = [
    re.compile(r"\bchapter\b"),
    re.compile(r"\bbook\b"),
    re.compile(r"\bpart\b"),
    re.compile(r"\bsection\b"),
    re.compile(r"\bscene\b"),
    re.compile(r"\bprologue\b"),
    re.compile(r"\bepilogue\b"),
    re.compile(r"\bintroduction\b"),
    re.compile(r"\bstory\b"),
]

_SUPPLEMENT_TEXT_KEYWORDS: List[tuple[str, float]] = [
    ("copyright", 1.2),
    ("all rights reserved", 1.1),
    ("isbn", 0.9),
    ("library of congress", 1.0),
    ("table of contents", 1.0),
    ("dedicated to", 0.8),
    ("acknowledg", 0.8),
    ("printed in", 0.6),
    ("permission", 0.6),
    ("publisher", 0.5),
    ("praise for", 0.9),
    ("also by", 0.9),
    ("glossary", 0.8),
    ("index", 0.8),
]


def _supplement_score(title: str, text: str, index: int) -> float:
    normalized_title = (title or "").lower()
    score = 0.0

    for pattern, weight in _SUPPLEMENT_TITLE_PATTERNS:
        if pattern.search(normalized_title):
            score += weight

    for pattern in _CONTENT_TITLE_PATTERNS:
        if pattern.search(normalized_title):
            score -= 2.0

    stripped_text = (text or "").strip()
    length = len(stripped_text)
    if length <= 150:
        score += 0.9
    elif length <= 400:
        score += 0.6
    elif length <= 800:
        score += 0.35

    lowercase_text = stripped_text.lower()
    for keyword, weight in _SUPPLEMENT_TEXT_KEYWORDS:
        if keyword in lowercase_text:
            score += weight

    if index == 0 and score > 0:
        score += 0.25

    return score


def _should_preselect_chapter(
    title: str,
    text: str,
    index: int,
    total_count: int,
) -> bool:
    if total_count <= 1:
        return True
    score = _supplement_score(title, text, index)
    return score < 1.9


def _ensure_at_least_one_chapter_enabled(chapters: List[Dict[str, Any]]) -> None:
    if not chapters:
        return
    if any(chapter.get("enabled") for chapter in chapters):
        return
    best_index = max(range(len(chapters)), key=lambda idx: chapters[idx].get("characters", 0))
    chapters[best_index]["enabled"] = True


def _service() -> ConversionService:
    return current_app.extensions["conversion_service"]


def _build_voice_catalog() -> List[Dict[str, str]]:
    catalog: List[Dict[str, str]] = []
    gender_map = {"f": "Female", "m": "Male"}
    for voice_id in VOICES_INTERNAL:
        prefix, _, rest = voice_id.partition("_")
        language_code = prefix[0] if prefix else "a"
        gender_code = prefix[1] if len(prefix) > 1 else ""
        catalog.append(
            {
                "id": voice_id,
                "language": language_code,
                "language_label": LANGUAGE_DESCRIPTIONS.get(language_code, language_code.upper()),
                "gender": gender_map.get(gender_code, "Unknown"),
                "gender_code": gender_code,
                "display_name": rest.replace("_", " ").title() if rest else voice_id,
            }
        )
    return catalog


def _template_options() -> Dict[str, Any]:
    current_settings = _load_settings()
    profiles = serialize_profiles()
    ordered_profiles = sorted(profiles.items())
    profile_options = []
    for name, entry in ordered_profiles:
        profile_options.append(
            {
                "name": name,
                "language": (entry or {}).get("language", ""),
                "formula": _formula_from_profile(entry or {}) or "",
            }
        )
    voice_catalog = _build_voice_catalog()
    return {
        "languages": LANGUAGE_DESCRIPTIONS,
        "voices": VOICES_INTERNAL,
        "subtitle_formats": SUBTITLE_FORMATS,
        "supported_langs_for_subs": SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION,
        "output_formats": SUPPORTED_SOUND_FORMATS,
        "voice_profiles": ordered_profiles,
        "voice_profile_options": profile_options,
        "separate_formats": ["wav", "flac", "mp3", "opus"],
    "voice_catalog": voice_catalog,
    "voice_catalog_map": {entry["id"]: entry for entry in voice_catalog},
        "sample_voice_texts": SAMPLE_VOICE_TEXTS,
        "voice_profiles_data": profiles,
        "speaker_configs": list_configs(),
        "chunk_levels": _CHUNK_LEVEL_OPTIONS,
        "speaker_modes": _SPEAKER_MODE_OPTIONS,
        "speaker_analysis_threshold": current_settings.get(
            "speaker_analysis_threshold", _DEFAULT_ANALYSIS_THRESHOLD
        ),
        "speaker_pronunciation_sentence": current_settings.get(
            "speaker_pronunciation_sentence", _settings_defaults()["speaker_pronunciation_sentence"]
        ),
    }


SAVE_MODE_LABELS = {
    "save_next_to_input": "Save next to input file",
    "save_to_desktop": "Save to Desktop",
    "choose_output_folder": "Choose output folder",
    "default_output": "Use default save location",
}

LEGACY_SAVE_MODE_MAP = {label: key for key, label in SAVE_MODE_LABELS.items()}

BOOLEAN_SETTINGS = {
    "replace_single_newlines",
    "use_gpu",
    "save_chapters_separately",
    "merge_chapters_at_end",
    "save_as_project",
    "generate_epub3",
}

FLOAT_SETTINGS = {"silence_between_chapters", "chapter_intro_delay"}
INT_SETTINGS = {"max_subtitle_words", "speaker_analysis_threshold"}


def _has_output_override() -> bool:
    return bool(os.environ.get("ABOGEN_OUTPUT_DIR") or os.environ.get("ABOGEN_OUTPUT_ROOT"))


def _settings_defaults() -> Dict[str, Any]:
    return {
        "output_format": "wav",
        "subtitle_format": "srt",
        "save_mode": "default_output" if _has_output_override() else "save_next_to_input",
        "default_voice": VOICES_INTERNAL[0] if VOICES_INTERNAL else "",
        "replace_single_newlines": False,
        "use_gpu": True,
        "save_chapters_separately": False,
        "merge_chapters_at_end": True,
        "save_as_project": False,
        "separate_chapters_format": "wav",
        "silence_between_chapters": 2.0,
        "chapter_intro_delay": 0.5,
        "max_subtitle_words": 50,
        "chunk_level": "paragraph",
        "speaker_mode": "single",
        "generate_epub3": False,
        "speaker_analysis_threshold": _DEFAULT_ANALYSIS_THRESHOLD,
        "speaker_pronunciation_sentence": "This is {{name}} speaking.",
        "speaker_random_languages": [],
    }


def _coerce_bool(value: Any, default: bool) -> bool:
    if isinstance(value, bool):
        return value
    if isinstance(value, str):
        return value.lower() in {"true", "1", "yes", "on"}
    if value is None:
        return default
    return bool(value)


def _coerce_float(value: Any, default: float) -> float:
    try:
        return max(0.0, float(value))
    except (TypeError, ValueError):
        return default


def _coerce_int(value: Any, default: int, *, minimum: int = 1, maximum: int = 200) -> int:
    try:
        parsed = int(value)
    except (TypeError, ValueError):
        return default
    return max(minimum, min(parsed, maximum))


def _normalize_save_mode(value: Any, default: str) -> str:
    if isinstance(value, str):
        if value in SAVE_MODE_LABELS:
            return value
        if value in LEGACY_SAVE_MODE_MAP:
            return LEGACY_SAVE_MODE_MAP[value]
    return default


def _normalize_setting_value(key: str, value: Any, defaults: Dict[str, Any]) -> Any:
    if key in BOOLEAN_SETTINGS:
        return _coerce_bool(value, defaults[key])
    if key in FLOAT_SETTINGS:
        return _coerce_float(value, defaults[key])
    if key in INT_SETTINGS:
        return _coerce_int(value, defaults[key])
    if key == "save_mode":
        return _normalize_save_mode(value, defaults[key])
    if key == "output_format":
        return value if value in SUPPORTED_SOUND_FORMATS else defaults[key]
    if key == "subtitle_format":
        valid = {item[0] for item in SUBTITLE_FORMATS}
        return value if value in valid else defaults[key]
    if key == "separate_chapters_format":
        if isinstance(value, str):
            normalized = value.lower()
            if normalized in {"wav", "flac", "mp3", "opus"}:
                return normalized
        return defaults[key]
    if key == "default_voice":
        if isinstance(value, str) and value in VOICES_INTERNAL:
            return value
        return defaults[key]
    if key == "chunk_level":
        if isinstance(value, str) and value in _CHUNK_LEVEL_VALUES:
            return value
        return defaults[key]
    if key == "speaker_mode":
        if isinstance(value, str) and value in _SPEAKER_MODE_VALUES:
            return value
        return defaults[key]
    if key == "speaker_random_languages":
        if isinstance(value, (list, tuple, set)):
            return [code for code in value if isinstance(code, str) and code in LANGUAGE_DESCRIPTIONS]
        if isinstance(value, str):
            parts = [item.strip().lower() for item in value.split(",") if item.strip()]
            return [code for code in parts if code in LANGUAGE_DESCRIPTIONS]
        return defaults.get(key, [])
    return value if value is not None else defaults.get(key)


def _load_settings() -> Dict[str, Any]:
    defaults = _settings_defaults()
    cfg = load_config() or {}
    settings: Dict[str, Any] = {}
    for key, default in defaults.items():
        raw_value = cfg.get(key, default)
        settings[key] = _normalize_setting_value(key, raw_value, defaults)
    return settings

def _formula_from_profile(entry: Dict[str, Any]) -> Optional[str]:
    voices = entry.get("voices") or []
    if not voices:
        return None
    total = sum(weight for _, weight in voices)
    if total <= 0:
        return None

    def _format_weight(value: float) -> str:
        normalized = value / total if total else 0.0
        return (f"{normalized:.4f}").rstrip("0").rstrip(".") or "0"

    parts = [f"{name}*{_format_weight(weight)}" for name, weight in voices if weight > 0]
    return "+".join(parts) if parts else None


def _resolve_voice_choice(
    language: str,
    base_voice: str,
    profile_name: str,
    custom_formula: str,
    profiles: Dict[str, Any],
) -> tuple[str, str, Optional[str]]:
    resolved_voice = base_voice
    resolved_language = language
    selected_profile = None

    if profile_name:
        entry = profiles.get(profile_name)
        formula = _formula_from_profile(entry or {}) if entry else None
        if formula:
            resolved_voice = formula
            selected_profile = profile_name
            profile_language = (entry or {}).get("language")
            if profile_language:
                resolved_language = profile_language

    if custom_formula:
        resolved_voice = custom_formula
        selected_profile = None

    return resolved_voice, resolved_language, selected_profile


def _persist_cover_image(extraction_result: Any, stored_path: Path) -> tuple[Optional[Path], Optional[str]]:
    cover_bytes = getattr(extraction_result, "cover_image", None)
    if not cover_bytes:
        return None, None

    mime = getattr(extraction_result, "cover_mime", None)
    extension = mimetypes.guess_extension(mime or "") or ".png"
    base_stem = Path(stored_path).stem or "cover"
    candidate = stored_path.parent / f"{base_stem}_cover{extension}"
    counter = 1
    while candidate.exists():
        candidate = stored_path.parent / f"{base_stem}_cover_{counter}{extension}"
        counter += 1

    try:
        candidate.write_bytes(cover_bytes)
    except OSError:
        return None, None

    return candidate, mime


def _parse_voice_formula(formula: str) -> List[tuple[str, float]]:
    parts = [segment.strip() for segment in formula.split("+") if segment.strip()]
    voices: List[tuple[str, float]] = []
    for part in parts:
        if "*" not in part:
            raise ValueError("Each component must be in the form voice*weight")
        name, weight_str = part.split("*", 1)
        name = name.strip()
        if name not in VOICES_INTERNAL:
            raise ValueError(f"Unknown voice '{name}'")
        try:
            weight = float(weight_str.strip())
        except ValueError as exc:  # pragma: no cover - validated via form
            raise ValueError(f"Invalid weight for {name}") from exc
        if weight <= 0:
            raise ValueError(f"Weight for {name} must be positive")
        voices.append((name, weight))
    total = sum(weight for _, weight in voices)
    if total <= 0:
        raise ValueError("Voice weights must sum to a positive value")
    return voices


def _sanitize_voice_entries(entries: Iterable[Any]) -> List[Dict[str, Any]]:
    sanitized: List[Dict[str, Any]] = []
    for entry in entries or []:
        if isinstance(entry, dict):
            voice_id = entry.get("id") or entry.get("voice")
            if not voice_id:
                continue
            enabled = entry.get("enabled", True)
            if not enabled:
                continue
            sanitized.append({"voice": voice_id, "weight": entry.get("weight")})
        elif isinstance(entry, (list, tuple)) and len(entry) >= 2:
            sanitized.append({"voice": entry[0], "weight": entry[1]})
    return sanitized


def _pairs_to_formula(pairs: Iterable[Tuple[str, float]]) -> Optional[str]:
    voices = [(voice, float(weight)) for voice, weight in pairs if float(weight) > 0]
    if not voices:
        return None
    total = sum(weight for _, weight in voices)
    if total <= 0:
        return None

    def _format_value(value: float) -> str:
        normalized = value / total if total else 0.0
        return (f"{normalized:.4f}").rstrip("0").rstrip(".") or "0"

    parts = [f"{voice}*{_format_value(weight)}" for voice, weight in voices]
    return "+".join(parts)


def _profiles_payload() -> Dict[str, Any]:
    return {"profiles": serialize_profiles()}


def _get_preview_pipeline(language: str, device: str):
    key = (language, device)
    with _preview_pipeline_lock:
        pipeline = _preview_pipelines.get(key)
        if pipeline is not None:
            return pipeline
        _, KPipeline = load_numpy_kpipeline()
        pipeline = KPipeline(lang_code=language, repo_id="hexgrad/Kokoro-82M", device=device)
        _preview_pipelines[key] = pipeline
        return pipeline


@web_bp.app_template_filter("datetimeformat")
def datetimeformat(value: float, fmt: str = "%Y-%m-%d %H:%M:%S") -> str:
    if not value:
        return "—"
    from datetime import datetime

    return datetime.fromtimestamp(value).strftime(fmt)


@web_bp.get("/")
def index() -> str:
    return render_template(
        "index.html",
        options=_template_options(),
        settings=_load_settings(),
        jobs_panel=_render_jobs_panel(),
    )


@web_bp.get("/queue")
def queue_page() -> ResponseReturnValue:
    return redirect(url_for("web.index", _anchor="queue"))


@web_bp.route("/settings", methods=["GET", "POST"])
def settings_page() -> ResponseReturnValue:
    options = _template_options()
    current_settings = _load_settings()

    if request.method == "POST":
        form = request.form
        defaults = _settings_defaults()
        updated: Dict[str, Any] = {}

        updated["output_format"] = _normalize_setting_value(
            "output_format", form.get("output_format"), defaults
        )
        updated["subtitle_format"] = _normalize_setting_value(
            "subtitle_format", form.get("subtitle_format"), defaults
        )
        updated["save_mode"] = _normalize_setting_value(
            "save_mode", form.get("save_mode"), defaults
        )
        updated["default_voice"] = _normalize_setting_value(
            "default_voice", form.get("default_voice"), defaults
        )
        for key in sorted(BOOLEAN_SETTINGS):
            updated[key] = _coerce_bool(form.get(key), False)
        updated["chunk_level"] = _normalize_setting_value(
            "chunk_level", form.get("chunk_level"), defaults
        )
        updated["speaker_mode"] = _normalize_setting_value(
            "speaker_mode", form.get("speaker_mode"), defaults
        )
        updated["separate_chapters_format"] = _normalize_setting_value(
            "separate_chapters_format", form.get("separate_chapters_format"), defaults
        )
        updated["silence_between_chapters"] = _coerce_float(
            form.get("silence_between_chapters"), defaults["silence_between_chapters"]
        )
        updated["chapter_intro_delay"] = _coerce_float(
            form.get("chapter_intro_delay"), defaults["chapter_intro_delay"]
        )
        updated["max_subtitle_words"] = _coerce_int(
            form.get("max_subtitle_words"), defaults["max_subtitle_words"]
        )
        updated["speaker_analysis_threshold"] = _coerce_int(
            form.get("speaker_analysis_threshold"),
            defaults["speaker_analysis_threshold"],
            minimum=1,
            maximum=25,
        )
        sentence_value = (form.get("speaker_pronunciation_sentence") or "").strip()
        if not sentence_value:
            sentence_value = defaults["speaker_pronunciation_sentence"]
        updated["speaker_pronunciation_sentence"] = sentence_value

        random_languages = [
            code.lower()
            for code in form.getlist("speaker_random_languages")
            if isinstance(code, str) and code.lower() in LANGUAGE_DESCRIPTIONS
        ]
        updated["speaker_random_languages"] = random_languages

        cfg = load_config() or {}
        cfg.update(updated)
        save_config(cfg)
        return redirect(url_for("web.settings_page", saved="1"))

    save_locations = [
        {"value": key, "label": label} for key, label in SAVE_MODE_LABELS.items()
    ]
    context = {
        "options": options,
        "settings": current_settings,
        "save_locations": save_locations,
        "default_output_dir": get_user_output_path(),
        "saved": request.args.get("saved") == "1",
    }
    return render_template("settings.html", **context)


@web_bp.get("/voices")
def voice_profiles_page() -> str:
    options = _template_options()
    return render_template("voices.html", options=options)


@web_bp.route("/speakers", methods=["GET", "POST"])
def speaker_configs_page() -> ResponseReturnValue:
    options = _template_options()
    configs = list_configs()
    message = None
    error = None

    if request.method == "POST":
        name, config_payload, errors = _extract_speaker_config_form(request.form)
        editing_payload = config_payload
        editing_name = name
        if errors:
            error = " ".join(errors)
            context = {
                "options": options,
                "configs": configs,
                "editing_name": editing_name,
                "editing": editing_payload,
                "message": message,
                "error": error,
            }
            return render_template("speakers.html", **context)
        upsert_config(name, config_payload)
        return redirect(url_for("web.speaker_configs_page", config=name, saved="1"))

    editing_name = request.args.get("config") or ""
    editing_payload = get_config(editing_name) if editing_name else None
    if editing_payload is None and configs:
        editing_name = configs[0]["name"]
        editing_payload = get_config(editing_name)
    if editing_payload is None:
        editing_payload = {
            "language": "a",
            "languages": [],
            "default_voice": "",
            "speakers": {},
            "notes": "",
            "version": 1,
        }

    if request.args.get("saved") == "1":
        message = "Speaker configuration saved."

    context = {
        "options": options,
        "configs": configs,
        "editing_name": editing_name,
        "editing": editing_payload,
        "message": message,
        "error": error,
    }
    return render_template("speakers.html", **context)


@web_bp.post("/speakers/<name>/delete")
def delete_speaker_config_route(name: str) -> ResponseReturnValue:
    delete_config(name)
    return redirect(url_for("web.speaker_configs_page"))


@web_bp.post("/voices")
def save_voice_profile_route() -> ResponseReturnValue:
    name = request.form.get("name", "").strip()
    language = request.form.get("language", "a").strip() or "a"
    formula = request.form.get("formula", "").strip()
    if not name or not formula:
        abort(400, "Name and formula are required")
    voices = _parse_voice_formula(formula)
    profiles = load_profiles()
    profiles[name] = {"voices": voices, "language": language}
    save_profiles(profiles)
    return redirect(url_for("web.voice_profiles_page"))


@web_bp.post("/voices/<name>/delete")
def delete_voice_profile_route(name: str) -> ResponseReturnValue:
    delete_profile(name)
    return redirect(url_for("web.voice_profiles_page"))


@api_bp.get("/voice-profiles")
def api_list_voice_profiles() -> ResponseReturnValue:
    return jsonify(_profiles_payload())


@api_bp.post("/voice-profiles")
def api_save_voice_profile() -> ResponseReturnValue:
    payload = request.get_json(force=True, silent=False)
    name = (payload.get("name") or "").strip()
    if not name:
        abort(400, "Profile name is required")

    original = (payload.get("originalName") or "").strip()
    language = (payload.get("language") or "a").strip() or "a"
    formula = (payload.get("formula") or "").strip()

    try:
        if formula:
            voices = _parse_voice_formula(formula)
        else:
            voices_raw = _sanitize_voice_entries(payload.get("voices", []))
            voices = normalize_voice_entries(voices_raw)
        if not voices:
            raise ValueError("At least one voice must be enabled with a weight above zero")
        save_profile(name, language=language, voices=voices)
        if original and original != name:
            remove_profile(original)
    except ValueError as exc:
        abort(400, str(exc))

    return jsonify({"ok": True, "profile": name, **_profiles_payload()})


@api_bp.delete("/voice-profiles/<name>")
def api_delete_voice_profile(name: str) -> ResponseReturnValue:
    remove_profile(name)
    return jsonify({"ok": True, **_profiles_payload()})


@api_bp.post("/voice-profiles/<name>/duplicate")
def api_duplicate_voice_profile(name: str) -> ResponseReturnValue:
    payload = request.get_json(silent=True) or {}
    new_name = (payload.get("name") or payload.get("new_name") or "").strip()
    if not new_name:
        abort(400, "Duplicate name is required")
    duplicate_profile(name, new_name)
    return jsonify({"ok": True, "profile": new_name, **_profiles_payload()})


@api_bp.post("/voice-profiles/import")
def api_import_voice_profiles() -> ResponseReturnValue:
    replace = False
    data: Optional[Dict[str, Any]] = None
    if "file" in request.files:
        file_storage = request.files["file"]
        try:
            file_storage.stream.seek(0)
            raw_bytes = file_storage.read()
            text_payload = raw_bytes.decode("utf-8")
            data = json.loads(text_payload)
        except UnicodeDecodeError as exc:
            abort(400, f"JSON file must be UTF-8 encoded: {exc}")
        except Exception as exc:  # pragma: no cover - defensive
            abort(400, f"Invalid JSON file: {exc}")
        replace = request.form.get("replace_existing") in {"true", "1", "on"}
    else:
        payload = request.get_json(force=True, silent=False)
        replace = bool(payload.get("replace_existing", False))
        data = payload.get("profiles") or payload.get("data") or payload
        if not isinstance(data, dict):
            data = None
    if data is None:
        abort(400, "Import payload must be a dictionary")
    data_dict = cast(Dict[str, Any], data)
    imported: List[str] = []
    try:
        imported = import_profiles_data(data_dict, replace_existing=replace)
    except ValueError as exc:
        abort(400, str(exc))
    return jsonify({"ok": True, "imported": imported, **_profiles_payload()})


@api_bp.get("/voice-profiles/export")
def api_export_voice_profiles() -> ResponseReturnValue:
    names_param = request.args.get("names")
    names = None
    if names_param:
        names = [name.strip() for name in names_param.split(",") if name.strip()]
    payload = export_profiles_payload(names)
    buffer = io.BytesIO()
    buffer.write(json.dumps(payload, indent=2).encode("utf-8"))
    buffer.seek(0)
    filename = request.args.get("filename") or "voice_profiles.json"
    return send_file(
        buffer,
        mimetype="application/json",
        as_attachment=True,
        download_name=filename,
    )


@api_bp.post("/voice-profiles/preview")
def api_preview_voice_mix() -> ResponseReturnValue:
    payload = request.get_json(force=True, silent=False)
    language = (payload.get("language") or "a").strip() or "a"
    text = (payload.get("text") or "").strip()
    speed = float(payload.get("speed", 1.0) or 1.0)
    try:
        requested_preview = float(payload.get("max_seconds", 60.0) or 60.0)
    except (TypeError, ValueError):
        requested_preview = 60.0
    max_seconds = max(1.0, min(60.0, requested_preview))
    profile_name = (payload.get("profile") or payload.get("profile_name") or "").strip()
    formula = (payload.get("formula") or "").strip()

    voices: List[Tuple[str, float]] = []
    if profile_name:
        profiles = load_profiles()
        entry = profiles.get(profile_name)
        if entry is None:
            abort(404, "Profile not found")
        if not isinstance(entry, dict):
            abort(400, "Profile data is invalid")
        entry_dict = cast(Dict[str, Any], entry)
        language = entry_dict.get("language", language)
        profile_voices = entry_dict.get("voices", [])
        for item in profile_voices:
            if isinstance(item, (list, tuple)) and len(item) >= 2:
                try:
                    voices.append((str(item[0]), float(item[1])))
                except (TypeError, ValueError):
                    continue
    else:
        try:
            if formula:
                voices = _parse_voice_formula(formula)
            else:
                voices_raw = _sanitize_voice_entries(payload.get("voices", []))
                voices = normalize_voice_entries(voices_raw)
        except ValueError as exc:
            abort(400, str(exc))

    if not voices:
        abort(400, "At least one voice must be provided for preview")

    if not text:
        text = SAMPLE_VOICE_TEXTS.get(language, SAMPLE_VOICE_TEXTS.get("a", "This is a sample of the selected voice."))

    settings = _load_settings()
    use_gpu_default = settings.get("use_gpu", True)
    if "use_gpu" in payload:
        use_gpu = _coerce_bool(payload.get("use_gpu"), use_gpu_default)
    else:
        use_gpu = use_gpu_default
    device = "cpu"
    if use_gpu:
        try:
            device = _select_device()
        except Exception:  # pragma: no cover - fallback
            device = "cpu"
            use_gpu = False

    pipeline: Any = None
    try:
        pipeline = _get_preview_pipeline(language, device)
    except Exception as exc:  # pragma: no cover - defensive guard
        abort(500, f"Failed to initialise preview pipeline: {exc}")
    if pipeline is None:  # pragma: no cover - defensive double-check
        abort(500, "Preview pipeline initialisation failed")

    voice_choice: Any = None
    if len(voices) == 1:
        voice_choice = voices[0][0]
    else:
        formula_value = _pairs_to_formula(voices)
        if not formula_value:
            abort(400, "Invalid voice weights provided")
        try:
            voice_choice = get_new_voice(pipeline, formula_value, use_gpu)
        except ValueError as exc:
            abort(400, str(exc))
    if voice_choice is None:
        abort(400, "Unable to resolve voice selection")

    segments = pipeline(
        text,
        voice=voice_choice,
        speed=speed,
        split_pattern=SPLIT_PATTERN,
    )

    audio_chunks: List[np.ndarray] = []
    accumulated = 0
    max_samples = int(max_seconds * SAMPLE_RATE)

    for segment in segments:
        graphemes = segment.graphemes.strip()
        if not graphemes:
            continue
        audio = _to_float32(segment.audio)
        if audio.size == 0:
            continue
        remaining = max_samples - accumulated
        if remaining <= 0:
            break
        if audio.shape[0] > remaining:
            audio = audio[:remaining]
        audio_chunks.append(audio)
        accumulated += audio.shape[0]
        if accumulated >= max_samples:
            break

    if not audio_chunks:
        abort(500, "Preview could not be generated")

    audio_data = np.concatenate(audio_chunks)
    buffer = io.BytesIO()
    sf.write(buffer, audio_data, SAMPLE_RATE, format="WAV")
    buffer.seek(0)
    response = send_file(
        buffer,
        mimetype="audio/wav",
        as_attachment=False,
        download_name="voice_preview.wav",
    )
    response.headers["Cache-Control"] = "no-store"
    return response


@api_bp.post("/speaker-preview")
def api_speaker_preview() -> ResponseReturnValue:
    payload = request.get_json(force=True, silent=False)
    text = (payload.get("text") or "").strip()
    voice_spec = (payload.get("voice") or "").strip()
    language = (payload.get("language") or "a").strip() or "a"
    speed_input = payload.get("speed", 1.0)
    try:
        speed = float(speed_input)
    except (TypeError, ValueError):
        speed = 1.0
    max_seconds_input = payload.get("max_seconds", 8.0)
    try:
        max_seconds = max(1.0, min(15.0, float(max_seconds_input)))
    except (TypeError, ValueError):
        max_seconds = 8.0

    if not text:
        abort(400, "Preview text is required")
    if not voice_spec:
        abort(400, "Voice selection is required")

    settings = _load_settings()
    use_gpu_default = settings.get("use_gpu", True)
    if "use_gpu" in payload:
        use_gpu = _coerce_bool(payload.get("use_gpu"), use_gpu_default)
    else:
        use_gpu = use_gpu_default

    device = "cpu"
    if use_gpu:
        try:
            device = _select_device()
        except Exception:  # pragma: no cover - fallback
            device = "cpu"
            use_gpu = False

    try:
        pipeline = _get_preview_pipeline(language, device)
    except Exception as exc:  # pragma: no cover - defensive guard
        abort(500, f"Failed to initialise preview pipeline: {exc}")
    if pipeline is None:  # pragma: no cover - defensive double-check
        abort(500, "Preview pipeline initialisation failed")

    voice_choice: Any = voice_spec
    if "*" in voice_spec:
        try:
            voice_choice = get_new_voice(pipeline, voice_spec, use_gpu)
        except ValueError as exc:
            abort(400, str(exc))

    segments = pipeline(
        text,
        voice=voice_choice,
        speed=speed,
        split_pattern=SPLIT_PATTERN,
    )

    audio_chunks: List[np.ndarray] = []
    accumulated = 0
    max_samples = int(max_seconds * SAMPLE_RATE)

    for segment in segments:
        graphemes = getattr(segment, "graphemes", "").strip()
        if not graphemes:
            continue
        audio = _to_float32(getattr(segment, "audio", None))
        if audio.size == 0:
            continue
        remaining = max_samples - accumulated
        if remaining <= 0:
            break
        if audio.shape[0] > remaining:
            audio = audio[:remaining]
        audio_chunks.append(audio)
        accumulated += audio.shape[0]
        if accumulated >= max_samples:
            break

    if not audio_chunks:
        abort(500, "Preview could not be generated")

    audio_data = np.concatenate(audio_chunks)
    buffer = io.BytesIO()
    sf.write(buffer, audio_data, SAMPLE_RATE, format="WAV")
    buffer.seek(0)
    response = send_file(
        buffer,
        mimetype="audio/wav",
        as_attachment=False,
        download_name="speaker_preview.wav",
    )
    response.headers["Cache-Control"] = "no-store"
    return response


@web_bp.post("/jobs")
def enqueue_job() -> ResponseReturnValue:
    service = _service()
    uploads_dir = Path(current_app.config["UPLOAD_FOLDER"])
    uploads_dir.mkdir(parents=True, exist_ok=True)

    file = request.files.get("source_file")
    text_input = request.form.get("source_text", "").strip()

    if not file and not text_input:
        return redirect(url_for("web.index"))

    stored_path: Path
    original_name: str

    if file and file.filename:
        filename = secure_filename(file.filename)
        if not filename:
            return redirect(url_for("web.index"))
        stored_path = uploads_dir / f"{uuid.uuid4().hex}_{filename}"
        file.save(stored_path)
        original_name = filename
    else:
        original_name = "direct_text.txt"
        stored_path = uploads_dir / f"{uuid.uuid4().hex}_{original_name}"
        stored_path.write_text(text_input, encoding="utf-8")

    extraction = None
    try:
        extraction = extract_from_path(stored_path)
    except Exception as exc:  # pragma: no cover - defensive
        try:
            stored_path.unlink(missing_ok=True)
        except Exception:
            pass
        abort(400, f"Unable to read the supplied content: {exc}")

    if extraction is None:  # pragma: no cover - defensive
        abort(400, "Unable to read the supplied content")

    assert extraction is not None

    cover_path, cover_mime = _persist_cover_image(extraction, stored_path)

    metadata_tags = extraction.metadata or {}
    total_chars = extraction.total_characters or calculate_text_length(extraction.combined_text)
    total_chapter_count = len(extraction.chapters)
    chapters_payload: List[Dict[str, Any]] = []
    for index, chapter in enumerate(extraction.chapters):
        enabled = _should_preselect_chapter(
            chapter.title,
            chapter.text,
            index,
            total_chapter_count,
        )
        chapters_payload.append(
            {
                "id": f"{index:04d}",
                "index": index,
                "title": chapter.title,
                "text": chapter.text,
                "characters": calculate_text_length(chapter.text),
                "enabled": enabled,
            }
        )

    if not chapters_payload:
        chapters_payload.append(
            {
                "id": "0000",
                "index": 0,
                "title": original_name,
                "text": "",
                "characters": 0,
                "enabled": True,
            }
        )

    _ensure_at_least_one_chapter_enabled(chapters_payload)

    profiles = load_profiles()
    settings = _load_settings()

    language = request.form.get("language", "a")
    base_voice = request.form.get("voice", "af_alloy")
    profile_selection = (request.form.get("voice_profile") or "__standard").strip()
    custom_formula_raw = request.form.get("voice_formula", "").strip()
    selected_speaker_config = (request.form.get("speaker_config") or "").strip()
    speaker_config_payload = get_config(selected_speaker_config) if selected_speaker_config else None
    config_randomize_default = False
    if isinstance(speaker_config_payload, Mapping):
        speakers_map = speaker_config_payload.get("speakers")
        if isinstance(speakers_map, Mapping):
            for config_entry in speakers_map.values():
                if isinstance(config_entry, Mapping) and config_entry.get("randomize"):
                    config_randomize_default = True
                    break

    if profile_selection in {"__standard", ""}:
        profile_name = ""
        custom_formula = ""
    elif profile_selection == "__formula":
        profile_name = ""
        custom_formula = custom_formula_raw
    else:
        profile_name = profile_selection
        custom_formula = ""

    voice, language, selected_profile = _resolve_voice_choice(
        language,
        base_voice,
        profile_name,
        custom_formula,
        profiles,
    )
    speed = float(request.form.get("speed", "1.0"))
    subtitle_mode = request.form.get("subtitle_mode", "Disabled")
    output_format = settings["output_format"]
    subtitle_format = settings["subtitle_format"]
    save_mode_key = settings["save_mode"]
    save_mode = SAVE_MODE_LABELS.get(save_mode_key, SAVE_MODE_LABELS["save_next_to_input"])
    replace_single_newlines = settings["replace_single_newlines"]
    use_gpu = settings["use_gpu"]
    save_chapters_separately = settings["save_chapters_separately"]
    merge_chapters_at_end = settings["merge_chapters_at_end"] or not save_chapters_separately
    save_as_project = settings["save_as_project"]
    separate_chapters_format = settings["separate_chapters_format"]
    silence_between_chapters = settings["silence_between_chapters"]
    chapter_intro_delay = settings["chapter_intro_delay"]
    max_subtitle_words = settings["max_subtitle_words"]

    chunk_level_default = str(settings.get("chunk_level", "paragraph")).strip().lower()
    raw_chunk_level = (request.form.get("chunk_level") or chunk_level_default).strip().lower()
    if raw_chunk_level not in _CHUNK_LEVEL_VALUES:
        raw_chunk_level = chunk_level_default if chunk_level_default in _CHUNK_LEVEL_VALUES else "paragraph"
    chunk_level_value = raw_chunk_level
    chunk_level_literal = cast(ChunkLevel, chunk_level_value)

    speaker_mode_default = str(settings.get("speaker_mode", "single")).strip().lower()
    raw_speaker_mode = (request.form.get("speaker_mode") or speaker_mode_default).strip().lower()
    if raw_speaker_mode not in _SPEAKER_MODE_VALUES:
        raw_speaker_mode = "single"
    speaker_mode_value = raw_speaker_mode

    generate_epub3_default = bool(settings.get("generate_epub3", False))
    generate_epub3 = _coerce_bool(request.form.get("generate_epub3"), generate_epub3_default)

    selected_chapter_sources = [entry for entry in chapters_payload if entry.get("enabled")]
    raw_chunks = build_chunks_for_chapters(selected_chapter_sources, level=chunk_level_literal)
    analysis_chunks = build_chunks_for_chapters(selected_chapter_sources, level="sentence")

    analysis_threshold = _coerce_int(
        settings.get("speaker_analysis_threshold"),
        _DEFAULT_ANALYSIS_THRESHOLD,
        minimum=1,
        maximum=25,
    )

    initial_analysis = speaker_mode_value == "multi"
    processed_chunks, speakers, analysis_payload, config_languages, _ = _prepare_speaker_metadata(
        chapters=selected_chapter_sources,
        chunks=raw_chunks,
        analysis_chunks=analysis_chunks,
        speaker_mode=speaker_mode_value,
        voice=voice,
        voice_profile=selected_profile or None,
        threshold=analysis_threshold,
        run_analysis=initial_analysis,
        speaker_config=speaker_config_payload,
        apply_config=bool(speaker_config_payload),
        allow_randomize=config_randomize_default,
    )

    pending = PendingJob(
        id=uuid.uuid4().hex,
        original_filename=original_name,
        stored_path=stored_path,
        language=language,
        voice=voice,
        speed=speed,
        use_gpu=use_gpu,
        subtitle_mode=subtitle_mode,
        output_format=output_format,
        save_mode=save_mode,
        output_folder=None,
        replace_single_newlines=replace_single_newlines,
        subtitle_format=subtitle_format,
        total_characters=total_chars,
        save_chapters_separately=save_chapters_separately,
        merge_chapters_at_end=merge_chapters_at_end,
        separate_chapters_format=separate_chapters_format,
        silence_between_chapters=silence_between_chapters,
        save_as_project=save_as_project,
        voice_profile=selected_profile or None,
        max_subtitle_words=max_subtitle_words,
        metadata_tags=metadata_tags,
        chapters=chapters_payload,
        created_at=time.time(),
        cover_image_path=cover_path,
        cover_image_mime=cover_mime,
        chapter_intro_delay=chapter_intro_delay,
        chunk_level=chunk_level_value,
        speaker_mode=speaker_mode_value,
        generate_epub3=generate_epub3,
        chunks=processed_chunks,
        speakers=speakers,
        speaker_analysis=analysis_payload,
        speaker_analysis_threshold=analysis_threshold,
        analysis_requested=initial_analysis,
    )

    service.store_pending_job(pending)
    pending.applied_speaker_config = selected_speaker_config or None
    if config_languages:
        pending.speaker_voice_languages = list(config_languages)
    elif isinstance(speaker_config_payload, Mapping):
        languages = speaker_config_payload.get("languages")
        if isinstance(languages, list):
            pending.speaker_voice_languages = [code for code in languages if isinstance(code, str)]
    return redirect(url_for("web.prepare_job", pending_id=pending.id))


@web_bp.get("/jobs/prepare/<pending_id>")
def prepare_job(pending_id: str) -> str:
    pending = _service().get_pending_job(pending_id)
    if not pending:
        abort(404)
    pending = cast(PendingJob, pending)
    return _render_prepare_page(pending, active_step="chapters")


@web_bp.post("/jobs/prepare/<pending_id>/analyze")
def analyze_pending_job(pending_id: str) -> ResponseReturnValue:
    service = _service()
    pending = service.get_pending_job(pending_id)
    if not pending:
        abort(404)
    pending = cast(PendingJob, pending)

    (
        chunk_level_literal,
        overrides,
        enabled_overrides,
        errors,
        selected_total,
        selected_config,
        randomize_requested,
        apply_config_requested,
        persist_config_requested,
    ) = _apply_prepare_form(pending, request.form)

    if errors:
        return _render_prepare_page(pending, error=" ".join(errors), active_step="chapters")

    if pending.speaker_mode != "multi":
        setattr(pending, "analysis_requested", False)
        pending.chunks = []
        pending.speaker_analysis = {}
        return _render_prepare_page(
            pending,
            error="Switch to multi-speaker mode to analyze speakers.",
            active_step="chapters",
        )

    if not enabled_overrides:
        setattr(pending, "analysis_requested", False)
        pending.chunks = []
        pending.speaker_analysis = {}
        return _render_prepare_page(
            pending,
            error="Select at least one chapter to analyze.",
            active_step="chapters",
        )

    raw_chunks = build_chunks_for_chapters(enabled_overrides, level=chunk_level_literal)
    analysis_chunks = build_chunks_for_chapters(enabled_overrides, level="sentence")
    analysis_chunks = build_chunks_for_chapters(enabled_overrides, level="sentence")
    analysis_chunks = build_chunks_for_chapters(enabled_overrides, level="sentence")

    existing_roster: Optional[Mapping[str, Any]]
    if getattr(pending, "analysis_requested", False):
        existing_roster = pending.speakers
    else:
        existing_roster = None

    config_name = pending.applied_speaker_config or selected_config
    speaker_config_payload = get_config(config_name) if config_name else None
    processed_chunks, roster, analysis_payload, config_languages, updated_config = _prepare_speaker_metadata(
        chapters=enabled_overrides,
        chunks=raw_chunks,
        analysis_chunks=analysis_chunks,
        speaker_mode=pending.speaker_mode,
        voice=pending.voice,
        voice_profile=pending.voice_profile,
        threshold=pending.speaker_analysis_threshold,
        existing_roster=existing_roster,
        run_analysis=True,
        speaker_config=speaker_config_payload,
        apply_config=apply_config_requested or bool(speaker_config_payload),
        allow_randomize=randomize_requested,
        persist_config=persist_config_requested,
    )

    pending.chunks = processed_chunks
    pending.speakers = roster
    pending.speaker_analysis = analysis_payload
    if config_languages:
        pending.speaker_voice_languages = list(config_languages)
    config_name = getattr(pending, "applied_speaker_config", None)
    if updated_config and isinstance(config_name, str) and config_name:
        configs = load_configs()
        configs[config_name] = updated_config
        save_configs(configs)
    setattr(pending, "analysis_requested", True)
    if selected_total:
        pending.total_characters = selected_total

    service.store_pending_job(pending)

    notice_message = "Speaker analysis updated."
    if persist_config_requested and config_name:
        notice_message = "Speaker analysis updated and configuration saved."
    return _render_prepare_page(pending, notice=notice_message, active_step="speakers")


@web_bp.post("/jobs/prepare/<pending_id>")
def finalize_job(pending_id: str) -> ResponseReturnValue:
    service = _service()
    pending = service.get_pending_job(pending_id)
    if not pending:
        abort(404)
    pending = cast(PendingJob, pending)

    (
        chunk_level_literal,
        overrides,
        enabled_overrides,
        errors,
        selected_total,
        selected_config,
        randomize_requested,
        apply_config_requested,
        persist_config_requested,
    ) = _apply_prepare_form(pending, request.form)

    if errors:
        return _render_prepare_page(
            pending,
            error=" ".join(errors),
            active_step=request.form.get("active_step") or "speakers",
        )

    if pending.speaker_mode != "multi":
        setattr(pending, "analysis_requested", False)

    if not enabled_overrides:
        pending.chunks = []
        return _render_prepare_page(
            pending,
            error="Select at least one chapter to convert.",
            active_step="chapters",
        )

    raw_chunks = build_chunks_for_chapters(enabled_overrides, level=chunk_level_literal)
    analysis_chunks = build_chunks_for_chapters(enabled_overrides, level="sentence")
    analysis_active = pending.speaker_mode == "multi" and getattr(pending, "analysis_requested", False)
    if analysis_active:
        existing_roster: Optional[Mapping[str, Any]] = pending.speakers
    else:
        narrator_only: Dict[str, Any] = {}
        if isinstance(pending.speakers, dict):
            narrator_payload = pending.speakers.get("narrator")
            if isinstance(narrator_payload, Mapping):
                narrator_only["narrator"] = dict(narrator_payload)
        existing_roster = narrator_only or None
    config_name = pending.applied_speaker_config or selected_config
    speaker_config_payload = get_config(config_name) if config_name else None
    processed_chunks, roster, analysis_payload, config_languages, updated_config = _prepare_speaker_metadata(
        chapters=enabled_overrides,
        chunks=raw_chunks,
        analysis_chunks=analysis_chunks,
        speaker_mode=pending.speaker_mode,
        voice=pending.voice,
        voice_profile=pending.voice_profile,
        threshold=pending.speaker_analysis_threshold,
        existing_roster=existing_roster,
        run_analysis=analysis_active,
        speaker_config=speaker_config_payload,
        apply_config=apply_config_requested or bool(speaker_config_payload),
        allow_randomize=randomize_requested,
        persist_config=persist_config_requested,
    )
    pending.chunks = processed_chunks
    pending.speakers = roster
    pending.speaker_analysis = analysis_payload
    if config_languages:
        pending.speaker_voice_languages = list(config_languages)
    config_name = getattr(pending, "applied_speaker_config", None)
    if updated_config and isinstance(config_name, str) and config_name:
        configs = load_configs()
        configs[config_name] = updated_config
        save_configs(configs)

    total_characters = selected_total or pending.total_characters

    service.pop_pending_job(pending_id)

    job = service.enqueue(
        original_filename=pending.original_filename,
        stored_path=pending.stored_path,
        language=pending.language,
        voice=pending.voice,
        speed=pending.speed,
        use_gpu=pending.use_gpu,
        subtitle_mode=pending.subtitle_mode,
        output_format=pending.output_format,
        save_mode=pending.save_mode,
        output_folder=pending.output_folder,
        replace_single_newlines=pending.replace_single_newlines,
        subtitle_format=pending.subtitle_format,
        total_characters=total_characters,
        chapters=overrides,
        metadata_tags=pending.metadata_tags,
        save_chapters_separately=pending.save_chapters_separately,
        merge_chapters_at_end=pending.merge_chapters_at_end,
        separate_chapters_format=pending.separate_chapters_format,
        silence_between_chapters=pending.silence_between_chapters,
        save_as_project=pending.save_as_project,
        voice_profile=pending.voice_profile,
        max_subtitle_words=pending.max_subtitle_words,
        cover_image_path=pending.cover_image_path,
        cover_image_mime=pending.cover_image_mime,
        chapter_intro_delay=pending.chapter_intro_delay,
        chunk_level=pending.chunk_level,
        chunks=processed_chunks,
        speakers=roster,
        speaker_mode=pending.speaker_mode,
        speaker_analysis=analysis_payload,
        speaker_analysis_threshold=pending.speaker_analysis_threshold,
        generate_epub3=pending.generate_epub3,
        analysis_requested=getattr(pending, "analysis_requested", False),
    )
    if config_languages:
        job.speaker_voice_languages = list(config_languages)
    config_name = getattr(pending, "applied_speaker_config", None)
    if updated_config and isinstance(config_name, str) and config_name:
        job.applied_speaker_config = config_name
        configs = load_configs()
        configs[config_name] = updated_config
        save_configs(configs)
    elif isinstance(config_name, str) and config_name:
        job.applied_speaker_config = config_name
        job.speaker_voice_languages = job.speaker_voice_languages or list(pending.speaker_voice_languages)

    return redirect(url_for("web.index", _anchor="queue"))


@web_bp.post("/jobs/prepare/<pending_id>/cancel")
def cancel_pending_job(pending_id: str) -> ResponseReturnValue:
    pending = _service().pop_pending_job(pending_id)
    if pending and pending.stored_path.exists():
        try:
            pending.stored_path.unlink()
        except OSError:
            pass
    if pending and pending.cover_image_path and pending.cover_image_path.exists():
        try:
            pending.cover_image_path.unlink()
        except OSError:
            pass
    return redirect(url_for("web.index", _anchor="queue"))


def _render_jobs_panel() -> str:
    jobs = _service().list_jobs()
    active_statuses = {JobStatus.PENDING, JobStatus.RUNNING, JobStatus.PAUSED}
    active_jobs = [job for job in jobs if job.status in active_statuses]
    active_jobs.sort(key=lambda job: ((job.queue_position or 10_000), -job.created_at))
    finished_jobs = [job for job in jobs if job.status not in active_statuses]
    return render_template(
        "partials/jobs.html",
        active_jobs=active_jobs,
        finished_jobs=finished_jobs[:5],
        total_finished=len(finished_jobs),
        JobStatus=JobStatus,
    )


def _render_prepare_page(
    pending: PendingJob,
    *,
    error: Optional[str] = None,
    notice: Optional[str] = None,
    active_step: Optional[str] = None,
) -> str:
    if not active_step:
        active_step = (
            request.form.get("active_step")
            if request.method == "POST"
            else request.args.get("step")
        ) or "chapters"
    return render_template(
        "prepare_job.html",
        pending=pending,
        options=_template_options(),
        settings=_load_settings(),
        error=error,
        notice=notice,
        active_step=active_step,
    )


@web_bp.get("/jobs/<job_id>")
def job_detail(job_id: str) -> str:
    job = _service().get_job(job_id)
    if not job:
        abort(404)
    return render_template(
        "job_detail.html",
        job=job,
        options=_template_options(),
    )


@web_bp.post("/jobs/<job_id>/pause")
def pause_job(job_id: str) -> ResponseReturnValue:
    _service().pause(job_id)
    if request.headers.get("HX-Request"):
        return _render_jobs_panel()
    return redirect(url_for("web.job_detail", job_id=job_id))


@web_bp.post("/jobs/<job_id>/resume")
def resume_job(job_id: str) -> ResponseReturnValue:
    _service().resume(job_id)
    if request.headers.get("HX-Request"):
        return _render_jobs_panel()
    return redirect(url_for("web.job_detail", job_id=job_id))


@web_bp.post("/jobs/<job_id>/cancel")
def cancel_job(job_id: str) -> ResponseReturnValue:
    _service().cancel(job_id)
    if request.headers.get("HX-Request"):
        return _render_jobs_panel()
    return redirect(url_for("web.job_detail", job_id=job_id))


@web_bp.post("/jobs/<job_id>/delete")
def delete_job(job_id: str) -> ResponseReturnValue:
    _service().delete(job_id)
    if request.headers.get("HX-Request"):
        return _render_jobs_panel()
    return redirect(url_for("web.index"))


@web_bp.post("/jobs/clear-finished")
def clear_finished_jobs() -> ResponseReturnValue:
    _service().clear_finished()
    if request.headers.get("HX-Request"):
        return _render_jobs_panel()
    return redirect(url_for("web.index", _anchor="queue"))


@web_bp.get("/jobs/<job_id>/download")
def download_job(job_id: str) -> ResponseReturnValue:
    job = _service().get_job(job_id)
    if job is None or job.status != JobStatus.COMPLETED:
        abort(404)
    result = getattr(job, "result", None)
    audio_path = getattr(result, "audio_path", None)
    if audio_path is None:
        abort(404)
    if not isinstance(audio_path, Path):  # pragma: no cover - sanity guard
        abort(404)
    audio_path_path = cast(Path, audio_path)
    if not audio_path_path.exists():
        abort(404)
    mime_type, _ = mimetypes.guess_type(str(audio_path_path))
    return send_file(
        audio_path_path,
        mimetype=mime_type or "application/octet-stream",
        as_attachment=True,
        download_name=audio_path_path.name,
    )


@web_bp.get("/partials/jobs")
def jobs_partial() -> str:
    return _render_jobs_panel()

@web_bp.get("/partials/jobs/<job_id>/logs")
def job_logs_partial(job_id: str) -> str:
    job = _service().get_job(job_id)
    if not job:
        abort(404)
    return render_template("partials/logs.html", job=job)


@api_bp.get("/jobs/<job_id>")
def job_json(job_id: str) -> ResponseReturnValue:
    job = _service().get_job(job_id)
    if job is None:
        abort(404)
    if not isinstance(job, Job):  # pragma: no cover - defensive guard
        abort(404)
    job_obj = cast(Job, job)
    payload = job_obj.as_dict()
    return jsonify(payload)