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:
193
scripts/scan_name_tokens.py
Normal file
193
scripts/scan_name_tokens.py
Normal 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
236
tests/test_name_tokens.py
Normal 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()
|
||||
184
tests/test_orphan_resolution.py
Normal file
184
tests/test_orphan_resolution.py
Normal 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")
|
||||
503
umbraaudio/analysis/name_tokens.py
Normal file
503
umbraaudio/analysis/name_tokens.py
Normal 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
|
||||
311
umbraaudio/analysis/orphan_resolution.py
Normal file
311
umbraaudio/analysis/orphan_resolution.py
Normal 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,
|
||||
}
|
||||
@@ -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);
|
||||
|
||||
Reference in New Issue
Block a user