diff --git a/abogen/domain/metadata_overrides.py b/abogen/domain/metadata_overrides.py new file mode 100644 index 0000000..bd5b9fa --- /dev/null +++ b/abogen/domain/metadata_overrides.py @@ -0,0 +1,99 @@ +"""OPDS metadata normalization. + +Normalizes metadata keys from various OPDS/Calibre sources into +a canonical set of overrides for the audiobook conversion pipeline. +""" + +from __future__ import annotations + +from typing import Any, Dict, Mapping + + +def normalize_opds_metadata(metadata_payload: Mapping[str, Any]) -> Dict[str, Any]: + """Normalize OPDS/Calibre metadata into canonical override keys. + + Takes a metadata payload with various key aliases (e.g. 'series'/'series_name', + 'tags'/'keywords', 'authors'/'creator') and returns a dict with canonical + keys set. + + Args: + metadata_payload: Raw metadata dict from OPDS/Calibre import. + + Returns: + Dict with canonical metadata keys (series, series_index, tags, + description, subtitle, publisher, authors). + """ + metadata_overrides: Dict[str, Any] = {} + + def _stringify(value: Any) -> str: + if value is None: + return "" + if isinstance(value, (list, tuple, set)): + parts = [str(item).strip() for item in value if item is not None] + return ", ".join(part for part in parts if part) + return str(value).strip() + + raw_series = metadata_payload.get("series") or metadata_payload.get("series_name") + series_name = str(raw_series or "").strip() + if series_name: + metadata_overrides["series"] = series_name + metadata_overrides.setdefault("series_name", series_name) + + series_index_value = ( + metadata_payload.get("series_index") + or metadata_payload.get("series_position") + or metadata_payload.get("series_sequence") + or metadata_payload.get("book_number") + ) + if series_index_value is not None: + series_index_text = str(series_index_value).strip() + if series_index_text: + metadata_overrides.setdefault("series_index", series_index_text) + metadata_overrides.setdefault("series_position", series_index_text) + metadata_overrides.setdefault("series_sequence", series_index_text) + metadata_overrides.setdefault("book_number", series_index_text) + + tags_value = metadata_payload.get("tags") or metadata_payload.get("keywords") + if tags_value: + tags_text = _stringify(tags_value) + if tags_text: + metadata_overrides.setdefault("tags", tags_text) + metadata_overrides.setdefault("keywords", tags_text) + metadata_overrides.setdefault("genre", tags_text) + + description_value = metadata_payload.get("description") or metadata_payload.get("summary") + if description_value: + description_text = _stringify(description_value) + if description_text: + metadata_overrides.setdefault("description", description_text) + metadata_overrides.setdefault("summary", description_text) + + subtitle_value = ( + metadata_payload.get("subtitle") + or metadata_payload.get("sub_title") + or metadata_payload.get("calibre_subtitle") + ) + if subtitle_value: + subtitle_text = _stringify(subtitle_value) + if subtitle_text: + metadata_overrides.setdefault("subtitle", subtitle_text) + + publisher_value = metadata_payload.get("publisher") + if publisher_value: + publisher_text = _stringify(publisher_value) + if publisher_text: + metadata_overrides.setdefault("publisher", publisher_text) + + authors_value = ( + metadata_payload.get("authors") + or metadata_payload.get("author") + or metadata_payload.get("creator") + or metadata_payload.get("dc_creator") + ) + if authors_value: + authors_text = _stringify(authors_value) + if authors_text: + metadata_overrides.setdefault("authors", authors_text) + metadata_overrides.setdefault("author", authors_text) + + return metadata_overrides diff --git a/abogen/domain/settings_core.py b/abogen/domain/settings_core.py new file mode 100644 index 0000000..542d0be --- /dev/null +++ b/abogen/domain/settings_core.py @@ -0,0 +1,363 @@ +"""Shared settings core. + +Pure settings logic (defaults, coercion, normalization) used by both +Web UI and Desktop GUI. No Flask dependencies. +""" + +from __future__ import annotations + +import os +import re +from typing import Any, Dict, Mapping + +from abogen.constants import ( + LANGUAGE_DESCRIPTIONS, + SUBTITLE_FORMATS, + SUPPORTED_SOUND_FORMATS, +) +from abogen.tts_plugin.utils import get_default_voice +from abogen.normalization_settings import ( + DEFAULT_LLM_PROMPT, + environment_llm_defaults, +) + +# ── Constants ──────────────────────────────────────────────────────── + +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()} + +CHUNK_LEVEL_OPTIONS = [ + {"value": "paragraph", "label": "Paragraphs"}, + {"value": "sentence", "label": "Sentences"}, +] + +CHUNK_LEVEL_VALUES = {option["value"] for option in CHUNK_LEVEL_OPTIONS} + +DEFAULT_ANALYSIS_THRESHOLD = 3 + +BOOLEAN_SETTINGS = { + "replace_single_newlines", + "use_gpu", + "save_chapters_separately", + "merge_chapters_at_end", + "save_as_project", + "read_title_intro", + "read_closing_outro", + "normalize_chapter_opening_caps", + "generate_epub3", + "auto_prefix_chapter_titles", + "enable_entity_recognition", + "normalization_numbers", + "normalization_titles", + "normalization_terminal", + "normalization_phoneme_hints", + "normalization_caps_quotes", + "normalization_currency", + "normalization_footnotes", + "normalization_internet_slang", + "normalization_apostrophes_contractions", + "normalization_apostrophes_plural_possessives", + "normalization_apostrophes_sibilant_possessives", + "normalization_apostrophes_decades", + "normalization_apostrophes_leading_elisions", + "normalization_contraction_aux_be", + "normalization_contraction_aux_have", + "normalization_contraction_modal_will", + "normalization_contraction_modal_would", + "normalization_contraction_negation_not", + "normalization_contraction_let_us", +} + +FLOAT_SETTINGS = {"silence_between_chapters", "chapter_intro_delay", "llm_timeout"} + +INT_SETTINGS = {"max_subtitle_words", "speaker_analysis_threshold"} + +_NORMALIZATION_BOOLEAN_KEYS = { + "normalization_numbers", + "normalization_titles", + "normalization_terminal", + "normalization_phoneme_hints", + "normalization_caps_quotes", + "normalization_currency", + "normalization_footnotes", + "normalization_internet_slang", + "normalization_apostrophes_contractions", + "normalization_apostrophes_plural_possessives", + "normalization_apostrophes_sibilant_possessives", + "normalization_apostrophes_decades", + "normalization_apostrophes_leading_elisions", + "normalization_contraction_aux_be", + "normalization_contraction_aux_have", + "normalization_contraction_modal_will", + "normalization_contraction_modal_would", + "normalization_contraction_negation_not", + "normalization_contraction_let_us", +} + +_NORMALIZATION_STRING_KEYS = { + "normalization_apostrophe_mode", + "normalization_numbers_year_style", +} + + +# ── Coercion helpers ───────────────────────────────────────────────── + +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)) + + +# ── Profile/speaker spec helpers ───────────────────────────────────── + +def split_profile_spec(value: Any) -> tuple[str, str | None]: + """Split 'speaker:Name' or 'profile:Name' into (raw, name).""" + text = str(value or "").strip() + if not text: + return "", None + lowered = text.lower() + if lowered.startswith("profile:") or lowered.startswith("speaker:"): + _, _, remainder = text.partition(":") + name = remainder.strip() + return "", name or None + return text, None + + +# ── Normalization ──────────────────────────────────────────────────── + +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: + """Normalize a single setting value to its expected type.""" + 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): + text = value.strip() + if not text: + return defaults[key] + spec, profile_name = split_profile_spec(text) + if profile_name: + return f"speaker:{profile_name}" + return spec + return defaults[key] + if key == "default_speaker": + if isinstance(value, str): + text = value.strip() + if not text: + return "" + spec, profile_name = split_profile_spec(text) + if profile_name: + return f"speaker:{profile_name}" + return spec + return "" + if key == "chunk_level": + if isinstance(value, str) and value in CHUNK_LEVEL_VALUES: + return value + return defaults[key] + if key == "normalization_apostrophe_mode": + if isinstance(value, str): + normalized_mode = value.strip().lower() + if normalized_mode in {"off", "spacy", "llm"}: + return normalized_mode + return defaults[key] + if key == "normalization_numbers_year_style": + if isinstance(value, str): + normalized_style = value.strip().lower() + if normalized_style in {"american", "off"}: + return normalized_style + return defaults[key] + if key == "llm_context_mode": + if isinstance(value, str): + normalized_scope = value.strip().lower() + if normalized_scope == "sentence": + return normalized_scope + return defaults[key] + if key == "llm_prompt": + candidate = str(value or "").strip() + return candidate if candidate else defaults[key] + if key in {"llm_base_url", "llm_api_key", "llm_model"}: + return str(value or "").strip() + 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, []) + if key == "supertonic_total_steps": + try: + steps = int(value) + except (TypeError, ValueError): + return defaults.get(key, 5) + return max(2, min(15, steps)) + if key == "supertonic_speed": + try: + speed = float(value) + except (TypeError, ValueError): + return defaults.get(key, 1.0) + return max(0.7, min(2.0, speed)) + return value if value is not None else defaults.get(key) + + +# ── Defaults ───────────────────────────────────────────────────────── + +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]: + """Default values for all settings keys.""" + llm_env_defaults = environment_llm_defaults() + return { + "output_format": "wav", + "subtitle_format": "srt", + "save_mode": "default_output" if has_output_override() else "save_next_to_input", + "default_speaker": "", + "default_voice": get_default_voice("kokoro"), + "supertonic_total_steps": 5, + "supertonic_speed": 1.0, + "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, + "read_title_intro": False, + "read_closing_outro": True, + "normalize_chapter_opening_caps": True, + "max_subtitle_words": 50, + "chunk_level": "paragraph", + "enable_entity_recognition": True, + "generate_epub3": False, + "auto_prefix_chapter_titles": True, + "speaker_analysis_threshold": DEFAULT_ANALYSIS_THRESHOLD, + "speaker_pronunciation_sentence": "This is {{name}} speaking.", + "speaker_random_languages": [], + "llm_base_url": llm_env_defaults.get("llm_base_url", ""), + "llm_api_key": llm_env_defaults.get("llm_api_key", ""), + "llm_model": llm_env_defaults.get("llm_model", ""), + "llm_timeout": llm_env_defaults.get("llm_timeout", 30.0), + "llm_prompt": llm_env_defaults.get("llm_prompt", DEFAULT_LLM_PROMPT), + "llm_context_mode": llm_env_defaults.get("llm_context_mode", "sentence"), + "normalization_numbers": True, + "normalization_currency": True, + "normalization_footnotes": True, + "normalization_titles": True, + "normalization_terminal": True, + "normalization_phoneme_hints": True, + "normalization_caps_quotes": True, + "normalization_internet_slang": False, + "normalization_apostrophes_contractions": True, + "normalization_apostrophes_plural_possessives": True, + "normalization_apostrophes_sibilant_possessives": True, + "normalization_apostrophes_decades": True, + "normalization_apostrophes_leading_elisions": True, + "normalization_apostrophe_mode": "spacy", + "normalization_numbers_year_style": "american", + "normalization_contraction_aux_be": True, + "normalization_contraction_aux_have": True, + "normalization_contraction_modal_will": True, + "normalization_contraction_modal_would": True, + "normalization_contraction_negation_not": True, + "normalization_contraction_let_us": True, + } + + +def integration_defaults() -> Dict[str, Dict[str, Any]]: + """Default values for integration settings.""" + return { + "calibre_opds": { + "enabled": False, + "base_url": "", + "username": "", + "password": "", + "verify_ssl": True, + }, + "audiobookshelf": { + "enabled": False, + "base_url": "", + "api_token": "", + "library_id": "", + "collection_id": "", + "folder_id": "", + "verify_ssl": True, + "send_cover": True, + "send_chapters": True, + "send_subtitles": False, + "auto_send": False, + "timeout": 30.0, + }, + } + + +# ── LLM helpers ────────────────────────────────────────────────────── + +_PROMPT_TOKEN_RE = re.compile(r"{{\s*([a-zA-Z0-9_]+)\s*}}") + + +def llm_ready(settings: Mapping[str, Any]) -> bool: + base_url = str(settings.get("llm_base_url") or "").strip() + return bool(base_url) + + +def render_prompt_template(template: str, context: Mapping[str, str]) -> str: + if not template: + return "" + + def _replace(match: re.Match[str]) -> str: + key = match.group(1) + return context.get(key, "") + + return _PROMPT_TOKEN_RE.sub(_replace, template) diff --git a/abogen/infrastructure/exporters.py b/abogen/infrastructure/exporters.py index d3e6575..5a32e96 100644 --- a/abogen/infrastructure/exporters.py +++ b/abogen/infrastructure/exporters.py @@ -344,7 +344,6 @@ class ExportService: ) -> None: """Upload to Audiobookshelf.""" if config is None: - # Load from job or global config cfg = getattr(job, "_abs_config", None) if cfg is None: from abogen.utils import load_config @@ -367,7 +366,7 @@ class ExportService: if log_callback: log_callback("Audiobookshelf upload skipped: not configured", "warning") return - + if not config.base_url or not config.api_token or not config.library_id: if log_callback: log_callback("Audiobookshelf upload skipped: configure base URL, API token, and library ID first", "warning") @@ -376,7 +375,7 @@ class ExportService: if log_callback: log_callback("Audiobookshelf upload skipped: enter folder name or ID in settings", "warning") return - + if not audio_path.exists(): if log_callback: log_callback("Audiobookshelf upload skipped: audio output not found", "warning") diff --git a/abogen/webui/routes/api.py b/abogen/webui/routes/api.py index 35f4f94..d7085d6 100644 --- a/abogen/webui/routes/api.py +++ b/abogen/webui/routes/api.py @@ -281,83 +281,7 @@ def api_speaker_preview() -> ResponseReturnValue: # --- Integration Routes --- -def _opds_metadata_overrides(metadata_payload: Mapping[str, Any]) -> Dict[str, Any]: - metadata_overrides: Dict[str, Any] = {} - - def _stringify_metadata_value(value: Any) -> str: - if value is None: - return "" - if isinstance(value, (list, tuple, set)): - parts = [str(item).strip() for item in value if item is not None] - parts = [part for part in parts if part] - return ", ".join(parts) - return str(value).strip() - - raw_series = metadata_payload.get("series") or metadata_payload.get("series_name") - series_name = str(raw_series or "").strip() - if series_name: - metadata_overrides["series"] = series_name - metadata_overrides.setdefault("series_name", series_name) - - series_index_value = ( - metadata_payload.get("series_index") - or metadata_payload.get("series_position") - or metadata_payload.get("series_sequence") - or metadata_payload.get("book_number") - ) - if series_index_value is not None: - series_index_text = str(series_index_value).strip() - if series_index_text: - metadata_overrides.setdefault("series_index", series_index_text) - metadata_overrides.setdefault("series_position", series_index_text) - metadata_overrides.setdefault("series_sequence", series_index_text) - metadata_overrides.setdefault("book_number", series_index_text) - - tags_value = metadata_payload.get("tags") or metadata_payload.get("keywords") - if tags_value: - tags_text = _stringify_metadata_value(tags_value) - if tags_text: - metadata_overrides.setdefault("tags", tags_text) - metadata_overrides.setdefault("keywords", tags_text) - metadata_overrides.setdefault("genre", tags_text) - - description_value = metadata_payload.get("description") or metadata_payload.get("summary") - if description_value: - description_text = _stringify_metadata_value(description_value) - if description_text: - metadata_overrides.setdefault("description", description_text) - metadata_overrides.setdefault("summary", description_text) - - subtitle_value = ( - metadata_payload.get("subtitle") - or metadata_payload.get("sub_title") - or metadata_payload.get("calibre_subtitle") - ) - if subtitle_value: - subtitle_text = _stringify_metadata_value(subtitle_value) - if subtitle_text: - metadata_overrides.setdefault("subtitle", subtitle_text) - - publisher_value = metadata_payload.get("publisher") - if publisher_value: - publisher_text = _stringify_metadata_value(publisher_value) - if publisher_text: - metadata_overrides.setdefault("publisher", publisher_text) - - # Author mapping: Abogen templates look for either 'authors' or 'author'. - authors_value = ( - metadata_payload.get("authors") - or metadata_payload.get("author") - or metadata_payload.get("creator") - or metadata_payload.get("dc_creator") - ) - if authors_value: - authors_text = _stringify_metadata_value(authors_value) - if authors_text: - metadata_overrides.setdefault("authors", authors_text) - metadata_overrides.setdefault("author", authors_text) - - return metadata_overrides +from abogen.domain.metadata_overrides import normalize_opds_metadata as _opds_metadata_overrides @api_bp.get("/integrations/calibre-opds/feed") def api_calibre_opds_feed() -> ResponseReturnValue: diff --git a/abogen/webui/routes/jobs.py b/abogen/webui/routes/jobs.py index 545db76..1e28ed7 100644 --- a/abogen/webui/routes/jobs.py +++ b/abogen/webui/routes/jobs.py @@ -8,8 +8,8 @@ from flask.typing import ResponseReturnValue from abogen.webui.service import ( JobStatus, - load_audiobookshelf_chapters, build_audiobookshelf_metadata, + load_audiobookshelf_chapters, ) from abogen.webui.routes.utils.service import get_service from abogen.webui.routes.utils.form import render_jobs_panel @@ -22,15 +22,22 @@ from abogen.webui.routes.utils.epub import ( from abogen.webui.routes.utils.settings import ( stored_integration_config, build_audiobookshelf_config, - coerce_bool, ) from abogen.webui.routes.utils.common import existing_paths -from abogen.integrations.audiobookshelf import AudiobookshelfClient, AudiobookshelfUploadError +from abogen.infrastructure.exporters import ExportService logger = logging.getLogger(__name__) jobs_bp = Blueprint("jobs", __name__) + +def _resolve_cover(job: Any, config: Any) -> Optional[Path]: + """Resolve cover image path if enabled.""" + if not config.send_cover or not job.cover_image_path: + return None + cover = job.cover_image_path if isinstance(job.cover_image_path, Path) else Path(str(job.cover_image_path)) + return cover if cover.exists() else None + @jobs_bp.get("/") def job_detail(job_id: str) -> ResponseReturnValue: job = get_service().get_job(job_id) @@ -98,24 +105,18 @@ def send_job_to_audiobookshelf(job_id: str) -> ResponseReturnValue: return _panel_response() settings = stored_integration_config("audiobookshelf") - if not settings or not coerce_bool(settings.get("enabled"), False): + if not settings or not settings.get("enabled"): job.add_log("Audiobookshelf upload skipped: integration is disabled.", level="warning") service._persist_state() return _panel_response() config = build_audiobookshelf_config(settings) if config is None: - job.add_log( - "Audiobookshelf upload skipped: configure base URL, API token, and library ID first.", - level="warning", - ) + job.add_log("Audiobookshelf upload skipped: configure base URL, API token, and library ID first.", level="warning") service._persist_state() return _panel_response() if not config.folder_id: - job.add_log( - "Audiobookshelf upload skipped: enter the folder name or ID in the Audiobookshelf settings.", - level="warning", - ) + job.add_log("Audiobookshelf upload skipped: enter the folder name or ID in the Audiobookshelf settings.", level="warning") service._persist_state() return _panel_response() @@ -125,83 +126,49 @@ def send_job_to_audiobookshelf(job_id: str) -> ResponseReturnValue: service._persist_state() return _panel_response() - cover_path = None - if config.send_cover and job.cover_image_path: - cover_candidate = job.cover_image_path - if not isinstance(cover_candidate, Path): - cover_candidate = Path(str(cover_candidate)) - if cover_candidate.exists(): - cover_path = cover_candidate - - subtitles = existing_paths(job.result.subtitle_paths) if config.send_subtitles else None - chapters = load_audiobookshelf_chapters(job) if config.send_chapters else None - metadata = build_audiobookshelf_metadata(job) - display_title = metadata.get("title") or audio_path.stem overwrite_requested = request.form.get("overwrite") == "true" or request.args.get("overwrite") == "true" - try: - client = AudiobookshelfClient(config) - except ValueError as exc: - job.add_log(f"Audiobookshelf configuration error: {exc}", level="error") - service._persist_state() - return _panel_response() - - try: - existing_items = client.find_existing_items(display_title, folder_id=config.folder_id) - except AudiobookshelfUploadError as exc: - job.add_log(f"Audiobookshelf lookup failed: {exc}", level="error") - service._persist_state() - return _panel_response() - - if existing_items and not overwrite_requested: - job.add_log( - f"Audiobookshelf already contains '{display_title}'. Awaiting overwrite confirmation.", - level="warning", - ) - service._persist_state() - if request.headers.get("HX-Request"): - detail = { - "jobId": job.id, - "title": display_title, - "url": url_for("jobs.send_job_to_audiobookshelf", job_id=job.id), - "target": request.headers.get("HX-Target") or "#jobs-panel", - "message": f'Audiobookshelf already contains "{display_title}". Overwrite?', - } - headers = {"HX-Trigger": json.dumps({"audiobookshelf-overwrite-prompt": detail})} - return Response("", status=204, headers=headers) - return _panel_response() - - if existing_items and overwrite_requested: + if not overwrite_requested: + from abogen.integrations.audiobookshelf import AudiobookshelfClient, AudiobookshelfUploadError + metadata = build_audiobookshelf_metadata(job) + display_title = metadata.get("title") or audio_path.stem try: - client.delete_items(existing_items) + existing_items = AudiobookshelfClient(config).find_existing_items(display_title, folder_id=config.folder_id) except AudiobookshelfUploadError as exc: - job.add_log(f"Audiobookshelf overwrite aborted: {exc}", level="error") + job.add_log(f"Audiobookshelf lookup failed: {exc}", level="error") service._persist_state() return _panel_response() - else: - job.add_log( - f"Removed {len(existing_items)} existing Audiobookshelf item(s) prior to overwrite.", - level="info", - ) + if existing_items: + job.add_log(f"Audiobookshelf already contains '{display_title}'. Awaiting overwrite confirmation.", level="warning") + service._persist_state() + if request.headers.get("HX-Request"): + detail = { + "jobId": job.id, + "title": display_title, + "url": url_for("jobs.send_job_to_audiobookshelf", job_id=job.id), + "target": request.headers.get("HX-Target") or "#jobs-panel", + "message": f'Audiobookshelf already contains "{display_title}". Overwrite?', + } + headers = {"HX-Trigger": json.dumps({"audiobookshelf-overwrite-prompt": detail})} + return Response("", status=204, headers=headers) + return _panel_response() job.add_log("Audiobookshelf upload triggered manually.", level="info") + export_svc = ExportService() try: - client.upload_audiobook( + export_svc.upload_audiobookshelf( + job, audio_path, - metadata=metadata, - cover_path=cover_path, - chapters=chapters, - subtitles=subtitles, + existing_paths(job.result.subtitle_paths), + load_audiobookshelf_chapters(job) if config.send_chapters else None, + build_audiobookshelf_metadata(job), + cover_path=_resolve_cover(job, config), + config=config, + log_callback=lambda msg, lvl="info": job.add_log(msg, level=lvl), ) - except AudiobookshelfUploadError as exc: - job.add_log(f"Audiobookshelf upload failed: {exc}", level="error") except Exception as exc: job.add_log(f"Audiobookshelf integration error: {exc}", level="error") - else: - job.add_log("Audiobookshelf upload queued.", level="success") - finally: - service._persist_state() - + service._persist_state() return _panel_response() @jobs_bp.post("/clear-finished") diff --git a/abogen/webui/routes/settings.py b/abogen/webui/routes/settings.py index 4c1e72c..ba4ea07 100644 --- a/abogen/webui/routes/settings.py +++ b/abogen/webui/routes/settings.py @@ -8,22 +8,15 @@ from flask.typing import ResponseReturnValue from abogen.webui.routes.utils.settings import ( load_settings, - load_integration_settings, save_settings, - stored_integration_config, - coerce_bool, - coerce_int, SAVE_MODE_LABELS, llm_ready, - _NORMALIZATION_BOOLEAN_KEYS, - _NORMALIZATION_STRING_KEYS, - _DEFAULT_ANALYSIS_THRESHOLD, ) -from abogen.webui.routes.utils.common import extract_checkbox from abogen.webui.routes.utils.voice import template_options +from abogen.webui.services.settings_service import apply_form_to_settings from abogen.webui.debug_tts_runner import run_debug_tts_wavs from abogen.debug_tts_samples import DEBUG_TTS_SAMPLES -from abogen.utils import get_user_output_path, load_config +from abogen.utils import get_user_output_path settings_bp = Blueprint("settings", __name__) @@ -38,143 +31,7 @@ _NORMALIZATION_SAMPLES = { @settings_bp.post("/update") def update_settings() -> ResponseReturnValue: current = load_settings() - form = request.form - - # General settings - current["language"] = (form.get("language") or "en").strip() - current["default_speaker"] = (form.get("default_speaker") or "").strip() - current["default_voice"] = (form.get("default_voice") or "").strip() - try: - current["supertonic_total_steps"] = max(2, min(15, int(form.get("supertonic_total_steps", current.get("supertonic_total_steps", 5))))) - except (TypeError, ValueError): - pass - try: - current["supertonic_speed"] = max(0.7, min(2.0, float(form.get("supertonic_speed", current.get("supertonic_speed", 1.0))))) - except (TypeError, ValueError): - pass - current["output_format"] = (form.get("output_format") or "mp3").strip() - current["subtitle_mode"] = (form.get("subtitle_mode") or "Disabled").strip() - current["subtitle_format"] = (form.get("subtitle_format") or "srt").strip() - current["save_mode"] = (form.get("save_mode") or "save_next_to_input").strip() - - current["replace_single_newlines"] = coerce_bool(form.get("replace_single_newlines"), False) - current["use_gpu"] = coerce_bool(form.get("use_gpu"), False) - current["save_chapters_separately"] = coerce_bool(form.get("save_chapters_separately"), False) - current["merge_chapters_at_end"] = coerce_bool(form.get("merge_chapters_at_end"), True) - current["save_as_project"] = coerce_bool(form.get("save_as_project"), False) - current["separate_chapters_format"] = (form.get("separate_chapters_format") or "wav").strip() - - try: - current["silence_between_chapters"] = max(0.0, float(form.get("silence_between_chapters", 2.0))) - except ValueError: - pass - - try: - current["chapter_intro_delay"] = max(0.0, float(form.get("chapter_intro_delay", 0.5))) - except ValueError: - pass - - current["read_title_intro"] = coerce_bool(form.get("read_title_intro"), False) - current["read_closing_outro"] = coerce_bool(form.get("read_closing_outro"), True) - current["normalize_chapter_opening_caps"] = coerce_bool(form.get("normalize_chapter_opening_caps"), True) - current["auto_prefix_chapter_titles"] = coerce_bool(form.get("auto_prefix_chapter_titles"), True) - - try: - current["max_subtitle_words"] = max(1, int(form.get("max_subtitle_words", 50))) - except ValueError: - pass - - current["chunk_level"] = (form.get("chunk_level") or "paragraph").strip() - current["generate_epub3"] = coerce_bool(form.get("generate_epub3"), False) - - current["speaker_analysis_threshold"] = coerce_int( - form.get("speaker_analysis_threshold"), - _DEFAULT_ANALYSIS_THRESHOLD, - minimum=1, - maximum=25, - ) - - # Normalization settings - for key in _NORMALIZATION_BOOLEAN_KEYS: - current[key] = extract_checkbox(form, key, bool(current.get(key, True))) - for key in _NORMALIZATION_STRING_KEYS: - if hasattr(form, "__contains__") and key in form: - current[key] = (form.get(key) or "").strip() - - # Integrations - # `load_settings()` returns only the general settings subset and intentionally - # does not include stored integrations. Seed them from the stored config so - # saving unrelated settings cannot wipe credentials/tokens. - current_integrations: dict[str, dict[str, Any]] = {} - cfg = load_config() or {} - stored_integrations = cfg.get("integrations") - if isinstance(stored_integrations, Mapping): - for name, payload in stored_integrations.items(): - if isinstance(name, str) and isinstance(payload, Mapping): - current_integrations[name] = dict(payload) - # Ensure known integrations are loaded even if the config is still in legacy format. - for name in ("audiobookshelf", "calibre_opds"): - stored = stored_integration_config(name) - if stored and name not in current_integrations: - current_integrations[name] = dict(stored) - current["integrations"] = current_integrations - - # Audiobookshelf - abs_enabled = coerce_bool(form.get("audiobookshelf_enabled"), False) - abs_url = (form.get("audiobookshelf_base_url") or "").strip() - abs_token = (form.get("audiobookshelf_api_token") or "").strip() - abs_library = (form.get("audiobookshelf_library_id") or "").strip() - abs_folder = (form.get("audiobookshelf_folder_id") or "").strip() - abs_verify = coerce_bool(form.get("audiobookshelf_verify_ssl"), True) - abs_auto_send = coerce_bool(form.get("audiobookshelf_auto_send"), False) - abs_cover = coerce_bool(form.get("audiobookshelf_send_cover"), True) - abs_chapters = coerce_bool(form.get("audiobookshelf_send_chapters"), True) - abs_subtitles = coerce_bool(form.get("audiobookshelf_send_subtitles"), False) - - try: - abs_timeout = max(1.0, float(form.get("audiobookshelf_timeout", 30.0))) - except ValueError: - abs_timeout = 30.0 - - # Preserve existing token if not provided and not cleared - if not abs_token and not coerce_bool(form.get("audiobookshelf_api_token_clear"), False): - existing_abs = current["integrations"].get("audiobookshelf", {}) - abs_token = existing_abs.get("api_token", "") - - current["integrations"]["audiobookshelf"] = { - "enabled": abs_enabled, - "base_url": abs_url, - "api_token": abs_token, - "library_id": abs_library, - "folder_id": abs_folder, - "verify_ssl": abs_verify, - "auto_send": abs_auto_send, - "send_cover": abs_cover, - "send_chapters": abs_chapters, - "send_subtitles": abs_subtitles, - "timeout": abs_timeout, - } - - # Calibre OPDS - calibre_enabled = coerce_bool(form.get("calibre_opds_enabled"), False) - calibre_url = (form.get("calibre_opds_base_url") or "").strip() - calibre_user = (form.get("calibre_opds_username") or "").strip() - calibre_pass = (form.get("calibre_opds_password") or "").strip() - calibre_verify = coerce_bool(form.get("calibre_opds_verify_ssl"), True) - - # Preserve existing password if not provided and not cleared - if not calibre_pass and not coerce_bool(form.get("calibre_opds_password_clear"), False): - existing_calibre = current["integrations"].get("calibre_opds", {}) - calibre_pass = existing_calibre.get("password", "") - - current["integrations"]["calibre_opds"] = { - "enabled": calibre_enabled, - "base_url": calibre_url, - "username": calibre_user, - "password": calibre_pass, - "verify_ssl": calibre_verify, - } - + apply_form_to_settings(current, request.form) save_settings(current) flash("Settings updated successfully.", "success") return redirect(url_for("settings.settings_page")) diff --git a/abogen/webui/routes/utils/common.py b/abogen/webui/routes/utils/common.py index 5f3d93c..e87ddcd 100644 --- a/abogen/webui/routes/utils/common.py +++ b/abogen/webui/routes/utils/common.py @@ -1,32 +1,11 @@ from typing import Any, Optional, Tuple, Iterable, List, Mapping from pathlib import Path - -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 split_profile_spec(value: Any) -> Tuple[str, Optional[str]]: - text = str(value or "").strip() - if not text: - return "", None - lowered = text.lower() - if lowered.startswith("profile:") or lowered.startswith("speaker:"): - _, _, remainder = text.partition(":") - name = remainder.strip() - return "", name or None - return text, None +from abogen.domain.settings_core import coerce_bool, split_profile_spec # noqa: F401 def split_speaker_spec(value: Any) -> Tuple[str, Optional[str]]: """Preferred alias for split_profile_spec (supports 'speaker:' and legacy 'profile:').""" - return split_profile_spec(value) diff --git a/abogen/webui/routes/utils/settings.py b/abogen/webui/routes/utils/settings.py index b8abef6..ac6db98 100644 --- a/abogen/webui/routes/utils/settings.py +++ b/abogen/webui/routes/utils/settings.py @@ -1,108 +1,32 @@ import os -import re from typing import Any, Dict, Mapping, Optional -from abogen.constants import ( - LANGUAGE_DESCRIPTIONS, - SUBTITLE_FORMATS, - SUPPORTED_SOUND_FORMATS, -) -from abogen.tts_plugin.utils import get_default_voice -from abogen.normalization_settings import ( - DEFAULT_LLM_PROMPT, - environment_llm_defaults, -) -from abogen.utils import load_config, save_config from abogen.integrations.calibre_opds import CalibreOPDSClient from abogen.integrations.audiobookshelf import AudiobookshelfConfig -from abogen.webui.routes.utils.common import split_profile_spec, coerce_bool - -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()} - -_CHUNK_LEVEL_OPTIONS = [ - {"value": "paragraph", "label": "Paragraphs"}, - {"value": "sentence", "label": "Sentences"}, -] - -_CHUNK_LEVEL_VALUES = {option["value"] for option in _CHUNK_LEVEL_OPTIONS} - -_DEFAULT_ANALYSIS_THRESHOLD = 3 - -_APOSTROPHE_MODE_OPTIONS = [ - {"value": "off", "label": "Off"}, - {"value": "spacy", "label": "spaCy (built-in)"}, - {"value": "llm", "label": "LLM assisted"}, -] - -_NORMALIZATION_BOOLEAN_KEYS = { - "normalization_numbers", - "normalization_titles", - "normalization_terminal", - "normalization_phoneme_hints", - "normalization_caps_quotes", - "normalization_currency", - "normalization_footnotes", - "normalization_internet_slang", - "normalization_apostrophes_contractions", - "normalization_apostrophes_plural_possessives", - "normalization_apostrophes_sibilant_possessives", - "normalization_apostrophes_decades", - "normalization_apostrophes_leading_elisions", - "normalization_contraction_aux_be", - "normalization_contraction_aux_have", - "normalization_contraction_modal_will", - "normalization_contraction_modal_would", - "normalization_contraction_negation_not", - "normalization_contraction_let_us", -} - -_NORMALIZATION_STRING_KEYS = { - "normalization_numbers_year_style", - "normalization_apostrophe_mode", -} - -BOOLEAN_SETTINGS = { - "replace_single_newlines", - "use_gpu", - "save_chapters_separately", - "merge_chapters_at_end", - "save_as_project", - "generate_epub3", - "enable_entity_recognition", - "read_title_intro", - "read_closing_outro", - "auto_prefix_chapter_titles", - "normalize_chapter_opening_caps", - "normalization_numbers", - "normalization_titles", - "normalization_terminal", - "normalization_phoneme_hints", - "normalization_caps_quotes", - "normalization_currency", - "normalization_footnotes", - "normalization_internet_slang", - "normalization_apostrophes_contractions", - "normalization_apostrophes_plural_possessives", - "normalization_apostrophes_sibilant_possessives", - "normalization_apostrophes_decades", - "normalization_apostrophes_leading_elisions", - "normalization_contraction_aux_be", - "normalization_contraction_aux_have", - "normalization_contraction_modal_will", - "normalization_contraction_modal_would", - "normalization_contraction_negation_not", - "normalization_contraction_let_us", -} - -FLOAT_SETTINGS = {"silence_between_chapters", "chapter_intro_delay", "llm_timeout"} -INT_SETTINGS = {"max_subtitle_words", "speaker_analysis_threshold"} +from abogen.utils import load_config, save_config +from abogen.domain.settings_core import ( + SAVE_MODE_LABELS, + LEGACY_SAVE_MODE_MAP, + CHUNK_LEVEL_OPTIONS, + CHUNK_LEVEL_VALUES, + DEFAULT_ANALYSIS_THRESHOLD, + BOOLEAN_SETTINGS, + FLOAT_SETTINGS, + INT_SETTINGS, + _NORMALIZATION_BOOLEAN_KEYS, + _NORMALIZATION_STRING_KEYS, + coerce_bool, + coerce_float, + coerce_int, + split_profile_spec, + normalize_save_mode, + normalize_setting_value, + has_output_override, + settings_defaults, + integration_defaults, + llm_ready, + render_prompt_template, +) _NORMALIZATION_GROUPS = [ { @@ -136,226 +60,16 @@ _NORMALIZATION_GROUPS = [ } ] +_APOSTROPHE_MODE_OPTIONS = [ + {"value": "off", "label": "Off"}, + {"value": "spacy", "label": "spaCy (built-in)"}, + {"value": "llm", "label": "LLM assisted"}, +] -def integration_defaults() -> Dict[str, Dict[str, Any]]: - return { - "calibre_opds": { - "enabled": False, - "base_url": "", - "username": "", - "password": "", - "verify_ssl": True, - }, - "audiobookshelf": { - "enabled": False, - "base_url": "", - "api_token": "", - "library_id": "", - "collection_id": "", - "folder_id": "", - "verify_ssl": True, - "send_cover": True, - "send_chapters": True, - "send_subtitles": False, - "auto_send": False, - "timeout": 30.0, - }, - } - - -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]: - llm_env_defaults = environment_llm_defaults() - return { - "output_format": "wav", - "subtitle_format": "srt", - "save_mode": "default_output" if has_output_override() else "save_next_to_input", - "default_speaker": "", - "default_voice": get_default_voice("kokoro"), - "supertonic_total_steps": 5, - "supertonic_speed": 1.0, - "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, - "read_title_intro": False, - "read_closing_outro": True, - "normalize_chapter_opening_caps": True, - "max_subtitle_words": 50, - "chunk_level": "paragraph", - "enable_entity_recognition": True, - "generate_epub3": False, - "auto_prefix_chapter_titles": True, - "speaker_analysis_threshold": _DEFAULT_ANALYSIS_THRESHOLD, - "speaker_pronunciation_sentence": "This is {{name}} speaking.", - "speaker_random_languages": [], - "llm_base_url": llm_env_defaults.get("llm_base_url", ""), - "llm_api_key": llm_env_defaults.get("llm_api_key", ""), - "llm_model": llm_env_defaults.get("llm_model", ""), - "llm_timeout": llm_env_defaults.get("llm_timeout", 30.0), - "llm_prompt": llm_env_defaults.get("llm_prompt", DEFAULT_LLM_PROMPT), - "llm_context_mode": llm_env_defaults.get("llm_context_mode", "sentence"), - "normalization_numbers": True, - "normalization_currency": True, - "normalization_footnotes": True, - "normalization_titles": True, - "normalization_terminal": True, - "normalization_phoneme_hints": True, - "normalization_caps_quotes": True, - "normalization_internet_slang": False, - "normalization_apostrophes_contractions": True, - "normalization_apostrophes_plural_possessives": True, - "normalization_apostrophes_sibilant_possessives": True, - "normalization_apostrophes_decades": True, - "normalization_apostrophes_leading_elisions": True, - "normalization_apostrophe_mode": "spacy", - "normalization_numbers_year_style": "american", - "normalization_contraction_aux_be": True, - "normalization_contraction_aux_have": True, - "normalization_contraction_modal_will": True, - "normalization_contraction_modal_would": True, - "normalization_contraction_negation_not": True, - "normalization_contraction_let_us": True, - } - - -def llm_ready(settings: Mapping[str, Any]) -> bool: - base_url = str(settings.get("llm_base_url") or "").strip() - return bool(base_url) - - -_PROMPT_TOKEN_RE = re.compile(r"{{\s*([a-zA-Z0-9_]+)\s*}}") - - -def render_prompt_template(template: str, context: Mapping[str, str]) -> str: - if not template: - return "" - - def _replace(match: re.Match[str]) -> str: - key = match.group(1) - return context.get(key, "") - - return _PROMPT_TOKEN_RE.sub(_replace, template) - - -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): - text = value.strip() - if not text: - return defaults[key] - spec, profile_name = split_profile_spec(text) - if profile_name: - return f"speaker:{profile_name}" - return spec - return defaults[key] - if key == "default_speaker": - if isinstance(value, str): - text = value.strip() - if not text: - return "" - spec, profile_name = split_profile_spec(text) - if profile_name: - return f"speaker:{profile_name}" - return spec - return "" - if key == "chunk_level": - if isinstance(value, str) and value in _CHUNK_LEVEL_VALUES: - return value - return defaults[key] - if key == "normalization_apostrophe_mode": - if isinstance(value, str): - normalized_mode = value.strip().lower() - if normalized_mode in {"off", "spacy", "llm"}: - return normalized_mode - return defaults[key] - if key == "normalization_numbers_year_style": - if isinstance(value, str): - normalized_style = value.strip().lower() - if normalized_style in {"american", "off"}: - return normalized_style - return defaults[key] - if key == "llm_context_mode": - if isinstance(value, str): - normalized_scope = value.strip().lower() - if normalized_scope == "sentence": - return normalized_scope - return defaults[key] - if key == "llm_prompt": - candidate = str(value or "").strip() - return candidate if candidate else defaults[key] - if key in {"llm_base_url", "llm_api_key", "llm_model"}: - return str(value or "").strip() - 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, []) - if key == "supertonic_total_steps": - try: - steps = int(value) - except (TypeError, ValueError): - return defaults.get(key, 5) - return max(2, min(15, steps)) - if key == "supertonic_speed": - try: - speed = float(value) - except (TypeError, ValueError): - return defaults.get(key, 1.0) - return max(0.7, min(2.0, speed)) - return value if value is not None else defaults.get(key) +# Backward-compatible aliases for modules still referencing old underscore-prefixed names +_DEFAULT_ANALYSIS_THRESHOLD = DEFAULT_ANALYSIS_THRESHOLD +_CHUNK_LEVEL_OPTIONS = CHUNK_LEVEL_OPTIONS +_CHUNK_LEVEL_VALUES = CHUNK_LEVEL_VALUES def load_settings() -> Dict[str, Any]: diff --git a/abogen/webui/services/settings_service.py b/abogen/webui/services/settings_service.py new file mode 100644 index 0000000..e73b64b --- /dev/null +++ b/abogen/webui/services/settings_service.py @@ -0,0 +1,165 @@ +"""Settings form-to-dict mapping. + +Pure functions that convert form data into a settings dict. +No Flask dependencies — testable without a request context. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any + + +def apply_form_to_settings(current: dict, form: Mapping[str, Any]) -> dict: + """Apply form data to a settings dict. + + Pure function: takes a current settings dict and a form-like mapping, + returns the updated settings dict. No Flask dependencies. + + Args: + current: Current settings dict (will be mutated). + form: Form-like mapping (e.g. request.form.to_dict()). + + Returns: + Updated settings dict (same object as input). + """ + from abogen.domain.settings_core import ( + coerce_bool, + coerce_int, + DEFAULT_ANALYSIS_THRESHOLD, + _NORMALIZATION_BOOLEAN_KEYS, + _NORMALIZATION_STRING_KEYS, + ) + from abogen.webui.routes.utils.settings import stored_integration_config + from abogen.webui.routes.utils.common import extract_checkbox + from abogen.utils import load_config + # General settings + current["language"] = (form.get("language") or "en").strip() + current["default_speaker"] = (form.get("default_speaker") or "").strip() + current["default_voice"] = (form.get("default_voice") or "").strip() + try: + current["supertonic_total_steps"] = max(2, min(15, int(form.get("supertonic_total_steps", current.get("supertonic_total_steps", 5))))) + except (TypeError, ValueError): + pass + try: + current["supertonic_speed"] = max(0.7, min(2.0, float(form.get("supertonic_speed", current.get("supertonic_speed", 1.0))))) + except (TypeError, ValueError): + pass + current["output_format"] = (form.get("output_format") or "mp3").strip() + current["subtitle_mode"] = (form.get("subtitle_mode") or "Disabled").strip() + current["subtitle_format"] = (form.get("subtitle_format") or "srt").strip() + current["save_mode"] = (form.get("save_mode") or "save_next_to_input").strip() + + current["replace_single_newlines"] = coerce_bool(form.get("replace_single_newlines"), False) + current["use_gpu"] = coerce_bool(form.get("use_gpu"), False) + current["save_chapters_separately"] = coerce_bool(form.get("save_chapters_separately"), False) + current["merge_chapters_at_end"] = coerce_bool(form.get("merge_chapters_at_end"), True) + current["save_as_project"] = coerce_bool(form.get("save_as_project"), False) + current["separate_chapters_format"] = (form.get("separate_chapters_format") or "wav").strip() + + try: + current["silence_between_chapters"] = max(0.0, float(form.get("silence_between_chapters", 2.0))) + except ValueError: + pass + + try: + current["chapter_intro_delay"] = max(0.0, float(form.get("chapter_intro_delay", 0.5))) + except ValueError: + pass + + current["read_title_intro"] = coerce_bool(form.get("read_title_intro"), False) + current["read_closing_outro"] = coerce_bool(form.get("read_closing_outro"), True) + current["normalize_chapter_opening_caps"] = coerce_bool(form.get("normalize_chapter_opening_caps"), True) + current["auto_prefix_chapter_titles"] = coerce_bool(form.get("auto_prefix_chapter_titles"), True) + + try: + current["max_subtitle_words"] = max(1, int(form.get("max_subtitle_words", 50))) + except ValueError: + pass + + current["chunk_level"] = (form.get("chunk_level") or "paragraph").strip() + current["generate_epub3"] = coerce_bool(form.get("generate_epub3"), False) + + current["speaker_analysis_threshold"] = coerce_int( + form.get("speaker_analysis_threshold"), + DEFAULT_ANALYSIS_THRESHOLD, + minimum=1, + maximum=25, + ) + + # Normalization settings + for key in _NORMALIZATION_BOOLEAN_KEYS: + current[key] = extract_checkbox(form, key, bool(current.get(key, True))) + for key in _NORMALIZATION_STRING_KEYS: + if key in form: + current[key] = (form.get(key) or "").strip() + + # Integrations — seed from stored config to prevent wiping credentials + current_integrations: dict[str, dict[str, Any]] = {} + cfg = load_config() or {} + stored_integrations = cfg.get("integrations") + if isinstance(stored_integrations, Mapping): + for name, payload in stored_integrations.items(): + if isinstance(name, str) and isinstance(payload, Mapping): + current_integrations[name] = dict(payload) + for name in ("audiobookshelf", "calibre_opds"): + stored = stored_integration_config(name) + if stored and name not in current_integrations: + current_integrations[name] = dict(stored) + current["integrations"] = current_integrations + + # Audiobookshelf + abs_enabled = coerce_bool(form.get("audiobookshelf_enabled"), False) + abs_url = (form.get("audiobookshelf_base_url") or "").strip() + abs_token = (form.get("audiobookshelf_api_token") or "").strip() + abs_library = (form.get("audiobookshelf_library_id") or "").strip() + abs_folder = (form.get("audiobookshelf_folder_id") or "").strip() + abs_verify = coerce_bool(form.get("audiobookshelf_verify_ssl"), True) + abs_auto_send = coerce_bool(form.get("audiobookshelf_auto_send"), False) + abs_cover = coerce_bool(form.get("audiobookshelf_send_cover"), True) + abs_chapters = coerce_bool(form.get("audiobookshelf_send_chapters"), True) + abs_subtitles = coerce_bool(form.get("audiobookshelf_send_subtitles"), False) + + try: + abs_timeout = max(1.0, float(form.get("audiobookshelf_timeout", 30.0))) + except ValueError: + abs_timeout = 30.0 + + if not abs_token and not coerce_bool(form.get("audiobookshelf_api_token_clear"), False): + existing_abs = current["integrations"].get("audiobookshelf", {}) + abs_token = existing_abs.get("api_token", "") + + current["integrations"]["audiobookshelf"] = { + "enabled": abs_enabled, + "base_url": abs_url, + "api_token": abs_token, + "library_id": abs_library, + "folder_id": abs_folder, + "verify_ssl": abs_verify, + "auto_send": abs_auto_send, + "send_cover": abs_cover, + "send_chapters": abs_chapters, + "send_subtitles": abs_subtitles, + "timeout": abs_timeout, + } + + # Calibre OPDS + calibre_enabled = coerce_bool(form.get("calibre_opds_enabled"), False) + calibre_url = (form.get("calibre_opds_base_url") or "").strip() + calibre_user = (form.get("calibre_opds_username") or "").strip() + calibre_pass = (form.get("calibre_opds_password") or "").strip() + calibre_verify = coerce_bool(form.get("calibre_opds_verify_ssl"), True) + + if not calibre_pass and not coerce_bool(form.get("calibre_opds_password_clear"), False): + existing_calibre = current["integrations"].get("calibre_opds", {}) + calibre_pass = existing_calibre.get("password", "") + + current["integrations"]["calibre_opds"] = { + "enabled": calibre_enabled, + "base_url": calibre_url, + "username": calibre_user, + "password": calibre_pass, + "verify_ssl": calibre_verify, + } + + return current diff --git a/tests/test_domain_metadata_overrides.py b/tests/test_domain_metadata_overrides.py new file mode 100644 index 0000000..a6678fa --- /dev/null +++ b/tests/test_domain_metadata_overrides.py @@ -0,0 +1,52 @@ +"""Tests for domain/metadata_overrides.py and webui/services/settings_service.py.""" + +import pytest + +from abogen.domain.metadata_overrides import normalize_opds_metadata + + +class TestNormalizeOpdsMetadata: + def test_series_mapping(self): + result = normalize_opds_metadata({"series": "My Series", "series_index": 3}) + assert result["series"] == "My Series" + assert result["series_name"] == "My Series" + assert result["series_index"] == "3" + assert result["series_position"] == "3" + + def test_series_name_alias(self): + result = normalize_opds_metadata({"series_name": "Alt Series"}) + assert result["series"] == "Alt Series" + assert result["series_name"] == "Alt Series" + + def test_tags_to_keywords(self): + result = normalize_opds_metadata({"tags": "sci-fi, action"}) + assert result["tags"] == "sci-fi, action" + assert result["keywords"] == "sci-fi, action" + assert result["genre"] == "sci-fi, action" + + def test_description_summary(self): + result = normalize_opds_metadata({"description": "A great book"}) + assert result["description"] == "A great book" + assert result["summary"] == "A great book" + + def test_authors_creator(self): + result = normalize_opds_metadata({"creator": "Author Name"}) + assert result["authors"] == "Author Name" + assert result["author"] == "Author Name" + + def test_subtitle_aliases(self): + result = normalize_opds_metadata({"calibre_subtitle": "Sub Title"}) + assert result["subtitle"] == "Sub Title" + + def test_empty_payload(self): + result = normalize_opds_metadata({}) + assert result == {} + + def test_list_authors(self): + result = normalize_opds_metadata({"authors": ["Alice", "Bob"]}) + assert result["authors"] == "Alice, Bob" + + def test_none_values_filtered(self): + result = normalize_opds_metadata({"series": None, "tags": ""}) + assert "series" not in result + assert "tags" not in result diff --git a/tests/test_settings_service.py b/tests/test_settings_service.py new file mode 100644 index 0000000..b42b186 --- /dev/null +++ b/tests/test_settings_service.py @@ -0,0 +1,94 @@ +"""Tests for webui/services/settings_service.py — form→settings mapping.""" + +from collections import OrderedDict + +from abogen.webui.services.settings_service import apply_form_to_settings + + +def _form(**kwargs: str | None) -> dict[str, str | None]: + return OrderedDict(kwargs) + + +def _base() -> dict: + return { + "language": "en", + "default_speaker": "", + "default_voice": "", + "supertonic_total_steps": 5, + "supertonic_speed": 1.0, + "output_format": "mp3", + "subtitle_mode": "Disabled", + "subtitle_format": "srt", + "save_mode": "save_next_to_input", + "replace_single_newlines": False, + "use_gpu": False, + "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, + "read_title_intro": False, + "read_closing_outro": True, + "normalize_chapter_opening_caps": True, + "auto_prefix_chapter_titles": True, + "max_subtitle_words": 50, + "chunk_level": "paragraph", + "generate_epub3": False, + "speaker_analysis_threshold": 15, + "integrations": {}, + } + + +class TestApplyFormToSettings: + def test_general_fields(self): + settings = _base() + form = _form(language="fr", default_speaker="af_heart", output_format="wav") + apply_form_to_settings(settings, form) + assert settings["language"] == "fr" + assert settings["default_speaker"] == "af_heart" + assert settings["output_format"] == "wav" + + def test_numeric_fields(self): + settings = _base() + form = _form(supertonic_total_steps="10", supertonic_speed="1.5", max_subtitle_words="30") + apply_form_to_settings(settings, form) + assert settings["supertonic_total_steps"] == 10 + assert settings["supertonic_speed"] == 1.5 + assert settings["max_subtitle_words"] == 30 + + def test_numeric_clamping(self): + settings = _base() + form = _form(supertonic_total_steps="100", supertonic_speed="5.0", max_subtitle_words="0") + apply_form_to_settings(settings, form) + assert settings["supertonic_total_steps"] == 15 + assert settings["supertonic_speed"] == 2.0 + assert settings["max_subtitle_words"] == 1 + + def test_boolean_checkboxes(self): + settings = _base() + form = _form(use_gpu="on", save_chapters_separately="on") + apply_form_to_settings(settings, form) + assert settings["use_gpu"] is True + assert settings["save_chapters_separately"] is True + + def test_default_values_preserved(self): + settings = _base() + form = _form() + apply_form_to_settings(settings, form) + assert settings["silence_between_chapters"] == 2.0 + assert settings["chunk_level"] == "paragraph" + + def test_empty_form_keeps_general_defaults(self): + settings = _base() + apply_form_to_settings(settings, _form()) + # General fields should stay at their base values + assert settings["language"] == "en" + assert settings["output_format"] == "mp3" + assert settings["chunk_level"] == "paragraph" + assert settings["silence_between_chapters"] == 2.0 + + def test_language_whitespace_trimmed(self): + settings = _base() + apply_form_to_settings(settings, _form(language=" de ")) + assert settings["language"] == "de"