Multi-pass attribution + web voice-maintenance banner + qwen demo branding
Attribution (from prior session, now committed): - llm_attributor: mechanical narration evidence, unnamed-role rules, wider context (12->20 lines, 8->40 utterances), uncertain flag, verify pass with 20-item batches and invented-name gate - reattribute_all: --min-id/--max-id, --dry-run, --no-verify, --verify-rounds, qa_accepted skip guard, per-chapter elapsed times - test_multipass_attribution: 19 tests Web banner: - voice_maintenance module: pid-liveness-checked marker file so a crashed run self-heals the banner away - routes/base.html: amber banner on every page while the LLM re-attribution run has Qwen3-TTS offline - reattribute_all: sets banner before switching models, clears in finally (also on failure) - test_voice_maintenance: 8 tests qwen3-tts submodule: UmbraAudio Voice Studio branding in web demo (icon data-uri + favicon logo).
This commit is contained in:
B.D. Murphy
Submodule Qwen3-TTS updated: 022e286b98...012e45298d
@@ -5,6 +5,7 @@ Preserves utterance IDs and mechanical segmentation; only updates speaker_name.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sqlite3
|
||||
import sys
|
||||
import time
|
||||
@@ -12,11 +13,63 @@ from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||||
|
||||
from umbraaudio.analysis.attribution_qa import (
|
||||
check_attribution,
|
||||
verify_conversation_flow,
|
||||
)
|
||||
from umbraaudio.analysis.llm_attributor import LLMAttributor
|
||||
from umbraaudio.config import AppConfig, LLMConfig
|
||||
from umbraaudio.config import load_config
|
||||
from umbraaudio.voice_maintenance import (
|
||||
clear_voice_maintenance,
|
||||
set_voice_maintenance,
|
||||
)
|
||||
from umbraaudio.voices.model_coordinator import ModelCoordinator, PipelineStage
|
||||
|
||||
|
||||
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 _qa_issue_count(utt_list: list[dict], chapter_id: int, alias_map: dict) -> int:
|
||||
qa = check_attribution(utt_list, chapter_id, alias_map)
|
||||
flow = verify_conversation_flow(utt_list, chapter_id, alias_map)
|
||||
return len(qa.issues) + len(flow.issues)
|
||||
|
||||
|
||||
def _report_dry_run(attributed: list[dict], original: list[dict]) -> None:
|
||||
"""Print speaker changes the run would make (no DB writes)."""
|
||||
orig_by_id = {u["id"]: u for u in original}
|
||||
changed = 0
|
||||
shown = 0
|
||||
for u in attributed:
|
||||
o = orig_by_id.get(u["id"])
|
||||
if o is None:
|
||||
continue
|
||||
before = o.get("speaker_name") or "narrator"
|
||||
after = u.get("speaker_name") or "narrator"
|
||||
if (before or "").lower() == (after or "").lower():
|
||||
continue
|
||||
changed += 1
|
||||
if shown < 5:
|
||||
text = (u.get("text") or "")[:70]
|
||||
print(f" utt {u.get('utterance_index')}: {before} -> {after} {text!r}")
|
||||
shown += 1
|
||||
print(f" dry-run: {changed} speaker change(s) (sample above)")
|
||||
|
||||
|
||||
def main():
|
||||
import argparse
|
||||
|
||||
@@ -33,6 +86,18 @@ def main():
|
||||
default=None,
|
||||
help="Comma-separated chapter IDs (e.g. '5,6,7')",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--min-id",
|
||||
type=int,
|
||||
default=None,
|
||||
help="Only chapters with id >= this value (e.g. 11 = 'chapters 11 to the end')",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--max-id",
|
||||
type=int,
|
||||
default=None,
|
||||
help="Only chapters with id <= this value",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--model",
|
||||
type=str,
|
||||
@@ -45,10 +110,24 @@ def main():
|
||||
default=1.0,
|
||||
help="Seconds to sleep between LLM batches",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--no-verify",
|
||||
action="store_true",
|
||||
help="Skip the multi-pass verification round (pass 3)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--verify-rounds",
|
||||
type=int,
|
||||
default=2,
|
||||
help="Max verification rounds for flagged lines (default 2)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--dry-run",
|
||||
action="store_true",
|
||||
help="Attribute in memory and report vs DB, but write NOTHING",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
from umbraaudio.config import load_config
|
||||
|
||||
cfg = load_config(args.project)
|
||||
config = cfg.llm
|
||||
|
||||
@@ -59,92 +138,144 @@ def main():
|
||||
print("Failed to switch models to CHAPTER_ANALYSIS stage")
|
||||
sys.exit(1)
|
||||
|
||||
db_path = cfg.db_path
|
||||
if not db_path.exists():
|
||||
print(f"Database not found: {db_path}")
|
||||
return
|
||||
|
||||
db = sqlite3.connect(str(db_path))
|
||||
db.row_factory = sqlite3.Row
|
||||
|
||||
if args.chapters:
|
||||
chapter_ids = [int(x.strip()) for x in args.chapters.split(",")]
|
||||
placeholders = ",".join("?" * len(chapter_ids))
|
||||
chapters = db.execute(
|
||||
f"SELECT id, title, chapter_index FROM chapters "
|
||||
f"WHERE id IN ({placeholders}) ORDER BY chapter_index",
|
||||
chapter_ids,
|
||||
).fetchall()
|
||||
else:
|
||||
chapters = db.execute(
|
||||
"SELECT id, title, chapter_index FROM chapters ORDER BY chapter_index"
|
||||
).fetchall()
|
||||
|
||||
print(f"Found {len(chapters)} chapters to attribute")
|
||||
print(f"Model: {args.model}")
|
||||
|
||||
attributor = LLMAttributor(config, model=args.model, data_dir=cfg.data_dir)
|
||||
total_dialogue = 0
|
||||
total_narrator_left = 0
|
||||
|
||||
for ch in chapters:
|
||||
ch_id = ch["id"]
|
||||
title = ch["title"] or f"Chapter {ch['chapter_index']}"
|
||||
|
||||
utts = db.execute(
|
||||
"SELECT id, utterance_index, text, type, speaker_name "
|
||||
"FROM utterances WHERE chapter_id=? ORDER BY utterance_index",
|
||||
(ch_id,),
|
||||
).fetchall()
|
||||
if not utts:
|
||||
print(f" {title}: no utterances, skipping")
|
||||
continue
|
||||
|
||||
utt_list = [dict(u) for u in utts]
|
||||
dialogue_before = sum(1 for u in utt_list if u.get("type") == "dialogue")
|
||||
narrator_before = sum(
|
||||
1
|
||||
for u in utt_list
|
||||
if u.get("type") == "dialogue"
|
||||
and (u.get("speaker_name") or "narrator").lower() == "narrator"
|
||||
)
|
||||
|
||||
attributed = attributor.attribute_speakers(utt_list, title, ch_id)
|
||||
time.sleep(0.5)
|
||||
|
||||
for u in attributed:
|
||||
db.execute(
|
||||
"UPDATE utterances SET speaker_name=?, attribution_confidence=?, attribution_reasoning=? WHERE id=?",
|
||||
(
|
||||
u.get("speaker_name", "Narrator"),
|
||||
u.get("attribution_confidence", u.get("confidence", 0.5)),
|
||||
u.get("attribution_reasoning", "") or "",
|
||||
u["id"],
|
||||
),
|
||||
)
|
||||
db.commit()
|
||||
|
||||
narrator_after = sum(
|
||||
1
|
||||
for u in attributed
|
||||
if u.get("type") == "dialogue"
|
||||
and (u.get("speaker_name") or "narrator").lower() == "narrator"
|
||||
)
|
||||
total_dialogue += dialogue_before
|
||||
total_narrator_left += narrator_after
|
||||
print(
|
||||
f" {title} (id={ch_id}): dialogue {dialogue_before}, "
|
||||
f"narrator-left {narrator_before} -> {narrator_after}"
|
||||
)
|
||||
|
||||
db.close()
|
||||
|
||||
coordinator.switch_to(PipelineStage.IDLE)
|
||||
|
||||
print(
|
||||
f"\nTotal: {total_dialogue} dialogue utterances, "
|
||||
f"{total_narrator_left} still attributed to narrator"
|
||||
set_voice_maintenance(
|
||||
"LLM speaker re-attribution is running — Qwen3-TTS is offline "
|
||||
"for the analysis. Voice re-generation is paused until the run finishes."
|
||||
)
|
||||
|
||||
try:
|
||||
db_path = cfg.db_path
|
||||
if not db_path.exists():
|
||||
print(f"Database not found: {db_path}")
|
||||
return
|
||||
|
||||
db = sqlite3.connect(str(db_path))
|
||||
db.row_factory = sqlite3.Row
|
||||
|
||||
alias_map = _build_alias_map(cfg.data_dir / "speakers.json")
|
||||
|
||||
if args.chapters:
|
||||
chapter_ids = [int(x.strip()) for x in args.chapters.split(",")]
|
||||
placeholders = ",".join("?" * len(chapter_ids))
|
||||
chapters = db.execute(
|
||||
f"SELECT id, title, chapter_index FROM chapters "
|
||||
f"WHERE id IN ({placeholders}) ORDER BY chapter_index",
|
||||
chapter_ids,
|
||||
).fetchall()
|
||||
else:
|
||||
chapters = db.execute(
|
||||
"SELECT id, title, chapter_index FROM chapters ORDER BY chapter_index"
|
||||
).fetchall()
|
||||
|
||||
if args.min_id is not None or args.max_id is not None:
|
||||
chapters = [
|
||||
ch
|
||||
for ch in chapters
|
||||
if (args.min_id is None or ch["id"] >= args.min_id)
|
||||
and (args.max_id is None or ch["id"] <= args.max_id)
|
||||
]
|
||||
|
||||
print(f"Found {len(chapters)} chapters to attribute")
|
||||
print(f"Model: {args.model}")
|
||||
print(f"Verify pass: {'ON' if not args.no_verify else 'OFF'} "
|
||||
f"(up to {args.verify_rounds} round(s))")
|
||||
if args.dry_run:
|
||||
print("DRY RUN: no changes will be written")
|
||||
|
||||
attributor = LLMAttributor(config, model=args.model, data_dir=cfg.data_dir)
|
||||
total_dialogue = 0
|
||||
total_narrator_left = 0
|
||||
total_qa_before = 0
|
||||
total_qa_after = 0
|
||||
|
||||
for ch in chapters:
|
||||
ch_id = ch["id"]
|
||||
title = ch["title"] or f"Chapter {ch['chapter_index']}"
|
||||
ch_start = time.time()
|
||||
|
||||
utts = db.execute(
|
||||
"SELECT id, utterance_index, text, type, speaker_name, "
|
||||
"qa_accepted, qa_accepted_sig "
|
||||
"FROM utterances WHERE chapter_id=? ORDER BY utterance_index",
|
||||
(ch_id,),
|
||||
).fetchall()
|
||||
if not utts:
|
||||
print(f" {title}: no utterances, skipping")
|
||||
continue
|
||||
|
||||
utt_list = [dict(u) for u in utts]
|
||||
dialogue_before = sum(1 for u in utt_list if u.get("type") == "dialogue")
|
||||
narrator_before = sum(
|
||||
1
|
||||
for u in utt_list
|
||||
if u.get("type") == "dialogue"
|
||||
and (u.get("speaker_name") or "narrator").lower() == "narrator"
|
||||
)
|
||||
qa_before = _qa_issue_count(utt_list, ch_id, alias_map)
|
||||
|
||||
attributed = attributor.attribute_speakers(
|
||||
utt_list,
|
||||
title,
|
||||
ch_id,
|
||||
verify=not args.no_verify,
|
||||
verify_rounds=args.verify_rounds,
|
||||
)
|
||||
time.sleep(0.5)
|
||||
ch_elapsed = time.time() - ch_start
|
||||
|
||||
accepted_skipped = 0
|
||||
for u in attributed:
|
||||
if u.get("qa_accepted"):
|
||||
accepted_skipped += 1
|
||||
continue
|
||||
if args.dry_run:
|
||||
continue
|
||||
db.execute(
|
||||
"UPDATE utterances SET speaker_name=?, attribution_confidence=?, "
|
||||
"attribution_reasoning=? WHERE id=?",
|
||||
(
|
||||
u.get("speaker_name", "Narrator"),
|
||||
u.get("attribution_confidence", u.get("confidence", 0.5)),
|
||||
u.get("attribution_reasoning", "") or "",
|
||||
u["id"],
|
||||
),
|
||||
)
|
||||
if not args.dry_run:
|
||||
db.commit()
|
||||
|
||||
if args.dry_run:
|
||||
_report_dry_run(attributed, utt_list)
|
||||
|
||||
narrator_after = sum(
|
||||
1
|
||||
for u in attributed
|
||||
if u.get("type") == "dialogue"
|
||||
and (u.get("speaker_name") or "narrator").lower() == "narrator"
|
||||
)
|
||||
qa_after = _qa_issue_count(attributed, ch_id, alias_map)
|
||||
total_dialogue += dialogue_before
|
||||
total_narrator_left += narrator_after
|
||||
total_qa_before += qa_before
|
||||
total_qa_after += qa_after
|
||||
print(
|
||||
f" {title} (id={ch_id}): dialogue {dialogue_before}, "
|
||||
f"narrator-left {narrator_before} -> {narrator_after}, "
|
||||
f"QA issues {qa_before} -> {qa_after}"
|
||||
f" ({ch_elapsed:.0f}s)"
|
||||
+ (f", accepted-skipped {accepted_skipped}" if accepted_skipped else "")
|
||||
)
|
||||
|
||||
db.close()
|
||||
|
||||
print(
|
||||
f"\nTotal: {total_dialogue} dialogue utterances, "
|
||||
f"{total_narrator_left} still attributed to narrator, "
|
||||
f"QA issues {total_qa_before} -> {total_qa_after}"
|
||||
)
|
||||
finally:
|
||||
clear_voice_maintenance()
|
||||
coordinator.switch_to(PipelineStage.IDLE)
|
||||
|
||||
print("Done!")
|
||||
|
||||
|
||||
|
||||
329
tests/test_multipass_attribution.py
Normal file
329
tests/test_multipass_attribution.py
Normal file
@@ -0,0 +1,329 @@
|
||||
"""Tests for the multi-pass speaker-attribution pipeline.
|
||||
|
||||
Covers the mechanical narration-evidence builder, speaker normalization for
|
||||
parenthetical qualifiers, target collection for the verification pass, and the
|
||||
deterministic-fix + LLM verification loop itself (with a stubbed LLM).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from umbraaudio.analysis.llm_attributor import LLMAttributor
|
||||
|
||||
|
||||
def _attributor_with_speakers(speakers: list[dict]) -> LLMAttributor:
|
||||
attr = LLMAttributor.__new__(LLMAttributor)
|
||||
attr._load_speakers = lambda: speakers # type: ignore[method-assign]
|
||||
return attr
|
||||
|
||||
|
||||
_SPEAKERS = [
|
||||
{
|
||||
"primary_name": "Kainda Walters Foster Murphy",
|
||||
"aliases": ["Princess Kainda", "Princess", "Kainda"],
|
||||
},
|
||||
{
|
||||
"primary_name": "Narrator",
|
||||
"aliases": ["narrator", "the narrator"],
|
||||
},
|
||||
{
|
||||
"primary_name": "Chief of the Guard",
|
||||
"aliases": ["Lead Guard"],
|
||||
},
|
||||
{
|
||||
"primary_name": "Nanny",
|
||||
"aliases": ["Niñera"],
|
||||
},
|
||||
{
|
||||
"primary_name": "Deal Hunter",
|
||||
"aliases": ["The AI", "Ship AI"],
|
||||
},
|
||||
{
|
||||
"primary_name": "Xylia",
|
||||
"aliases": [],
|
||||
},
|
||||
]
|
||||
|
||||
_NARRATION = "narration"
|
||||
_DIALOGUE = "dialogue"
|
||||
|
||||
|
||||
def _utt(idx: int, text: str, utt_type: str = _NARRATION, speaker: str | None = None) -> dict:
|
||||
u: dict = {
|
||||
"utterance_index": idx,
|
||||
"text": text,
|
||||
"type": utt_type,
|
||||
}
|
||||
if speaker is not None:
|
||||
u["speaker_name"] = speaker
|
||||
return u
|
||||
|
||||
|
||||
class TestMechanicalEvidence:
|
||||
def test_name_tag_from_preceding_narration(self) -> None:
|
||||
attr = _attributor_with_speakers(_SPEAKERS)
|
||||
alias_map = attr._build_alias_map()
|
||||
scene = [
|
||||
_utt(1, "Kainda asked quietly."),
|
||||
_utt(2, "Have you seen the vault?", _DIALOGUE),
|
||||
]
|
||||
evidence = attr._build_mechanical_evidence(scene, [scene[1]], alias_map)
|
||||
entry = evidence[2]
|
||||
assert entry["tag"] == "Kainda Walters Foster Murphy"
|
||||
assert entry["tag_kind"] == "name"
|
||||
assert "smiled" in entry["narr_before"] or "asked" in entry["narr_before"]
|
||||
|
||||
def test_role_tag(self) -> None:
|
||||
attr = _attributor_with_speakers(_SPEAKERS)
|
||||
alias_map = attr._build_alias_map()
|
||||
scene = [
|
||||
_utt(1, "The lead guard replied."),
|
||||
_utt(2, "Nobody enters tonight.", _DIALOGUE),
|
||||
]
|
||||
evidence = attr._build_mechanical_evidence(scene, [scene[1]], alias_map)
|
||||
entry = evidence[2]
|
||||
assert entry["tag_kind"] == "role"
|
||||
assert entry["tag"].lower() == "the lead guard"
|
||||
|
||||
def test_pronoun_tag(self) -> None:
|
||||
attr = _attributor_with_speakers(_SPEAKERS)
|
||||
alias_map = attr._build_alias_map()
|
||||
scene = [
|
||||
_utt(1, "She asks, looking nervous."),
|
||||
_utt(2, "Where is Father?", _DIALOGUE),
|
||||
]
|
||||
evidence = attr._build_mechanical_evidence(scene, [scene[1]], alias_map)
|
||||
entry = evidence[2]
|
||||
assert entry["tag_kind"] == "pronoun"
|
||||
assert entry["tag"] == "she"
|
||||
|
||||
def test_addressee_detected(self) -> None:
|
||||
attr = _attributor_with_speakers(_SPEAKERS)
|
||||
alias_map = attr._build_alias_map()
|
||||
scene = [
|
||||
_utt(1, "Nanny stepped closer."),
|
||||
_utt(2, "Princess, you must run.", _DIALOGUE),
|
||||
]
|
||||
evidence = attr._build_mechanical_evidence(scene, [scene[1]], alias_map)
|
||||
entry = evidence[2]
|
||||
assert entry["addressees"] == ["Kainda Walters Foster Murphy"]
|
||||
|
||||
def test_no_evidence_when_prev_is_dialogue(self) -> None:
|
||||
attr = _attributor_with_speakers(_SPEAKERS)
|
||||
alias_map = attr._build_alias_map()
|
||||
scene = [
|
||||
_utt(1, "First line.", _DIALOGUE, "Kainda Walters Foster Murphy"),
|
||||
_utt(2, "Second line.", _DIALOGUE),
|
||||
]
|
||||
evidence = attr._build_mechanical_evidence(scene, [scene[1]], alias_map)
|
||||
entry = evidence[2]
|
||||
assert entry["tag"] is None
|
||||
assert entry["narr_before"] == ""
|
||||
|
||||
|
||||
class TestNormalizeSpeakerParenthetical:
|
||||
def test_strips_parenthetical_qualifier_to_known_name(self) -> None:
|
||||
attr = _attributor_with_speakers(_SPEAKERS)
|
||||
alias_map = attr._build_alias_map()
|
||||
assert (
|
||||
attr._normalize_speaker("Xylia (Voice (Dragon))", alias_map)
|
||||
== "Xylia"
|
||||
)
|
||||
|
||||
def test_strips_scene_qualifier_to_known_name(self) -> None:
|
||||
attr = _attributor_with_speakers(_SPEAKERS)
|
||||
alias_map = attr._build_alias_map()
|
||||
assert (
|
||||
attr._normalize_speaker("Nanny (Scene 14)", alias_map) == "Nanny"
|
||||
)
|
||||
|
||||
def test_keeps_unknown_name_with_parens_unchanged(self) -> None:
|
||||
attr = _attributor_with_speakers(_SPEAKERS)
|
||||
alias_map = attr._build_alias_map()
|
||||
assert attr._normalize_speaker("Somebody (Guard)", alias_map) == "Somebody (Guard)"
|
||||
|
||||
def test_strips_quotes(self) -> None:
|
||||
attr = _attributor_with_speakers(_SPEAKERS)
|
||||
alias_map = attr._build_alias_map()
|
||||
assert attr._normalize_speaker('"Kainda"', alias_map) == "Kainda Walters Foster Murphy"
|
||||
|
||||
|
||||
class TestCollectVerifyTargets:
|
||||
def test_flags_narrator_left_dialogue(self) -> None:
|
||||
attr = _attributor_with_speakers(_SPEAKERS)
|
||||
alias_map = attr._build_alias_map()
|
||||
utterances = [
|
||||
_utt(1, "Nanny speaks."),
|
||||
_utt(2, "Line one.", _DIALOGUE, "narrator"),
|
||||
]
|
||||
targets = attr._collect_verify_targets(utterances, 6, alias_map)
|
||||
assert 2 in targets
|
||||
|
||||
def test_flags_tag_mismatch(self) -> None:
|
||||
attr = _attributor_with_speakers(_SPEAKERS)
|
||||
alias_map = attr._build_alias_map()
|
||||
utterances = [
|
||||
_utt(1, "Nanny asked"),
|
||||
_utt(2, "Line one.", _DIALOGUE, "Kainda Walters Foster Murphy"),
|
||||
]
|
||||
targets = attr._collect_verify_targets(utterances, 6, alias_map)
|
||||
assert 2 in targets
|
||||
assert "speech_tag_mismatch" in targets[2]
|
||||
|
||||
def test_flags_low_confidence(self) -> None:
|
||||
attr = _attributor_with_speakers(_SPEAKERS)
|
||||
alias_map = attr._build_alias_map()
|
||||
utterances = [
|
||||
_utt(1, "Some narration."),
|
||||
_utt(2, "Confident line.", _DIALOGUE, "Nanny"),
|
||||
_utt(3, "Low confidence.", _DIALOGUE, "Nanny"),
|
||||
]
|
||||
utterances[1]["attribution_confidence"] = 0.75
|
||||
utterances[2]["attribution_confidence"] = 0.55
|
||||
targets = attr._collect_verify_targets(utterances, 6, alias_map)
|
||||
assert 3 in targets # 0.55 < 0.6 → flagged
|
||||
assert 2 not in targets # 0.75 ≥ 0.6 → clean
|
||||
|
||||
def test_flags_uncertain_flag(self) -> None:
|
||||
attr = _attributor_with_speakers(_SPEAKERS)
|
||||
alias_map = attr._build_alias_map()
|
||||
utterances = [
|
||||
_utt(1, "Some narration."),
|
||||
_utt(2, "Line one.", _DIALOGUE, "Nanny"),
|
||||
]
|
||||
utterances[1]["attribution_confidence"] = 0.8
|
||||
utterances[1]["attribution_uncertain"] = True
|
||||
targets = attr._collect_verify_targets(utterances, 6, alias_map)
|
||||
assert 2 in targets
|
||||
assert "uncertain" in targets[2]
|
||||
|
||||
|
||||
class TestVerifyAttributions:
|
||||
def test_applies_llm_correction(self) -> None:
|
||||
attr = _attributor_with_speakers(_SPEAKERS)
|
||||
alias_map = attr._build_alias_map()
|
||||
|
||||
def fake_llm(prompt: str) -> list[dict]:
|
||||
return [
|
||||
{
|
||||
"index": 4,
|
||||
"speaker": "Nanny",
|
||||
"confidence": 0.95,
|
||||
"reasoning": "Nanny is the nurse",
|
||||
}
|
||||
]
|
||||
|
||||
attr._call_llm = fake_llm # type: ignore[method-assign]
|
||||
|
||||
utterances = [
|
||||
_utt(1, "Nanny entered the room."),
|
||||
_utt(2, "Who are you?", _DIALOGUE, "Kainda Walters Foster Murphy"),
|
||||
_utt(3, "The nurse."),
|
||||
_utt(4, "I am here to help you.", _DIALOGUE, "Kainda Walters Foster Murphy"),
|
||||
]
|
||||
profiles = "- Kainda Walters Foster Murphy\n- Nanny"
|
||||
result = attr._verify_attributions(
|
||||
utterances, 6, "Chapter Six", profiles, alias_map, max_rounds=1
|
||||
)
|
||||
assert result[3]["speaker_name"] == "Nanny"
|
||||
|
||||
def test_keeps_correct_attribution_unchanged(self) -> None:
|
||||
attr = _attributor_with_speakers(_SPEAKERS)
|
||||
alias_map = attr._build_alias_map()
|
||||
|
||||
def fake_llm(prompt: str) -> list[dict]:
|
||||
return [{"index": 2, "speaker": "Nanny", "confidence": 0.9, "reasoning": ""}]
|
||||
|
||||
attr._call_llm = fake_llm # type: ignore[method-assign]
|
||||
|
||||
utterances = [
|
||||
_utt(1, "Nanny spoke."),
|
||||
_utt(2, "Stay still.", _DIALOGUE, "Nanny"),
|
||||
]
|
||||
profiles = "- Nanny"
|
||||
result = attr._verify_attributions(
|
||||
utterances, 6, "Chapter Six", profiles, alias_map, max_rounds=1
|
||||
)
|
||||
assert result[1]["speaker_name"] == "Nanny"
|
||||
|
||||
def test_rejects_narrator_correction(self) -> None:
|
||||
attr = _attributor_with_speakers(_SPEAKERS)
|
||||
alias_map = attr._build_alias_map()
|
||||
|
||||
def fake_llm(prompt: str) -> list[dict]:
|
||||
return [{"index": 2, "speaker": "Narrator", "confidence": 0.9, "reasoning": ""}]
|
||||
|
||||
attr._call_llm = fake_llm # type: ignore[method-assign]
|
||||
|
||||
utterances = [
|
||||
_utt(1, "Nanny spoke."),
|
||||
_utt(2, "Stay still.", _DIALOGUE, "Nanny"),
|
||||
]
|
||||
profiles = "- Nanny"
|
||||
result = attr._verify_attributions(
|
||||
utterances, 6, "Chapter Six", profiles, alias_map, max_rounds=1
|
||||
)
|
||||
assert result[1]["speaker_name"] == "Nanny"
|
||||
|
||||
def test_deterministic_fix_applied_before_llm(self) -> None:
|
||||
attr = _attributor_with_speakers(_SPEAKERS)
|
||||
alias_map = attr._build_alias_map()
|
||||
|
||||
def fake_llm(prompt: str) -> list[dict]:
|
||||
return []
|
||||
|
||||
attr._call_llm = fake_llm # type: ignore[method-assign]
|
||||
|
||||
# "Nanny asked" is directly before the line → fix_high_confidence corrects.
|
||||
utterances = [
|
||||
_utt(1, "Nanny asked"),
|
||||
_utt(2, "Who is there?", _DIALOGUE, "Kainda Walters Foster Murphy"),
|
||||
]
|
||||
profiles = "- Nanny\n- Kainda Walters Foster Murphy"
|
||||
result = attr._verify_attributions(
|
||||
utterances, 6, "Chapter Six", profiles, alias_map, max_rounds=1
|
||||
)
|
||||
assert result[1]["speaker_name"] == "Nanny"
|
||||
|
||||
def test_rejects_invented_name(self) -> None:
|
||||
attr = _attributor_with_speakers(_SPEAKERS)
|
||||
alias_map = attr._build_alias_map()
|
||||
|
||||
def fake_llm(prompt: str) -> list[dict]:
|
||||
# Digits make it fail both the alias map and the role-name heuristic.
|
||||
return [{"index": 2, "speaker": "Guard 5", "confidence": 0.9, "reasoning": ""}]
|
||||
|
||||
attr._call_llm = fake_llm # type: ignore[method-assign]
|
||||
|
||||
utterances = [
|
||||
_utt(1, "Nanny spoke."),
|
||||
_utt(2, "Stay still.", _DIALOGUE, "Nanny"),
|
||||
]
|
||||
utterances[1]["attribution_confidence"] = 0.5
|
||||
profiles = "- Nanny"
|
||||
result = attr._verify_attributions(
|
||||
utterances, 6, "Chapter Six", profiles, alias_map, max_rounds=1
|
||||
)
|
||||
assert result[1]["speaker_name"] == "Nanny"
|
||||
|
||||
def test_verify_targets_batched_at_twenty(self) -> None:
|
||||
attr = _attributor_with_speakers(_SPEAKERS)
|
||||
alias_map = attr._build_alias_map()
|
||||
calls: list[str] = []
|
||||
|
||||
def fake_llm(prompt: str) -> list[dict]:
|
||||
calls.append(prompt)
|
||||
return []
|
||||
|
||||
attr._call_llm = fake_llm # type: ignore[method-assign]
|
||||
|
||||
# 25 narrator-left dialogue lines → 25 targets → 20 + 5 in two calls.
|
||||
utterances = [_utt(1, "Some narration.")]
|
||||
for i in range(2, 27):
|
||||
utterances.append(_utt(i, f"Line {i}.", _DIALOGUE, "narrator"))
|
||||
profiles = "- Nanny"
|
||||
attr._verify_attributions(
|
||||
utterances, 6, "Chapter Six", profiles, alias_map, max_rounds=1
|
||||
)
|
||||
assert len(calls) == 2
|
||||
assert '"index": 2' in calls[0]
|
||||
assert '"index": 22' not in calls[0]
|
||||
assert '"index": 22' in calls[1]
|
||||
75
tests/test_voice_maintenance.py
Normal file
75
tests/test_voice_maintenance.py
Normal file
@@ -0,0 +1,75 @@
|
||||
"""Tests for the web-UI voice-maintenance banner marker.
|
||||
|
||||
The marker is a small JSON file whose ``pid`` field gates the banner: a
|
||||
run that died without clearing the file must not leave the banner up
|
||||
forever, so ``active_voice_maintenance`` checks the owner is still alive.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import pytest
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
from umbraaudio import voice_maintenance as vm
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def marker(tmp_path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
||||
path = tmp_path / "voice_maintenance.json"
|
||||
monkeypatch.setattr(vm, "MAINTENANCE_FILE", path)
|
||||
return path
|
||||
|
||||
|
||||
def test_set_writes_marker_with_live_pid(marker: Path) -> None:
|
||||
vm.set_voice_maintenance("analysis in progress")
|
||||
assert marker.exists()
|
||||
data = json.loads(marker.read_text())
|
||||
assert data["pid"] == os.getpid()
|
||||
assert "analysis" in data["reason"]
|
||||
|
||||
|
||||
def test_active_returns_marker_while_owner_alive(marker: Path) -> None:
|
||||
vm.set_voice_maintenance("analysis in progress")
|
||||
active = vm.active_voice_maintenance()
|
||||
assert active is not None
|
||||
assert active["pid"] == os.getpid()
|
||||
|
||||
|
||||
def test_active_none_when_no_marker(marker: Path) -> None:
|
||||
assert vm.active_voice_maintenance() is None
|
||||
|
||||
|
||||
def test_clear_removes_marker(marker: Path) -> None:
|
||||
vm.set_voice_maintenance("analysis in progress")
|
||||
vm.clear_voice_maintenance()
|
||||
assert not marker.exists()
|
||||
assert vm.active_voice_maintenance() is None
|
||||
|
||||
|
||||
def test_clear_missing_marker_is_idempotent(marker: Path) -> None:
|
||||
vm.clear_voice_maintenance()
|
||||
vm.clear_voice_maintenance()
|
||||
|
||||
|
||||
def test_active_none_when_owner_dead(marker: Path) -> None:
|
||||
marker.write_text(
|
||||
json.dumps({"pid": 2**24, "started_at": 1, "reason": "dead owner"})
|
||||
)
|
||||
assert vm.active_voice_maintenance() is None
|
||||
|
||||
|
||||
def test_active_none_on_corrupt_marker(marker: Path) -> None:
|
||||
marker.write_text("{not json")
|
||||
assert vm.active_voice_maintenance() is None
|
||||
|
||||
|
||||
def test_active_none_on_bad_pid_type(marker: Path) -> None:
|
||||
marker.write_text(
|
||||
json.dumps({"pid": "not-an-int", "started_at": 1, "reason": "bad"})
|
||||
)
|
||||
assert vm.active_voice_maintenance() is None
|
||||
@@ -13,6 +13,14 @@ from openai import OpenAI
|
||||
from umbraaudio.config import LLMConfig
|
||||
from umbraaudio.analysis.context_analyzer import ContextAnalyzer, AttributionConfidence
|
||||
from umbraaudio.analysis.tone_analyzer import ToneAnalyzer, SceneEmotionalProfile
|
||||
from umbraaudio.analysis.attribution_qa import (
|
||||
SPEECH_VERBS as _QA_SPEECH_VERBS,
|
||||
_detect_addressee as _qa_detect_addressee,
|
||||
_match_speech_tag as _qa_match_speech_tag,
|
||||
check_attribution as _qa_check_attribution,
|
||||
fix_high_confidence as _qa_fix_high_confidence,
|
||||
verify_conversation_flow as _qa_verify_flow,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -23,6 +31,36 @@ _PRONOUNS = frozenset(
|
||||
}
|
||||
)
|
||||
|
||||
# Matches a TAG narration line whose subject is a pronoun: "He said,", "She asks."
|
||||
_PRONOUN_TAG = re.compile(
|
||||
r"^\s*(H[ei]|She|They)\s+"
|
||||
r"(says|said|asks|asked|replies|replied|whispers|whispered|shouts|shouted|"
|
||||
r"mutters|muttered|calls|called|answers|answered)\b",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
# Words allowed inside a Title Case ROLE name (rule 3g / 15).
|
||||
_ROLE_WORDS = frozenset(
|
||||
{"the", "a", "an", "of", "in", "to", "for", "and", "or", "on", "at", "de"}
|
||||
)
|
||||
|
||||
|
||||
def _looks_like_role(name: str) -> bool:
|
||||
"""True when a name plausibly is a Title Case ROLE name, not a personal name."""
|
||||
if not name or len(name) > 40:
|
||||
return False
|
||||
if re.search(r"[()\d]", name):
|
||||
return False
|
||||
words = [w for w in re.split(r"[\s-]+", name.strip()) if w]
|
||||
if not 1 <= len(words) <= 6:
|
||||
return False
|
||||
for w in words:
|
||||
if w.lower() in _ROLE_WORDS:
|
||||
continue
|
||||
if not w[:1].isupper():
|
||||
return False
|
||||
return True
|
||||
|
||||
# Enhanced prompt with context awareness and confidence scoring
|
||||
PROMPT_TEMPLATE = """You are a dialogue attribution expert for audiobook production.
|
||||
|
||||
@@ -49,6 +87,18 @@ You are attributing dialogue for: {scene_label}
|
||||
## Recent Conversation Flow (context before this batch)
|
||||
{position_context}
|
||||
|
||||
## Mechanical Narration Evidence
|
||||
For each utterance below, these machine-extracted fields came straight from the
|
||||
manuscript (the story's own words, not guesses):
|
||||
- "narr_before": the narration text immediately before this dialogue line, verbatim
|
||||
- "tag": the speaker named or tagged by that narration ("Kainda", "the lead guard",
|
||||
"She"), or null when the narration tags no speaker
|
||||
- "tag_kind": "name" (exact character name), "role" (role/title), "pronoun"
|
||||
(He/She/They), or null
|
||||
- "addressees": names or titles the dialogue itself addresses at its start, if any
|
||||
|
||||
These fields are the STRONGEST signal available. Prefer them over dialogue content.
|
||||
|
||||
## Utterances to Attribute
|
||||
{utterances_json}
|
||||
|
||||
@@ -76,6 +126,17 @@ You are attributing dialogue for: {scene_label}
|
||||
11. CRITICAL — every returned speaker must be the ACTUAL character speaking. Assigning the wrong speaker ruins the audiobook. When genuinely uncertain, prefer an unnamed role speaker over a random named character.
|
||||
12. If the Character Profiles include an AI/computer character (e.g. a ship AI), attribute dialogue to it only when the text indicates an AI or computer is speaking (e.g., "ear device", "ship AI", "computer voice"). Do NOT default undetermined utterances to an AI character if another character is more likely.
|
||||
13. KEYWORD NON-MATCHING: When a character's name appears in the TEXT of a dialogue line (e.g., "The ship is the Kestrel", "I know Captain Smith", "We are near the Kestrel"), it does NOT mean that character is speaking. The speaker is determined by the preceding narration tag, NOT by character names mentioned inside the dialogue. A character can be TALKED ABOUT without being the speaker. Only attribute to a character when a narration tag or dialogue tag explicitly identifies them as the speaker.
|
||||
14. EVIDENCE PRIORITY (order of preference):
|
||||
(a) a "name" tag in narr_before → that character speaks
|
||||
(b) a "role" tag → the role speaker (rule 3g)
|
||||
(c) a "pronoun" tag → resolve by rule 3f
|
||||
(d) an addressee listed in "addressees" → the addressee is NOT the speaker (rules 3j/3k)
|
||||
(e) scene flow and dialogue content as a last resort
|
||||
Never override an explicit tag with a guess from dialogue content (rule 13).
|
||||
15. ROLE-NAME CONSISTENCY: once you pick a speaker for a recurring role ("the guard",
|
||||
"a pirate"), keep using the SAME speaker name for that role throughout the chapter.
|
||||
Do NOT emit scene-qualified variants like "Guard (Scene 2)" or invent a new role
|
||||
name per line. Reuse the existing role speaker from the profiles.
|
||||
|
||||
## Scene Attribution Rules (CRITICAL)
|
||||
- Chapters can contain multiple scenes separated by scene breaks (<<<>>>)
|
||||
@@ -93,6 +154,59 @@ Return ONLY a JSON array of objects, one per utterance, each with:
|
||||
- "speaker": the correct speaker name (use primary_name from profiles, MUST be a character name, NOT "Narrator")
|
||||
- "confidence": your confidence in this attribution (0.0 - 1.0)
|
||||
- "reasoning": brief explanation of key evidence (max 50 chars)
|
||||
- "uncertain": true ONLY when no narration evidence exists and you had to guess
|
||||
from content; this flags the line for a later verification pass
|
||||
|
||||
Return ONLY the JSON array, no other text."""
|
||||
|
||||
|
||||
# Second-pass prompt: verify and correct attributions a first pass already made.
|
||||
# The LLM is shown only the lines a mechanical QA pass flagged (or low-confidence
|
||||
# lines), each with its mechanical narration evidence, and asked to confirm or fix.
|
||||
VERIFY_TEMPLATE = """You are a dialogue attribution verifier for audiobook production.
|
||||
|
||||
A first pass already attributed speakers to the lines below, but a mechanical check
|
||||
flagged some as possibly wrong. For EACH line decide whether the current attribution
|
||||
is correct; if not, give the correct speaker.
|
||||
|
||||
## Character Profiles
|
||||
{character_profiles}
|
||||
|
||||
## Chapter Title
|
||||
{chapter_title}
|
||||
|
||||
## Lines to Verify
|
||||
{items_json}
|
||||
|
||||
## Rules
|
||||
1. The narration IMMEDIATELY BEFORE a line (the "narr_before" field,
|
||||
VERBATIM from the manuscript) is the STRONGEST signal. A "name" tag
|
||||
there decides the speaker.
|
||||
2. A "role" tag ("the lead guard", "a pirate") → use the matching role
|
||||
speaker's EXACT primary_name from the Character Profiles, else a
|
||||
descriptive Title Case role name. NEVER switch to an unrelated named
|
||||
character.
|
||||
3. A "pronoun" tag ("He says", "She says") → resolve to the most recently
|
||||
mentioned character of that gender in the recent context, within the
|
||||
SAME scene.
|
||||
4. The addressee of a line is NOT its speaker ("Yes, Captain" is spoken TO
|
||||
the Captain). A character never addresses themself by name.
|
||||
5. Repeated lines ("The voice calls again, ...") keep the first
|
||||
occurrence's speaker.
|
||||
6. NEVER use "Narrator" for dialogue when any character evidence exists.
|
||||
7. Use EXACT primary_names from the profiles, never invent names. Keep
|
||||
role names consistent (no "Guard (Scene 2)" variants).
|
||||
8. The "issue" field explains why the line was flagged — treat it as a
|
||||
hint, not a verdict. Confirm the current speaker when it is right.
|
||||
9. Keep the EXACT "index" values from the input. Return one object per
|
||||
line.
|
||||
|
||||
## Output Format
|
||||
Return ONLY a JSON array of objects, one per input line, in the same order, each with:
|
||||
- "index": the line's exact "index" value (number)
|
||||
- "speaker": the CORRECT speaker primary_name (the same as "current" if it is already right)
|
||||
- "confidence": your confidence in this attribution (0.0 - 1.0)
|
||||
- "reasoning": brief explanation (max 60 chars)
|
||||
|
||||
Return ONLY the JSON array, no other text."""
|
||||
|
||||
@@ -173,14 +287,27 @@ class LLMAttributor:
|
||||
def _normalize_speaker(self, name: str, alias_map: dict[str, str]) -> str:
|
||||
if not name:
|
||||
return name
|
||||
lower = name.strip().lower()
|
||||
raw = name.strip().strip("\"'“”")
|
||||
lower = raw.lower()
|
||||
if lower in ("narrator", "the narrator"):
|
||||
return "narrator"
|
||||
if lower in _PRONOUNS:
|
||||
return "narrator"
|
||||
if lower in alias_map:
|
||||
return alias_map[lower]
|
||||
return name.strip()
|
||||
# Strip a trailing parenthetical qualifier ("Xylia (Voice (Dragon))" → "Xylia")
|
||||
# and re-check the base name. "Guard (Scene 2)" stays as-is when it IS a
|
||||
# primary, but LLM reinventions like "Guard (Scene 3)" collapse to "Guard".
|
||||
match = re.match(r"^(?P<base>.+?)\s*\([^()]*\)\s*$", raw)
|
||||
if match:
|
||||
base = match.group("base").strip()
|
||||
if base and base.lower() in alias_map:
|
||||
return alias_map[base.lower()]
|
||||
if base := re.match(r"^(?P<base>[^(]+)\s*\(.*$", raw):
|
||||
base_name = base.group("base").strip()
|
||||
if base_name.lower() in alias_map:
|
||||
return alias_map[base_name.lower()]
|
||||
return raw
|
||||
|
||||
def attribute_speakers(
|
||||
self,
|
||||
@@ -188,6 +315,8 @@ class LLMAttributor:
|
||||
chapter_title: str,
|
||||
chapter_id: int,
|
||||
chapter_data: dict[str, Any] | None = None,
|
||||
verify: bool = True,
|
||||
verify_rounds: int = 2,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""
|
||||
Multi-pass speaker attribution with context analysis.
|
||||
@@ -197,6 +326,8 @@ class LLMAttributor:
|
||||
chapter_title: Title of the chapter
|
||||
chapter_id: Chapter ID
|
||||
chapter_data: Optional full chapter data with source_text for context analysis
|
||||
verify: Run the deterministic-fix + LLM verification pass afterwards
|
||||
verify_rounds: Max verification rounds (stop early on no change)
|
||||
"""
|
||||
profiles = self._load_character_profiles()
|
||||
alias_map = self._build_alias_map()
|
||||
@@ -246,6 +377,7 @@ class LLMAttributor:
|
||||
speaker_map: dict[int, str] = {}
|
||||
confidence_map: dict[int, float] = {}
|
||||
reasoning_map: dict[int, str] = {}
|
||||
uncertain_map: dict[int, bool] = {}
|
||||
|
||||
SPEECH_VERB_SUFFIXES = (
|
||||
" says", " says,", " said", " said,", " asks", " asks,",
|
||||
@@ -291,12 +423,24 @@ class LLMAttributor:
|
||||
|
||||
for batch_idx, batch in enumerate(batches):
|
||||
batch_data = []
|
||||
evidence_map = self._build_mechanical_evidence(
|
||||
scene_utterances, batch, alias_map
|
||||
)
|
||||
for u in batch:
|
||||
evidence = evidence_map.get(u["utterance_index"])
|
||||
if evidence is None:
|
||||
evidence = {
|
||||
"narr_before": "",
|
||||
"tag": None,
|
||||
"tag_kind": None,
|
||||
"addressees": [],
|
||||
}
|
||||
batch_data.append(
|
||||
{
|
||||
"index": u["utterance_index"],
|
||||
"text": u["text"][:200],
|
||||
"current_speaker": u.get("speaker_name", "Narrator"),
|
||||
**evidence,
|
||||
}
|
||||
)
|
||||
|
||||
@@ -356,6 +500,7 @@ class LLMAttributor:
|
||||
speaker_map[idx] = speaker
|
||||
confidence_map[idx] = confidence
|
||||
reasoning_map[idx] = reasoning
|
||||
uncertain_map[idx] = bool(a.get("uncertain", False))
|
||||
|
||||
if batch_idx < len(batches) - 1:
|
||||
time.sleep(1)
|
||||
@@ -385,6 +530,24 @@ class LLMAttributor:
|
||||
|
||||
updated = self._apply_speaker_fallback(updated)
|
||||
|
||||
# Store the "uncertain" flag from pass 1 so the verification pass knows
|
||||
# which lines had no mechanical evidence to lean on.
|
||||
for u in updated:
|
||||
idx = u.get("utterance_index")
|
||||
if idx is not None and uncertain_map.get(idx):
|
||||
u["attribution_uncertain"] = True
|
||||
|
||||
# Pass 3: mechanical auto-fix + LLM verification of flagged lines
|
||||
if verify:
|
||||
updated = self._verify_attributions(
|
||||
updated,
|
||||
chapter_id,
|
||||
chapter_title,
|
||||
profiles,
|
||||
alias_map,
|
||||
max_rounds=verify_rounds,
|
||||
)
|
||||
|
||||
# Store confidence scores in context analysis for exception review
|
||||
context_analysis["attribution_confidences"] = [
|
||||
{
|
||||
@@ -405,6 +568,267 @@ class LLMAttributor:
|
||||
|
||||
return updated
|
||||
|
||||
def _build_mechanical_evidence(
|
||||
self,
|
||||
scene_utterances: list[dict[str, Any]],
|
||||
batch: list[dict[str, Any]],
|
||||
alias_map: dict[str, str],
|
||||
) -> dict[int, dict[str, Any]]:
|
||||
"""Extract machine-readable narration evidence for each dialogue line.
|
||||
|
||||
For every dialogue utterance in ``batch``, walk back through the scene
|
||||
looking at narration BEFORE the line (up to 2 narration lines, staying
|
||||
inside the same scene) and pull out:
|
||||
- ``narr_before``: the nearest narration text before the line
|
||||
- ``tag``: speaker named by a speech tag ("Kainda", "the lead guard", "She")
|
||||
- ``tag_kind``: "name", "role", or "pronoun"
|
||||
- ``addressees``: whom the dialogue itself addresses at its start
|
||||
"""
|
||||
evidence: dict[int, dict[str, Any]] = {}
|
||||
dialogue_positions: dict[int, int] = {}
|
||||
for pos, u in enumerate(scene_utterances):
|
||||
if u.get("type") == "dialogue":
|
||||
dialogue_positions[u["utterance_index"]] = pos
|
||||
|
||||
for u in batch:
|
||||
idx = u.get("utterance_index")
|
||||
if idx is None:
|
||||
idx = u.get("index")
|
||||
pos = dialogue_positions.get(idx)
|
||||
if pos is None:
|
||||
continue
|
||||
entry: dict[str, Any] = {
|
||||
"narr_before": "",
|
||||
"tag": None,
|
||||
"tag_kind": None,
|
||||
"addressees": [],
|
||||
}
|
||||
|
||||
# Walk back through narration lines (max 2), staying in the scene.
|
||||
narr_texts: list[str] = []
|
||||
j = pos - 1
|
||||
while j >= 0:
|
||||
prev = scene_utterances[j]
|
||||
prev_text = (prev.get("text") or "").strip()
|
||||
if prev_text == "<<<>>>" or (prev.get("type") or "").lower() != "narration":
|
||||
break
|
||||
narr_texts.append(prev_text)
|
||||
if len(narr_texts) >= 2:
|
||||
break
|
||||
j -= 1
|
||||
|
||||
if narr_texts:
|
||||
entry["narr_before"] = narr_texts[0][:220]
|
||||
|
||||
# Find the first narration line with a speech tag (nearest wins).
|
||||
for narr_text in narr_texts:
|
||||
if _QA_SPEECH_VERBS.search(narr_text):
|
||||
kind, value, _resolved = _qa_match_speech_tag(narr_text, alias_map)
|
||||
if kind == "name" and value:
|
||||
entry["tag"] = value
|
||||
entry["tag_kind"] = "name"
|
||||
break
|
||||
if kind == "role" and value:
|
||||
entry["tag"] = value
|
||||
entry["tag_kind"] = "role"
|
||||
break
|
||||
pm = _PRONOUN_TAG.match(narr_text)
|
||||
if pm:
|
||||
entry["tag"] = pm.group(1).lower()
|
||||
entry["tag_kind"] = "pronoun"
|
||||
break
|
||||
|
||||
addressee = _qa_detect_addressee(u.get("text") or "", alias_map)
|
||||
if addressee:
|
||||
entry["addressees"].append(addressee)
|
||||
|
||||
evidence[idx] = entry
|
||||
|
||||
return evidence
|
||||
|
||||
def _collect_verify_targets(
|
||||
self,
|
||||
utterances: list[dict[str, Any]],
|
||||
chapter_id: int,
|
||||
alias_map: dict[str, str],
|
||||
) -> dict[int, str]:
|
||||
"""Collect utterance indices the verification pass should re-check.
|
||||
|
||||
Returns a map of ``utterance_index -> reason`` for every dialogue line
|
||||
that (a) a mechanical QA pass flagged, (b) is still attributed to the
|
||||
narrator, or (c) has low confidence or the ``uncertain`` flag from
|
||||
pass 1.
|
||||
"""
|
||||
qa = _qa_check_attribution(utterances, chapter_id, alias_map)
|
||||
flow = _qa_verify_flow(utterances, chapter_id, alias_map)
|
||||
|
||||
targets: dict[int, str] = {}
|
||||
for issue in qa.issues + flow.issues:
|
||||
reason = issue.evidence or issue.issue_type
|
||||
targets.setdefault(issue.utterance_index, f"{issue.issue_type}: {reason}")
|
||||
|
||||
for u in utterances:
|
||||
if u.get("type") != "dialogue":
|
||||
continue
|
||||
idx = u.get("utterance_index")
|
||||
if idx is None or idx in targets:
|
||||
continue
|
||||
speaker = (u.get("speaker_name") or "narrator").lower()
|
||||
confidence = float(u.get("attribution_confidence") or 0.5)
|
||||
if speaker == "narrator":
|
||||
targets[idx] = "dialogue still attributed to narrator"
|
||||
elif confidence < 0.6:
|
||||
targets[idx] = f"low confidence ({confidence:.2f})"
|
||||
elif u.get("attribution_uncertain"):
|
||||
targets[idx] = "uncertain: no narration evidence in first pass"
|
||||
|
||||
return targets
|
||||
|
||||
def _verify_attributions(
|
||||
self,
|
||||
utterances: list[dict[str, Any]],
|
||||
chapter_id: int,
|
||||
chapter_title: str,
|
||||
profiles: str,
|
||||
alias_map: dict[str, str],
|
||||
max_rounds: int = 2,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Deterministic auto-fix + LLM verification of flagged attributions.
|
||||
|
||||
Round loop:
|
||||
1. Apply ``fix_high_confidence`` (exact resolved name-tag mismatches).
|
||||
2. Collect targets via mechanical QA (+ narrator-left, low confidence).
|
||||
3. Ask the LLM to confirm or correct each target line (VERIFY_TEMPLATE).
|
||||
4. Apply only corrections that normalize to a known speaker.
|
||||
Stops when a round makes no changes or no targets remain.
|
||||
"""
|
||||
if max_rounds <= 0:
|
||||
return utterances
|
||||
|
||||
current = [dict(u) for u in utterances]
|
||||
|
||||
for round_no in range(1, max_rounds + 1):
|
||||
current = _qa_fix_high_confidence(current, alias_map)
|
||||
targets = self._collect_verify_targets(current, chapter_id, alias_map)
|
||||
if not targets:
|
||||
break
|
||||
|
||||
corrections = self._verify_round_llm(
|
||||
current,
|
||||
targets,
|
||||
chapter_id,
|
||||
chapter_title,
|
||||
profiles,
|
||||
alias_map,
|
||||
)
|
||||
if not corrections:
|
||||
break
|
||||
|
||||
changed = 0
|
||||
by_idx = {int(c["index"]): c for c in corrections if c.get("index") is not None}
|
||||
for u in current:
|
||||
if u.get("type") != "dialogue":
|
||||
continue
|
||||
idx = u.get("utterance_index")
|
||||
if idx is None or idx not in by_idx:
|
||||
continue
|
||||
corr = by_idx[idx]
|
||||
new_speaker = str(corr.get("speaker") or "").strip()
|
||||
if not new_speaker:
|
||||
continue
|
||||
new_speaker = self._normalize_speaker(new_speaker, alias_map)
|
||||
if new_speaker.lower() in ("narrator", "the narrator"):
|
||||
continue
|
||||
# Gate: only accept known speakers (profiles/aliases) or
|
||||
# plausible Title Case role names — rejects invented names
|
||||
# and scene-qualified variants the normalizer kept verbatim.
|
||||
if new_speaker.lower() not in alias_map and not _looks_like_role(
|
||||
new_speaker
|
||||
):
|
||||
logger.debug(
|
||||
"Chapter %d: verify rejected invented name %r for utt %s",
|
||||
chapter_id,
|
||||
new_speaker,
|
||||
idx,
|
||||
)
|
||||
continue
|
||||
old_speaker = u.get("speaker_name") or "narrator"
|
||||
if new_speaker.lower() == old_speaker.lower():
|
||||
continue
|
||||
u["speaker_name"] = new_speaker
|
||||
u["attribution_confidence"] = float(corr.get("confidence") or 0.7)
|
||||
u["attribution_reasoning"] = (
|
||||
"verified: " + str(corr.get("reasoning") or "")[:100]
|
||||
)
|
||||
changed += 1
|
||||
|
||||
if changed == 0:
|
||||
break
|
||||
logger.info(
|
||||
"Chapter %d: verification round %d corrected %d target(s)",
|
||||
chapter_id,
|
||||
round_no,
|
||||
changed,
|
||||
)
|
||||
|
||||
return current
|
||||
|
||||
def _verify_round_llm(
|
||||
self,
|
||||
utterances: list[dict[str, Any]],
|
||||
targets: dict[int, str],
|
||||
chapter_id: int,
|
||||
chapter_title: str,
|
||||
profiles: str,
|
||||
alias_map: dict[str, str],
|
||||
) -> list[dict[str, Any]] | None:
|
||||
"""One LLM call over the target lines using VERIFY_TEMPLATE."""
|
||||
items: list[dict[str, Any]] = []
|
||||
target_positions: dict[int, int] = {}
|
||||
for pos, u in enumerate(utterances):
|
||||
if u.get("utterance_index") in targets:
|
||||
target_positions[u["utterance_index"]] = pos
|
||||
|
||||
for idx in sorted(target_positions):
|
||||
pos = target_positions[idx]
|
||||
u = utterances[pos]
|
||||
items.append(
|
||||
{
|
||||
"index": idx,
|
||||
"current": u.get("speaker_name") or "narrator",
|
||||
"text": (u.get("text") or "")[:200],
|
||||
"issue": targets[idx],
|
||||
"narr_before": "",
|
||||
"tag": None,
|
||||
"tag_kind": None,
|
||||
"addressees": [],
|
||||
}
|
||||
)
|
||||
|
||||
if not items:
|
||||
return None
|
||||
|
||||
# Add narration evidence (same extraction as pass 1) for context.
|
||||
evidence_map = self._build_mechanical_evidence(utterances, items, alias_map)
|
||||
for item in items:
|
||||
item.update(evidence_map.get(item["index"], {}))
|
||||
|
||||
results: list[dict[str, Any]] = []
|
||||
batch_size = 20
|
||||
for start in range(0, len(items), batch_size):
|
||||
chunk = items[start : start + batch_size]
|
||||
prompt = VERIFY_TEMPLATE.format(
|
||||
character_profiles=profiles,
|
||||
chapter_title=chapter_title,
|
||||
items_json=json.dumps(chunk, indent=2),
|
||||
)
|
||||
batch_result = self._call_llm(prompt)
|
||||
if batch_result:
|
||||
results.extend(batch_result)
|
||||
if start + batch_size < len(items):
|
||||
time.sleep(1)
|
||||
return results or None
|
||||
|
||||
def _build_character_presence_context(self, chapter_id: int, scene_idx: int) -> str:
|
||||
"""Build context about which characters are present in this scene."""
|
||||
scenes = self._current_context.get("scene_contexts", {}).get(chapter_id, [])
|
||||
|
||||
58
umbraaudio/voice_maintenance.py
Normal file
58
umbraaudio/voice_maintenance.py
Normal file
@@ -0,0 +1,58 @@
|
||||
"""Web-UI maintenance banner state for long-running offline phases.
|
||||
|
||||
During the LLM speaker re-attribution run the Mac's Qwen3-TTS models must
|
||||
stay unloaded (16 GB RAM), so voice re-generation is temporarily offline.
|
||||
The run script records that state here, the web routes surface it through
|
||||
a template global, and this module's pid-liveness check makes a crashed
|
||||
run self-heal the banner away instead of lingering until a human notices.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
|
||||
from umbraaudio.config import PROJECT_BASE
|
||||
|
||||
MAINTENANCE_FILE = PROJECT_BASE / ".umbra" / "voice_maintenance.json"
|
||||
|
||||
|
||||
def set_voice_maintenance(reason: str) -> None:
|
||||
"""Record that voice re-generation is temporarily unavailable."""
|
||||
payload = {
|
||||
"pid": os.getpid(),
|
||||
"started_at": int(time.time()),
|
||||
"reason": reason,
|
||||
}
|
||||
MAINTENANCE_FILE.parent.mkdir(parents=True, exist_ok=True)
|
||||
tmp = MAINTENANCE_FILE.with_suffix(".json.tmp")
|
||||
tmp.write_text(json.dumps(payload, indent=2))
|
||||
tmp.replace(MAINTENANCE_FILE)
|
||||
|
||||
|
||||
def clear_voice_maintenance() -> None:
|
||||
"""Remove the maintenance marker so the banner disappears."""
|
||||
with contextlib.suppress(FileNotFoundError):
|
||||
MAINTENANCE_FILE.unlink()
|
||||
|
||||
|
||||
def active_voice_maintenance() -> dict | None:
|
||||
"""Return the marker dict while a live process owns it, else None.
|
||||
|
||||
The pid check means a run that died without clearing the file stops
|
||||
showing the banner as soon as its process is gone.
|
||||
"""
|
||||
try:
|
||||
with open(MAINTENANCE_FILE) as f:
|
||||
data = json.load(f)
|
||||
except (FileNotFoundError, json.JSONDecodeError, OSError):
|
||||
return None
|
||||
pid = data.get("pid")
|
||||
if not isinstance(pid, int) or pid <= 0:
|
||||
return None
|
||||
try:
|
||||
os.kill(pid, 0)
|
||||
except (ProcessLookupError, PermissionError):
|
||||
return None
|
||||
return data
|
||||
@@ -53,6 +53,17 @@ def _active_project_for_template(request: Request) -> dict:
|
||||
|
||||
templates.env.globals["active_project"] = _active_project_for_template
|
||||
|
||||
|
||||
def _voice_maintenance_for_template() -> dict | None:
|
||||
"""Expose the active voice-maintenance marker (or None) to templates."""
|
||||
|
||||
from umbraaudio.voice_maintenance import active_voice_maintenance
|
||||
|
||||
return active_voice_maintenance()
|
||||
|
||||
|
||||
templates.env.globals["voice_maintenance"] = _voice_maintenance_for_template
|
||||
|
||||
from umbraaudio.generation.audio_effects import (
|
||||
apply_effects_to_file_by_name,
|
||||
load_speaker_distortion_effects,
|
||||
|
||||
@@ -94,6 +94,13 @@
|
||||
<a href="/m4b-meta">m4b Meta</a>
|
||||
<a href="/projects" style="margin-left: auto; border-color: #a78bfa;">Projects</a>
|
||||
</nav>
|
||||
{% set vm = voice_maintenance() %}
|
||||
{% if vm %}
|
||||
<div style="padding:0.75rem 1rem; margin-bottom:1rem; background:#1a1a2e; border:1px solid #b45309; border-left:4px solid #f59e0b; border-radius:6px">
|
||||
<strong style="color:#f59e0b">⚠ Voice re-generation unavailable</strong>
|
||||
<span style="color:#94a3b8; font-size:0.85rem"> — {{ vm.reason }} Voice generation will fall back to Edge-TTS (loses raspy/texture) until the run finishes.</span>
|
||||
</div>
|
||||
{% endif %}
|
||||
{% block content %}{% endblock %}
|
||||
</body>
|
||||
</html>
|
||||
|
||||
Reference in New Issue
Block a user