9bda553311
Runs after a reviewer finishes a book and answers the two questions that
gate the next stage:
1. Are there utterances with no speaker?
2. Now that speakers have been defined, renamed, merged or deleted, are
there utterances that should change?
Read-only - it never edits the database, and a test asserts the files are
byte-identical afterwards. Output is a markdown table and a CSV, one row per
finding with chapter, utterance number and a text preview, so each item is
locatable rather than a count.
Deal Hunter, after the completed review (56 speakers, 0 still needing
review, 4319 utterances):
- Dialogue with no speaker: 0. The 1803 speakerless rows are narration
and scene breaks, which is correct and is reported as a number rather
than silently dropped.
- Utterances attributed to a deleted speaker: 0.
- One speaker stored under two spellings: 6 rows across 3 speakers
(narrator/Narrator, Kainda/Princess Kainda, Willamina Wake/Captain).
All resolve to one voice so the audio is correct, but a per-speaker
export would split them into six - caught before the translation stage,
not after.
- 6 speech-tag mismatches where the narration immediately before the line
names a different speaker.
- 10 dialogue rows voiced by the narrator: 9 are quoted labels ("Reactor.",
"operations") that are correctly read by the narrator, 1 (ch28 utt46) is
a real segmentation error. Classified advisory so the one real defect is
not buried under nine false positives.
build_alias_map moved from scripts/qa_attribution.py into attribution_qa.py
so the CLI and the audit cannot disagree about what a speech tag resolves to;
qa_attribution output verified byte-identical.
13 new tests. Suite: 721 passed, 8 pre-existing.
567 lines
19 KiB
Python
567 lines
19 KiB
Python
"""Mechanical cross-check of speaker attribution against speech tags.
|
|
|
|
Runs after LLM attribution as a cheap, deterministic verification pass. It does
|
|
not attempt full coreference resolution; it only flags high-signal cases so a
|
|
human (or the LLM re-run) can fix them:
|
|
|
|
- ``speech_tag_mismatch``: narration immediately before a dialogue names a
|
|
speaker ("X says,") and the dialogue is attributed to someone else.
|
|
- ``narrator_dialogue``: a dialogue line is still attributed to the narrator.
|
|
- ``role_tag_guessed``: narration tags the speaker with an unnamed role ("the
|
|
lead guard", "a pirate") but the line is attributed to a named character with
|
|
no lexical connection to that role.
|
|
- ``address_mismatch``: dialogue addresses someone by name/title ("Captain,"
|
|
"Hunter,") and the attributed speaker IS that person — a character never
|
|
addresses themselves.
|
|
- ``narration_speaker_mismatch``: narration within 2 preceding lines names a
|
|
speaker via speech verb, but the dialogue is attributed to someone else.
|
|
|
|
``--fix`` only auto-applies ``speech_tag_mismatch`` (the high-confidence case).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
SPEECH_VERBS = re.compile(
|
|
r"\b(says|said|asks|asked|replies|replied|whispers|whispered|shouts|shouted|"
|
|
r"mutters|muttered|calls|called|answers|answered|tells|told|explains|explained|"
|
|
r"saying|telling|responding)\b",
|
|
)
|
|
|
|
_SUBJECT_PROPER = re.compile(
|
|
r"\b([A-Z][a-zA-Z]+(?:\s+[A-Z][a-zA-Z]+)*)\s+"
|
|
r"(says|said|asks|asked|replies|replied|whispers|whispered|shouts|shouted|"
|
|
r"mutters|muttered|calls|called|answers|answered|tells|told|explains|explained)\b",
|
|
)
|
|
|
|
_SUBJECT_LOWER = re.compile(
|
|
r"\b([Tt]he|[Aa]|[Aa]n|[Tt]his|[Tt]that)\s+([a-z][a-z-]+(?:\s+[a-z][a-z-]+){0,1})\s+"
|
|
r"(says|said|asks|asked|replies|replied|whispers|whispered|shouts|shouted|"
|
|
r"mutters|muttered|calls|called|answers|answered|tells|told|explains|explained)\b",
|
|
)
|
|
|
|
ROLE_KEYWORDS = frozenset(
|
|
{
|
|
"guard",
|
|
"guards",
|
|
"doctor",
|
|
"pilot",
|
|
"nurse",
|
|
"medic",
|
|
"robot",
|
|
"robots",
|
|
"officer",
|
|
"officers",
|
|
"crew",
|
|
"security",
|
|
"voice",
|
|
"fugitive",
|
|
"pirate",
|
|
"pirates",
|
|
"dealer",
|
|
"servant",
|
|
"informant",
|
|
"interrogator",
|
|
"questioner",
|
|
"nanny",
|
|
"warbot",
|
|
"warbots",
|
|
"messenger",
|
|
"courier",
|
|
"driver",
|
|
"attendant",
|
|
"staff",
|
|
"engineer",
|
|
"mechanic",
|
|
"sailor",
|
|
"soldier",
|
|
"patrol",
|
|
"suit",
|
|
"figure",
|
|
"person",
|
|
"stranger",
|
|
"woman",
|
|
"man",
|
|
"girl",
|
|
"boy",
|
|
"child",
|
|
"speaker",
|
|
"intruder",
|
|
"enemy",
|
|
"friend",
|
|
"employer",
|
|
"client",
|
|
"customer",
|
|
"steward",
|
|
"valet",
|
|
}
|
|
)
|
|
|
|
PRONOUN_SUBJECTS = frozenset({"he", "she", "they", "it", "we", "you", "his", "her", "their", "i"})
|
|
|
|
_CONJUNCTIONS = frozenset(
|
|
{
|
|
"and",
|
|
"or",
|
|
"but",
|
|
"then",
|
|
"while",
|
|
"when",
|
|
"so",
|
|
"that",
|
|
"who",
|
|
"with",
|
|
"which",
|
|
"from",
|
|
"before",
|
|
"after",
|
|
}
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class QAIssue:
|
|
"""One flagged speaker-attribution problem for review."""
|
|
|
|
chapter_id: int
|
|
utterance_index: int
|
|
issue_type: str
|
|
speaker: str
|
|
expected: str | None = None
|
|
confidence: float = 0.5
|
|
evidence: str = ""
|
|
text: str = ""
|
|
|
|
|
|
@dataclass
|
|
class QAResult:
|
|
"""Outcome of running the cross-check over a chapter."""
|
|
|
|
chapter_id: int
|
|
issues: list[QAIssue] = field(default_factory=list)
|
|
checked: int = 0
|
|
|
|
def has(self, issue_type: str) -> bool:
|
|
return any(i.issue_type == issue_type for i in self.issues)
|
|
|
|
def by_type(self, issue_type: str) -> list[QAIssue]:
|
|
return [i for i in self.issues if i.issue_type == issue_type]
|
|
|
|
|
|
def _canonicalize(name: str, alias_map: dict[str, str]) -> str:
|
|
lower = name.strip().lower()
|
|
if lower in ("narrator", "the narrator"):
|
|
return "narrator"
|
|
return alias_map.get(lower, name.strip())
|
|
|
|
|
|
def _subject_words(subject: str) -> set[str]:
|
|
words = set(re.findall(r"[a-z]+", subject.lower()))
|
|
words.discard("the")
|
|
words.discard("a")
|
|
words.discard("an")
|
|
words.discard("this")
|
|
words.discard("that")
|
|
return words
|
|
|
|
|
|
def _match_speech_tag(text: str, alias_map: dict[str, str]) -> tuple[str | None, str | None, bool]:
|
|
"""Return ``(kind, value, resolved)`` for the speaker named by a tag line.
|
|
|
|
``kind`` is ``"name"`` (value = canonical speaker name) or ``"role"``
|
|
(value = "the lead guard"-style role phrase). ``resolved`` is True only for
|
|
name tags whose raw name is present in ``alias_map`` (safe to auto-fix).
|
|
Returns ``(None, None, False)`` when the line carries no resolvable tag.
|
|
Pronoun subjects ("He says,") are left to the LLM's coreference pass.
|
|
"""
|
|
proper = _SUBJECT_PROPER.search(text)
|
|
if proper:
|
|
raw = proper.group(1)
|
|
if raw.strip().lower() not in PRONOUN_SUBJECTS:
|
|
expected = _canonicalize(raw, alias_map)
|
|
if expected and expected != "narrator":
|
|
return "name", expected, raw.strip().lower() in alias_map
|
|
return None, None, False
|
|
|
|
lower = _SUBJECT_LOWER.search(text)
|
|
if lower:
|
|
words = lower.group(2).strip()
|
|
last = words.split()[-1].lower()
|
|
if last not in _CONJUNCTIONS:
|
|
role = f"{lower.group(1)} {words}"
|
|
if any(kw in _subject_words(role) for kw in ROLE_KEYWORDS):
|
|
return "role", role, False
|
|
return None, None, False
|
|
|
|
return None, None, False
|
|
|
|
|
|
def _names_match(tag: str, speaker: str, alias_map: dict[str, str]) -> tuple[bool, bool]:
|
|
"""Return ``(matches, resolved)`` for a name tag vs the assigned speaker.
|
|
|
|
``resolved`` is True when the tag name is a known alias/primary, making an
|
|
exact canonical comparison safe to auto-fix. Unresolved titles ("Wakam",
|
|
"Chief Felucca") fall back to word overlap against both names.
|
|
"""
|
|
tag = tag.strip()
|
|
speaker = speaker.strip()
|
|
canonical_tag = alias_map.get(tag.lower(), tag)
|
|
canonical_speaker = alias_map.get(speaker.lower(), speaker)
|
|
if canonical_tag.lower() == canonical_speaker.lower():
|
|
return True, True
|
|
resolved = tag.lower() in alias_map
|
|
overlap = bool(_subject_words(canonical_tag) & _subject_words(canonical_speaker))
|
|
return overlap, resolved
|
|
|
|
|
|
def _detect_addressee(text: str, alias_map: dict[str, str]) -> str | None:
|
|
"""Detect if dialogue addresses someone by name or title at the start.
|
|
|
|
Patterns: "Hunter, kill the connection" / "Captain, we need to jump" /
|
|
"Kainda, please." / "\"Yes, Captain."
|
|
Returns the canonical speaker name if an addressee is detected.
|
|
"""
|
|
stripped = text.strip().strip("\u201c\u201d\"'")
|
|
address_re = re.compile(
|
|
r"^(?:Yes,\s*|No,\s*|Well,\s*|So,\s*|Okay,\s*|Alright,\s*)?"
|
|
r"([A-Z][a-zA-Z]+(?:\s+[A-Z][a-zA-Z]+){0,2})\b",
|
|
)
|
|
m = address_re.match(stripped)
|
|
if m:
|
|
raw = m.group(1)
|
|
canonical = alias_map.get(raw.lower(), raw)
|
|
if canonical and canonical.lower() not in ("narrator", "the narrator"):
|
|
return canonical
|
|
|
|
title_re = re.compile(
|
|
r"^(?:Yes,\s*|No,\s*|Well,\s*|So,\s*|Okay,\s*|Alright,\s*)?"
|
|
r"(Captain|Princess|Hunter|Doctor|Sir|Ma'am|Admiral|Chief)\b",
|
|
)
|
|
m2 = title_re.match(stripped)
|
|
if m2:
|
|
title = m2.group(1)
|
|
canonical = alias_map.get(title.lower(), title)
|
|
if canonical and canonical.lower() not in ("narrator",):
|
|
return canonical
|
|
|
|
return None
|
|
|
|
|
|
def _find_narrated_speaker(
|
|
utterances: list[dict[str, Any]], dialogue_index: int, alias_map: dict[str, str]
|
|
) -> str | None:
|
|
"""Look back from dialogue_index for a narrated speaker via speech verb.
|
|
|
|
Only checks the immediately preceding utterance — a narration tag at i-2
|
|
was already consumed by the dialogue at i-1.
|
|
|
|
Returns the canonical speaker name if narration explicitly names who is
|
|
speaking before this dialogue line.
|
|
"""
|
|
j = dialogue_index - 1
|
|
if j < 0:
|
|
return None
|
|
|
|
prev = utterances[j]
|
|
prev_text = (prev.get("text") or "").strip()
|
|
if prev_text == "<<<>>>":
|
|
return None
|
|
if (prev.get("type") or "").lower() != "narration":
|
|
return None
|
|
if not SPEECH_VERBS.search(prev_text):
|
|
return None
|
|
|
|
kind, value, resolved = _match_speech_tag(prev_text, alias_map)
|
|
if kind == "name" and value:
|
|
return value
|
|
|
|
complex_narr_re = re.compile(
|
|
r"\b([A-Z][a-zA-Z]+(?:\s+[A-Z][a-zA-Z]+){0,2})\s+"
|
|
r"(?:nods|pauses|looks|checks|hits|turns|looks|smiles|frowns|shakes|holds|grabs|pushes|leans|stands|sits|walks|continues|starts|stops)\b",
|
|
)
|
|
cm = complex_narr_re.search(prev_text)
|
|
if cm:
|
|
raw = cm.group(1)
|
|
canonical = alias_map.get(raw.lower(), raw)
|
|
if canonical and canonical.lower() not in ("narrator", "the narrator"):
|
|
return canonical
|
|
|
|
return None
|
|
|
|
|
|
def build_alias_map(speakers_path) -> dict[str, str]:
|
|
"""Lowercase name/alias -> canonical ``primary_name``, plus first-word keys.
|
|
|
|
First-word entries let a speech tag like "Wakam" resolve to "Trader
|
|
Wakam". Lives here rather than in the CLI so every caller agrees on what a
|
|
tag can resolve to; scripts/qa_attribution.py imports this.
|
|
"""
|
|
import json
|
|
from pathlib import Path as _Path
|
|
|
|
path = _Path(speakers_path)
|
|
if not path.exists():
|
|
return {}
|
|
try:
|
|
data = json.loads(path.read_text())
|
|
except (OSError, ValueError):
|
|
return {}
|
|
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
|
|
first = primary.split()[0].lower() if primary.split() else ""
|
|
if first and first not in alias_map:
|
|
alias_map[first] = primary
|
|
return alias_map
|
|
|
|
|
|
def check_attribution(
|
|
utterances: list[dict[str, Any]],
|
|
chapter_id: int,
|
|
alias_map: dict[str, str] | None = None,
|
|
) -> QAResult:
|
|
"""Flag attribution problems in a chapter's utterance list.
|
|
|
|
``utterances`` must be ordered by ``utterance_index`` and contain ``text``,
|
|
``type``, and ``speaker_name``. ``alias_map`` maps lowercase names/aliases to
|
|
canonical ``primary_name`` values.
|
|
"""
|
|
alias_map = alias_map or {}
|
|
result = QAResult(chapter_id=chapter_id)
|
|
|
|
for i, u in enumerate(utterances):
|
|
if (u.get("type") or "").lower() != "dialogue":
|
|
continue
|
|
|
|
result.checked += 1
|
|
text = (u.get("text") or "").strip()
|
|
speaker = (u.get("speaker_name") or "").strip() or "narrator"
|
|
idx = int(u.get("utterance_index") or 0)
|
|
|
|
if speaker.lower() == "narrator":
|
|
result.issues.append(
|
|
QAIssue(
|
|
chapter_id=chapter_id,
|
|
utterance_index=idx,
|
|
issue_type="narrator_dialogue",
|
|
speaker=speaker,
|
|
confidence=0.95,
|
|
evidence="Dialogue line still attributed to the narrator.",
|
|
text=text[:150],
|
|
)
|
|
)
|
|
continue
|
|
|
|
# Apply a speech tag only from the immediately preceding narration
|
|
# line. A tag never spans a dialogue line (shared-tag dialogue blocks
|
|
# like "Kainda asks," + reply belong to whoever follows next).
|
|
if i == 0:
|
|
continue
|
|
prev = utterances[i - 1]
|
|
prev_text = (prev.get("text") or "").strip()
|
|
if prev_text == "<<<>>>":
|
|
continue
|
|
if (prev.get("type") or "").lower() != "narration":
|
|
continue
|
|
if not SPEECH_VERBS.search(prev_text):
|
|
continue
|
|
|
|
kind, value, resolved = _match_speech_tag(prev_text, alias_map)
|
|
if kind == "name":
|
|
assert value is not None
|
|
matches, tag_resolved = _names_match(value, speaker, alias_map)
|
|
if not matches:
|
|
result.issues.append(
|
|
QAIssue(
|
|
chapter_id=chapter_id,
|
|
utterance_index=idx,
|
|
issue_type="speech_tag_mismatch",
|
|
speaker=speaker,
|
|
expected=value,
|
|
confidence=0.9 if tag_resolved else 0.7,
|
|
evidence=f"Nearest tag: '{prev_text[:80]}'",
|
|
text=text[:150],
|
|
)
|
|
)
|
|
elif kind == "role":
|
|
assert value is not None
|
|
role_words = _subject_words(value)
|
|
speaker_words = _subject_words(speaker)
|
|
if not (role_words & speaker_words):
|
|
result.issues.append(
|
|
QAIssue(
|
|
chapter_id=chapter_id,
|
|
utterance_index=idx,
|
|
issue_type="role_tag_guessed",
|
|
speaker=speaker,
|
|
expected=value.title(),
|
|
confidence=0.5,
|
|
evidence=f"Nearest tag: '{prev_text[:80]}'",
|
|
text=text[:150],
|
|
)
|
|
)
|
|
|
|
for i, u in enumerate(utterances):
|
|
if (u.get("type") or "").lower() != "dialogue":
|
|
continue
|
|
text = (u.get("text") or "").strip()
|
|
speaker = (u.get("speaker_name") or "").strip() or "narrator"
|
|
idx = int(u.get("utterance_index") or 0)
|
|
if speaker.lower() == "narrator":
|
|
continue
|
|
canonical_speaker = _canonicalize(speaker, alias_map)
|
|
|
|
addressee = _detect_addressee(text, alias_map)
|
|
if addressee and canonical_speaker == addressee:
|
|
result.issues.append(
|
|
QAIssue(
|
|
chapter_id=chapter_id,
|
|
utterance_index=idx,
|
|
issue_type="address_mismatch",
|
|
speaker=speaker,
|
|
expected=None,
|
|
confidence=0.85,
|
|
evidence=f"Dialogue addresses '{addressee}' but speaker is also '{addressee}'",
|
|
text=text[:150],
|
|
)
|
|
)
|
|
|
|
narrated_speaker = _find_narrated_speaker(utterances, i, alias_map)
|
|
if narrated_speaker:
|
|
matches, _ = _names_match(narrated_speaker, speaker, alias_map)
|
|
if not matches:
|
|
already_flagged = any(
|
|
iss.issue_type == "speech_tag_mismatch"
|
|
and iss.utterance_index == idx
|
|
for iss in result.issues
|
|
)
|
|
if not already_flagged:
|
|
result.issues.append(
|
|
QAIssue(
|
|
chapter_id=chapter_id,
|
|
utterance_index=idx,
|
|
issue_type="narration_speaker_mismatch",
|
|
speaker=speaker,
|
|
expected=narrated_speaker,
|
|
confidence=0.80,
|
|
evidence=f"Narration names '{narrated_speaker}' as speaker",
|
|
text=text[:150],
|
|
)
|
|
)
|
|
|
|
return result
|
|
|
|
|
|
def fix_high_confidence(
|
|
utterances: list[dict[str, Any]], alias_map: dict[str, str]
|
|
) -> list[dict[str, Any]]:
|
|
"""Auto-fix only exact speech-tag mismatches (speaker == tagged name)."""
|
|
result = [dict(u) for u in utterances]
|
|
for i, u in enumerate(result):
|
|
if (u.get("type") or "").lower() != "dialogue" or i == 0:
|
|
continue
|
|
speaker = (u.get("speaker_name") or "").strip()
|
|
prev = result[i - 1]
|
|
prev_text = (prev.get("text") or "").strip()
|
|
if prev_text == "<<<>>>":
|
|
continue
|
|
if (prev.get("type") or "").lower() != "narration":
|
|
continue
|
|
if not SPEECH_VERBS.search(prev_text):
|
|
continue
|
|
kind, value, resolved = _match_speech_tag(prev_text, alias_map)
|
|
if kind == "name" and resolved:
|
|
assert value is not None
|
|
matches, _ = _names_match(value, speaker, alias_map)
|
|
if not matches:
|
|
result[i]["speaker_name"] = value
|
|
return result
|
|
|
|
|
|
def verify_conversation_flow(
|
|
utterances: list[dict[str, Any]],
|
|
chapter_id: int,
|
|
alias_map: dict[str, str] | None = None,
|
|
) -> QAResult:
|
|
"""Post-attribution check for conversation flow anomalies.
|
|
|
|
Flags:
|
|
- ``long_monologue``: 4+ consecutive dialogue lines from the same speaker
|
|
without narration breaks (likely misattribution in a dialogue scene).
|
|
- ``broken_alternation``: in a rapid A-B-A-B dialogue, a third speaker
|
|
appears without narration indicating the switch.
|
|
"""
|
|
alias_map = alias_map or {}
|
|
result = QAResult(chapter_id=chapter_id)
|
|
|
|
dialogue_runs: list[tuple[str, int, int]] = []
|
|
current_speaker = None
|
|
run_start = None
|
|
|
|
for i, u in enumerate(utterances):
|
|
utt_type = (u.get("type") or "").lower()
|
|
text = (u.get("text") or "").strip()
|
|
|
|
if text == "<<<>>>" or utt_type == "narration":
|
|
if current_speaker is not None and run_start is not None:
|
|
dialogue_runs.append((current_speaker, run_start, i - 1))
|
|
current_speaker = None
|
|
run_start = None
|
|
continue
|
|
|
|
if utt_type != "dialogue":
|
|
continue
|
|
|
|
speaker = (u.get("speaker_name") or "").strip() or "narrator"
|
|
canonical = _canonicalize(speaker, alias_map)
|
|
|
|
if canonical != current_speaker:
|
|
if current_speaker is not None and run_start is not None:
|
|
dialogue_runs.append((current_speaker, run_start, i - 1))
|
|
current_speaker = canonical
|
|
run_start = i
|
|
|
|
if current_speaker is not None and run_start is not None:
|
|
dialogue_runs.append((current_speaker, run_start, len(utterances) - 1))
|
|
|
|
for speaker, start_idx, end_idx in dialogue_runs:
|
|
run_length = end_idx - start_idx + 1
|
|
if run_length >= 4:
|
|
first_u = utterances[start_idx]
|
|
idx = int(first_u.get("utterance_index") or 0)
|
|
has_narration_between = False
|
|
for k in range(start_idx, end_idx + 1):
|
|
if k < len(utterances):
|
|
txt = (utterances[k].get("text") or "").strip()
|
|
if txt == "<<<>>>":
|
|
has_narration_between = True
|
|
break
|
|
if not has_narration_between:
|
|
texts = []
|
|
for k in range(start_idx, min(start_idx + 3, end_idx + 1)):
|
|
if k < len(utterances):
|
|
texts.append(utterances[k].get("text", "")[:60])
|
|
result.issues.append(
|
|
QAIssue(
|
|
chapter_id=chapter_id,
|
|
utterance_index=idx,
|
|
issue_type="long_monologue",
|
|
speaker=speaker,
|
|
expected=None,
|
|
confidence=0.5,
|
|
evidence=f"{run_length} consecutive lines by same speaker",
|
|
text=" | ".join(texts)[:150],
|
|
)
|
|
)
|
|
|
|
return result
|