UmbraAudio/scripts/scan_speaker_duplicates.py

#!/usr/bin/env python3
# Copyright 2026 Umbra Tools (umbratools.com)
# SPDX-License-Identifier: Apache-2.0

"""Scan for duplicate speakers and report them ranked by how safe the merge is.

Read-only by default. Nothing is merged without ``--apply`` and an explicit group choice.

    python scripts/scan_speaker_duplicates.py --project "Last Dwarf"
    python scripts/scan_speaker_duplicates.py --project "Last Dwarf" --json
    python scripts/scan_speaker_duplicates.py --project "Last Dwarf" --verdict auto
    python scripts/scan_speaker_duplicates.py --project "Last Dwarf" \
        --apply "The Girl helping <- The Girl, One Girl Helping"
    python scripts/scan_speaker_duplicates.py --project "Last Dwarf" --apply-unresolved

Verdicts:

  auto     - a member exists only as a scanned name with no lines and the surviving name is
             a strong variant of it. Safe to act on.
  review   - plausible, but two people were actually heard, or the similarity is only
             suggestive. A human decides.
  blocked  - two members have approved voices, or two different voice references. Merging
             would destroy something. Never automatic.

An approved voice is never the thing a merge discards: survivor selection prefers it, and
``--apply`` carries any approval across to the survivor rather than dropping it.
"""
from __future__ import annotations

import argparse
import json
import sqlite3
import sys
from pathlib import Path

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

from umbraaudio.analysis.speaker_dedup import (
    candidate_groups,
    load_speaker_facts,
    merge_group,
    profile_names,
    unresolved_speaker_names,
)
from umbraaudio.config import load_config


def _print_report(facts, groups, unresolved) -> None:  # noqa: ANN001
    speaking = [f for f in facts.values() if f.utterances > 0]
    print(f"profiles {len(facts)}  |  with lines {len(speaking)}  |  "
          f"approved voices {sum(1 for f in facts.values() if f.approved)}")
    print(f"duplicate groups {len(groups)}  |  "
          f"unresolved spoken names {len(unresolved)}")

    for verdict in ("auto", "review", "blocked"):
        subset = [g for g in groups if g.verdict == verdict]
        if not subset:
            continue
        print(f"\n{'=' * 78}\n{verdict.upper()}  ({len(subset)} group(s))\n{'=' * 78}")
        for group in subset:
            print(f"\n  survivor: {group.survivor}")
            for reason in group.survivor_reasons:
                print(f"      - {reason}")
            print(f"  members ({len(group.members)}), score {group.score}:")
            for name in group.members:
                f = facts[name]
                flags = []
                if f.approved:
                    flags.append("APPROVED")
                if f.has_ref:
                    flags.append("ref")
                if f.is_designation:
                    flags.append("designation")
                mark = " ->" if name == group.survivor else "   "
                print(f"      {mark} {name:34s} {f.utterances:4d} utt  "
                      f"{f.chapters:2d} ch  [{', '.join(flags) or '-'}]")
            print(f"  signals: {'; '.join(group.signals)}")
            for reason in group.reasons:
                print(f"  why {verdict}: {reason}")

    if unresolved:
        print(f"\n{'=' * 78}\nUNRESOLVED SPOKEN NAMES (case-only duplicates)\n{'=' * 78}")
        for name, count, chapters in unresolved:
            print(f"      {name!r:36s} {count:4d} utt  {chapters:2d} ch  "
                  "-> exists only as a differently-cased profile")


def _apply(data_dir: Path, spec: str, force: bool) -> int:
    """``"Survivor <- A, B"``. ``force`` overrides the conflicting-voice refusal."""
    if "<-" not in spec:
        print("--apply needs: 'Survivor <- A, B'", file=sys.stderr)
        return 1
    survivor, rest = spec.split("<-", 1)
    survivor = survivor.strip()
    duplicates = [d.strip() for d in rest.split(",") if d.strip()]

    if force:
        # A human has decided these are dead profiles or the same person, so the
        # conflicting-voice guard is deliberately bypassed. Print exactly what is being
        # overridden so the decision is on the record.
        from umbraaudio.analysis.orphan_resolution import merge_speakers

        facts = load_speaker_facts(data_dir)
        clashes = [
            f"{name}={f.ref_sha256}"
            for name in duplicates
            if (f := facts.get(name)) and f.has_ref and f.ref_sha256
        ]
        if clashes:
            print(f"FORCED: overriding conflicting voice references: {', '.join(clashes)}")
            print("        those reference FILES are left untouched on disk.")
        db = sqlite3.connect(str(data_dir / "project.db"))
        db.row_factory = sqlite3.Row
        try:
            for name in duplicates:
                if name in profile_names(data_dir):
                    r = merge_speakers(db, data_dir, name, survivor)
                    print(f"  merged profile {name!r} into {survivor!r} "
                          f"({r.get('utterances_changed', 0)} utt re-pointed)")
                else:
                    cur = db.execute(
                        "UPDATE utterances SET speaker_name=? "
                        "WHERE speaker_name=? AND lower(type)='dialogue'",
                        (survivor, name),
                    )
                    print(f"  re-pointed spoken-only {name!r} -> {survivor!r} "
                          f"({cur.rowcount or 0} dialogue utt)")
            db.commit()
        finally:
            db.close()
        return 0

    db = sqlite3.connect(str(data_dir / "project.db"))
    db.row_factory = sqlite3.Row
    try:
        result = merge_group(db, data_dir, survivor, duplicates)
    finally:
        db.close()
    if not result.get("ok"):
        print(f"REFUSED: {result.get('reason')}", file=sys.stderr)
        for detail in result.get("detail", []) or []:
            print(f"  {detail}", file=sys.stderr)
        print("  re-run with --force if you have decided this by hand", file=sys.stderr)
        return 2
    print(f"merged into {result['survivor']}: "
          f"{', '.join(result['merged'] + result['repointed_only'])}")
    if result["repointed_only"]:
        print(f"  (re-pointed only, no profile existed: "
              f"{', '.join(result['repointed_only'])})")
    print(f"  utterances re-pointed: {result['utterances_changed']}")
    if result["approvals_carried"]:
        print(f"  approvals carried across from: {', '.join(result['approvals_carried'])}")
    return 0


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--project", default=None)
    parser.add_argument("--json", action="store_true")
    parser.add_argument("--verdict", choices=("auto", "review", "blocked"), default=None)
    parser.add_argument("--apply", default=None, metavar="'Survivor <- A, B'")
    parser.add_argument("--force", action="store_true",
                        help="override the conflicting-voice refusal; use when a human "
                             "has already decided (e.g. dead profiles)")
    parser.add_argument("--apply-unresolved", action="store_true",
                        help="re-point case-variant spoken names at their profile")
    args = parser.parse_args()

    cfg = load_config(args.project)
    data_dir = cfg.data_dir

    if args.apply:
        return _apply(data_dir, args.apply, args.force)

    if args.apply_unresolved:
        facts = load_speaker_facts(data_dir)
        # Compare against profile names only: a spoken name that differs from a profile
        # by case IS the typo class, even though it also appears in a candidate group.
        known = profile_names(data_dir)
        db = sqlite3.connect(str(data_dir / "project.db"))
        db.row_factory = sqlite3.Row
        try:
            unresolved = unresolved_speaker_names(db, known)
            changed = 0
            for name, _count, _chapters in unresolved:
                target = next(k for k in known if k.casefold() == name.casefold())
                cursor = db.execute(
                    "UPDATE utterances SET speaker_name=? WHERE speaker_name=?",
                    (target, name),
                )
                changed += cursor.rowcount or 0
                print(f"  {name!r} -> {target!r}  ({changed} rows so far)")
            db.commit()
        finally:
            db.close()
        print(f"re-pointed {changed} utterance(s)")
        return 0

    facts = load_speaker_facts(data_dir)
    groups = candidate_groups(facts)
    if args.verdict:
        groups = [g for g in groups if g.verdict == args.verdict]

    db = sqlite3.connect(f"file:{(data_dir / 'project.db').resolve()}?mode=ro", uri=True)
    db.row_factory = sqlite3.Row
    try:
        unresolved = unresolved_speaker_names(db, profile_names(data_dir))
    finally:
        db.close()

    if args.json:
        print(json.dumps(
            {
                "profiles": len(facts),
                "groups": [g.as_dict() for g in groups],
                "unresolved": [
                    {"name": n, "utterances": c, "chapters": ch} for n, c, ch in unresolved
                ],
                "speakers": {k: v.as_dict() for k, v in sorted(facts.items())},
            },
            indent=2, sort_keys=True,
        ))
    else:
        _print_report(facts, groups, unresolved)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())