UmbraAudio/scripts/remigrate_speakers.py

# Copyright 2026 Umbra Tools (umbratools.com)
# SPDX-License-Identifier: Apache-2.0

"""Deterministic re-merge of existing discovery chapter data with fixed logic.

Regenerates .umbra/speakers.json from .umbra/discovery/discovery_pass1.json
chapter results using the corrected _merge_chapter_results (no non-speaker
promotion, shared-specific-alias dedup, LLM alias evidence respected).

The discovery pass1 chapter results are unchanged (LLM per-chapter output);
only the deterministic merge/dedup post-processing is re-applied.
"""

from __future__ import annotations

import json
import shutil
import sys
from datetime import datetime
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent.parent))

from umbraaudio.analysis.character_discovery import (
    ChapterSpeakers,
    CharacterDiscoverer,
)
from umbraaudio.config import load_config


def main() -> None:
    import argparse

    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--project", type=str, default=None, help="Project name or path"
    )
    args = parser.parse_args()

    cfg = load_config(args.project)
    data_dir = cfg.data_dir
    pass1_path = data_dir / "discovery" / "discovery_pass1.json"
    speakers_path = data_dir / "speakers.json"

    with open(pass1_path) as f:
        pass1 = json.load(f)

    crs = []
    for cr in pass1.get("chapter_results", []):
        crs.append(
            ChapterSpeakers(
                chapter_index=cr.get("chapter_index", 0),
                chapter_title=cr.get("chapter_title", ""),
                speaking_characters=cr.get("speaking_characters", []),
                non_speaking_characters=cr.get("non_speaking_characters", []),
                validation_notes=cr.get("validation_notes", []),
            )
        )

    cd = CharacterDiscoverer.__new__(CharacterDiscoverer)
    cd.TITLE_WORDS = CharacterDiscoverer.TITLE_WORDS

    speakers = cd._merge_chapter_results(crs, pass1.get("book_title", ""), 0)

    has_narrator = any(s.is_narrator for s in speakers)
    if not has_narrator:
        from umbraaudio.analysis.character_discovery import (
            CommunicationStyle,
            Demographics,
            Speaker,
        )

        narrator = Speaker(
            primary_name="Narrator",
            aliases=["narrator", "the narrator"],
            demographics=Demographics(type="Human", gender="Male"),
            vocal_traits=["deep", "resonant", "calm", "authoritative"],
            communication_style=CommunicationStyle(
                primary_register="Formal",
                style_shifts_or_notes=(
                    "Steady, even-toned throughout. Never rushes."
                ),
            ),
            description=(
                "Deep, resonant male voice. Calm, authoritative tone. "
                "Rich baritone timbre, slow and deliberate pacing, "
                "like a late-night radio host."
            ),
            is_narrator=True,
        )
        speakers.insert(0, narrator)

    result = {
        "speakers": [s.to_dict() for s in speakers],
        "book_title": pass1.get("book_title", ""),
        "total_chapters": pass1.get("total_chapters", 0),
        "total_words": pass1.get("total_words", 0),
    }

    if speakers_path.exists():
        backup = (
            data_dir
            / "archives"
            / "speakers_remigrated"
            / f"{datetime.now():%Y%m%d_%H%M%S}.json"
        )
        backup.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(speakers_path, backup)
        print(f"Backed up previous speakers.json -> {backup}")
        _port_enriched_fields(backup, result)

    with open(speakers_path, "w") as f:
        json.dump(result, f, indent=2)

    print(f"Wrote {len(speakers)} speakers to {speakers_path}")
    for s in speakers:
        n = s.primary_name
        if "kainda" in n.lower() or "capek" in n.lower():
            print(f"  {n}: ch{s.chapters_appeared} aliases={s.aliases}")

    _reconcile_voice_profiles(data_dir, speakers)


def _port_enriched_fields(backup: Path, result: dict) -> None:
    with open(backup) as f:
        old = json.load(f)

    old_by_name = {s["primary_name"].lower(): s for s in old.get("speakers", [])}
    old_by_alias: dict[str, dict] = {}
    for s in old.get("speakers", []):
        for alias in s.get("aliases", []):
            old_by_alias.setdefault(alias.lower(), s)

    for speaker in result["speakers"]:
        n = speaker["primary_name"]
        candidates = []
        if n.lower() in old_by_name:
            candidates.append(old_by_name[n.lower()])
        if n.lower() in old_by_alias:
            candidates.append(old_by_alias[n.lower()])
        candidates.extend(
            s
            for key, s in old_by_name.items()
            if key != n.lower() and key in n.lower()
        )
        old_speaker = next(
            (s for s in candidates if s.get("description") or s.get("sample_dialogue")),
            candidates[0] if candidates else None,
        )
        if not old_speaker:
            continue
        for field in ("description", "sample_dialogue"):
            if not speaker.get(field) and old_speaker.get(field):
                speaker[field] = old_speaker[field]


def _reconcile_voice_profiles(data_dir: Path, speakers: list) -> None:
    profiles_path = data_dir / "voice_profiles.json"
    if not profiles_path.exists():
        return

    with open(profiles_path) as f:
        profiles = json.load(f)

    current_names = {s.primary_name for s in speakers}

    merged = {}
    for s in speakers:
        name = s.primary_name
        if name in profiles:
            merged[name] = profiles[name]

    alias_to_canonical: dict[str, str] = {}
    for s in speakers:
        for alias in s.aliases:
            alias_to_canonical.setdefault(alias.lower(), s.primary_name)

    for old_name in list(profiles.keys()):
        if old_name in merged:
            continue
        canonical = alias_to_canonical.get(old_name.lower())
        if canonical is None:
            canonical = next(
                (
                    n
                    for n in current_names
                    if n.lower() != old_name.lower()
                    and old_name.lower() in n.lower()
                ),
                None,
            )
        if canonical and canonical not in merged:
            merged[canonical] = profiles[old_name]
        elif canonical:
            for key, value in profiles[old_name].items():
                if value and not merged[canonical].get(key):
                    merged[canonical][key] = value

    with open(profiles_path, "w") as f:
        json.dump(merged, f, indent=2)
    print(f"Reconciled voice profiles: {len(merged)} kept (was {len(profiles)})")


if __name__ == "__main__":
    main()