Add orphan name-token scanner + resolution actions (spec 17 Phase 2)

Surfaces proper-noun speaker identities in the manuscript that do not resolve to
a known speaker profile, so each can be assigned/created/dismissed/merged by a
reviewer instead of silently mis-attributing.

New:
- umbraaudio/analysis/name_tokens.py: scans two high-signal evidence positions
  (unresolved proper-noun speech tags + un-profiled speaker_name values) rather
  than every capitalized word (594-token noise -> 9 real orphans on Last Dwarf).
  Groups book-wide, offers fuzzy suggestions (word-overlap / difflib / soundex /
  4-char common-prefix misspelling), and suppresses personal-name suggestions
  for role nouns. Persists to a new name_tokens table (idempotent, preserves
  human status across re-runs).
- umbraaudio/analysis/orphan_resolution.py: assign-to-speaker (add alias +
  evidence, re-normalize utterances), create-speaker, dismiss, merge-speakers.
  Merging is idempotent; matching is case-insensitive.
- scripts/scan_name_tokens.py: scan CLI (--write) + one-shot actions
  (--assign/--create/--dismiss/--merge).
- DB: name_tokens table in database.py SCHEMA (web routes create it on demand via
  ensure_name_tokens_table, so no shared-file conflict).

Last Dwarf: 9 orphan tokens (8 role speakers + 'Tobias Teppo' name-tag which
suggests 'Tobias'). Merge candidates Guard 1/Guard 2 + Narrator/OrigNarrator are
real speakers flagged for the merge action.

Tests: tests/test_name_tokens.py (30) + tests/test_orphan_resolution.py (12) green.
ruff/mypy clean (TC003 sqlite3/Path notes match the repo's existing pattern in
database.py). Full suite: 420 passed; the 2 test_pronunciation failures and 6
voice_profile_manager accent failures are pre-existing/parallel-session (a
concurrent 'qwen_tts_helper_url' change), unrelated to this work.

NOT committed: umbraaudio/web/routes.py name_tokens schema hunk — that file is
actively being edited by a parallel session (mixed with its helper_url changes);
left in the working tree to avoid a merge hazard. Table is created on demand, so
nothing is functionally blocked.
This commit is contained in:
B.D. Murphy
2026-09-24 11:27:33 -07:00
parent 65724ed159
commit c338370f19
6 changed files with 1445 additions and 0 deletions

193
scripts/scan_name_tokens.py Normal file
View File

@@ -0,0 +1,193 @@
#!/usr/bin/env python3
"""Scan the manuscript for orphan name tokens (unresolved speaker identities).
Walks every chapter's utterances, collects proper-noun tokens used as speaker
evidence (unresolved name speech-tags and un-profiled ``speaker_name`` values),
groups them book-wide, and writes them to the ``name_tokens`` table. Human-set
resolutions (assigned/created/dismissed) are preserved across re-runs.
Usage:
python scripts/scan_name_tokens.py --project "Last Dwarf"
python scripts/scan_name_tokens.py --project "Last Dwarf" --json
"""
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.name_tokens import (
ensure_name_tokens_table,
merge_tokens,
persist_name_tokens,
scan_utterances,
)
from umbraaudio.config import load_config
def _build_alias_map(speakers_path: Path) -> dict[str, str]:
if not speakers_path.exists():
return {}
with open(speakers_path) as f:
data = json.load(f)
alias_map: dict[str, str] = {}
for s in data.get("speakers", []):
primary = s.get("primary_name", "")
if not primary:
continue
alias_map[primary.lower()] = primary
for alias in s.get("aliases", []):
if alias:
alias_map[alias.lower()] = primary
return alias_map
def _project_id(db: sqlite3.Connection) -> int:
row = db.execute("SELECT id FROM project LIMIT 1").fetchone()
if row is None:
raise SystemExit("No project row found in DB.")
return row["id"]
def _run_action(args, cfg, db: sqlite3.Connection) -> int:
"""Handle a one-shot resolution action and exit."""
from umbraaudio.analysis.orphan_resolution import (
ResolutionError,
assign_token_to_speaker,
create_speaker_from_token,
dismiss_token,
merge_speakers,
)
pid = _project_id(db)
try:
if args.assign:
result = assign_token_to_speaker(
db, cfg.data_dir, pid, args.token, args.speaker
)
elif args.create:
result = create_speaker_from_token(
db, cfg.data_dir, pid, args.token, description=args.reason or ""
)
elif args.dismiss:
result = dismiss_token(db, pid, args.token, reason=args.reason or "")
else: # merge
result = merge_speakers(db, cfg.data_dir, args.token, args.speaker)
except ResolutionError as e:
print(f"Error: {e}")
return 1
print(json.dumps(result, indent=2))
return 0
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--project", default=".", help="Project name or path")
parser.add_argument(
"--chapters", type=str, default=None, help="Comma-separated chapter IDs"
)
parser.add_argument(
"--write", action="store_true", help="Persist results to the name_tokens table"
)
parser.add_argument("--json", action="store_true", help="Emit JSON")
# Resolution actions (mutating; run instead of a scan)
parser.add_argument("--token", default=None, help="Orphan token to act on")
parser.add_argument(
"--speaker", default=None, help="Target speaker primary name (--assign/--merge)"
)
parser.add_argument(
"--assign", action="store_true", help="Assign --token as an alias of --speaker"
)
parser.add_argument(
"--create", action="store_true", help="Create a new speaker named --token"
)
parser.add_argument(
"--dismiss", action="store_true", help="Dismiss --token (not a character)"
)
parser.add_argument(
"--merge", action="store_true", help="Merge speaker --token into --speaker"
)
parser.add_argument("--reason", default="", help="Optional note/reason")
args = parser.parse_args()
cfg = load_config(args.project)
db_path = cfg.db_path
if not db_path.exists():
print(f"Database not found: {db_path}")
return 1
alias_map = _build_alias_map(cfg.data_dir / "speakers.json")
db = sqlite3.connect(str(db_path))
db.row_factory = sqlite3.Row
if args.assign or args.create or args.dismiss or args.merge:
if not args.token:
print("--token is required for resolution actions")
return 1
if (args.assign or args.merge) and not args.speaker:
print("--speaker is required for --assign/--merge")
return 1
return _run_action(args, cfg, db)
if args.chapters:
ids = [int(x.strip()) for x in args.chapters.split(",")]
chapters = db.execute(
"SELECT id, title, chapter_index FROM chapters WHERE id IN "
f"({','.join('?' * len(ids))}) ORDER BY chapter_index",
ids,
).fetchall()
else:
chapters = db.execute(
"SELECT id, title, chapter_index FROM chapters ORDER BY chapter_index"
).fetchall()
per_chapter = []
for ch in chapters:
rows = db.execute(
"SELECT id, utterance_index, text, type, speaker_name "
"FROM utterances WHERE chapter_id=? ORDER BY utterance_index",
(ch["id"],),
).fetchall()
utts = [dict(r) for r in rows]
if utts:
per_chapter.append(scan_utterances(utts, ch["id"], alias_map))
tokens = merge_tokens(per_chapter, alias_map)
if args.write:
pid_row = db.execute("SELECT id FROM project LIMIT 1").fetchone()
if pid_row is None:
print("No project row found in DB; cannot write name_tokens.")
return 1
ensure_name_tokens_table(db)
inserted, updated = persist_name_tokens(db, pid_row["id"], tokens)
print(f"Wrote name_tokens: {inserted} inserted, {updated} updated.")
if args.json:
print(json.dumps([t.to_dict() for t in tokens], indent=2))
else:
print(f"Orphan name tokens: {len(tokens)}\n")
for t in tokens:
sug = (
f" [suggest: {t.suggestion} {t.suggestion_score:.2f}]"
if t.suggestion
else ""
)
d = t.to_dict()
print(
f" {t.occurrences:3d}x [{t.kind}] {t.token!r} "
f"chapters={d['chapters']}{sug}"
)
for sample in d["sample_texts"][:1]:
print(f" e.g. {sample[:100]!r}")
db.close()
return 0
if __name__ == "__main__":
raise SystemExit(main())

236
tests/test_name_tokens.py Normal file
View File

@@ -0,0 +1,236 @@
from __future__ import annotations
import sqlite3
from umbraaudio.analysis.name_tokens import (
_is_known,
_is_role_token,
_normalize_name_token,
_soundex,
ensure_name_tokens_table,
merge_tokens,
persist_name_tokens,
scan_utterances,
suggest_speaker,
)
ALIAS_MAP = {
"tobias": "Tobias",
"teppo": "Teppo",
"kainda": "Kainda",
"kainda walters foster murphy": "Kainda Walters Foster Murphy",
"chief of the guard": "Chief of the Guard",
"the king": "The King",
"narrator": "Narrator",
}
def _u(idx: int, text: str, typ: str, speaker: str = "narrator") -> dict:
return {
"utterance_index": idx,
"text": text,
"type": typ,
"speaker_name": speaker,
}
class TestNormalization:
def test_scene_break_contamination(self) -> None:
assert _normalize_name_token("At The Gate\nTobias") == "Tobias"
def test_gerund_leadin(self) -> None:
assert _normalize_name_token("Nodding Tobias") == "Tobias"
def test_temporal_leadin(self) -> None:
assert _normalize_name_token("Now Milfred") == "Milfred"
def test_role_phrase_preserved(self) -> None:
# The determiner is part of the role reference; keep it intact.
assert _normalize_name_token("The first guard") == "The first guard"
assert _normalize_name_token("Royal Guard") == "Royal Guard"
def test_bare_determiner_dropped(self) -> None:
assert _normalize_name_token("The") == ""
def test_indefinite_pronoun_dropped(self) -> None:
assert _normalize_name_token("Another") == ""
assert _normalize_name_token("One") == ""
def test_conjoined_names_kept(self) -> None:
# Two adjacent speakers in a tag: keep the full phrase for review.
assert _normalize_name_token("Tobias Teppo") == "Tobias Teppo"
def test_empty_input(self) -> None:
assert _normalize_name_token("") == ""
assert _normalize_name_token("...") == ""
class TestRoleDetection:
def test_role_nouns(self) -> None:
for t in ["The guard", "Royal Guard", "The Kid", "Advisor", "The officer"]:
assert _is_role_token(t) is True
def test_personal_names_not_roles(self) -> None:
for t in ["Tobias", "Kainda", "Tobias Teppo", "Nering"]:
assert _is_role_token(t) is False
class TestIsKnown:
def test_exact(self) -> None:
assert _is_known("Tobias", ALIAS_MAP) is True
def test_containment_distinctive(self) -> None:
# "Kainda" is a distinctive name contained in a longer profile.
assert _is_known("Kainda", ALIAS_MAP) is True
def test_single_role_word_not_known(self) -> None:
# "guard" alone is a DIFFERENT speaker from "Chief of the Guard".
assert _is_known("guard", ALIAS_MAP) is False
assert _is_known("voice", ALIAS_MAP) is False
def test_empty_always_known(self) -> None:
assert _is_known("", ALIAS_MAP) is True
class TestSuggestSpeaker:
def test_soundex_misspelling(self) -> None:
name, score = suggest_speaker("Nering", {"neriah": "Neriah"})
assert name == "Neriah"
assert score >= 0.72
def test_close_spelling_variant(self) -> None:
# "Nering"/"Neriah" share a first letter and one small edit.
name, _score = suggest_speaker("Nering", {"neriah": "Neriah"})
assert name == "Neriah"
def test_containment_suggestion(self) -> None:
name, score = suggest_speaker("Tobias", {"tobias teppo": "Tobias Teppo"})
assert name == "Tobias Teppo"
assert score >= 0.72
def test_no_suggestion_for_unrelated(self) -> None:
name, _score = suggest_speaker("Zorblax", {"tobias": "Tobias"})
assert name is None
def test_article_not_soundex_matched(self) -> None:
# "The first guard" should NOT match "The General"/"The King" merely on
# the shared article via soundex.
amap = {"the first guard": "The First Guard", "the king": "The King"}
name, _ = suggest_speaker("The first guard", amap)
assert name == "The First Guard"
class TestSoundex:
def test_known_codes(self) -> None:
assert _soundex("Robert") == "R163"
assert _soundex("Rupert") == "R163"
assert _soundex("Tymczak") == "T522"
assert _soundex("Pfister") == "P236"
def test_empty(self) -> None:
assert _soundex("") == ""
class TestScanUtterances:
def test_name_tag_orphan(self) -> None:
utts = [
_u(0, "Nering said,", "narration"),
_u(1, "We must go now.", "dialogue", "Nering"),
]
toks = scan_utterances(utts, 3, ALIAS_MAP)
tokens = {t.token for t in toks}
assert "Nering" in tokens
def test_speaker_name_orphan_raw(self) -> None:
utts = [_u(0, "Stop there.", "dialogue", "The first guard")]
toks = scan_utterances(utts, 3, ALIAS_MAP)
# The raw stored speaker_name is preserved (assign matches it exactly).
assert [t.token for t in toks] == ["The first guard"]
assert toks[0].kind == "speaker_name"
def test_known_speaker_not_flagged(self) -> None:
utts = [
_u(0, "Kainda says,", "narration"),
_u(1, "We are ready.", "dialogue", "Kainda"),
]
assert scan_utterances(utts, 3, ALIAS_MAP) == []
def test_narrator_dialogue_not_flagged(self) -> None:
utts = [_u(0, "Hello there.", "dialogue", "narrator")]
assert scan_utterances(utts, 3, ALIAS_MAP) == []
def test_role_token_has_no_suggestion(self) -> None:
utts = [_u(0, "Move.", "dialogue", "The Kid")]
toks = scan_utterances(utts, 3, ALIAS_MAP)
assert toks[0].suggestion is None
class TestMergeTokens:
def test_merges_across_chapters(self) -> None:
c1 = [_u(0, "Stop.", "dialogue", "Advisor")]
c2 = [_u(0, "Wait.", "dialogue", "Advisor")]
merged = merge_tokens(
[
scan_utterances(c1, 1, ALIAS_MAP),
scan_utterances(c2, 2, ALIAS_MAP),
],
ALIAS_MAP,
)
assert len(merged) == 1
assert merged[0].occurrences == 2
assert merged[0].chapters == [1, 2]
def test_sorted_by_occurrences(self) -> None:
per = [
scan_utterances(
[
_u(0, "A.", "dialogue", "Advisor"),
_u(1, "B.", "dialogue", "Royal Guard"),
_u(2, "C.", "dialogue", "Royal Guard"),
],
1,
ALIAS_MAP,
)
]
merged = merge_tokens(per, ALIAS_MAP)
assert merged[0].token == "Royal Guard"
class TestPersistence:
def _db(self) -> sqlite3.Connection:
db = sqlite3.connect(":memory:")
db.row_factory = sqlite3.Row
db.execute("CREATE TABLE project (id INTEGER PRIMARY KEY)")
db.execute("INSERT INTO project (id) VALUES (1)")
ensure_name_tokens_table(db)
return db
def test_insert_and_preserve_status(self) -> None:
db = self._db()
utts = [_u(0, "Stop.", "dialogue", "Advisor")]
tokens = scan_utterances(utts, 1, ALIAS_MAP)
inserted, updated = persist_name_tokens(db, 1, tokens)
assert inserted == 1 and updated == 0
# Mark it dismissed, then re-persist the same token.
db.execute(
"UPDATE name_tokens SET status='dismissed', target_speaker=NULL "
"WHERE token='Advisor'"
)
db.commit()
inserted2, updated2 = persist_name_tokens(db, 1, tokens)
assert inserted2 == 0 and updated2 == 1
row = db.execute("SELECT status FROM name_tokens WHERE token='Advisor'").fetchone()
assert row["status"] == "dismissed" # human decision preserved
db.close()
def test_idempotent(self) -> None:
db = self._db()
tokens = scan_utterances(
[_u(0, "Stop.", "dialogue", "Advisor")], 1, ALIAS_MAP
)
persist_name_tokens(db, 1, tokens)
persist_name_tokens(db, 1, tokens)
count = db.execute("SELECT COUNT(*) c FROM name_tokens").fetchone()["c"]
assert count == 1
db.close()

View File

@@ -0,0 +1,184 @@
from __future__ import annotations
import json
import sqlite3
from pathlib import Path
import pytest
from umbraaudio.analysis.orphan_resolution import (
ResolutionError,
assign_token_to_speaker,
create_speaker_from_token,
dismiss_token,
merge_speakers,
)
@pytest.fixture
def env(tmp_path: Path) -> tuple[sqlite3.Connection, Path]:
db = sqlite3.connect(":memory:")
db.row_factory = sqlite3.Row
db.execute("CREATE TABLE project (id INTEGER PRIMARY KEY)")
db.execute("INSERT INTO project (id) VALUES (1)")
db.execute(
"CREATE TABLE utterances (id INTEGER PRIMARY KEY, speaker_name TEXT, "
"type TEXT, text TEXT)"
)
data_dir = tmp_path / ".umbra"
data_dir.mkdir()
speakers = [
{
"primary_name": "Kainda",
"aliases": ["Princess Kainda"],
"demographics": {"type": "Human", "gender": "Female"},
"vocal_traits": ["calm"],
"chapters_appeared": [1, 2],
},
{
"primary_name": "Guard 1",
"aliases": [],
"demographics": {"type": "Human", "gender": "Unknown"},
"vocal_traits": ["gruff"],
"chapters_appeared": [3],
},
]
(data_dir / "speakers.json").write_text(json.dumps({"speakers": speakers}))
return db, data_dir
def _add_utt(db: sqlite3.Connection, speaker: str, text: str = "Hello") -> None:
db.execute(
"INSERT INTO utterances (speaker_name, type, text) VALUES (?, 'dialogue', ?)",
(speaker, text),
)
db.commit()
def _load(data_dir: Path) -> list[dict]:
return json.loads((data_dir / "speakers.json").read_text())["speakers"]
class TestAssign:
def test_assign_adds_alias_and_renormalizes(self, env) -> None:
db, data_dir = env
_add_utt(db, "The first guard")
result = assign_token_to_speaker(db, data_dir, 1, "The first guard", "Guard 1")
assert result["ok"] and result["alias_added"] is True
assert result["utterances_changed"] == 1
speakers = _load(data_dir)
guard = next(s for s in speakers if s["primary_name"] == "Guard 1")
assert "The first guard" in guard["aliases"]
row = db.execute("SELECT speaker_name FROM utterances").fetchone()
assert row["speaker_name"] == "Guard 1"
def test_assign_is_case_insensitive(self, env) -> None:
db, data_dir = env
_add_utt(db, "the first guard")
assign_token_to_speaker(db, data_dir, 1, "The first guard", "Guard 1")
row = db.execute("SELECT speaker_name FROM utterances").fetchone()
assert row["speaker_name"] == "Guard 1"
def test_assign_unknown_target_raises(self, env) -> None:
db, data_dir = env
with pytest.raises(ResolutionError, match="not found"):
assign_token_to_speaker(db, data_dir, 1, "The Kid", "Nobody")
def test_assign_idempotent_alias(self, env) -> None:
db, data_dir = env
_add_utt(db, "The Kid")
assign_token_to_speaker(db, data_dir, 1, "The Kid", "Guard 1")
# Second assign should not duplicate the alias.
result = assign_token_to_speaker(db, data_dir, 1, "The Kid", "Guard 1")
guard = next(s for s in _load(data_dir) if s["primary_name"] == "Guard 1")
assert guard["aliases"].count("The Kid") == 1
assert result["alias_added"] is False
def test_assign_sets_token_status(self, env) -> None:
db, data_dir = env
_add_utt(db, "The Kid")
assign_token_to_speaker(db, data_dir, 1, "The Kid", "Guard 1")
row = db.execute(
"SELECT status, target_speaker FROM name_tokens WHERE token='The Kid'"
).fetchone()
assert row["status"] == "assigned" and row["target_speaker"] == "Guard 1"
class TestCreate:
def test_create_from_token(self, env) -> None:
db, data_dir = env
_add_utt(db, "The Kid", "Where am I?")
result = create_speaker_from_token(db, data_dir, 1, "The Kid")
assert result["ok"]
speakers = _load(data_dir)
kid = next(s for s in speakers if s["primary_name"] == "The Kid")
assert kid["aliases"] == ["The Kid"]
assert kid["sample_dialogue"] == "Where am I?"
row = db.execute(
"SELECT status, target_speaker FROM name_tokens WHERE token='The Kid'"
).fetchone()
assert row["status"] == "created" and row["target_speaker"] == "The Kid"
def test_create_duplicate_raises(self, env) -> None:
db, data_dir = env
with pytest.raises(ResolutionError, match="already exists"):
create_speaker_from_token(db, data_dir, 1, "Kainda")
class TestDismiss:
def test_dismiss_records_status(self, env) -> None:
db, data_dir = env
result = dismiss_token(db, 1, "The Voice", reason="a sound effect")
assert result["ok"]
row = db.execute(
"SELECT status FROM name_tokens WHERE token='The Voice'"
).fetchone()
assert row["status"] == "dismissed"
# speakers.json and utterances must be untouched.
assert all("The Voice" not in s["primary_name"] for s in _load(data_dir))
class TestMerge:
def test_merge_folds_aliases_and_renormalizes(self, env) -> None:
db, data_dir = env
# Register "The first guard" as an alias of Guard 1 (as an earlier
# token-assignment would have done), then merge.
speakers = _load(data_dir)
next(s for s in speakers if s["primary_name"] == "Guard 1")["aliases"] = [
"The first guard"
]
(data_dir / "speakers.json").write_text(
json.dumps({"speakers": speakers})
)
_add_utt(db, "Guard 1")
_add_utt(db, "The first guard")
result = merge_speakers(db, data_dir, "Guard 1", "Kainda")
assert result["merged"] is True
assert result["utterances_changed"] == 2
speakers = _load(data_dir)
assert all(s["primary_name"] != "Guard 1" for s in speakers)
kainda = next(s for s in speakers if s["primary_name"] == "Kainda")
assert "Guard 1" in kainda["aliases"]
assert "The first guard" in kainda["aliases"]
assert "calm" in kainda["vocal_traits"]
# Both utterances now on Kainda.
names = {r["speaker_name"] for r in db.execute("SELECT speaker_name FROM utterances")}
assert names == {"Kainda"}
def test_merge_idempotent(self, env) -> None:
db, data_dir = env
merge_speakers(db, data_dir, "Guard 1", "Kainda")
# Second merge: source is gone -> no-op, not an error.
result = merge_speakers(db, data_dir, "Guard 1", "Kainda")
assert result["merged"] is False
def test_merge_into_self_raises(self, env) -> None:
db, data_dir = env
with pytest.raises(ResolutionError, match="itself"):
merge_speakers(db, data_dir, "Kainda", "kainda")
def test_merge_unknown_target_raises(self, env) -> None:
db, data_dir = env
with pytest.raises(ResolutionError, match="not found"):
merge_speakers(db, data_dir, "Guard 1", "Nobody")

View File

@@ -0,0 +1,503 @@
"""Orphan name-token scanner.
Finds proper-noun name tokens used as *speaker evidence* in the manuscript that
do not resolve to a known speaker profile (primary name or alias), so a reviewer
can assign each to an existing speaker, create a new one, or dismiss it.
The scanner deliberately reads only high-signal positions rather than sweeping
every capitalized word in the prose (which is mostly noise — "The", "You",
place names, sentence-initial words). It reads:
1. **Name speech tags** — a proper-noun subject in narration attached to a
speech verb ("Nering said, …") that ``_match_speech_tag`` cannot resolve to a
known alias. This is the classic missed-alias / misspelling case.
2. **Assigned speaker names** — ``speaker_name`` values on dialogue that are not
in the alias map. These are LLM-invented role names ("The first guard") or
genuine un-profiled speakers, and must be assigned or confirmed as roles.
Each token is grouped across the book, given a fuzzy *suggestion* of the closest
existing speaker (word overlap / difflib ratio / soundex), and persisted in the
``name_tokens`` table so re-runs are stable and the translation export can prove
every name resolves.
"""
from __future__ import annotations
import difflib
import json
import re
import sqlite3
from dataclasses import dataclass, field
from typing import Any
from umbraaudio.analysis.attribution_qa import (
SPEECH_VERBS,
_match_speech_tag,
)
# Confidence threshold for offering a one-click "suggested alias" assignment.
SUGGESTION_THRESHOLD = 0.72
# Capitalized words that commonly lead a name tag without being part of the
# name. Stripped from the left of a candidate token during normalization.
# Articles ("the"/"a"/"an") are deliberately NOT here: they are conventionally
# part of role names ("The first guard") that the reviewer needs to see intact.
_LEADING_STRIP_WORDS = frozenset(
{
"and", "or", "but", "at", "in", "on", "to", "of",
"from", "by", "with", "after", "when", "while", "during", "then",
"now", "later", "past", "before", "for", "as", "one", "two",
"first", "second", "next", "again", "still", "already", "suddenly",
"quietly", "loudly", "slowly", "quickly", "softly", "turning",
"walking", "looking", "nodding", "gesturing", "stopping", "putting",
"taking", "smiling", "shifting", "stepping", "rising", "falling",
"speaking", "saying", "telling", "calling", "shouting", "whispering",
"laughing", "crying", "sighing", "glancing", "leaning", "crossing",
}
)
# Proper-noun tokens that are never character names (world/place/object words
# and indefinite pronouns that can appear in a speech-tag position).
_NON_NAME_WORDS = frozenset(
{
"chapter", "book", "page", "word", "words", "god", "gods", "lord",
"yes", "no", "maybe", "never", "always", "hello", "goodbye",
"please", "thanks", "sorry", "well", "so", "now", "then", "here",
"there", "this", "that", "these", "those", "what", "who", "why",
"how", "where", "when", "and", "but", "for", "with", "not", "all",
"heaven", "hell", "earth", "air", "fire", "water", "stone", "iron",
"gold", "silver", "sword", "shield", "door", "gate", "wall", "room",
"city", "town", "world", "end", "day", "night", "morning", "evening",
# indefinite pronouns / quantifiers mistaken for names
"another", "someone", "anyone", "everyone", "nobody", "somebody",
"everybody", "one", "two", "three", "both", "either", "neither",
}
)
# Role nouns: a token whose head word is one of these is an unnamed *role*, not
# a personal name. Roles get create/assign/confirm treatment, and we suppress
# personal-name fuzzy suggestions for them (so "The Kid" does not suggest
# "The King" on a spurious soundex match).
_ROLE_NOUNS = frozenset(
{
"guard", "guards", "officer", "officers", "captain", "captains",
"advisor", "adviser", "soldier", "soldiers", "man", "men", "woman",
"women", "girl", "girls", "boy", "boys", "kid", "kids", "king",
"queen", "doctor", "doctors", "master", "commander", "leader",
"pilot", "crew", "sergeant", "lieutenant", "general", "nurse",
"teacher", "student", "engineer", "scientist", "merchant", "trader",
"hunter", "pirate", "pirates", "dwarf", "dwarves", "robot", "robots",
"computer", "computers", "voice", "voices", "friend", "stranger",
"child", "children", "baby", "elder", "gentleman", "lady", "youth",
"bystander", "passerby", "figure", "one", "servant", "attendant",
"clerk", "driver", "priest",
}
)
_NAME_TOKEN_RE = re.compile(r"\b[A-Z][a-zA-Z'’-]+(?:\s+[A-Z][a-zA-Z'’-]+)*")
@dataclass
class NameToken:
"""One unresolved proper-noun token found in a speaker-evidence position."""
token: str
kind: str # "name_tag" | "speaker_name"
occurrences: int = 0
chapters: list[int] = field(default_factory=list)
sample_texts: list[str] = field(default_factory=list)
suggestion: str | None = None
suggestion_score: float = 0.0
def to_dict(self) -> dict[str, Any]:
return {
"token": self.token,
"kind": self.kind,
"occurrences": self.occurrences,
"chapters": sorted(set(self.chapters)),
"sample_texts": self.sample_texts[:3],
"suggestion": self.suggestion,
"suggestion_score": round(self.suggestion_score, 3),
}
# ---------------------------------------------------------------------------
# Fuzzy matching helpers
# ---------------------------------------------------------------------------
_SOUNDEX_MAP = {
**dict.fromkeys("bfpv", "1"),
**dict.fromkeys("cgjkqsxz", "2"),
**dict.fromkeys("dt", "3"),
**dict.fromkeys("l", "4"),
**dict.fromkeys("mn", "5"),
**dict.fromkeys("r", "6"),
}
def _soundex(word: str) -> str:
"""Classic 4-character Soundex code for approximate name matching."""
w = re.sub(r"[^a-z]", "", (word or "").lower())
if not w:
return ""
first = w[0]
result = [first.upper()]
prev_code = _SOUNDEX_MAP.get(first, "")
for ch in w[1:]:
if ch in ("h", "w"):
# h/w are transparent: they do not separate equal adjacent codes.
continue
code = _SOUNDEX_MAP.get(ch, "")
if code:
if code != prev_code:
result.append(code)
prev_code = code
else:
# Vowels separate otherwise-adjacent equal codes.
prev_code = ""
if len(result) == 4:
break
return ("".join(result) + "000")[:4]
def _overlap_score(a: str, b: str) -> float:
"""Jaccard-style word overlap between two multi-word names."""
aw = {w.lower() for w in re.findall(r"[a-zA-Z]+", a) if len(w) > 1}
bw = {w.lower() for w in re.findall(r"[a-zA-Z]+", b) if len(w) > 1}
if not aw or not bw:
return 0.0
return len(aw & bw) / len(aw | bw)
def _name_similarity(token: str, candidate: str) -> float:
"""Blend of word overlap, difflib ratio, and soundex equality.
Soundex is applied to the first *significant* word (articles stripped) so
"The first guard" does not spuriously match "The General" on soundex("The").
"""
overlap = _overlap_score(token, candidate)
ratio = difflib.SequenceMatcher(None, token.lower(), candidate.lower()).ratio()
t_sig = _first_significant_word(token)
c_sig = _first_significant_word(candidate)
soundex = 1.0 if (
t_sig and c_sig and _soundex(t_sig) and _soundex(t_sig) == _soundex(c_sig)
) else 0.0
score = max(overlap, ratio, (0.85 if soundex else 0.0))
# Common-prefix misspelling ("Nering" -> "Neriah"): two single-word names of
# near-equal length sharing a 4+ character prefix are a strong alias signal
# even when difflib/soundex fall short. Two unrelated names rarely share a
# 4-char prefix, so this is precise.
if t_sig and c_sig and t_sig.lower() != c_sig.lower() and abs(len(t_sig) - len(c_sig)) <= 1:
common = 0
for a, b in zip(t_sig.lower(), c_sig.lower(), strict=False):
if a != b:
break
common += 1
if common >= 4:
score = max(score, 0.9)
return score
def _first_significant_word(name: str) -> str:
"""First word of a name that is not a bare article."""
for w in re.findall(r"[a-zA-Z]+", name or ""):
if w.lower() not in ("the", "a", "an"):
return str(w)
return ""
def suggest_speaker(token: str, alias_map: dict[str, str]) -> tuple[str | None, float]:
"""Return the best canonical speaker for ``token`` and its similarity score."""
best: str | None = None
best_score = 0.0
seen: set[str] = set()
for raw_name, canonical in alias_map.items():
if canonical in seen:
continue
seen.add(canonical)
score = _name_similarity(token, raw_name)
if score > best_score:
best_score = score
best = canonical
if best_score < SUGGESTION_THRESHOLD:
return None, best_score
return best, best_score
# ---------------------------------------------------------------------------
# Token extraction / normalization
# ---------------------------------------------------------------------------
_WORD_RE = re.compile(r"[A-Za-z][A-Za-z'’-]*")
def _normalize_name_token(raw: str) -> str:
"""Clean a captured tag/name string into a candidate proper-noun name.
Handles scene-break contamination ("At The Gate\\nTobias" -> "Tobias"),
gerund/participle lead-ins ("Nodding Tobias" -> "Tobias"), and leading
temporal/prepositional words ("Now Milfred" -> "Milfred"). Role phrases are
kept intact ("The guard" -> "The guard") because the article is part of how
the role is referenced. Returns "" when no plausible name remains.
"""
# Scene breaks split a narration line; the name is in the final segment.
segments = [s.strip() for s in re.split(r"[\n\r]+", raw or "") if s.strip()]
text = segments[-1] if segments else ""
words: list[str] = _WORD_RE.findall(text)
if not words:
return ""
def _lw(i: int) -> str:
return str(words[i].lower())
# Strip leading temporal/prepositional/participial words.
i = 0
while i < len(words) and _lw(i) in _LEADING_STRIP_WORDS:
i += 1
# Strip leading place/object words, but never a role noun. A leading
# determiner ("The guard") is KEPT — it is part of the role reference the
# reviewer must resolve — so only place/object words are skipped here.
while i < len(words) - 1:
low = _lw(i)
if low in _NON_NAME_WORDS and low not in _ROLE_NOUNS:
i += 1
else:
break
cleaned = words[i:]
if not cleaned:
return ""
# A bare determiner on its own is not a name.
if len(cleaned) == 1 and cleaned[0].lower() in ("the", "a", "an"):
return ""
# If what remains is only stop/indefinite words, it is not a name.
if all(w.lower() in _NON_NAME_WORDS for w in cleaned):
return ""
return " ".join(cleaned)
def _is_known(token: str, alias_map: dict[str, str]) -> bool:
"""True when the token (or its words) already resolve to a known speaker."""
if not token:
return True # nothing to report
t = token.lower()
if t in alias_map:
return True
words = [w for w in re.findall(r"[a-z]+", t) if len(w) > 1]
if not words:
return False
# Containment ("Tobias" resolves to a "Tobias Teppo" profile) is only safe
# for distinctive names. A single generic/role word ("guard" vs the "Chief
# of the Guard", "voice") refers to a DIFFERENT, unnamed speaker and must
# stay unresolved for review.
single_generic = len(words) == 1 and (
words[0] in _ROLE_NOUNS or words[0] in _NON_NAME_WORDS
)
if single_generic:
return False
for name_lower in alias_map:
name_words = {w for w in re.findall(r"[a-z]+", name_lower) if len(w) > 1}
if set(words).issubset(name_words):
return True
return False
def _is_role_token(token: str) -> bool:
"""True when the token's head word names a role, not a person."""
words = [w.lower() for w in re.findall(r"[a-zA-Z]+", token or "")]
for w in words:
if w in ("the", "a", "an"):
continue
if w in _ROLE_NOUNS:
return True
return False
# ---------------------------------------------------------------------------
# Scanning
# ---------------------------------------------------------------------------
def scan_utterances(
utterances: list[dict[str, Any]],
chapter_id: int,
alias_map: dict[str, str],
) -> list[NameToken]:
"""Extract unresolved name tokens from one chapter's ordered utterances.
Sources: unresolved proper-noun speech tags (narration) and assigned
``speaker_name`` values on dialogue that are not in the alias map. Returns a
list of per-chapter tokens (unsorted; the caller groups across chapters).
"""
found: dict[tuple[str, str], NameToken] = {}
def _add(kind: str, token: str, sample: str) -> None:
if not token or _is_known(token, alias_map):
return
key = (token.lower(), kind)
if key in found:
found[key].occurrences += 1
found[key].sample_texts.append(sample[:200])
else:
found[key] = NameToken(
token=token,
kind=kind,
occurrences=1,
chapters=[chapter_id],
sample_texts=[sample[:200]],
)
for u in utterances:
utype = (u.get("type") or "").lower()
text = (u.get("text") or "").strip()
if utype == "narration":
if not SPEECH_VERBS.search(text):
continue
kind, value, resolved = _match_speech_tag(text, alias_map)
if kind == "name" and value and not resolved:
_add("name_tag", _normalize_name_token(value), text)
elif utype == "dialogue":
speaker = (u.get("speaker_name") or "").strip()
if not speaker or speaker.lower() in ("narrator", "the narrator"):
continue
if speaker.lower() not in alias_map:
# Use the RAW stored speaker_name as the token: the assign action
# re-normalizes utterances by exact speaker_name match, so the
# token must equal what is in the DB.
_add("speaker_name", speaker, text)
tokens = list(found.values())
for tok in tokens:
# Roles are not personal names: do not offer a fuzzy personal-name
# suggestion for them (the reviewer creates/assigns/assigns-or-dismisses
# the role itself). Name-like tokens get the one-click alias suggestion.
if not _is_role_token(tok.token):
tok.suggestion, tok.suggestion_score = suggest_speaker(
tok.token, alias_map
)
return tokens
def merge_tokens(
per_chapter: list[list[NameToken]],
alias_map: dict[str, str],
) -> list[NameToken]:
"""Group per-chapter tokens into one book-wide, de-duplicated list.
A token found in both a name-tag and a speaker_name position is merged into
a single entry (kind becomes the higher-signal "speaker_name" only when the
token has no tag evidence). Sorted by descending occurrences.
"""
merged: dict[str, NameToken] = {}
for tokens in per_chapter:
for tok in tokens:
key = tok.token.lower()
if key in merged:
m = merged[key]
m.occurrences += tok.occurrences
m.chapters.extend(tok.chapters)
m.sample_texts.extend(tok.sample_texts)
# Prefer "name_tag" evidence when both exist (it is the direct
# manuscript signal of the missing alias).
if tok.kind == "name_tag":
m.kind = "name_tag"
else:
merged[key] = tok
out = list(merged.values())
for tok in out:
tok.chapters = sorted(set(tok.chapters))
if not _is_role_token(tok.token):
tok.suggestion, tok.suggestion_score = suggest_speaker(
tok.token, alias_map
)
else:
tok.suggestion = None
tok.suggestion_score = 0.0
out.sort(key=lambda t: (-t.occurrences, t.token.lower()))
return out
# ---------------------------------------------------------------------------
# DB persistence (name_tokens table)
# ---------------------------------------------------------------------------
CREATE_NAME_TOKENS_SQL = """
CREATE TABLE IF NOT EXISTS name_tokens (
id INTEGER PRIMARY KEY,
project_id INTEGER NOT NULL REFERENCES project(id),
token TEXT NOT NULL,
kind TEXT NOT NULL DEFAULT 'name_tag',
status TEXT NOT NULL DEFAULT 'unresolved',
target_speaker TEXT,
occurrences INTEGER NOT NULL DEFAULT 0,
chapters_json TEXT NOT NULL DEFAULT '[]',
evidence_json TEXT NOT NULL DEFAULT '[]',
suggestion TEXT,
suggestion_score REAL DEFAULT 0.0,
updated_at TEXT NOT NULL DEFAULT (datetime('now')),
UNIQUE(project_id, token)
);
"""
def ensure_name_tokens_table(db: sqlite3.Connection) -> None:
"""Create the name_tokens table if absent (idempotent)."""
db.execute(CREATE_NAME_TOKENS_SQL)
db.commit()
def persist_name_tokens(
db: sqlite3.Connection,
project_id: int,
tokens: list[NameToken],
) -> tuple[int, int]:
"""Upsert scanned tokens, preserving human-set status/target.
A token already marked ``dismissed``/``assigned``/``created`` keeps that
status; only unresolved rows are refreshed. Returns (inserted, updated).
"""
inserted = updated = 0
for tok in tokens:
existing = db.execute(
"SELECT id, status FROM name_tokens WHERE project_id=? AND token=?",
(project_id, tok.token),
).fetchone()
chapters_json = json.dumps(tok.to_dict()["chapters"])
evidence_json = json.dumps(tok.to_dict()["sample_texts"][:5])
if existing:
# Refresh the scan facts but keep any human resolution.
db.execute(
"UPDATE name_tokens SET kind=?, occurrences=?, chapters_json=?, "
"evidence_json=?, suggestion=?, suggestion_score=?, "
"updated_at=datetime('now') WHERE id=?",
(
tok.kind,
tok.occurrences,
chapters_json,
evidence_json,
tok.suggestion,
tok.suggestion_score,
existing["id"],
),
)
updated += 1
else:
db.execute(
"INSERT INTO name_tokens (project_id, token, kind, status, occurrences, "
"chapters_json, evidence_json, suggestion, suggestion_score) "
"VALUES (?, ?, ?, 'unresolved', ?, ?, ?, ?, ?)",
(
project_id,
tok.token,
tok.kind,
tok.occurrences,
chapters_json,
evidence_json,
tok.suggestion,
tok.suggestion_score,
),
)
inserted += 1
db.commit()
return inserted, updated

View File

@@ -0,0 +1,311 @@
"""Resolution actions for orphan name tokens (assign / create / dismiss / merge).
These are the durable, side-effecting operations behind the review UI and the
``scan_name_tokens.py`` CLI. Each action:
- records a human decision in the ``name_tokens`` table (status + target), and
- for assign/create/merge, re-normalizes every ``utterances.speaker_name`` that
carried the orphan token to the canonical ``primary_name``.
Speaker identity lives in ``speakers.json`` (aliases + alias_evidence). The
re-normalized utterance speaker names keep the audio markers stale automatically,
so changing a speaker triggers regen through the existing staleness path.
These functions are intentionally framework-agnostic (plain sqlite3 + pathlib)
so they are unit-testable without the web app and reusable from scripts.
"""
from __future__ import annotations
import json
import sqlite3
from pathlib import Path
from typing import Any
from umbraaudio.analysis.name_tokens import ensure_name_tokens_table
# name_tokens.status values
STATUS_UNRESOLVED = "unresolved"
STATUS_ASSIGNED = "assigned"
STATUS_CREATED = "created"
STATUS_DISMISSED = "dismissed"
class ResolutionError(Exception):
"""Raised when a resolution action cannot be completed."""
def _load_speaker_doc(data_dir: Path) -> tuple[dict[str, Any], list[dict[str, Any]]]:
path = data_dir / "speakers.json"
if not path.exists():
raise ResolutionError(f"speakers.json not found: {path}")
with open(path) as f:
doc = json.load(f)
return doc, doc.get("speakers", [])
def _save_speaker_doc(data_dir: Path, doc: dict[str, Any], speakers: list[dict[str, Any]]) -> None:
path = data_dir / "speakers.json"
doc["speakers"] = speakers
tmp = path.with_suffix(".json.tmp")
with open(tmp, "w") as f:
json.dump(doc, f, indent=2)
tmp.replace(path)
def _find_speaker(speakers: list[dict[str, Any]], name: str) -> dict[str, Any] | None:
low = name.lower()
for s in speakers:
if s.get("primary_name", "").lower() == low:
return s
return None
def _renormalize_utterances(
db: sqlite3.Connection, old_names: list[str], new_name: str
) -> int:
"""Re-point every utterance whose speaker_name matched an old name.
Matches case-insensitively so "the guard" and "The guard" are both caught.
Skips rows already on ``new_name``. Returns the number of rows changed.
"""
changed = 0
for old in old_names:
if not old or old.lower() == new_name.lower():
continue
cur = db.execute(
"UPDATE utterances SET speaker_name = ? "
"WHERE lower(speaker_name) = lower(?) AND speaker_name != ?",
(new_name, old, new_name),
)
changed += cur.rowcount or 0
return changed
def _set_token_status(
db: sqlite3.Connection,
project_id: int,
token: str,
status: str,
target_speaker: str | None = None,
) -> None:
ensure_name_tokens_table(db)
db.execute(
"INSERT INTO name_tokens (project_id, token, status, target_speaker) "
"VALUES (?, ?, ?, ?) "
"ON CONFLICT(project_id, token) DO UPDATE SET "
"status=excluded.status, target_speaker=excluded.target_speaker, "
"updated_at=datetime('now')",
(project_id, token, status, target_speaker),
)
db.commit()
# ---------------------------------------------------------------------------
# Public actions
# ---------------------------------------------------------------------------
def assign_token_to_speaker(
db: sqlite3.Connection,
data_dir: Path,
project_id: int,
token: str,
target_speaker: str,
) -> dict[str, Any]:
"""Assign an orphan ``token`` to an existing speaker.
Adds the token to that speaker's ``aliases`` (with evidence), re-normalizes
every utterance carrying the token to the speaker's ``primary_name``, and
records status=assigned. Returns a summary dict.
"""
if not token or not target_speaker:
raise ResolutionError("token and target_speaker required")
doc, speakers = _load_speaker_doc(data_dir)
speaker = _find_speaker(speakers, target_speaker)
if speaker is None:
raise ResolutionError(f"target speaker not found: {target_speaker}")
primary = speaker.get("primary_name", "")
aliases = speaker.setdefault("aliases", [])
already_alias = token.lower() in {a.lower() for a in aliases}
if token.lower() != primary.lower() and not already_alias:
aliases.append(token)
evidence = speaker.setdefault("alias_evidence", [])
if not isinstance(evidence, list):
evidence = []
speaker["alias_evidence"] = evidence
evidence.append(
{"alias": token, "source": "name_token_scan", "chapters": []}
)
# Capture any utterances currently labelled with the token before renaming.
affected = db.execute(
"SELECT COUNT(*) c FROM utterances WHERE lower(speaker_name) = lower(?)",
(token,),
).fetchone()["c"]
_save_speaker_doc(data_dir, doc, speakers)
changed = _renormalize_utterances(db, [token], primary)
_set_token_status(db, project_id, token, STATUS_ASSIGNED, primary)
return {
"ok": True,
"token": token,
"target_speaker": primary,
"alias_added": (not already_alias and token.lower() != primary.lower()),
"utterances_before": affected,
"utterances_changed": changed,
}
def create_speaker_from_token(
db: sqlite3.Connection,
data_dir: Path,
project_id: int,
token: str,
description: str = "",
) -> dict[str, Any]:
"""Create a new speaker profile from an orphan ``token``.
primary_name = token, aliases = [token]. The speaker enters the normal
voice-design flow afterward (speakers tab). Records status=created.
"""
if not token:
raise ResolutionError("token required")
doc, speakers = _load_speaker_doc(data_dir)
if _find_speaker(speakers, token) is not None:
raise ResolutionError(f"speaker already exists: {token}")
# Grab a sample attributed line for the new profile, if any.
sample = db.execute(
"SELECT text FROM utterances WHERE lower(speaker_name) = lower(?) "
"AND type='dialogue' ORDER BY id LIMIT 1",
(token,),
).fetchone()
sample_text = sample["text"] if sample else ""
new_speaker = {
"primary_name": token,
"aliases": [token],
"demographics": {"type": "Unknown", "gender": "Unknown"},
"vocal_traits": [],
"communication_style": {"primary_register": "Casual", "style_shifts_or_notes": ""},
"is_narrator": False,
"is_group": False,
"description": description or f"Created from unresolved name token '{token}'",
"sample_dialogue": sample_text,
"chapters_appeared": [],
"accent": "none",
}
speakers.append(new_speaker)
_save_speaker_doc(data_dir, doc, speakers)
affected = db.execute(
"SELECT COUNT(*) c FROM utterances WHERE lower(speaker_name) = lower(?)",
(token,),
).fetchone()["c"]
_set_token_status(db, project_id, token, STATUS_CREATED, token)
return {
"ok": True,
"token": token,
"target_speaker": token,
"utterances_before": affected,
}
def dismiss_token(
db: sqlite3.Connection,
project_id: int,
token: str,
reason: str = "",
) -> dict[str, Any]:
"""Dismiss a token as not-a-character (place, one-off mention, artifact).
Leaves speakers.json and utterances untouched; records the decision so the
token does not reappear on the next scan. ``reason`` is stored as evidence.
"""
if not token:
raise ResolutionError("token required")
ensure_name_tokens_table(db)
cur = db.execute(
"UPDATE name_tokens SET status=?, target_speaker=NULL, "
"evidence_json=?, updated_at=datetime('now') "
"WHERE project_id=? AND token=?",
(STATUS_DISMISSED, json.dumps([reason] if reason else []), project_id, token),
)
if (cur.rowcount or 0) == 0:
# Row not present: create it so the dismissal is recorded.
db.execute(
"INSERT INTO name_tokens (project_id, token, status, evidence_json) "
"VALUES (?, ?, ?, ?)",
(project_id, token, STATUS_DISMISSED, json.dumps([reason] if reason else [])),
)
db.commit()
return {"ok": True, "token": token, "status": STATUS_DISMISSED}
def merge_speakers(
db: sqlite3.Connection,
data_dir: Path,
source_primary: str,
target_primary: str,
) -> dict[str, Any]:
"""Merge two speaker profiles (the reverse of assign).
Folds the source's primary name + aliases into the target, re-normalizes all
utterances, and removes the source profile. Idempotent: merging an already
-removed source is a no-op. Returns a summary dict.
"""
if not source_primary or not target_primary:
raise ResolutionError("source_primary and target_primary required")
if source_primary.lower() == target_primary.lower():
raise ResolutionError("cannot merge speaker into itself")
doc, speakers = _load_speaker_doc(data_dir)
source = _find_speaker(speakers, source_primary)
target = _find_speaker(speakers, target_primary)
if target is None:
raise ResolutionError(f"target speaker not found: {target_primary}")
if source is None:
# Already merged (idempotent no-op).
return {"ok": True, "merged": False, "reason": "source already removed"}
source_name = source.get("primary_name", "")
target_name = target.get("primary_name", "")
old_names = [n for n in [source_name, *source.get("aliases", [])] if n]
# Fold aliases into the target (keep the longer alias set).
target_aliases = target.setdefault("aliases", [])
for alias in [source_name, *source.get("aliases", [])]:
if alias and alias.lower() != target_name.lower() and alias not in target_aliases:
target_aliases.append(alias)
# Fill in missing demographics/traits from the source.
src_demos = source.get("demographics", {})
tgt_demos = target.setdefault("demographics", {})
if src_demos.get("type") and src_demos["type"] != "Human":
tgt_demos.setdefault("type", src_demos["type"])
if src_demos.get("gender") and src_demos["gender"] != "Unknown":
tgt_demos.setdefault("gender", src_demos["gender"])
for trait in source.get("vocal_traits", []):
if trait not in target.setdefault("vocal_traits", []):
target["vocal_traits"].append(trait)
target["chapters_appeared"] = sorted(
set(source.get("chapters_appeared", [])) | set(target.get("chapters_appeared", []))
)
speakers = [s for s in speakers if s is not source]
_save_speaker_doc(data_dir, doc, speakers)
changed = _renormalize_utterances(db, old_names, target_name)
return {
"ok": True,
"merged": True,
"source": source_name,
"target": target_name,
"utterances_changed": changed,
}

View File

@@ -140,6 +140,24 @@ CREATE TABLE IF NOT EXISTS review_decisions (
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS name_tokens (
id INTEGER PRIMARY KEY,
project_id INTEGER NOT NULL REFERENCES project(id),
token TEXT NOT NULL,
kind TEXT NOT NULL DEFAULT 'name_tag',
status TEXT NOT NULL DEFAULT 'unresolved',
target_speaker TEXT,
occurrences INTEGER NOT NULL DEFAULT 0,
chapters_json TEXT NOT NULL DEFAULT '[]',
evidence_json TEXT NOT NULL DEFAULT '[]',
suggestion TEXT,
suggestion_score REAL DEFAULT 0.0,
updated_at TEXT NOT NULL DEFAULT (datetime('now')),
UNIQUE(project_id, token)
);
CREATE INDEX IF NOT EXISTS idx_name_tokens_status ON name_tokens(status);
CREATE INDEX IF NOT EXISTS idx_name_tokens_project ON name_tokens(project_id);
CREATE INDEX IF NOT EXISTS idx_utterances_chapter ON utterances(chapter_id);
CREATE INDEX IF NOT EXISTS idx_utterances_character ON utterances(character_id);
CREATE INDEX IF NOT EXISTS idx_generated_audio_utterance ON generated_audio(utterance_id);