umbragit/UmbraAudio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Duplicate-speaker scan that refuses to eat an approved voice (spec 23)

Browse Source 
Last Dwarf has accumulated duplicate speakers from successive discovery and
designation scans: The Girl / The Girl helping / One Girl Helping, Guard 1 / The
first guard, Walking oldest / The Oldest, plus three lowercase spoken names that
differ from their profiles by one letter. The operator wants them eliminated and
was explicit that created and approved voices must not be lost.

umbraaudio/analysis/speaker_dedup.py
  Read-only fact gathering (profiles, ledger, voice refs, chapter spread) and a
  matcher that explains itself: determiner and ordinal normalization, superlative
  stemming, and three VETO rules, because string similarity alone would merge
  people who are not the same person:
    - distinguishing modifiers: The Oldest is not The Tallest; Guard 1 is not Guard 2
    - role nouns: Soldier (Rider) is not One Rider
    - proper names need ratio >= 0.88: Anfisa (0.77 from Tanisha) is a different
      character; Rielan/Rieland at 0.93 is a misspelling
  Verdicts auto / review / blocked. A group with two different voice references is
  blocked, because merging would discard a generated voice. Survivor selection
  prefers an approval, then a voice reference, then utterance volume.

Two rules were written and then removed, recorded in the spec so nobody re-adds
them: a "shared token plus role noun" rule proposed The Oldest = The Old Woman,
and a "single shared token" rule proposed Scum leader = The Leader. A missed
duplicate costs one manual merge; a wrong one silently rewrites a speaker.

Two real bugs, both found by running against a COPY of Last Dwarf rather than by
reading code:

1. orphan_resolution.merge_speakers never touches voice_status.json, so merging
   an approved speaker into an unapproved target would have deleted the approval
   record silently. merge_group() now carries the approval to the survivor,
   drops the loser's entry, and reports approvals_carried.
2. The first re-point was unscoped and moved 5,981 rows instead of 3: `narrator`
   is the labelling convention for ~5,978 NARRATION rows too, and renaming them
   would invalidate thousands of clips because every marker records the speaker
   exactly. The re-point is now dialogue-only. Narration casing is a separate,
   deliberate decision and is flagged as such, not taken by a duplicate scan.

Verified on a copy: 93 -> 84 profiles, 23 utterances re-pointed, narration
untouched, 72 -> 72 voice ref files with zero removed, 13 -> 13 approvals with
zero lost.

Last Dwarf scan: 10 auto, 2 review, 4 blocked. Blocked groups are exactly the
ones the operator must decide - The Girl (two voices for one person), Officer,
the three-way Narrator family, and Border Soldier vs Soldier (Rider).

Nothing was applied to the live project. tests/test_speaker_dedup.py: 37 tests,
vetoes weighted as heavily as positives. Full suite 575 passed / 8 failed, the
same 8 pre-existing failures. ruff clean.
This commit is contained in:
B.D. Murphy
2026-10-04 18:36:37 -07:00
parent 9f3a0a4b18
commit 714b11367a
5 changed files with 1343 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

185
scripts/scan_speaker_duplicates.py Normal file
View File

@@ -0,0 +1,185 @@
#!/usr/bin/env python3
"""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) -> int:
"""``"Survivor <- A, B"``"""
if "<-" not in spec:
print("--apply needs: 'Survivor <- A, B'", file=sys.stderr)
return 1
survivor, rest = spec.split("<-", 1)
survivors = survivor.strip()
duplicates = [d.strip() for d in rest.split(",") if d.strip()]
db = sqlite3.connect(str(data_dir / "project.db"))
db.row_factory = sqlite3.Row
try:
result = merge_group(db, data_dir, survivors, 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)
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("--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)
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())

156
specs/23-speaker-duplicate-detection.md Normal file
View File

@@ -0,0 +1,156 @@
# Spec 23 — Duplicate Speaker Detection and Approval-Safe Merging
- **Status:** implemented (scanner + merge wrapper + tests). **Nothing applied to `Last Dwarf`
yet** — the scan is a report, and the merge is a deliberate act.
- **Date:** 2026-10-03
- **Trigger:** operator review of `Last Dwarf`, which has accumulated duplicate speakers from
successive scans. *"The last scan created several new speakers. I have created and approved
voices for several characters. I don't want to lose those voices."*
- **Companions:** `specs/17` (speaker confidence, designations), `specs/22` (releases).
---
## 1. The two requirements, and how each is enforced
| requirement | how it is enforced |
|---|---|
| **Do not lose an approved voice.** | Survivor selection *prefers* an approved voice and a voice reference above everything else. `merge_group()` carries an approval across to the survivor instead of dropping it. And a group with **two** different voice references is `blocked` — never merged, because merging would silently discard one generated voice. |
| **Better scan: eliminate duplicates among non-minor characters.** | A silent twin is the evidence. A designation invented by one scan and re-invented by another exists only as a name with **zero lines**; if the name also matches a speaker who *is* heard, they are one person. That is the operator's single-chapter rule, made mechanical. |
## 2. What the matching actually knows
String similarity is not enough. `manifest.md` has no opinion about who "the tallest" is, but
the scan can encode what a human knows:
| signal | meaning | example |
|---|---|---|
| determiner / numeral prefix stripped | same name, different lead-in | `The Sergeant` = `Sergeant`, `One Man` = `The Man` |
| ordinals normalized to digits | same enumeration, two styles | `The first guard` = `Guard 1` |
| superlatives stemmed | emphasis varies, identity does not | `The biggest one` = `The Big One` |
| **activity modifiers** | same person, described by what they are doing | `The Girl` = `The Girl helping` (0.95) |
| **distinguishing modifiers — a VETO** | different members of one crowd | `The Oldest` ≠ `The Tallest`, `Guard 1` ≠ `Guard 2` |
| **role nouns — a VETO** | different roles that share a word | `Soldier (Rider)` ≠ `One Rider` |
| **proper names need ratio ≥ 0.88** | real names must be genuinely close | `Anfisa` ≠ `Tanisha` (0.77); `Rielan` ≈ `Rieland` (0.93) |
The vetoes are the point. Two rules that were **written and then removed** are recorded here
so nobody re-adds them:
- A *shared token plus a narrower role noun* rule proposed `The Oldest` = `The Old Woman`
because both contain "old". They are different characters. Removed.
- A *single shared token* rule proposed `Scum leader` = `The Leader`. Removed; a shared word
is not an identity.
A missed duplicate costs one manual merge. A wrong one silently rewrites a speaker. The scan
is tuned to find fewer, not more.
## 3. Verdicts
| verdict | meaning | action |
|---|---|---|
| `auto` | every member is a strong variant of the survivor, and the other members have **zero lines** | safe to apply unattended |
| `review` | plausible, but two people were actually **heard**, or the weakest member is only suggestive | a human decides |
| `blocked` | two members have approved voices, or two **different voice references** | never automatic — a human must choose which voice survives |
## 4. `Last Dwarf` scan, 2026-10-03
96 facts (93 profiles + 3 spoken-only names), 62 with lines, 13 approved voices.
### AUTO — 10 groups, apply-ready
| survivor | duplicates | why |
|---|---|---|
| `Scum country boy` | `scum country boy` (7 utt) | case-only |
| `Scum leader` | `scum leader` (10 utt) | case-only |
| `Guard 2` | `The second guard` (3 utt) | determiner |
| `Lieutenant` | `One Lieutenant` | determiner |
| `The Man` | `One Man` (14 utt) | determiner |
| `Soldier` | `One Soldier`, `The Soldier` (20 utt) | determiner |
| `Sergeant` | `The Sergeant` (14 utt) | determiner |
| `Walking oldest` | `The Oldest` | activity modifier |
| `Walking Tallest` | `The Tallest` | activity modifier |
| `Walking Youngest` | `The Youngest` | activity modifier |
### REVIEW — 2 groups, your call
| group | question |
|---|---|
| `Guard 1` ← `The first guard` | both have lines (6 and 4). One chapter. Your single-chapter rule says probably the same. |
| `The Big One` ← `The Biggest One` | both silent; nothing to move, so merging is cosmetic. |
### BLOCKED — 4 groups, conflicting voices
| group | conflict | the decision |
|---|---|---|
| `Narrator` ← `Narratorold`, `OrigNarrator`, `narrator` | three different ref WAVs; `Narrator` is **approved** | three stale narrator profiles, and 3 dialogue rows still labelled lowercase `narrator` |
| `The Girl helping` ← `The Girl`, `One Girl Helping` | two different ref WAVs | **your example.** Same person, two generated voices. Which voice survives? |
| `The Officer` ← `Officer` | two different ref WAVs | `The Officer` has 24 utt / 4 ch; `Officer` has 5 / 2. Probably one person. |
| `Border Soldier` ← `Soldier (Rider)` | two different ref WAVs; `Border Soldier` is **approved** | probably two different riders |
## 5. Two real bugs the verification found
Run against a **copy** of `Last Dwarf`, not the live project. Both would have been invisible in
a code review.
1. **An approval would have vanished.** `orphan_resolution.merge_speakers` folds aliases and
re-points utterances but **never touches `voice_status.json`**. Merging an approved speaker
into a target that had no ledger entry would have deleted the approval record silently.
`merge_group()` now copies the approval to the survivor, removes the loser's entry, and
reports `approvals_carried`.
2. **5,981 rows moved instead of 3.** `narrator` is the labelling convention for ~5,978
**narration** rows as well as the 3 dialogue rows. The first re-point was unscoped, so
merging `narrator` into `Narrator` would have renamed every narration row — invalidating
thousands of clips, because every staleness marker records the speaker exactly. The
re-point is now `lower(type)='dialogue'` **only**.
Worth flagging separately: `Last Dwarf` uses lowercase `narrator` for 5,978 narration rows
and `Narrator` for 9. That inconsistency is real and `specs/16` §5a says scene breaks must
be `Narrator`. Harmonising it is a deliberate decision with a large regeneration cost —
**not** something a duplicate scan should do on its own.
## 6. Verified outcome on a copy
```
BEFORE profiles=93 refs=72 approved=13
AFTER profiles=84 refs=72 approved=13
utterances re-pointed: 23 (was 6001 before the dialogue-only fix)
type counts unchanged: True dialogue 3918 / narration 6016 / scene_break 81
VOICE REF FILES 72 -> 72, removed 0
APPROVALS lost: 0
```
Nine profiles removed, **no voice file touched, no approval lost**, narration untouched.
## 7. How to use it
```bash
# Report only (default). Nothing is written.
python scripts/scan_speaker_duplicates.py --project "Last Dwarf"
python scripts/scan_speaker_duplicates.py --project "Last Dwarf" --verdict auto
python scripts/scan_speaker_duplicates.py --project "Last Dwarf" --json
# Apply one group (explicit, by name).
python scripts/scan_speaker_duplicates.py --project "Last Dwarf" \
--apply "Sergeant <- The Sergeant"
# Re-point case-variant spoken names at their profile.
python scripts/scan_speaker_duplicates.py --project "Last Dwarf" --apply-unresolved
```
## 8. Consequences of applying, stated plainly
- The 23 re-pointed utterances become **speaker-stale**; their clips regenerate on the next
chapter build. Nothing else regenerates.
- Merged profiles keep their voice references **on disk**. The loser's file is not deleted —
the next voice-design pass for that book should decide whether to prune orphans.
- A merged speaker's `chapters_appeared` widens, and the loser's name becomes an alias, so
the merge is reversible by editing one JSON file.
## 9. Open, for the operator
1. **The Girl.** `The Girl` and `The Girl helping` are the same person with two generated
voices. Which ref survives — or regenerate one voice for both?
2. **Officer.** Same question, 24/4 against 5/2.
3. **Narrator.** `Narrator` (approved) versus `Narratorold` and `OrigNarrator` — three refs.
Are the latter two dead profiles from an earlier book state?
4. **Guard 1.** Merge `The first guard` in, or are they two different guards?
5. **The Big One / The Biggest One.** Merge cosmetically, or leave both?

6
specs/README.md
View File

@@ -19,6 +19,12 @@ Index of all specification documents for the UmbraAudio audiobook generation sys
| [10-cli-orchestrator.md](10-cli-orchestrator.md) | CLI | Click commands, full pipeline orchestration |
| [11-configuration-dependencies.md](11-configuration-dependencies.md) | Config | Pydantic settings, dependencies, build system |
| [12-quality.md](12-quality.md) | Quality | MOS, WER, consistency, ACX compliance |
| [18-UmbraTrans.md](18-UmbraTrans.md) | Multilingual | Meaning-preserving translation, target voices, and multilingual audiobook production |
| [19-enterprise-control-plane.md](19-enterprise-control-plane.md) | Enterprise | Multi-tenant projects, access control, artifact lineage, and commercial voice library |
| [20-enterprise-database-integration.md](20-enterprise-database-integration.md) | Enterprise DB | Shared-PostgreSQL contract, blockers, cutover phases P1-P4 |
| [21-umbra-system-nodes-tasks-and-changes.md](21-umbra-system-nodes-tasks-and-changes.md) | System | Worker nodes, task assignment, cross-product change management, UI topology |
| [22-releases-and-the-publish-flag.md](22-releases-and-the-publish-flag.md) | Release | The publish flag, pinned artifact sets, post-publication change, and the Deal Hunter lesson |
| [23-speaker-duplicate-detection.md](23-speaker-duplicate-detection.md) | Speakers | Duplicate-speaker scan, approval-safe merging, and the crowd veto |
## Quick Reference

334
tests/test_speaker_dedup.py Normal file
View File

@@ -0,0 +1,334 @@
"""Tests for duplicate-speaker detection and approval-safe merging.
The rules that matter here are the ones that REFUSE to merge. A missed duplicate is a small
manual fix; a wrong merge silently rewrites a speaker and can invalidate a voice someone
paid for. So the veto cases are tested as carefully as the positive ones.
"""
from __future__ import annotations
import json
import sqlite3
from typing import TYPE_CHECKING
import pytest
from umbraaudio.analysis.speaker_dedup import (
MIN_SIMILARITY,
SpeakerFacts,
candidate_groups,
classify,
content_tokens,
is_designation,
load_speaker_facts,
merge_group,
normalize_name,
similarity,
unresolved_speaker_names,
)
if TYPE_CHECKING:
from pathlib import Path
def _f(name: str, **kwargs) -> SpeakerFacts: # noqa: ANN003
return SpeakerFacts(name=name, **kwargs)
def _db(path: Path, rows: list[tuple]) -> sqlite3.Connection: # noqa: ANN001
con = sqlite3.connect(str(path))
con.executescript(
"CREATE TABLE chapters (id INTEGER PRIMARY KEY);"
"CREATE TABLE utterances (id INTEGER PRIMARY KEY, chapter_id INTEGER, "
"utterance_index INTEGER, text TEXT, type TEXT, speaker_name TEXT);"
)
for i, (chapter_id, idx, utype, speaker) in enumerate(rows, start=1):
con.execute(
"INSERT INTO utterances VALUES (?,?,?,?,?,?)",
(i, chapter_id, idx, "line", utype, speaker),
)
con.commit()
return con
class TestNormalisation:
def test_determiner_is_dropped(self) -> None:
assert normalize_name("The Girl") == normalize_name("Girl")
assert normalize_name("One Girl Helping") == "girl helping"
def test_ordinals_become_digits(self) -> None:
""""The first guard" and "Guard 1" are the same person written two ways."""
assert normalize_name("The first guard") == normalize_name("Guard 1")
assert normalize_name("The second guard") == normalize_name("Guard 2")
def test_superlatives_are_stemmed(self) -> None:
assert normalize_name("The biggest one") == normalize_name("The Big One")
def test_parentheses_do_not_split_identity(self) -> None:
assert normalize_name("Aide (Royal)") == "aide royal"
def test_is_designation(self) -> None:
assert is_designation("The Girl helping")
assert is_designation("Sergeant")
assert not is_designation("Hallr")
assert not is_designation("Sergeant Atonal")
def test_content_tokens_keeps_the_distinguishing_word(self) -> None:
assert content_tokens("Walking oldest") == {"walking", "old"}
class TestVetoes:
"""Rules that must REFUSE a merge."""
def test_different_members_of_a_crowd_are_not_twins(self) -> None:
"""The oldest man and the tallest man are two people."""
score, signals = similarity(_f("The Oldest"), _f("The Tallest"))
assert score < 0.5
assert any("LIKELY DIFFERENT PEOPLE" in s for s in signals)
def test_crowd_veto_survives_stemming(self) -> None:
score, signals = similarity(_f("Walking oldest"), _f("Walking Tallest"))
assert score < 0.5
assert any("LIKELY DIFFERENT PEOPLE" in s for s in signals)
def test_guard_one_is_not_guard_two(self) -> None:
score, _ = similarity(_f("Guard 1"), _f("Guard 2"))
assert score < 0.5
def test_different_roles_are_not_twins(self) -> None:
"""A rider and a soldier-rider share the word "rider" and are different roles."""
score, _ = similarity(_f("Soldier (Rider)"), _f("One Rider"))
assert score < MIN_SIMILARITY, "must stay below the merge threshold"
def test_two_proper_names_need_a_high_bar(self) -> None:
""""Anfisa" and "Tanisha" are 0.77 similar as strings and are different people."""
assert similarity(_f("Anfisa"), _f("Tanisha"))[0] == 0.0
# ...but a genuine misspelling of one name still pairs up.
assert similarity(_f("Rielan"), _f("Rieland"))[0] >= 0.88
def test_oldest_and_old_woman_are_not_merged(self) -> None:
"""Regression: a 'shared token + role noun' rule used to propose this."""
assert similarity(_f("The Oldest"), _f("Old Woman"))[0] < 0.72
class TestPositives:
def test_case_variant_is_identical(self) -> None:
assert similarity(_f("Scum leader"), _f("scum leader"))[0] == 1.0
def test_determiner_variant_is_strong(self) -> None:
score, _ = similarity(_f("The Sergeant"), _f("Sergeant"))
assert score >= 0.93
def test_activity_modifier_is_the_same_person(self) -> None:
score, signals = similarity(_f("The Girl"), _f("The Girl helping"))
assert score >= 0.9
assert any("activity only" in s for s in signals)
def test_ordinal_wording_is_the_same_person(self) -> None:
assert similarity(_f("The first guard"), _f("Guard 1"))[0] >= 0.99
class TestSurvivorChoice:
def test_approval_is_never_sacrificed(self) -> None:
group = [
_f("Big One", utterances=400, chapters=20),
_f("The Biggest One", utterances=0, approved=True),
]
facts = {f.name: f for f in group}
groups = candidate_groups(facts)
assert groups and groups[0].survivor == "The Biggest One"
def test_voice_reference_is_preferred_over_volume(self) -> None:
"""A created reference outranks raw utterance count."""
group = [_f("One Sergeant", utterances=50), _f("Sergeant", utterances=1, has_ref=True)]
facts = {f.name: f for f in group}
assert candidate_groups(facts)[0].survivor == "Sergeant"
def test_volume_wins_when_nothing_else_does(self) -> None:
group = [_f("One Soldier", utterances=3), _f("Soldier", utterances=90)]
facts = {f.name: f for f in group}
assert candidate_groups(facts)[0].survivor == "Soldier"
class TestClassification:
def test_silent_twin_is_auto(self) -> None:
group = [_f("Sergeant", utterances=14), _f("The Sergeant", utterances=0)]
verdict, _ = classify(group, 0.99)
assert verdict == "auto"
def test_two_speakers_heard_is_review(self) -> None:
group = [_f("The Girl", utterances=4), _f("The Girl helping", utterances=4)]
verdict, _ = classify(group, 0.99)
assert verdict == "review"
def test_two_approved_voices_is_blocked(self) -> None:
group = [_f("A", utterances=5, approved=True), _f("B", utterances=3, approved=True)]
verdict, reasons = classify(group, 1.0)
assert verdict == "blocked"
assert any("approved voice" in r for r in reasons)
def test_conflicting_voice_references_block(self) -> None:
group = [
_f("A", utterances=9, has_ref=True, ref_sha256="aaaa"),
_f("B", utterances=4, has_ref=True, ref_sha256="bbbb"),
]
verdict, reasons = classify(group, 1.0)
assert verdict == "blocked"
assert any("conflicting voice references" in r for r in reasons)
def test_weak_chain_cannot_be_auto(self) -> None:
"""A strong pair plus a weak third member must not launder into 'auto'."""
group = [_f("Walking oldest", utterances=5), _f("The Oldest", utterances=0)]
verdict, _ = classify(group, 0.80)
assert verdict == "review"
class TestUnresolvedNames:
def test_case_variant_spoken_name_is_reported(self, tmp_path: Path) -> None:
db = _db(tmp_path / "p.db", [(1, 0, "dialogue", "scum leader")])
found = unresolved_speaker_names(db, {"Scum leader"})
assert found == [("scum leader", 1, 1)]
db.close()
def test_exact_match_is_not_reported(self, tmp_path: Path) -> None:
db = _db(tmp_path / "p.db", [(1, 0, "dialogue", "Hallr")])
assert unresolved_speaker_names(db, {"Hallr"}) == []
db.close()
def test_genuinely_unknown_name_is_not_reported(self, tmp_path: Path) -> None:
db = _db(tmp_path / "p.db", [(1, 0, "dialogue", "Manfred")])
assert unresolved_speaker_names(db, {"Hallr"}) == []
db.close()
def _project(tmp_path: Path, profiles: list[str], ledger: dict, rows: list[tuple],
refs: set[str] | None = None) -> Path:
data = tmp_path / ".umbra"
(data / "voices").mkdir(parents=True, exist_ok=True)
(data / "speakers.json").write_text(
json.dumps({"speakers": [{"primary_name": n} for n in profiles]})
)
(data / "voice_status.json").write_text(json.dumps(ledger))
for ref in refs or set():
(data / "voices" / f"{ref.replace(' ', '_')}_ref.wav").write_bytes(ref.encode())
_db(data / "project.db", rows).close()
return data
class TestMergeSafety:
def test_approval_is_carried_to_the_survivor(self, tmp_path: Path) -> None:
data = _project(
tmp_path,
profiles=["Keep", "Lose"],
ledger={"Lose": "approved"},
rows=[(1, 0, "dialogue", "Lose")],
)
con = sqlite3.connect(str(data / "project.db"))
result = merge_group(con, data, "Keep", ["Lose"])
con.close()
assert result["ok"] is True
ledger = json.loads((data / "voice_status.json").read_text())
assert ledger["Keep"] == "approved"
assert "Lose" not in ledger
assert result["approvals_carried"] == ["Lose"]
def test_repoint_never_touches_narration(self, tmp_path: Path) -> None:
"""Regression: an unscoped UPDATE moved 5,978 narration rows on Last Dwarf."""
data = _project(
tmp_path,
profiles=["Narrator"],
ledger={"Narrator": "approved"},
rows=[
(1, 0, "dialogue", "narrator"),
(1, 1, "narration", "narrator"),
(1, 2, "narration", "narrator"),
(1, 3, "scene_break", "Narrator"),
],
)
con = sqlite3.connect(str(data / "project.db"))
result = merge_group(con, data, "Narrator", ["narrator"])
remaining = con.execute(
"SELECT COUNT(*) FROM utterances WHERE speaker_name='narrator'"
).fetchone()[0]
moved = con.execute(
"SELECT COUNT(*) FROM utterances "
"WHERE speaker_name='Narrator' AND lower(type)='dialogue'"
).fetchone()[0]
con.close()
assert result["ok"] is True
assert result["repointed_only"] == ["narrator"]
assert moved == 1
assert remaining == 2, "narration rows must not be re-pointed"
def test_profile_merge_removes_the_duplicate_profile(self, tmp_path: Path) -> None:
data = _project(
tmp_path,
profiles=["Sergeant", "The Sergeant"],
ledger={},
rows=[(1, 0, "dialogue", "The Sergeant")],
)
con = sqlite3.connect(str(data / "project.db"))
result = merge_group(con, data, "Sergeant", ["The Sergeant"])
con.close()
assert result["merged"] == ["The Sergeant"]
names = {
p["primary_name"]
for p in json.loads((data / "speakers.json").read_text())["speakers"]
}
assert "The Sergeant" not in names
assert "Sergeant" in names
def test_conflicting_voices_are_refused_before_any_write(self, tmp_path: Path) -> None:
data = _project(
tmp_path,
profiles=["Officer", "The Officer"],
ledger={},
rows=[(1, 0, "dialogue", "Officer")],
refs={"Officer", "The Officer"},
)
before = (data / "speakers.json").read_text()
con = sqlite3.connect(str(data / "project.db"))
result = merge_group(con, data, "The Officer", ["Officer"])
con.close()
assert result["ok"] is False
assert result["reason"] == "conflicting voice references"
assert (data / "speakers.json").read_text() == before
def test_voice_files_are_never_deleted(self, tmp_path: Path) -> None:
"""A merge folds identity; it does not throw away a generated reference."""
data = _project(
tmp_path,
profiles=["Soldier", "One Soldier"],
ledger={},
rows=[(1, 0, "dialogue", "One Soldier")],
refs={"Soldier", "One Soldier"},
)
con = sqlite3.connect(str(data / "project.db"))
merge_group(con, data, "Soldier", ["One Soldier"])
con.close()
assert (data / "voices" / "One_Soldier_ref.wav").exists()
assert (data / "voices" / "Soldier_ref.wav").exists()
def test_load_speaker_facts_is_read_only(tmp_path: Path) -> None:
data = _project(tmp_path, ["Hallr"], {"Hallr": "approved"}, [(1, 0, "dialogue", "Hallr")])
before = (data / "speakers.json").read_text()
facts = load_speaker_facts(data)
assert facts["Hallr"].approved is True
assert facts["Hallr"].utterances == 1
assert (data / "speakers.json").read_text() == before
def test_spoken_only_name_becomes_a_fact_so_it_can_be_reported(tmp_path: Path) -> None:
data = _project(tmp_path, ["Hallr"], {}, [(1, 0, "dialogue", "scum leader")])
facts = load_speaker_facts(data)
assert "scum leader" in facts
assert facts["scum leader"].utterances == 1
@pytest.mark.parametrize("name", ["", " ", "()"])
def test_degenerate_names_do_not_crash(name: str) -> None:
facts = _f(name)
assert normalize_name(name) == ""
assert is_designation(name) is False
assert similarity(facts, _f("Hallr"))[0] == 0.0

662
umbraaudio/analysis/speaker_dedup.py Normal file
View File

@@ -0,0 +1,662 @@
"""Duplicate-speaker detection that will not eat an approved voice.
The problem
-----------
Successive passes of discovery and the designation scanner each invent names independently.
``Last Dwarf`` ended up with 93 profiles, 62 distinct spoken names, and families like::
The Girl / The Girl helping / One Girl Helping
Guard 1 / The first guard
Walking oldest / Walking Tallest / Walking Youngest
The Oldest / The Tallest / The Youngest / The Young One
Scum leader (profile) / scum leader (10 utterances, no profile)
Narrator (approved) / OrigNarrator / Narratorold / narrator (3 utterances)
Some of those are one person and some are genuinely different people, and the difference
matters: a merge that picks the wrong survivor silently discards a voice someone spent real
money and time approving.
The two rules this module enforces
----------------------------------
1. **Never lose an approval.** An approved voice is the most valuable thing in a speaker
record. Survivor selection prefers it, and a merge *carries the approval across* rather
than dropping it. ``merge_speakers`` in ``orphan_resolution.py`` does not do this - that is
a live bug, fixed here via :func:`merge_group` which wraps it.
2. **Never guess between two real people.** A similarity score proposes; it does not decide.
Every candidate group is classified ``auto`` / ``review`` / ``blocked`` and only ``auto``
is safe to act on unattended.
Why single-chapter is strong evidence
-------------------------------------
The operator's rule: *if one of these speakers is only in one chapter they are probably the
same.* It holds because an unnamed designation ("The Girl helping", "Walking tallest") is
invented independently every time a scene is analysed. Two such names in the same book almost
never denote two different unnamed people - the second scan simply failed to reuse the first
name. The evidence used here is therefore **co-occurrence**: when one candidate has zero
utterances, it exists only as a scanned name and is almost certainly the other one's twin.
"""
from __future__ import annotations
import difflib
import hashlib
import json
import sqlite3
from collections import defaultdict
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
# Tokens that mark a name as a *designation* rather than a proper name. A designation is
# cheap to invent twice and expensive to confuse between two people.
DESIGNATION_TOKENS = frozenset(
{
"the", "a", "an", "one", "two", "three", "some", "another", "unnamed",
"unidentified", "unknown", "voice", "speaker", "man", "woman", "boy", "girl",
"guy", "person", "soldier", "guard", "officer", "sergeant", "kid",
"group", "pair", "leader", "rider", "general", "captain", "commander",
}
)
# What a person is DOING. "The Girl" and "The Girl helping" are the same person described
# two ways, so this class is evidence FOR a merge.
ACTIVITY_MODIFIERS = frozenset(
{
"helping", "walking", "holding", "standing", "wearing", "leading", "sitting",
"sleeping", "following", "fighting", "talking", "carrying", "watching",
"speaking", "listening", "running", "waiting",
}
)
# Words that DISTINGUISH one member of a crowd from another. "The oldest" and "the tallest"
# are two different men in the same group, so this class is evidence AGAINST a merge - and
# it is the single most valuable thing this scan knows that a string comparison cannot.
# Listed in their STEMMED form, because normalize_name() has already turned "tallest" into
# "tall"; digits count too, so "Guard 1" and "Guard 2" are not twins.
DISTINGUISHING_MODIFIERS = frozenset(
{
"tall", "short", "small", "big", "large", "young", "old", "near", "close", "far",
"left", "right", "front", "back", "1", "2", "3", "4", "5",
}
)
# Enumerations are the SAME person written two ways: "The first guard" == "Guard 1".
# Ordinals are therefore normalized to digits before anything is compared.
_ORDINALS = {"first": "1", "second": "2", "third": "3", "fourth": "4"}
# Superlatives are stemmed so "the biggest one" and "the big one" can be compared, while
# "the oldest" (old) and "the tallest" (tall) stay distinguishable.
_SUPERLATIVES = {
"oldest": "old", "tallest": "tall", "biggest": "big", "smallest": "small",
"youngest": "young", "largest": "large", "shortest": "short",
}
# Leading determiners/numerals stripped before comparing. "The Girl" and "Girl" are the same
# candidate; "One Girl Helping" reduces to "girl helping".
_PREFIXES = ("the ", "a ", "an ", "one ", "two ", "three ", "some ", "another ")
# Plain role nouns, kept separate from ACTIVITY/DISTINGUISHING because they describe WHO
# rather than what they are doing or which member of a crowd they are.
ROLE_WORDS = frozenset(
{
"man", "woman", "boy", "girl", "kid", "guy", "person", "soldier", "guard",
"officer", "sergeant", "captain", "general", "lieutenant", "commander",
"leader", "rider", "speaker", "voice", "scout", "archer", "mage", "smith",
"narrator", "troll", "soldiers", "guards",
}
)
MIN_SIMILARITY = 0.72
STRONG_SIMILARITY = 0.88
AUTO_SIMILARITY = 0.93
def normalize_name(name: str) -> str:
"""Lowercase, drop a leading determiner/numeral, normalize ordinals and superlatives."""
text = " ".join((name or "").strip().lower().split())
text = text.replace("(", " ").replace(")", " ")
text = "".join(ch if ch.isalnum() or ch.isspace() else " " for ch in text)
text = " ".join(text.split())
changed = True
while changed:
changed = False
for prefix in _PREFIXES:
if text.startswith(prefix):
text = text[len(prefix) :].strip()
changed = True
mapped = [_ORDINALS.get(t, _SUPERLATIVES.get(t, t)) for t in text.split()]
return " ".join(sorted(mapped))
def content_tokens(name: str) -> set[str]:
"""Every meaningful token, minus designation noise and the determiner."""
tokens = set()
for token in normalize_name(name).split():
if token in ("s", "1x"):
continue
if token.endswith("'s"):
token = token[:-2]
tokens.add(token)
return tokens
def distinguishing_conflict(a_tokens: set[str], b_tokens: set[str]) -> list[str]:
"""Modifier classes that mean these are DIFFERENT members of the same crowd."""
reasons: list[str] = []
only_a = a_tokens - b_tokens
only_b = b_tokens - a_tokens
for side, extra in (("A", only_a), ("B", only_b)):
distinguishing = extra & DISTINGUISHING_MODIFIERS
if distinguishing:
other = b_tokens if side == "A" else a_tokens
distinguishing = distinguishing - other
if distinguishing:
reasons.append(
f"{side} names {sorted(distinguishing)} - "
f"{sorted(a_tokens)} vs {sorted(b_tokens)}"
)
return reasons
def is_designation(name: str) -> bool:
"""True when the name describes a role rather than naming a person.
A designation is built entirely from role words and modifiers. If ANY token looks like a
proper name, the character has a name: "Sergeant Atonal" is a person, "Sergeant" is not.
Getting this backwards matters because it decides whether two similar names are one
person's sloppy re-description or two real people.
"""
tokens = normalize_name(name).split()
if not tokens:
return False
known = (
DESIGNATION_TOKENS | ACTIVITY_MODIFIERS | DISTINGUISHING_MODIFIERS | ROLE_WORDS
)
return all(t in known for t in tokens)
def _sha256(path: Path) -> str | None:
try:
return hashlib.sha256(path.read_bytes()).hexdigest()[:16]
except OSError:
return None
@dataclass
class SpeakerFacts:
"""Everything the merge decision needs, gathered once."""
name: str
utterances: int = 0
chapters: int = 0
approved: bool = False
has_ref: bool = False
ref_sha256: str | None = None
aliases: list[str] = field(default_factory=list)
flagged: bool = False
@property
def is_designation(self) -> bool:
return is_designation(self.name)
@property
def normalized(self) -> str:
return normalize_name(self.name)
@property
def tokens(self) -> set[str]:
return content_tokens(self.name)
def as_dict(self) -> dict[str, Any]:
return {
"name": self.name,
"utterances": self.utterances,
"chapters": self.chapters,
"approved": self.approved,
"has_voice_ref": self.has_ref,
"voice_ref_sha256": self.ref_sha256,
"aliases": self.aliases,
"is_designation": self.is_designation,
"normalized": self.normalized,
}
def profile_names(data_dir: Path) -> set[str]:
"""Every name that has a speaker profile (no synthesized spoken-only names)."""
doc = json.loads((Path(data_dir) / "speakers.json").read_text(encoding="utf-8"))
return {
str(p.get("primary_name", ""))
for p in (doc.get("speakers", []) or [])
if isinstance(p, dict) and p.get("primary_name")
}
def load_speaker_facts(data_dir: Path) -> dict[str, SpeakerFacts]:
"""Read-only: profiles + ledger + voice refs + chapter spread. One pass."""
data_dir = Path(data_dir)
doc = json.loads((data_dir / "speakers.json").read_text(encoding="utf-8"))
profiles = doc.get("speakers", []) or []
ledger_path = data_dir / "voice_status.json"
ledger: dict[str, Any] = {}
if ledger_path.exists():
try:
ledger = json.loads(ledger_path.read_text(encoding="utf-8"))
except ValueError:
ledger = {}
if isinstance(ledger.get("speakers"), dict):
ledger = ledger["speakers"]
spread: dict[str, tuple[int, int]] = defaultdict(lambda: (0, 0))
db_path = data_dir / "project.db"
if db_path.exists():
con = sqlite3.connect(f"file:{db_path.resolve()}?mode=ro", uri=True)
try:
for name, count, chapters in con.execute(
"SELECT speaker_name, COUNT(*), COUNT(DISTINCT chapter_id) FROM utterances "
"WHERE lower(type)='dialogue' AND speaker_name IS NOT NULL "
"GROUP BY speaker_name"
):
spread[str(name)] = (int(count), int(chapters))
finally:
con.close()
voices_dir = data_dir / "voices"
facts: dict[str, SpeakerFacts] = {}
for profile in profiles:
if not isinstance(profile, dict):
continue
name = str(profile.get("primary_name", ""))
if not name:
continue
count, chapters = spread.get(name, (0, 0))
ref = voices_dir / f"{name.replace(' ', '_')}_ref.wav"
if not ref.exists():
# Nested refs exist for some speakers (rglob is how the generator resolves them).
matches = list(voices_dir.rglob(f"*{name.replace(' ', '_')}*_ref.wav")) if (
voices_dir.is_dir()
) else []
ref = matches[0] if matches else ref
status = ledger.get(name, "pending")
facts[name] = SpeakerFacts(
name=name,
utterances=count,
chapters=chapters,
approved=str(status).casefold() == "approved",
has_ref=ref.exists(),
ref_sha256=_sha256(ref) if ref.exists() else None,
aliases=[str(a) for a in profile.get("aliases") or []],
flagged=bool(profile.get("flagged")),
)
# A spoken name with no profile still counts, so it can be reported as unresolved.
for name, (count, chapters) in spread.items():
if name and name not in facts:
status = ledger.get(name, "pending")
facts[name] = SpeakerFacts(
name=name,
utterances=count,
chapters=chapters,
approved=str(status).casefold() == "approved",
)
return facts
def similarity(a: SpeakerFacts, b: SpeakerFacts) -> tuple[float, list[str]]:
"""Score two names and say WHICH signals fired. Explainability is the point.
Order matters. The distinguishing-modifier check comes early and is allowed to *veto*:
"the oldest" and "the tallest" are two men in one crowd, and no amount of string
similarity should propose merging them.
"""
if a.name.strip().casefold() == b.name.strip().casefold():
return 1.0, ["identical apart from case/whitespace"]
if a.normalized and a.normalized == b.normalized:
return 0.99, [f"same after normalization ({a.normalized!r})"]
a_tokens, b_tokens = a.tokens, b.tokens
conflict = distinguishing_conflict(a_tokens, b_tokens)
if conflict:
return 0.25, ["LIKELY DIFFERENT PEOPLE: " + "; ".join(conflict)]
shared = a_tokens & b_tokens
extra_a = a_tokens - b_tokens
extra_b = b_tokens - a_tokens
if shared and (extra_a <= ACTIVITY_MODIFIERS) and (extra_b <= ACTIVITY_MODIFIERS):
return 0.95, [
f"same person described by activity only (shared {sorted(shared)}, "
f"extra {sorted(extra_a | extra_b)})"
]
# NOTE: a "shared token plus a narrower role noun" rule was tried here and removed.
# It proposed "The Oldest" ~ "The Old Woman" because both contain "old", and they are
# different people - the oldest of a crowd and an old woman are not the same character.
# A missed merge is recoverable by hand; a wrong one silently rewrites a speaker, so the
# rule that finds fewer duplicates is the better rule.
ratio = difflib.SequenceMatcher(None, a.normalized, b.normalized).ratio()
if shared and ratio >= 0.80:
return round(min(0.9, ratio), 3), [
f"share {sorted(shared)} and are similar (ratio {ratio:.2f})"
]
# Two real names must be genuinely close. "Anfisa" and "Tanisha" score 0.77 on raw
# spelling similarity and are almost certainly different people, so proper names are held
# to a higher bar than designations.
if not a.is_designation and not b.is_designation:
if ratio >= 0.88:
return round(ratio, 3), [f"near-identical proper name (ratio {ratio:.2f})"]
return 0.0, []
if ratio >= MIN_SIMILARITY:
return round(ratio - 0.05, 3), [f"similar spelling (ratio {ratio:.2f})"]
if shared:
return round(min(0.65, 0.4 + 0.1 * len(shared)), 3), [
f"share only {sorted(shared)} - probably different roles"
]
return 0.0, []
@dataclass
class DuplicateGroup:
"""One suspected family of duplicate speakers."""
members: list[str]
score: float
signals: list[str]
verdict: str = "review"
reasons: list[str] = field(default_factory=list)
survivor: str | None = None
survivor_reasons: list[str] = field(default_factory=list)
def as_dict(self) -> dict[str, Any]:
return {
"members": self.members,
"survivor": self.survivor,
"verdict": self.verdict,
"score": self.score,
"signals": self.signals,
"reasons": self.reasons,
"survivor_reasons": self.survivor_reasons,
}
def _survivor_rank(facts: SpeakerFacts) -> tuple:
"""Lower sorts better. Order encodes what must not be lost, in priority order."""
return (
0 if facts.approved else 1, # an approval is never sacrificed
0 if facts.has_ref else 1, # a created voice reference is never sacrificed
-facts.utterances, # then: whoever is heard most
-facts.chapters, # then: whoever spans the most chapters
1 if facts.is_designation else 0, # then: a real name over a designation
len(facts.tokens) * -1, # then: the more specific designation
facts.name, # stable tiebreak
)
def choose_survivor(group: list[SpeakerFacts]) -> tuple[SpeakerFacts, list[str]]:
"""Pick who survives a merge, and explain why."""
ranked = sorted(group, key=_survivor_rank)
winner = ranked[0]
reasons: list[str] = []
approved = [f for f in group if f.approved]
if approved:
reasons.append(
f"keeps the approved voice ({', '.join(sorted(f.name for f in approved))})"
)
with_ref = [f for f in group if f.has_ref]
if with_ref:
reasons.append(f"has a voice reference ({', '.join(sorted(f.name for f in with_ref))})")
if winner.utterances:
reasons.append(f"heard most ({winner.utterances} utterances across "
f"{winner.chapters} chapters)")
else:
reasons.append("is the only member with any lines at all")
if not winner.is_designation:
reasons.append("is a proper name rather than a designation")
return winner, reasons
def _voice_conflicts(group: list[SpeakerFacts]) -> list[str]:
"""Two members with DIFFERENT voice references is a human decision, never a merge."""
with_ref = [f for f in group if f.has_ref and f.ref_sha256]
distinct = {f.ref_sha256 for f in with_ref}
if len(with_ref) > 1 and len(distinct) > 1:
return [
"conflicting voice references: "
+ ", ".join(f"{f.name}={f.ref_sha256}" for f in with_ref)
]
return []
def classify(group: list[SpeakerFacts], score: float) -> tuple[str, list[str]]:
"""``auto`` only when every member is a strong variant *of the survivor*.
Judged per member, not for the group as a whole: two members can be twins of each other
while a third only shares a word with them. A chain must not launder a weak link into a
confident merge.
"""
reasons: list[str] = []
approved = [f for f in group if f.approved]
if len(approved) > 1:
reasons.append(
"more than one member has an approved voice: merging would discard one"
)
return "blocked", reasons
conflicts = _voice_conflicts(group)
if conflicts:
reasons.extend(conflicts)
return "blocked", reasons
speaking = [f for f in group if f.utterances > 0]
silent = [f for f in group if f.utterances == 0]
if len(speaking) > 1:
if all(f.is_designation for f in speaking):
reasons.append(
"two designations both have lines; likely one person, but both were heard"
)
else:
reasons.append("both members have lines and are not both designations")
return "review", reasons
if not speaking:
reasons.append("no member has any lines; merging would move nothing")
return "review", reasons
if not silent:
reasons.append("no silent twin found")
return "review", reasons
reasons.append(
f"{len(silent)} member(s) exist only as scanned names with no lines, so they are "
"almost certainly the speaking member's twin"
)
if score >= AUTO_SIMILARITY:
reasons.append(f"every member is a strong variant (weakest score {score})")
return "auto", reasons
reasons.append(
f"the weakest member only scores {score} against the survivor; confirm before merging"
)
return "review", reasons
def candidate_groups(
facts: dict[str, SpeakerFacts], min_score: float = MIN_SIMILARITY
) -> list[DuplicateGroup]:
"""Group suspects by transitive similarity. Read-only, deterministic."""
names = sorted(facts)
parent: dict[str, str] = {n: n for n in names}
def find(x: str) -> str:
while parent[x] != x:
parent[x] = parent[parent[x]]
x = parent[x]
return x
def union(a: str, b: str) -> None:
ra, rb = find(a), find(b)
if ra != rb:
parent[max(ra, rb)] = min(ra, rb)
best: dict[tuple[str, str], tuple[float, list[str]]] = {}
for i, a in enumerate(names):
for b in names[i + 1 :]:
score, signals = similarity(facts[a], facts[b])
if score >= min_score and signals:
best[(a, b)] = (score, signals)
union(a, b)
clusters: dict[str, list[str]] = defaultdict(list)
for name in names:
clusters[find(name)].append(name)
groups: list[DuplicateGroup] = []
for members in clusters.values():
if len(members) < 2:
continue
member_set = set(members)
pairs = [
(score, signals)
for (a, b), (score, signals) in best.items()
if a in member_set and b in member_set
]
if not pairs:
continue
score = max(s for s, _ in pairs)
signals = sorted({sig for _, sigs in pairs for sig in sigs})
group = DuplicateGroup(members=sorted(members), score=score, signals=signals)
group.verdict, group.reasons = classify([facts[m] for m in members], score)
winner, why = choose_survivor([facts[m] for m in members])
group.survivor = winner.name
group.survivor_reasons = why
groups.append(group)
groups.sort(key=lambda g: (g.verdict != "auto", -g.score, g.members[0]))
return groups
def unresolved_speaker_names(
db: sqlite3.Connection, known: set[str]
) -> list[tuple[str, int, int]]:
"""Spoken names with no profile entry, matched case-insensitively against profiles.
This is the cheapest, safest duplicate class of all: ``scum leader`` (10 utterances) is
``Scum leader`` (a profile) with a different capital. Nothing about identity is inferred -
the only difference is the case of one letter.
"""
rows = db.execute(
"SELECT speaker_name, COUNT(*), COUNT(DISTINCT chapter_id) FROM utterances "
"WHERE lower(type)='dialogue' AND speaker_name IS NOT NULL GROUP BY speaker_name"
).fetchall()
folded = {k.casefold(): k for k in known}
out: list[tuple[str, int, int]] = []
for name, count, chapters in rows:
clean = str(name).strip()
if clean in known:
continue
match = folded.get(clean.casefold())
if match and match != clean:
out.append((clean, int(count), int(chapters)))
return sorted(out)
def merge_group(
db: sqlite3.Connection,
data_dir: Path,
survivor: str,
duplicates: list[str],
) -> dict[str, Any]:
"""Merge duplicates into the survivor **without losing an approval**.
Two different operations hide behind "merge", and conflating them is a bug:
* The duplicate **has a profile** -> ``merge_speakers`` folds its aliases and traits in
and drops the profile.
* The duplicate is **only a spoken string** (``scum leader`` where the profile is
``Scum leader``) -> there is no profile to fold, so the only correct action is to
re-point the utterances. ``merge_speakers`` raises on this, and rightly so: compared
case-insensitively the names ARE the same and it refuses to merge a speaker with
itself. Found the hard way, on a copy of ``Last Dwarf``.
``orphan_resolution.merge_speakers`` never touches ``voice_status.json``, so an approved
voice on a losing member would vanish silently. This wrapper copies any approval across
first and reports it.
Refuses when two members disagree about their voice: that is a human decision.
"""
from umbraaudio.analysis.orphan_resolution import merge_speakers
data_dir = Path(data_dir)
facts = load_speaker_facts(data_dir)
profiles = profile_names(data_dir)
group = [facts[n] for n in [survivor, *duplicates] if n in facts]
if len(group) < 2:
return {"ok": False, "reason": "member not found"}
conflicts = _voice_conflicts(group)
if conflicts:
return {"ok": False, "reason": "conflicting voice references", "detail": conflicts}
ledger_path = data_dir / "voice_status.json"
ledger: dict[str, Any] = {}
if ledger_path.exists():
try:
ledger = json.loads(ledger_path.read_text(encoding="utf-8"))
except ValueError:
ledger = {}
nested = isinstance(ledger.get("speakers"), dict)
entries = ledger["speakers"] if nested else ledger
carried: list[str] = []
if str(entries.get(survivor, "")).casefold() != "approved":
for name in duplicates:
if str(entries.get(name, "")).casefold() == "approved":
entries[survivor] = "approved"
carried.append(name)
merged_profiles: list[str] = []
repointed: list[str] = []
changed = 0
for name in duplicates:
if name in profiles:
result = merge_speakers(db, data_dir, name, survivor)
if result.get("merged"):
merged_profiles.append(str(result.get("source")))
changed += int(result.get("utterances_changed", 0))
else:
# Spoken-only string: no profile exists, so just move the utterances.
#
# DIALOGUE ONLY, deliberately. `narrator` is the labelling convention for ~5,978
# NARRATION rows as well as the 3 dialogue rows carrying it, and an unscoped
# UPDATE moves all of them - invalidating thousands of clips, because every
# marker records the speaker exactly. Found by running this against a copy of
# Last Dwarf and watching 5,981 rows move instead of 3. Narration casing is a
# separate, deliberate decision; a duplicate scan must not make it.
cursor = db.execute(
"UPDATE utterances SET speaker_name=? "
"WHERE speaker_name=? AND lower(type)='dialogue'",
(survivor, name),
)
changed += cursor.rowcount or 0
repointed.append(name)
if name in entries:
# The losing member must not keep an approval record; the approval now lives on
# the survivor and is documented in the publication record at release time.
entries.pop(name, None)
if carried:
tmp = ledger_path.with_suffix(".json.tmp")
tmp.write_text(json.dumps(ledger, indent=2, sort_keys=True) + "\n", encoding="utf-8")
tmp.replace(ledger_path)
db.commit()
return {
"ok": True,
"survivor": survivor,
"merged": merged_profiles,
"repointed_only": repointed,
"utterances_changed": changed,
"approvals_carried": carried,
"ledger_shape_preserved": "speakers" if nested else "flat",
}